From b76d36f3b91c886863aefe4756e2c6a6e6c143bf Mon Sep 17 00:00:00 2001 From: "M.Bozorgmehr" Date: Wed, 14 Oct 2026 10:27:18 +0330 Subject: [PATCH 1/9] feat(workflows): add opt-in output capture to command and prompt steps Command and prompt steps stream the integration CLI's output to the terminal, so steps..output.stdout and stderr were always empty and later steps could not parse or branch on agent output. `capture: true` retains the CLI's textual stdout and stderr in the step output instead. Each stream is capped by `capture_max_bytes` (default 1 MiB, -1 for no limit), counted in raw bytes before UTF-8 decoding and drained past the cap so the CLI never blocks on a full pipe. `stdout_truncated`/`stderr_truncated` report size truncation only; `timed_out` and `interrupted` report a stopped process, whose partial output is kept. A failed captured step reports a short summary as its error, so captured stderr never reaches log.jsonl or the terminal. dispatch_command() gains `output_json` and `capture_max_bytes` so capture requests text output without changing existing callers, including Copilot's override. Captured command steps get a 600-second default timeout; streaming command steps and prompt timeouts are unchanged. Closes #4613 Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 --- docs/reference/workflows.md | 116 ++++- src/specify_cli/integrations/_capture.py | 203 ++++++++ src/specify_cli/integrations/base.py | 33 +- .../integrations/copilot/__init__.py | 18 +- src/specify_cli/workflows/step/_capture.py | 114 +++++ .../workflows/step/command/__init__.py | 87 +++- .../workflows/step/prompt/__init__.py | 63 ++- tests/integrations/test_capture.py | 289 +++++++++++ tests/workflows/test_step_capture.py | 457 ++++++++++++++++++ workflows/README.md | 26 + 10 files changed, 1391 insertions(+), 15 deletions(-) create mode 100644 src/specify_cli/integrations/_capture.py create mode 100644 src/specify_cli/workflows/step/_capture.py create mode 100644 tests/integrations/test_capture.py create mode 100644 tests/workflows/test_step_capture.py diff --git a/docs/reference/workflows.md b/docs/reference/workflows.md index 9187265a84..c7cdbf1e7e 100644 --- a/docs/reference/workflows.md +++ b/docs/reference/workflows.md @@ -1054,6 +1054,115 @@ or paused command is resumed, the complete dispatch configuration is re-resolved from the current inputs, so values supplied with `workflow resume --input` take effect consistently. A resume without updated inputs reproduces the same values. +### Capturing Command and Prompt Output + +`command` and `prompt` steps stream the integration CLI's output to the +terminal, so `steps..output.stdout` and `steps..output.stderr` are empty +strings. Set `capture: true` to retain that output in the step result instead, +where later steps can read it: + +```yaml +- id: verdict + type: prompt + prompt: "Reply with exactly READY or BLOCKED for {{ inputs.spec }}" + capture: true + capture_max_bytes: 4096 # Optional: per-stream limit (default 1048576) + timeout: 120 # Optional: seconds before the CLI is killed + +- id: branch + type: if + condition: "{{ steps.verdict.output.stdout | contains('READY') }}" + then: + - id: plan + command: speckit.plan +``` + +| Field | Applies to | Meaning | +| ------------------- | ------------------- | --------------------------------------------------------------------------------------------------- | +| `capture` | `command`, `prompt` | `true` captures stdout and stderr instead of streaming them. Must be a literal boolean; default `false`. | +| `capture_max_bytes` | `command`, `prompt` | Limit for each captured stream: a positive integer, or `-1` for no limit. Default `1048576` (1 MiB). Requires `capture: true`. | +| `timeout` | `command`, `prompt` | Seconds before the CLI is killed. For `command` it applies only with `capture: true` (default `600`); a streaming `command` step has no timeout. For `prompt` it always applies (default `300`). | + +A captured step prints nothing while it runs. Its output gains these fields: + +| Output field | Meaning | +| ------------------ | -------------------------------------------------------------------------- | +| `stdout`, `stderr` | The captured text. | +| `stdout_truncated` | `true` when stdout exceeded `capture_max_bytes` and the rest was discarded. | +| `stderr_truncated` | The same for stderr. | +| `timed_out` | `true` when the CLI was killed at `timeout`; `exit_code` is `-1`. | +| `interrupted` | `true` when the run was interrupted (Ctrl+C); `exit_code` is `130`. | + +**What the text is.** Capture requests the integration CLI's normal text +output, the same output a streaming step prints. It is not a structured format: +it can include progress or tool-use lines as well as the final answer, it +differs between integrations, and it is valid JSON only if the agent printed +nothing but JSON. + +**Limits and encoding.** `capture_max_bytes` counts raw bytes on each stream +before decoding. Bytes past the limit are read and discarded so the CLI is never +blocked, and the matching `*_truncated` field is set. Output is decoded as +UTF-8: invalid bytes become U+FFFD, and a character split by the limit, a +timeout, or an interruption is dropped rather than replaced. Line endings are +normalised to `\n`. + +With `capture_max_bytes: -1` nothing is discarded and both `*_truncated` fields +stay `false`. The whole output is then held in memory and written to run state, +so an agent that prints a large amount grows both without bound. Use it only +for output you know is small. + +**Check completeness before using the output.** Filters such as `from_json` do +not look at these fields; they parse whatever text is there. Output is +incomplete when `stdout_truncated` is `true`, and also when `timed_out` or +`interrupted` is `true` — a stopped CLI leaves partial output even with no size +limit. Guard the use: + +```yaml +- id: parse + type: if + condition: "{{ steps.verdict.output.exit_code == 0 and not steps.verdict.output.stdout_truncated }}" + then: + - id: use + type: shell + run: "echo parsed" +``` + +**Failures.** A non-zero exit, a timeout, or an interruption fails the step and +keeps whatever was captured in `output.stdout` and `output.stderr`. The step's +error is a short summary (`Command exited with code 2`, +`Prompt timed out after 120 seconds.`), never the captured text. A failed step +stops the run, so set `continue_on_error: true` on the capturing step when a +later step needs to inspect the failure: + +```yaml +- id: verdict + type: prompt + prompt: "Reply with exactly READY or BLOCKED" + capture: true + timeout: 120 + continue_on_error: true + +- id: on-timeout + type: if + condition: "{{ steps.verdict.output.timed_out }}" + then: + - id: review + type: gate + message: "The verdict step timed out. Approve to continue without it." +``` + +> ⚠️ **Captured output is stored in plain text and is not redacted.** It is +> written to the run's `state.json` so that `specify workflow resume` can replay +> it, and it stays there after the run. Anything the agent prints — including +> secrets it read from files or the environment — is persisted. Captured text is +> not printed to the terminal and is not written to `log.jsonl`. Do not enable +> capture on a step whose output may contain credentials unless the run +> directory is protected accordingly. + +Captured text is agent output and therefore untrusted; see +[Interpolation and shell safety](#interpolation-and-shell-safety) before using +it in a `shell` step. + ## Expressions Steps can reference inputs and previous step outputs using `{{ expression }}` syntax: @@ -1114,7 +1223,7 @@ Expressions are resolved by **plain string substitution** — the value of `{{ . If an interpolated value can contain characters like `;`, `|`, `&`, `$( )`, backticks, or quotes, it can change or extend the command that actually runs. This matters most when the value is not fully under the workflow author's control: - **Workflow `inputs.*`** — supplied by whoever runs the workflow. -- **A prior step's output**, e.g. `{{ steps.plan.output.stdout }}` — for a `prompt` step this is **text produced by the AI agent**, which can in turn be influenced by files, tickets, or web content the agent read. Treat agent output as untrusted when it flows into a `shell` step. +- **A prior step's output**, e.g. `{{ steps.plan.output.stdout }}` — for a `command` or `prompt` step with [`capture: true`](#capturing-command-and-prompt-output) this is **text produced by the AI agent**, which can in turn be influenced by files, tickets, or web content the agent read. Treat agent output as untrusted when it flows into a `shell` step. There is **no shell-escaping filter** in the expression language and **no sandbox** around a `shell` step, so none of the practices below can be treated as a guarantee that a hostile value is neutralised. The only reliable control is to constrain what an interpolated value *can* be, and to keep values you cannot constrain out of `run` fields entirely. Scrutinise every `run` field that interpolates a value you do not control, and at minimum: @@ -1153,7 +1262,10 @@ Shell steps automatically receive the following environment variables: Each workflow run persists its state at `.specify/workflows/runs//`: -- `state.json` — current run state and step progress +- `state.json` — current run state and step progress, including each step's + output. Output captured with + [`capture: true`](#capturing-command-and-prompt-output) is stored here in + plain text. - `inputs.json` — resolved input values - `log.jsonl` — step-by-step execution log diff --git a/src/specify_cli/integrations/_capture.py b/src/specify_cli/integrations/_capture.py new file mode 100644 index 0000000000..0f11fd53dd --- /dev/null +++ b/src/specify_cli/integrations/_capture.py @@ -0,0 +1,203 @@ +"""Bounded capture of an integration CLI's stdout and stderr. + +``subprocess.run(capture_output=True)`` buffers everything a process writes, +decodes it with the locale encoding, and loses the output on a timeout. Captured +workflow dispatch needs the opposite on all three: a per-stream byte cap, a +fixed UTF-8 decoding, and whatever was read before a timeout or interruption. +""" + +from __future__ import annotations + +import codecs +import subprocess +import threading +import time +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any + +DEFAULT_CAPTURE_MAX_BYTES = 1024 * 1024 +UNLIMITED_CAPTURE = -1 + +_READ_CHUNK_BYTES = 64 * 1024 +# After a kill, a descendant that inherited the pipes can keep them open; wait +# this long for end-of-file before returning what has been read so far. +_DRAIN_GRACE_SECONDS = 2.0 + + +def capture_max_bytes_is_valid(value: Any) -> bool: + """Return whether *value* is a positive int or ``-1`` (unlimited). + + bool is a subclass of int, but ``capture_max_bytes: true`` is a config + error rather than a one-byte limit, so it is rejected explicitly. + """ + if isinstance(value, bool) or not isinstance(value, int): + return False + return value > 0 or value == UNLIMITED_CAPTURE + + +@dataclass(frozen=True) +class CapturedProcess: + """Outcome of one captured process. + + ``stdout_truncated``/``stderr_truncated`` report size-based truncation + only. Output can also be incomplete because the process was stopped; + ``timed_out`` and ``interrupted`` report that separately. + """ + + exit_code: int + stdout: str + stderr: str + stdout_truncated: bool = False + stderr_truncated: bool = False + timed_out: bool = False + interrupted: bool = False + + def as_dispatch_result(self) -> dict[str, Any]: + """Return the ``dispatch_command`` result dict for this process.""" + return { + "exit_code": self.exit_code, + "stdout": self.stdout, + "stderr": self.stderr, + "stdout_truncated": self.stdout_truncated, + "stderr_truncated": self.stderr_truncated, + "timed_out": self.timed_out, + "interrupted": self.interrupted, + } + + +class _Sink: + """Retain at most *limit* bytes of a stream; count the rest as truncated.""" + + def __init__(self, limit: int) -> None: + self._limit = limit + self._data = bytearray() + self._truncated = False + self._lock = threading.Lock() + + def feed(self, chunk: bytes) -> None: + with self._lock: + if self._limit != UNLIMITED_CAPTURE: + room = self._limit - len(self._data) + if len(chunk) > room: + self._truncated = True + chunk = chunk[:room] + self._data += chunk + + def snapshot(self) -> tuple[bytes, bool]: + with self._lock: + return bytes(self._data), self._truncated + + +def _drain(pipe: Any, sink: _Sink) -> None: + """Read *pipe* to end-of-file, including past the sink's limit. + + Reading continues after the cap so the process never blocks writing to a + full pipe; the sink discards what it does not retain. + """ + try: + while chunk := pipe.read1(_READ_CHUNK_BYTES): + sink.feed(chunk) + except (OSError, ValueError): + pass + finally: + try: + pipe.close() + except OSError: + pass + + +def _join(readers: Sequence[threading.Thread], deadline: float | None) -> bool: + """Join *readers* until *deadline*; return whether all reached end-of-file.""" + for reader in readers: + if deadline is None: + reader.join() + else: + reader.join(max(0.0, deadline - time.monotonic())) + return not any(reader.is_alive() for reader in readers) + + +def _decode(data: bytes, *, complete: bool) -> str: + """Decode captured bytes as UTF-8, replacing invalid sequences with U+FFFD. + + When the capture is not *complete* (cut at the byte limit, or the process + was stopped), a trailing partial multibyte sequence is dropped rather than + replaced, so a cut never invents a replacement character. Line endings are + normalised to ``\\n`` as ``subprocess`` text mode does. + """ + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + text = decoder.decode(data, final=complete) + return text.replace("\r\n", "\n").replace("\r", "\n") + + +def run_captured( + args: Sequence[str], + *, + cwd: str | None = None, + timeout: float | None = None, + max_bytes: int = DEFAULT_CAPTURE_MAX_BYTES, +) -> CapturedProcess: + """Run *args* and capture stdout and stderr, each capped at *max_bytes*. + + *max_bytes* counts raw bytes before decoding; ``-1`` disables the cap. A + timeout or ``KeyboardInterrupt`` kills the process and is reported on the + result (exit code ``-1`` or ``130``) together with the output read so far, + instead of being raised. ``OSError`` from starting the process propagates. + """ + if not capture_max_bytes_is_valid(max_bytes): + raise ValueError( + f"max_bytes must be a positive integer or -1, got {max_bytes!r}." + ) + + proc = subprocess.Popen( + args, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=cwd, + ) + sinks = (_Sink(max_bytes), _Sink(max_bytes)) + readers = [ + threading.Thread(target=_drain, args=(pipe, sink), daemon=True) + for pipe, sink in zip((proc.stdout, proc.stderr), sinks) + ] + for reader in readers: + reader.start() + + deadline = None if timeout is None else time.monotonic() + timeout + timed_out = interrupted = False + try: + try: + proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + timed_out = True + else: + # The process can exit while a descendant still holds the pipes + # open; the same deadline bounds the wait for end-of-file. + timed_out = not _join(readers, deadline) + except KeyboardInterrupt: + interrupted = True + + if timed_out or interrupted: + proc.kill() + proc.wait() + _join(readers, time.monotonic() + _DRAIN_GRACE_SECONDS) + + (stdout, stdout_truncated), (stderr, stderr_truncated) = ( + sink.snapshot() for sink in sinks + ) + stopped = timed_out or interrupted + if interrupted: + exit_code = 130 + elif timed_out: + exit_code = -1 + else: + exit_code = proc.returncode + return CapturedProcess( + exit_code=exit_code, + stdout=_decode(stdout, complete=not (stopped or stdout_truncated)), + stderr=_decode(stderr, complete=not (stopped or stderr_truncated)), + stdout_truncated=stdout_truncated, + stderr_truncated=stderr_truncated, + timed_out=timed_out, + interrupted=interrupted, + ) diff --git a/src/specify_cli/integrations/base.py b/src/specify_cli/integrations/base.py index 6eee0513d6..b3dd0abc34 100644 --- a/src/specify_cli/integrations/base.py +++ b/src/specify_cli/integrations/base.py @@ -32,6 +32,7 @@ from .._toml_string import escape_toml_basic as _escape_toml_basic from .._toml_string import has_illegal_toml_control as _has_illegal_toml_control from ..events import install_integration_events, remove_integration_events +from ._capture import run_captured if TYPE_CHECKING: from .manifest import IntegrationManifest @@ -413,6 +414,8 @@ def dispatch_command( stream: bool = True, integration_args: Sequence[str] | None = None, integration_options: Mapping[str, Any] | None = None, + output_json: bool | None = None, + capture_max_bytes: int | None = None, ) -> dict[str, Any]: """Dispatch a Spec Kit command through this integration's CLI. @@ -429,20 +432,40 @@ def dispatch_command( piped directly to the terminal so the user sees live output. When ``False``, output is captured and returned in the dict. + *output_json* selects the CLI's output format independently of + *stream*. ``None`` (the default) keeps the historical coupling: + text when streaming, structured output when capturing. Pass + ``False`` with ``stream=False`` to capture the CLI's textual + stdout. + + *capture_max_bytes* selects bounded capture and requires + ``stream=False``: each stream is capped at that many raw bytes + (``-1`` for no cap) and decoded as UTF-8. The result then also + carries ``stdout_truncated``, ``stderr_truncated``, ``timed_out``, + and ``interrupted``, and a timeout or interruption is reported + there with the output read so far instead of being raised. + ``None`` (the default) keeps the unbounded capture, which raises + ``subprocess.TimeoutExpired``. + Returns a dict with ``exit_code``, ``stdout``, and ``stderr``. Raises ``NotImplementedError`` if the integration does not support CLI dispatch. """ import subprocess + if stream and capture_max_bytes is not None: + raise ValueError("capture_max_bytes requires stream=False.") + self.validate_runtime_config(integration_args, integration_options) prompt = self._build_dispatch_prompt(command_name, args, project_root) # When streaming to the terminal, request text output so the # user sees readable output instead of raw JSONL events. + if output_json is None: + output_json = not stream exec_args = self.build_exec_args( prompt, model=model, - output_json=not stream, + output_json=output_json, integration_args=integration_args, integration_options=integration_options, project_root=project_root, @@ -489,6 +512,14 @@ def dispatch_command( "stderr": "", } + if capture_max_bytes is not None: + return run_captured( + exec_args, + cwd=cwd, + timeout=timeout, + max_bytes=capture_max_bytes, + ).as_dispatch_result() + result = subprocess.run( exec_args, capture_output=True, diff --git a/src/specify_cli/integrations/copilot/__init__.py b/src/specify_cli/integrations/copilot/__init__.py index d997db85d4..c41f55e87a 100644 --- a/src/specify_cli/integrations/copilot/__init__.py +++ b/src/specify_cli/integrations/copilot/__init__.py @@ -29,6 +29,7 @@ import typer +from .._capture import run_captured from ..base import IntegrationBase, IntegrationOption, SkillsIntegration from ..manifest import IntegrationManifest @@ -368,6 +369,8 @@ def dispatch_command( stream: bool = True, integration_args: Sequence[str] | None = None, integration_options: Mapping[str, Any] | None = None, + output_json: bool | None = None, + capture_max_bytes: int | None = None, ) -> dict[str, Any]: """Dispatch via ``--agent speckit.`` instead of slash-commands. @@ -380,6 +383,11 @@ def dispatch_command( """ import subprocess + if stream and capture_max_bytes is not None: + raise ValueError("capture_max_bytes requires stream=False.") + if output_json is None: + output_json = not stream + self.validate_runtime_config(integration_args, integration_options) stem = command_name @@ -412,7 +420,7 @@ def dispatch_command( cli_args.append("--yolo") if model: cli_args.extend(["--model", model]) - if not stream: + if output_json: cli_args.extend(["--output-format", "json"]) cwd = str(project_root) if project_root else None @@ -436,6 +444,14 @@ def dispatch_command( "stderr": "", } + if capture_max_bytes is not None: + return run_captured( + cli_args, + cwd=cwd, + timeout=timeout, + max_bytes=capture_max_bytes, + ).as_dispatch_result() + result = subprocess.run( cli_args, capture_output=True, diff --git a/src/specify_cli/workflows/step/_capture.py b/src/specify_cli/workflows/step/_capture.py new file mode 100644 index 0000000000..b67700ed5a --- /dev/null +++ b/src/specify_cli/workflows/step/_capture.py @@ -0,0 +1,114 @@ +"""Shared ``capture`` contract for the ``command`` and ``prompt`` steps. + +Both steps stream the integration CLI's output to the terminal by default. +``capture: true`` retains stdout and stderr in the step output instead, capped +per stream by ``capture_max_bytes``. This module holds the parts of that +contract the two steps must agree on: validation, the output fields, and the +failure summary. +""" + +from __future__ import annotations + +import math +from typing import Any + +from specify_cli.integrations._capture import ( + DEFAULT_CAPTURE_MAX_BYTES, + capture_max_bytes_is_valid, +) + + +def capture_errors(config: dict[str, Any], label: str) -> list[str]: + """Return errors for the ``capture``/``capture_max_bytes`` fields. + + Shared by execute() and validate() so both paths reject the same values + with the same message. *label* names the step kind (``"Command"``). + """ + step_id = config.get("id", "?") + errors: list[str] = [] + capture = config.get("capture", False) + if "capture" in config and not isinstance(capture, bool): + errors.append( + f"{label} step {step_id!r}: 'capture' must be true or false, " + f"got {capture!r}." + ) + if "capture_max_bytes" in config: + max_bytes = config["capture_max_bytes"] + if not capture_max_bytes_is_valid(max_bytes): + errors.append( + f"{label} step {step_id!r}: 'capture_max_bytes' must be a " + f"positive integer or -1 (no limit), got {max_bytes!r}." + ) + elif capture is not True: + # A limit on a step that streams would silently do nothing. + errors.append( + f"{label} step {step_id!r}: 'capture_max_bytes' requires " + f"'capture: true'." + ) + return errors + + +def capture_limit(config: dict[str, Any]) -> int | None: + """Return the per-stream byte limit, or ``None`` when the step streams. + + Call only after :func:`capture_errors` returned no errors. + """ + if config.get("capture", False) is not True: + return None + return config.get("capture_max_bytes", DEFAULT_CAPTURE_MAX_BYTES) + + +def timeout_error(config: dict[str, Any], label: str) -> str | None: + """Return an error message if ``config['timeout']`` is invalid, else None. + + Mirrors the shell and prompt steps: bool is rejected although it is an int + subclass, and non-finite or oversized numbers are rejected because they + would raise inside the subprocess wait. + """ + if "timeout" not in config: + return None + timeout = config["timeout"] + try: + valid = ( + not isinstance(timeout, bool) + and isinstance(timeout, (int, float)) + and math.isfinite(timeout) + and timeout > 0 + ) + except OverflowError: + valid = False + if valid: + return None + return ( + f"{label} step {config.get('id', '?')!r}: 'timeout' must be a " + f"positive number of seconds, got {timeout!r}." + ) + + +def record_capture(output: dict[str, Any], dispatch_result: dict[str, Any]) -> None: + """Add the capture-only completeness fields to a step's *output*. + + ``*_truncated`` report the byte limit only; ``timed_out`` and + ``interrupted`` report a stopped process, whose output is partial even + without a limit. + """ + for field in ("stdout_truncated", "stderr_truncated", "timed_out", "interrupted"): + output[field] = bool(dispatch_result.get(field, False)) + + +def capture_failure( + dispatch_result: dict[str, Any], + *, + subject: str, + timeout: float, +) -> str: + """Return the error summary for a failed captured dispatch. + + The step error is written to ``log.jsonl`` and printed, so it never + carries captured stderr; that text stays in ``output.stderr``. + """ + if dispatch_result.get("interrupted"): + return "Interrupted by user" + if dispatch_result.get("timed_out"): + return f"{subject} timed out after {timeout} seconds." + return f"{subject} exited with code {dispatch_result['exit_code']}" diff --git a/src/specify_cli/workflows/step/command/__init__.py b/src/specify_cli/workflows/step/command/__init__.py index f836e91929..770757e981 100644 --- a/src/specify_cli/workflows/step/command/__init__.py +++ b/src/specify_cli/workflows/step/command/__init__.py @@ -2,6 +2,7 @@ from __future__ import annotations +import inspect import shutil from pathlib import Path from typing import Any @@ -9,6 +10,15 @@ from specify_cli.integrations.installer import project_dispatch from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus from specify_cli.workflows.expressions import evaluate_expression +from specify_cli.workflows.step._capture import ( + capture_errors, + capture_failure, + capture_limit, + record_capture, + timeout_error, +) + +_DEFAULT_CAPTURE_TIMEOUT = 600 class CommandStep(StepBase): @@ -24,7 +34,11 @@ class CommandStep(StepBase): CLI output is streamed to the terminal for live progress. ``output.exit_code`` is always captured and can be referenced by later steps (e.g. ``{{ steps.specify.output.exit_code }}``). - Full ``stdout``/``stderr`` capture is a planned enhancement. + ``capture: true`` retains the CLI's textual stdout and stderr in + ``output.stdout``/``output.stderr`` instead of streaming them, + capped per stream by ``capture_max_bytes`` and bounded by + ``timeout`` (default 600 seconds). ``timeout`` applies only to a + captured step; a streaming step has none. """ type_key = "command" @@ -134,6 +148,23 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: return StepResult(status=StepStatus.FAILED, error=runtime_config) integration_args, integration_options = runtime_config + # Same rationale as the guards above: an unvalidated run must fail the + # step on a malformed capture/timeout rather than dispatch with a + # coerced default or crash inside the subprocess wait. 'timeout' only + # applies to a captured step, so it is only checked for one. + capture_config_errors = capture_errors(config, "Command") + if capture_config_errors: + return StepResult( + status=StepStatus.FAILED, error=capture_config_errors[0] + ) + capture_max_bytes = capture_limit(config) + timeout = _DEFAULT_CAPTURE_TIMEOUT + if capture_max_bytes is not None: + invalid_timeout = timeout_error(config, "Command") + if invalid_timeout is not None: + return StepResult(status=StepStatus.FAILED, error=invalid_timeout) + timeout = config.get("timeout", _DEFAULT_CAPTURE_TIMEOUT) + args_str = str(resolved_input.get("args", "")) output: dict[str, Any] = { "command": command, @@ -157,6 +188,8 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: context, integration_args, integration_options, + capture_max_bytes=capture_max_bytes, + timeout=timeout, ) except ValueError as exc: output["exit_code"] = 1 @@ -172,11 +205,22 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: output["stdout"] = dispatch_result["stdout"] output["stderr"] = dispatch_result["stderr"] output["dispatched"] = True + if capture_max_bytes is not None: + record_capture(output, dispatch_result) if dispatch_result["exit_code"] != 0: + if capture_max_bytes is not None: + error = capture_failure( + dispatch_result, subject="Command", timeout=timeout + ) + else: + error = ( + dispatch_result["stderr"] + or f"Command exited with code {dispatch_result['exit_code']}" + ) return StepResult( status=StepStatus.FAILED, output=output, - error=dispatch_result["stderr"] or f"Command exited with code {dispatch_result['exit_code']}", + error=error, ) return StepResult( status=StepStatus.COMPLETED, @@ -209,6 +253,9 @@ def _try_dispatch( context: StepContext, integration_args: list[str], integration_options: dict[str, Any], + *, + capture_max_bytes: int | None = None, + timeout: float = _DEFAULT_CAPTURE_TIMEOUT, ) -> dict[str, Any] | None: """Invoke *command* by name through the integration CLI. @@ -217,9 +264,14 @@ def _try_dispatch( markdown agents, ``/speckit-specify`` for skills agents), then executes the CLI non-interactively. + With *capture_max_bytes* set, the CLI's textual output is captured + (bounded by that limit and by *timeout*) instead of streamed. + Returns the dispatch result dict, or ``None`` if dispatch is not possible (integration not found, CLI not installed, or - dispatch not supported). + dispatch not supported). Raises ``ValueError`` if capture is + requested from an integration whose ``dispatch_command`` predates + the capture keywords. """ if not integration_key or not isinstance(integration_key, str): # A non-string integration (a list/dict/expression that resolved to @@ -257,6 +309,27 @@ def _try_dispatch( if cli_path is None and fallback_cli_path is None: return None + capture_kwargs: dict[str, Any] = {} + if capture_max_bytes is not None: + capture_kwargs = { + "stream": False, + "output_json": False, + "timeout": timeout, + "capture_max_bytes": capture_max_bytes, + } + # An installed integration may override dispatch_command() with + # the signature from before these keywords existed; calling it + # would raise a raw TypeError and take down the run. + parameters = inspect.signature(impl.dispatch_command).parameters + if not any( + parameter.kind is inspect.Parameter.VAR_KEYWORD + for parameter in parameters.values() + ) and not all(name in parameters for name in capture_kwargs): + raise ValueError( + f"integration {integration_key!r} does not support " + f"'capture: true'." + ) + try: return impl.dispatch_command( command, @@ -265,6 +338,7 @@ def _try_dispatch( model=model, integration_args=integration_args, integration_options=integration_options, + **capture_kwargs, ) except (NotImplementedError, OSError): return None @@ -386,4 +460,11 @@ def validate(self, config: dict[str, Any]) -> list[str]: f"Command step {config.get('id', '?')!r}: 'model' must be a " f"string, got {type(model).__name__}." ) + errors.extend(capture_errors(config, "Command")) + # 'timeout' is only applied to a captured step, so a streaming step's + # value is left unchecked as it always was. + if config.get("capture") is True: + invalid_timeout = timeout_error(config, "Command") + if invalid_timeout is not None: + errors.append(invalid_timeout) return errors diff --git a/src/specify_cli/workflows/step/prompt/__init__.py b/src/specify_cli/workflows/step/prompt/__init__.py index 1d6b988c5a..0177260c67 100644 --- a/src/specify_cli/workflows/step/prompt/__init__.py +++ b/src/specify_cli/workflows/step/prompt/__init__.py @@ -7,9 +7,16 @@ from pathlib import Path from typing import Any +from specify_cli.integrations._capture import run_captured from specify_cli.integrations.installer import project_dispatch from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus from specify_cli.workflows.expressions import evaluate_expression +from specify_cli.workflows.step._capture import ( + capture_errors, + capture_failure, + capture_limit, + record_capture, +) class PromptStep(StepBase): @@ -25,8 +32,9 @@ class PromptStep(StepBase): CLI output is streamed to the terminal for live progress. ``output.exit_code`` is always captured and can be referenced - by later steps. Full response text capture is a planned - enhancement. + by later steps. ``capture: true`` retains the CLI's textual + stdout and stderr in ``output.stdout``/``output.stderr`` instead + of streaming them, capped per stream by ``capture_max_bytes``. Example YAML:: @@ -99,10 +107,24 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: if timeout_error is not None: return StepResult(status=StepStatus.FAILED, error=timeout_error) + # Same rationale: fail the step on a malformed capture config rather + # than silently streaming or capturing with a coerced limit. + capture_config_errors = capture_errors(config, "Prompt") + if capture_config_errors: + return StepResult( + status=StepStatus.FAILED, error=capture_config_errors[0] + ) + capture_max_bytes = capture_limit(config) + # Attempt CLI dispatch timeout = config.get("timeout", 300) dispatch_result = self._try_dispatch( - prompt, integration, model, context, timeout=timeout + prompt, + integration, + model, + context, + timeout=timeout, + capture_max_bytes=capture_max_bytes, ) output: dict[str, Any] = { @@ -116,14 +138,22 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: output["stdout"] = dispatch_result["stdout"] output["stderr"] = dispatch_result["stderr"] output["dispatched"] = True + if capture_max_bytes is not None: + record_capture(output, dispatch_result) if dispatch_result["exit_code"] != 0: + if capture_max_bytes is not None: + error = capture_failure( + dispatch_result, subject="Prompt", timeout=timeout + ) + else: + error = ( + dispatch_result["stderr"] + or f"Prompt exited with code {dispatch_result['exit_code']}" + ) return StepResult( status=StepStatus.FAILED, output=output, - error=( - dispatch_result["stderr"] - or f"Prompt exited with code {dispatch_result['exit_code']}" - ), + error=error, ) return StepResult( status=StepStatus.COMPLETED, @@ -189,8 +219,13 @@ def _try_dispatch( model: str | None, context: StepContext, timeout: int = 300, + capture_max_bytes: int | None = None, ) -> dict[str, Any] | None: - """Dispatch *prompt* directly through the integration CLI.""" + """Dispatch *prompt* directly through the integration CLI. + + With *capture_max_bytes* set, stdout and stderr are captured (bounded + by that limit) and returned instead of streamed to the terminal. + """ if not integration_key or not isinstance(integration_key, str) or not prompt: # A non-string integration would raise TypeError: unhashable type # from get_integration's dict lookup and abort the run; treat it as @@ -242,6 +277,17 @@ def _try_dispatch( import subprocess + if capture_max_bytes is not None: + try: + return run_captured( + exec_args, + cwd=str(project_root), + timeout=timeout, + max_bytes=capture_max_bytes, + ).as_dispatch_result() + except OSError: + return None + try: result = subprocess.run( exec_args, @@ -309,4 +355,5 @@ def validate(self, config: dict[str, Any]) -> list[str]: timeout_error = self._timeout_error(config) if timeout_error is not None: errors.append(timeout_error) + errors.extend(capture_errors(config, "Prompt")) return errors diff --git a/tests/integrations/test_capture.py b/tests/integrations/test_capture.py new file mode 100644 index 0000000000..c1da3a01cf --- /dev/null +++ b/tests/integrations/test_capture.py @@ -0,0 +1,289 @@ +"""Bounded subprocess capture and the captured ``dispatch_command`` contract.""" + +from __future__ import annotations + +import subprocess +import sys +import time + +import pytest + +from specify_cli.integrations import get_integration +from specify_cli.integrations import _capture +from specify_cli.integrations._capture import ( + DEFAULT_CAPTURE_MAX_BYTES, + CapturedProcess, + run_captured, +) + + +def _python(source: str, *args: str) -> list[str]: + return [sys.executable, "-u", "-c", source, *args] + + +class TestRunCaptured: + def test_captures_both_streams_and_exit_code(self): + result = run_captured( + _python( + "import sys; print('out'); sys.stderr.write('err'); sys.exit(3)" + ) + ) + assert result == CapturedProcess(exit_code=3, stdout="out\n", stderr="err") + + def test_default_limit_is_one_mebibyte(self): + assert DEFAULT_CAPTURE_MAX_BYTES == 1024 * 1024 + + def test_finite_limit_truncates_only_the_stream_that_exceeds_it(self): + # 3 MB is far beyond any pipe buffer: the process only exits 0 if the + # capped stream keeps being drained after the limit is reached. + result = run_captured( + _python( + "import sys; sys.stdout.write('a' * 3_000_000); " + "sys.stderr.write('e' * 10)" + ), + max_bytes=1000, + timeout=60, + ) + assert result.exit_code == 0 + assert result.stdout == "a" * 1000 + assert result.stdout_truncated is True + assert result.stderr == "e" * 10 + assert result.stderr_truncated is False + assert result.timed_out is False + + def test_output_exactly_at_the_limit_is_not_truncated(self): + result = run_captured( + _python("import sys; sys.stdout.write('a' * 64)"), max_bytes=64 + ) + assert result.stdout == "a" * 64 + assert result.stdout_truncated is False + + def test_unlimited_capture_retains_everything_without_flags(self): + result = run_captured( + _python("import sys; sys.stdout.write('a' * 3_000_000)"), + max_bytes=-1, + timeout=60, + ) + assert len(result.stdout) == 3_000_000 + assert result.stdout_truncated is False + assert result.stderr_truncated is False + + def test_limit_counts_raw_bytes_and_drops_a_split_character(self): + # Two 3-byte characters with a 4-byte limit: the second is cut after + # its first byte and dropped, not turned into a replacement character. + result = run_captured( + _python("import sys; sys.stdout.buffer.write('€€'.encode('utf-8'))"), + max_bytes=4, + ) + assert result.stdout == "€" + assert result.stdout_truncated is True + + def test_invalid_utf8_is_replaced(self): + result = run_captured( + _python(r"import sys; sys.stdout.buffer.write(b'ok\xff\xfeend')") + ) + assert result.stdout == "ok��end" + assert result.stdout_truncated is False + + def test_complete_output_ending_mid_character_is_replaced_not_dropped(self): + result = run_captured( + _python("import sys; sys.stdout.buffer.write('€'.encode('utf-8')[:2])") + ) + assert result.stdout == "�" + + def test_line_endings_are_normalised(self): + result = run_captured( + _python(r"import sys; sys.stdout.buffer.write(b'a\r\nb\rc\n')") + ) + assert result.stdout == "a\nb\nc\n" + + def test_timeout_keeps_partial_output_and_is_not_size_truncation(self): + started = time.monotonic() + result = run_captured( + _python("import time; print('partial'); time.sleep(60)"), + timeout=1, + max_bytes=-1, + ) + assert time.monotonic() - started < 30 + assert result.timed_out is True + assert result.exit_code == -1 + assert result.stdout == "partial\n" + assert result.stdout_truncated is False + assert result.interrupted is False + + def test_interrupt_kills_the_process_and_keeps_partial_output( + self, monkeypatch + ): + class InterruptedOnce(subprocess.Popen): + def wait(self, timeout=None): + if not getattr(self, "_interrupt_raised", False): + self._interrupt_raised = True + # Let the child write before the simulated Ctrl+C. + time.sleep(1) + raise KeyboardInterrupt + return super().wait(timeout) + + monkeypatch.setattr(_capture.subprocess, "Popen", InterruptedOnce) + started = time.monotonic() + result = run_captured( + _python("import time; print('partial'); time.sleep(60)") + ) + assert time.monotonic() - started < 30 + assert result.interrupted is True + assert result.exit_code == 130 + assert result.stdout == "partial\n" + assert result.timed_out is False + + @pytest.mark.parametrize("max_bytes", [0, -2, True, False, 1.5, "10", None]) + def test_invalid_limit_is_rejected_before_starting_a_process( + self, max_bytes, monkeypatch + ): + def fail(*args, **kwargs): + raise AssertionError("process must not start") + + monkeypatch.setattr(_capture.subprocess, "Popen", fail) + with pytest.raises(ValueError, match="positive integer or -1"): + run_captured(_python("pass"), max_bytes=max_bytes) + + def test_missing_executable_raises_oserror(self, tmp_path): + with pytest.raises(OSError): + run_captured([str(tmp_path / "no-such-cli")]) + + +class TestDispatchCommandCapture: + """``output_json`` and ``capture_max_bytes`` on ``dispatch_command``.""" + + @pytest.fixture + def claude(self, monkeypatch): + impl = get_integration("claude") + monkeypatch.setattr( + type(impl), + "build_exec_args", + lambda self, prompt, *, output_json=True, **kwargs: _python( + "import sys; print(sys.argv[1], sys.argv[2])", + "json" if output_json else "text", + prompt, + ), + ) + return impl + + def test_bounded_capture_returns_text_and_completeness_fields(self, claude): + result = claude.dispatch_command( + "speckit.plan", + stream=False, + output_json=False, + capture_max_bytes=1024, + timeout=60, + ) + assert result == { + "exit_code": 0, + "stdout": "text /speckit-plan\n", + "stderr": "", + "stdout_truncated": False, + "stderr_truncated": False, + "timed_out": False, + "interrupted": False, + } + + def test_existing_captured_callers_keep_structured_output(self, claude): + """``stream=False`` alone still requests JSON and returns three keys.""" + result = claude.dispatch_command("speckit.plan", stream=False, timeout=60) + assert result == { + "exit_code": 0, + "stdout": "json /speckit-plan\n", + "stderr": "", + } + + def test_existing_captured_callers_still_raise_on_timeout(self, monkeypatch): + impl = get_integration("claude") + monkeypatch.setattr( + type(impl), + "build_exec_args", + lambda self, prompt, **kwargs: _python("import time; time.sleep(60)"), + ) + with pytest.raises(subprocess.TimeoutExpired): + impl.dispatch_command("speckit.plan", stream=False, timeout=1) + + def test_bounded_capture_reports_timeout_instead_of_raising(self, monkeypatch): + impl = get_integration("claude") + monkeypatch.setattr( + type(impl), + "build_exec_args", + lambda self, prompt, **kwargs: _python( + "import time; print('partial'); time.sleep(60)" + ), + ) + result = impl.dispatch_command( + "speckit.plan", stream=False, capture_max_bytes=-1, timeout=1 + ) + assert result["timed_out"] is True + assert result["exit_code"] == -1 + assert result["stdout"] == "partial\n" + + def test_streaming_dispatch_still_requests_text(self, claude, capfd): + result = claude.dispatch_command("speckit.plan") + assert result == {"exit_code": 0, "stdout": "", "stderr": ""} + assert capfd.readouterr().out == "text /speckit-plan\n" + + def test_capture_limit_with_streaming_is_rejected(self, claude): + with pytest.raises(ValueError, match="requires stream=False"): + claude.dispatch_command("speckit.plan", capture_max_bytes=1024) + + +class TestCopilotDispatchCapture: + """Copilot overrides ``dispatch_command`` and builds its own argv.""" + + @pytest.fixture + def recorded(self, monkeypatch): + from specify_cli.integrations import copilot as copilot_module + + calls: list[tuple[list[str], dict]] = [] + + def fake_run_captured(args, **kwargs): + calls.append((list(args), kwargs)) + return CapturedProcess(exit_code=0, stdout="text", stderr="") + + def fake_run(args, **kwargs): + calls.append((list(args), kwargs)) + return subprocess.CompletedProcess(args, 0, stdout="{}", stderr="") + + monkeypatch.setattr(copilot_module, "run_captured", fake_run_captured) + monkeypatch.setattr("subprocess.run", fake_run) + return calls + + def test_bounded_text_capture_omits_json_format(self, recorded): + result = get_integration("copilot").dispatch_command( + "speckit.plan", + stream=False, + output_json=False, + capture_max_bytes=2048, + timeout=7, + ) + (args, kwargs), = recorded + assert "--output-format" not in args + assert kwargs["max_bytes"] == 2048 + assert kwargs["timeout"] == 7 + assert result["stdout"] == "text" + assert result["stdout_truncated"] is False + + def test_existing_captured_callers_keep_json_format(self, recorded): + result = get_integration("copilot").dispatch_command( + "speckit.plan", stream=False + ) + (args, kwargs), = recorded + assert args[args.index("--output-format") + 1] == "json" + assert kwargs["capture_output"] is True + assert result == {"exit_code": 0, "stdout": "{}", "stderr": ""} + + def test_streaming_omits_json_format(self, recorded): + get_integration("copilot").dispatch_command("speckit.plan") + (args, kwargs), = recorded + assert "--output-format" not in args + assert "capture_output" not in kwargs + + def test_capture_limit_with_streaming_is_rejected(self, recorded): + with pytest.raises(ValueError, match="requires stream=False"): + get_integration("copilot").dispatch_command( + "speckit.plan", capture_max_bytes=1024 + ) + assert recorded == [] diff --git a/tests/workflows/test_step_capture.py b/tests/workflows/test_step_capture.py new file mode 100644 index 0000000000..d25cfd11fe --- /dev/null +++ b/tests/workflows/test_step_capture.py @@ -0,0 +1,457 @@ +"""``capture: true`` on the ``command`` and ``prompt`` workflow steps.""" + +from __future__ import annotations + +import json +import sys + +import pytest +import yaml + +from specify_cli.integrations import get_integration +from specify_cli.integrations._capture import CapturedProcess +from specify_cli.integrations.base import IntegrationBase +from specify_cli.workflows.base import StepContext, StepStatus +from specify_cli.workflows.engine import WorkflowEngine +from specify_cli.workflows.step.command import CommandStep +from specify_cli.workflows.step.prompt import PromptStep + +STEPS = { + "command": (CommandStep, {"command": "speckit.plan"}), + "prompt": (PromptStep, {"type": "prompt", "prompt": "Summarise the plan"}), +} +LABELS = {"command": "Command", "prompt": "Prompt"} +CAPTURE_FIELDS = ("stdout_truncated", "stderr_truncated", "timed_out", "interrupted") + + +def _config(kind: str, **fields) -> dict: + return {"id": "ask", **STEPS[kind][1], **fields} + + +def _execute(kind: str, project_dir, **fields): + context = StepContext(default_integration="claude", project_root=str(project_dir)) + return STEPS[kind][0]().execute(_config(kind, **fields), context) + + +@pytest.fixture +def fake_cli(monkeypatch, project_dir): + """Make the ``claude`` integration run a Python script as its CLI. + + The script receives the requested output format (``json``/``text``) as + ``sys.argv[1]`` and counts its invocations in ``calls.txt``. + """ + + def install(body: str) -> None: + script = project_dir / "fake_cli.py" + script.write_text( + "import sys, time\n" + f"open({str(project_dir / 'calls.txt')!r}, 'a').write('x')\n" + body, + encoding="utf-8", + ) + monkeypatch.setattr( + type(get_integration("claude")), + "build_exec_args", + lambda self, prompt, *, output_json=True, **kwargs: [ + sys.executable, + "-u", + str(script), + "json" if output_json else "text", + ], + ) + + install.calls = lambda: len((project_dir / "calls.txt").read_text()) + return install + + +@pytest.mark.parametrize("kind", ["command", "prompt"]) +class TestCaptureValidation: + @pytest.mark.parametrize( + "fields", + [ + {}, + {"capture": False}, + {"capture": True}, + {"capture": True, "capture_max_bytes": 1}, + {"capture": True, "capture_max_bytes": -1}, + {"capture": True, "timeout": 30}, + ], + ) + def test_valid_configs_pass(self, kind, fields): + assert STEPS[kind][0]().validate(_config(kind, **fields)) == [] + + @pytest.mark.parametrize("capture", ["true", 1, 0, None, []]) + def test_non_boolean_capture_is_rejected(self, kind, capture): + errors = STEPS[kind][0]().validate(_config(kind, capture=capture)) + assert errors == [ + f"{LABELS[kind]} step 'ask': 'capture' must be true or false, " + f"got {capture!r}." + ] + + @pytest.mark.parametrize("max_bytes", [0, -2, True, False, 1.5, "1024", None]) + def test_invalid_limit_is_rejected(self, kind, max_bytes): + errors = STEPS[kind][0]().validate( + _config(kind, capture=True, capture_max_bytes=max_bytes) + ) + assert errors == [ + f"{LABELS[kind]} step 'ask': 'capture_max_bytes' must be a positive " + f"integer or -1 (no limit), got {max_bytes!r}." + ] + + @pytest.mark.parametrize("fields", [{}, {"capture": False}]) + def test_limit_without_capture_is_rejected(self, kind, fields): + errors = STEPS[kind][0]().validate( + _config(kind, capture_max_bytes=1024, **fields) + ) + assert errors == [ + f"{LABELS[kind]} step 'ask': 'capture_max_bytes' requires " + f"'capture: true'." + ] + + @pytest.mark.parametrize( + "fields", + [ + {"capture": "yes"}, + {"capture": True, "capture_max_bytes": 0}, + {"capture_max_bytes": 1024}, + {"capture": True, "timeout": -1}, + ], + ) + def test_unvalidated_execute_fails_the_step_without_dispatching( + self, kind, fields, project_dir, fake_cli + ): + fake_cli("print('ran')\n") + result = _execute(kind, project_dir, **fields) + assert result.status == StepStatus.FAILED + assert "'ask'" in result.error + assert not (project_dir / "calls.txt").exists() + + +class TestCommandTimeoutValidation: + @pytest.mark.parametrize("timeout", [0, -5, True, "60", float("inf")]) + def test_invalid_timeout_is_rejected_when_capturing(self, timeout): + errors = CommandStep().validate( + _config("command", capture=True, timeout=timeout) + ) + assert errors == [ + f"Command step 'ask': 'timeout' must be a positive number of " + f"seconds, got {timeout!r}." + ] + + def test_streaming_command_timeout_stays_unchecked(self): + """A streaming command never applied 'timeout'; validation is unchanged.""" + assert CommandStep().validate(_config("command", timeout="5m")) == [] + + +@pytest.mark.parametrize("kind", ["command", "prompt"]) +class TestCaptureExecution: + def test_captures_text_output_without_printing_it( + self, kind, project_dir, fake_cli, capfd + ): + fake_cli("print('format=' + sys.argv[1]); sys.stderr.write('note')\n") + result = _execute(kind, project_dir, capture=True) + + assert result.status == StepStatus.COMPLETED + assert result.error is None + assert result.output["exit_code"] == 0 + assert result.output["stdout"] == "format=text\n" + assert result.output["stderr"] == "note" + assert result.output["dispatched"] is True + assert {field: result.output[field] for field in CAPTURE_FIELDS} == dict.fromkeys( + CAPTURE_FIELDS, False + ) + printed = capfd.readouterr() + assert "format=text" not in printed.out + printed.err + + @pytest.mark.parametrize("fields", [{}, {"capture": False}]) + def test_streaming_is_unchanged_without_capture( + self, kind, fields, project_dir, fake_cli, capfd + ): + fake_cli("print('format=' + sys.argv[1]); sys.stderr.write('note')\n") + result = _execute(kind, project_dir, **fields) + + assert result.status == StepStatus.COMPLETED + assert result.output["stdout"] == "" + assert result.output["stderr"] == "" + assert not set(CAPTURE_FIELDS) & set(result.output) + printed = capfd.readouterr() + assert printed.out == "format=text\n" + assert printed.err == "note" + + def test_non_zero_exit_keeps_output_and_summarises_the_error( + self, kind, project_dir, fake_cli + ): + fake_cli("print('half done'); sys.stderr.write('token=SECRET'); sys.exit(7)\n") + result = _execute(kind, project_dir, capture=True) + + assert result.status == StepStatus.FAILED + assert result.error == f"{LABELS[kind]} exited with code 7" + assert result.output["exit_code"] == 7 + assert result.output["stdout"] == "half done\n" + assert result.output["stderr"] == "token=SECRET" + assert result.output["timed_out"] is False + + def test_streaming_non_zero_exit_error_is_unchanged( + self, kind, project_dir, fake_cli + ): + fake_cli("sys.exit(7)\n") + result = _execute(kind, project_dir) + assert result.status == StepStatus.FAILED + assert result.error == f"{LABELS[kind]} exited with code 7" + assert result.output["stderr"] == "" + + def test_finite_limit_truncates_and_flags(self, kind, project_dir, fake_cli): + fake_cli("sys.stdout.write('a' * 500_000); sys.stderr.write('e' * 4)\n") + result = _execute(kind, project_dir, capture=True, capture_max_bytes=16) + + assert result.status == StepStatus.COMPLETED + assert result.output["stdout"] == "a" * 16 + assert result.output["stdout_truncated"] is True + assert result.output["stderr"] == "eeee" + assert result.output["stderr_truncated"] is False + + def test_unlimited_capture_never_flags_truncation( + self, kind, project_dir, fake_cli + ): + fake_cli("sys.stdout.write('a' * 2_000_000)\n") + result = _execute(kind, project_dir, capture=True, capture_max_bytes=-1) + + assert result.status == StepStatus.COMPLETED + assert len(result.output["stdout"]) == 2_000_000 + assert result.output["stdout_truncated"] is False + + def test_non_utf8_output_is_decoded_with_replacement( + self, kind, project_dir, fake_cli + ): + fake_cli("sys.stdout.buffer.write('caf\\u00e9 '.encode('utf-8') + b'\\xff')\n") + result = _execute(kind, project_dir, capture=True) + assert result.output["stdout"] == "café \ufffd" + + def test_timeout_fails_the_step_with_partial_output( + self, kind, project_dir, fake_cli + ): + fake_cli("print('partial'); sys.stderr.write('slow'); time.sleep(60)\n") + result = _execute( + kind, project_dir, capture=True, capture_max_bytes=-1, timeout=1 + ) + + assert result.status == StepStatus.FAILED + assert result.error == f"{LABELS[kind]} timed out after 1 seconds." + assert result.output["exit_code"] == -1 + assert result.output["timed_out"] is True + assert result.output["stdout"] == "partial\n" + assert result.output["stderr"] == "slow" + # Partial because the process was stopped, not because of the limit. + assert result.output["stdout_truncated"] is False + + def test_interruption_fails_the_step_with_partial_output( + self, kind, project_dir, fake_cli, monkeypatch + ): + fake_cli("print('unused')\n") + interrupted = CapturedProcess( + exit_code=130, stdout="partial", stderr="noise", interrupted=True + ) + for module in ( + "specify_cli.integrations.base", + "specify_cli.workflows.step.prompt", + ): + monkeypatch.setattr( + f"{module}.run_captured", lambda *args, **kwargs: interrupted + ) + result = _execute(kind, project_dir, capture=True) + + assert result.status == StepStatus.FAILED + assert result.error == "Interrupted by user" + assert result.output["exit_code"] == 130 + assert result.output["interrupted"] is True + assert result.output["stdout"] == "partial" + assert result.output["stderr"] == "noise" + + +class TestCommandCaptureDispatch: + def test_default_timeout_is_600_seconds(self, project_dir, fake_cli, monkeypatch): + fake_cli("print('unused')\n") + seen: dict = {} + + def fake_run_captured(args, **kwargs): + seen.update(kwargs) + return CapturedProcess(exit_code=0, stdout="", stderr="") + + monkeypatch.setattr( + "specify_cli.integrations.base.run_captured", fake_run_captured + ) + _execute("command", project_dir, capture=True) + assert seen["timeout"] == 600 + assert seen["max_bytes"] == 1024 * 1024 + + def test_integration_without_capture_keywords_fails_cleanly( + self, project_dir, fake_cli, monkeypatch + ): + """An installed integration may still override the older signature.""" + fake_cli("print('ran')\n") + + def legacy_dispatch( + self, + command_name, + args="", + *, + project_root=None, + model=None, + timeout=600, + stream=True, + integration_args=None, + integration_options=None, + ): + return IntegrationBase.dispatch_command( + self, command_name, args, project_root=project_root, stream=stream + ) + + monkeypatch.setattr( + type(get_integration("claude")), "dispatch_command", legacy_dispatch + ) + captured = _execute("command", project_dir, capture=True) + assert captured.status == StepStatus.FAILED + assert captured.error == ( + "Command step 'ask': integration 'claude' does not support " + "'capture: true'." + ) + assert not (project_dir / "calls.txt").exists() + + streamed = _execute("command", project_dir) + assert streamed.status == StepStatus.COMPLETED + + +def _install_workflow(project_dir, steps: list[dict], inputs: dict | None = None): + definition = { + "schema_version": "1.0", + "workflow": { + "id": "wf", + "name": "WF", + "version": "1.0.0", + "integration": "claude", + }, + "steps": steps, + } + if inputs: + definition["inputs"] = inputs + workflow_dir = project_dir / ".specify" / "workflows" / "wf" + workflow_dir.mkdir(parents=True, exist_ok=True) + (workflow_dir / "workflow.yml").write_text( + yaml.safe_dump(definition), encoding="utf-8" + ) + engine = WorkflowEngine(project_dir) + return engine, engine.load_workflow("wf") + + +def _run_dir(project_dir, state): + return project_dir / ".specify" / "workflows" / "runs" / state.run_id + + +@pytest.mark.parametrize("kind", ["command", "prompt"]) +class TestCaptureInWorkflows: + """The composition gap from issue #4613, before and after ``capture``.""" + + @staticmethod + def _steps(kind: str, **fields) -> list[dict]: + return [ + _config(kind, **fields), + { + "id": "use", + "type": "shell", + "run": "echo verdict={{ steps.ask.output.stdout }}", + }, + ] + + def test_without_capture_downstream_sees_no_output( + self, kind, project_dir, fake_cli + ): + fake_cli("sys.stdout.write('READY')\n") + engine, definition = _install_workflow(project_dir, self._steps(kind)) + state = engine.execute(definition, {}) + + assert state.status.value == "completed" + assert state.step_results["use"]["output"]["stdout"].strip() == "verdict=" + + def test_with_capture_downstream_consumes_the_output( + self, kind, project_dir, fake_cli + ): + fake_cli("sys.stdout.write('READY')\n") + engine, definition = _install_workflow( + project_dir, self._steps(kind, capture=True) + ) + state = engine.execute(definition, {}) + + assert state.status.value == "completed" + assert state.step_results["use"]["output"]["stdout"].strip() == "verdict=READY" + + def test_captured_output_survives_resume_without_redispatch( + self, kind, project_dir, fake_cli + ): + fake_cli("sys.stdout.write('READY')\n") + steps = self._steps(kind, capture=True) + steps.insert( + 1, + { + "id": "review", + "type": "gate", + "message": "Continue?", + "options": ["approve", "reject"], + "verdict_input": "verdict", + }, + ) + engine, definition = _install_workflow( + project_dir, steps, inputs={"verdict": {"type": "string", "default": ""}} + ) + paused = engine.execute(definition, {}) + assert paused.status.value == "paused" + assert "use" not in paused.step_results + + resumed = WorkflowEngine(project_dir).resume( + paused.run_id, inputs={"verdict": "approve"} + ) + assert resumed.status.value == "completed" + assert resumed.step_results["use"]["output"]["stdout"].strip() == "verdict=READY" + assert fake_cli.calls() == 1 + + def test_captured_stderr_is_persisted_in_state_but_kept_out_of_the_log( + self, kind, project_dir, fake_cli, capfd + ): + fake_cli("sys.stderr.write('token=SECRET'); sys.exit(2)\n") + engine, definition = _install_workflow( + project_dir, self._steps(kind, capture=True) + ) + state = engine.execute(definition, {}) + + assert state.status.value == "failed" + run_dir = _run_dir(project_dir, state) + saved = json.loads((run_dir / "state.json").read_text(encoding="utf-8")) + assert saved["step_results"]["ask"]["output"]["stderr"] == "token=SECRET" + assert "SECRET" not in (run_dir / "log.jsonl").read_text(encoding="utf-8") + printed = capfd.readouterr() + assert "SECRET" not in printed.out + printed.err + + def test_continue_on_error_lets_downstream_inspect_a_failed_capture( + self, kind, project_dir, fake_cli + ): + fake_cli("print('partial'); time.sleep(60)\n") + steps = [ + _config(kind, capture=True, timeout=1, continue_on_error=True), + { + "id": "recover", + "type": "if", + "condition": "{{ steps.ask.output.timed_out }}", + "then": [ + { + "id": "report", + "type": "shell", + "run": "echo partial={{ steps.ask.output.stdout }}", + } + ], + }, + ] + engine, definition = _install_workflow(project_dir, steps) + state = engine.execute(definition, {}) + + assert state.status.value == "completed" + assert state.step_results["ask"]["output"]["exit_code"] == -1 + report = state.step_results["report"]["output"]["stdout"] + assert report.strip() == "partial=partial" diff --git a/workflows/README.md b/workflows/README.md index 642aa6ebbe..c71010892a 100644 --- a/workflows/README.md +++ b/workflows/README.md @@ -151,6 +151,32 @@ Send an arbitrary inline prompt to an integration CLI (no command file needed): integration: claude ``` +### Capturing Command and Prompt Output + +Command and prompt steps stream the CLI's output to the terminal, so their +`output.stdout` and `output.stderr` are empty. Set `capture: true` to retain the +output for later steps instead: + +```yaml +- id: verdict + type: prompt + prompt: "Reply with exactly READY or BLOCKED for {{ inputs.spec }}" + capture: true + capture_max_bytes: 4096 # Optional: per-stream limit (default 1 MiB, -1 = no limit) + timeout: 120 # Optional: seconds before the CLI is killed +``` + +A captured step prints nothing while it runs. Output past the limit is +discarded and reported in `output.stdout_truncated` / `output.stderr_truncated`; +a timeout or interruption fails the step, sets `output.timed_out` or +`output.interrupted`, and keeps the partial output. Check those fields before +parsing the text. + +> ⚠️ Captured output is stored unredacted in the run's `state.json`, and it is +> untrusted agent text. See +> [Capturing Command and Prompt Output](../docs/reference/workflows.md#capturing-command-and-prompt-output) +> for the full contract. + ### Shell Steps Run a shell command and capture output: From dec5943ac60fd5dff4972c93e8b2695bad3f27a5 Mon Sep 17 00:00:00 2001 From: "M.Bozorgmehr" Date: Wed, 14 Oct 2026 17:55:05 +0330 Subject: [PATCH 2/9] fix(workflows): kill the whole process tree when a captured step stops On a timeout or Ctrl+C, captured dispatch killed only the integration CLI. Processes it had started kept running and could keep changing the project after the step failed, and when they held stdout/stderr the reader threads stayed blocked. The CLI now leads its own session (POSIX) or job object (Windows), and the whole tree is killed before the readers are joined. Also normalise CRLF in the tests that compare streamed terminal output, which failed on Windows. Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 --- docs/reference/workflows.md | 4 +- src/specify_cli/integrations/_capture.py | 144 +++++++++++++++++++---- tests/integrations/test_capture.py | 87 +++++++++++++- tests/workflows/test_step_capture.py | 3 +- 4 files changed, 213 insertions(+), 25 deletions(-) diff --git a/docs/reference/workflows.md b/docs/reference/workflows.md index c7cdbf1e7e..b68844a6d0 100644 --- a/docs/reference/workflows.md +++ b/docs/reference/workflows.md @@ -1081,7 +1081,7 @@ where later steps can read it: | ------------------- | ------------------- | --------------------------------------------------------------------------------------------------- | | `capture` | `command`, `prompt` | `true` captures stdout and stderr instead of streaming them. Must be a literal boolean; default `false`. | | `capture_max_bytes` | `command`, `prompt` | Limit for each captured stream: a positive integer, or `-1` for no limit. Default `1048576` (1 MiB). Requires `capture: true`. | -| `timeout` | `command`, `prompt` | Seconds before the CLI is killed. For `command` it applies only with `capture: true` (default `600`); a streaming `command` step has no timeout. For `prompt` it always applies (default `300`). | +| `timeout` | `command`, `prompt` | Seconds before the CLI is killed; with `capture: true`, every process it started is killed too. For `command` it applies only with `capture: true` (default `600`); a streaming `command` step has no timeout. For `prompt` it always applies (default `300`). | A captured step prints nothing while it runs. Its output gains these fields: @@ -1090,7 +1090,7 @@ A captured step prints nothing while it runs. Its output gains these fields: | `stdout`, `stderr` | The captured text. | | `stdout_truncated` | `true` when stdout exceeded `capture_max_bytes` and the rest was discarded. | | `stderr_truncated` | The same for stderr. | -| `timed_out` | `true` when the CLI was killed at `timeout`; `exit_code` is `-1`. | +| `timed_out` | `true` when the CLI and its processes were killed at `timeout`; `exit_code` is `-1`. | | `interrupted` | `true` when the run was interrupted (Ctrl+C); `exit_code` is `130`. | **What the text is.** Capture requests the integration CLI's normal text diff --git a/src/specify_cli/integrations/_capture.py b/src/specify_cli/integrations/_capture.py index 0f11fd53dd..50c3332ec8 100644 --- a/src/specify_cli/integrations/_capture.py +++ b/src/specify_cli/integrations/_capture.py @@ -9,6 +9,8 @@ from __future__ import annotations import codecs +import os +import signal import subprocess import threading import time @@ -117,6 +119,102 @@ def _join(readers: Sequence[threading.Thread], deadline: float | None) -> bool: return not any(reader.is_alive() for reader in readers) +class _ProcessTree: + """Run a process so that it and every descendant can be killed together. + + Killing only the CLI would leave the tools it started running, still + changing the project and holding the pipes open after the step has + failed. On POSIX the CLI leads a new session and the whole process group + is killed; on Windows it is assigned to a job object that is terminated. + A separate session also keeps the terminal's Ctrl+C from reaching the CLI + directly: the interrupt is handled here, by killing the tree. + """ + + def __init__(self, args: Sequence[str], cwd: str | None) -> None: + isolation: dict[str, Any] = ( + {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} + if os.name == "nt" + else {"start_new_session": True} + ) + self.proc = subprocess.Popen( + args, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + cwd=cwd, + **isolation, + ) + self._job = _assign_job(self.proc) if os.name == "nt" else None + + def kill(self) -> None: + """Kill the process and its descendants, then reap the process.""" + if self._job is not None: + _terminate_job(self._job) + elif os.name != "nt": + try: + os.killpg(self.proc.pid, signal.SIGKILL) + except OSError: + pass + try: + self.proc.kill() + except OSError: + pass + self.proc.wait() + + def close(self) -> None: + if self._job is not None: + _close_handle(self._job) + self._job = None + + +def _kernel32() -> Any: + import ctypes + from ctypes import wintypes + + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32.CreateJobObjectW.argtypes = (wintypes.LPVOID, wintypes.LPCWSTR) + kernel32.CreateJobObjectW.restype = wintypes.HANDLE + kernel32.AssignProcessToJobObject.argtypes = (wintypes.HANDLE, wintypes.HANDLE) + kernel32.AssignProcessToJobObject.restype = wintypes.BOOL + kernel32.TerminateJobObject.argtypes = (wintypes.HANDLE, wintypes.UINT) + kernel32.TerminateJobObject.restype = wintypes.BOOL + kernel32.CloseHandle.argtypes = (wintypes.HANDLE,) + kernel32.CloseHandle.restype = wintypes.BOOL + return kernel32 + + +def _assign_job(proc: subprocess.Popen) -> Any: + """Put *proc* in a new job object; return the job, or None on failure. + + Processes the CLI starts later join the job automatically. Without a job + only the CLI itself can be killed. + """ + try: + kernel32 = _kernel32() + job = kernel32.CreateJobObjectW(None, None) + if not job: + return None + if not kernel32.AssignProcessToJobObject(job, int(proc._handle)): + kernel32.CloseHandle(job) + return None + return job + except (AttributeError, OSError): + return None + + +def _terminate_job(job: Any) -> None: + try: + _kernel32().TerminateJobObject(job, 1) + except OSError: + pass + + +def _close_handle(handle: Any) -> None: + try: + _kernel32().CloseHandle(handle) + except OSError: + pass + + def _decode(data: bytes, *, complete: bool) -> str: """Decode captured bytes as UTF-8, replacing invalid sequences with U+FFFD. @@ -142,19 +240,16 @@ def run_captured( *max_bytes* counts raw bytes before decoding; ``-1`` disables the cap. A timeout or ``KeyboardInterrupt`` kills the process and is reported on the result (exit code ``-1`` or ``130``) together with the output read so far, - instead of being raised. ``OSError`` from starting the process propagates. + instead of being raised, after the process and its descendants are killed. + ``OSError`` from starting the process propagates. """ if not capture_max_bytes_is_valid(max_bytes): raise ValueError( f"max_bytes must be a positive integer or -1, got {max_bytes!r}." ) - proc = subprocess.Popen( - args, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - cwd=cwd, - ) + tree = _ProcessTree(args, cwd) + proc = tree.proc sinks = (_Sink(max_bytes), _Sink(max_bytes)) readers = [ threading.Thread(target=_drain, args=(pipe, sink), daemon=True) @@ -167,20 +262,27 @@ def run_captured( timed_out = interrupted = False try: try: - proc.wait(timeout=timeout) - except subprocess.TimeoutExpired: - timed_out = True - else: - # The process can exit while a descendant still holds the pipes - # open; the same deadline bounds the wait for end-of-file. - timed_out = not _join(readers, deadline) - except KeyboardInterrupt: - interrupted = True - - if timed_out or interrupted: - proc.kill() - proc.wait() - _join(readers, time.monotonic() + _DRAIN_GRACE_SECONDS) + try: + proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + timed_out = True + else: + # The process can exit while a descendant still holds the + # pipes open; the same deadline bounds the wait for + # end-of-file. + timed_out = not _join(readers, deadline) + except KeyboardInterrupt: + interrupted = True + + if timed_out or interrupted: + # Killing the whole tree closes every write end of the pipes, so + # the readers reach end-of-file. Only a descendant that left the + # session or job could keep them open; the grace period bounds + # that wait. + tree.kill() + _join(readers, time.monotonic() + _DRAIN_GRACE_SECONDS) + finally: + tree.close() (stdout, stdout_truncated), (stderr, stderr_truncated) = ( sink.snapshot() for sink in sinks diff --git a/tests/integrations/test_capture.py b/tests/integrations/test_capture.py index c1da3a01cf..f9748cb180 100644 --- a/tests/integrations/test_capture.py +++ b/tests/integrations/test_capture.py @@ -4,6 +4,7 @@ import subprocess import sys +import threading import time import pytest @@ -21,6 +22,43 @@ def _python(source: str, *args: str) -> list[str]: return [sys.executable, "-u", "-c", source, *args] +# Writes to the heartbeat file every 0.1s until it is killed. +_HEARTBEAT = ( + "import sys, time\n" + "while True:\n" + " with open(sys.argv[1], 'a') as f: f.write('.')\n" + " time.sleep(0.1)\n" +) + + +def _spawn_descendant(heartbeat, then: str) -> list[str]: + """Return a CLI that starts a heartbeat descendant, then runs *then*. + + The descendant inherits stdout and stderr, so it also holds the pipes. + """ + return _python( + "import subprocess, sys, time\n" + "subprocess.Popen([sys.executable, '-c', sys.argv[1], sys.argv[2]])\n" + "while not __import__('os').path.exists(sys.argv[2]): time.sleep(0.05)\n" + "print('started')\n" + then, + _HEARTBEAT, + str(heartbeat), + ) + + +def _assert_stopped(heartbeat) -> None: + """Assert the heartbeat descendant is dead and no reader is left running.""" + time.sleep(0.5) + size = heartbeat.stat().st_size + time.sleep(1) + assert heartbeat.stat().st_size == size, "descendant still running" + assert not [ + thread + for thread in threading.enumerate() + if thread.name.endswith("(_drain)") + ], "reader threads still running" + + class TestRunCaptured: def test_captures_both_streams_and_exit_code(self): result = run_captured( @@ -134,6 +172,52 @@ def wait(self, timeout=None): assert result.stdout == "partial\n" assert result.timed_out is False + def test_timeout_kills_descendants_and_stops_readers(self, tmp_path): + heartbeat = tmp_path / "heartbeat" + result = run_captured( + _spawn_descendant(heartbeat, "time.sleep(60)"), timeout=5 + ) + assert result.timed_out is True + assert result.stdout == "started\n" + _assert_stopped(heartbeat) + + def test_descendant_holding_the_pipes_after_exit_is_killed(self, tmp_path): + """The CLI exits, but its descendant keeps the pipes open.""" + heartbeat = tmp_path / "heartbeat" + started = time.monotonic() + result = run_captured(_spawn_descendant(heartbeat, ""), timeout=5) + assert time.monotonic() - started < 30 + assert result.timed_out is True + assert result.stdout == "started\n" + _assert_stopped(heartbeat) + + def test_interrupt_kills_descendants_and_stops_readers( + self, tmp_path, monkeypatch + ): + class InterruptedOnce(subprocess.Popen): + def wait(self, timeout=None): + if not getattr(self, "_interrupt_raised", False): + self._interrupt_raised = True + while not heartbeat.exists(): + time.sleep(0.05) + raise KeyboardInterrupt + return super().wait(timeout) + + heartbeat = tmp_path / "heartbeat" + monkeypatch.setattr(_capture.subprocess, "Popen", InterruptedOnce) + result = run_captured(_spawn_descendant(heartbeat, "time.sleep(60)")) + assert result.interrupted is True + assert result.exit_code == 130 + _assert_stopped(heartbeat) + + def test_completed_process_is_not_killed(self, monkeypatch): + def fail(self): + raise AssertionError("a completed process must not be killed") + + monkeypatch.setattr(_capture._ProcessTree, "kill", fail) + result = run_captured(_python("print('done')"), timeout=30) + assert result == CapturedProcess(exit_code=0, stdout="done\n", stderr="") + @pytest.mark.parametrize("max_bytes", [0, -2, True, False, 1.5, "10", None]) def test_invalid_limit_is_rejected_before_starting_a_process( self, max_bytes, monkeypatch @@ -223,7 +307,8 @@ def test_bounded_capture_reports_timeout_instead_of_raising(self, monkeypatch): def test_streaming_dispatch_still_requests_text(self, claude, capfd): result = claude.dispatch_command("speckit.plan") assert result == {"exit_code": 0, "stdout": "", "stderr": ""} - assert capfd.readouterr().out == "text /speckit-plan\n" + printed = capfd.readouterr().out.replace("\r\n", "\n") + assert printed == "text /speckit-plan\n" def test_capture_limit_with_streaming_is_rejected(self, claude): with pytest.raises(ValueError, match="requires stream=False"): diff --git a/tests/workflows/test_step_capture.py b/tests/workflows/test_step_capture.py index d25cfd11fe..2fca03a769 100644 --- a/tests/workflows/test_step_capture.py +++ b/tests/workflows/test_step_capture.py @@ -174,7 +174,8 @@ def test_streaming_is_unchanged_without_capture( assert result.output["stderr"] == "" assert not set(CAPTURE_FIELDS) & set(result.output) printed = capfd.readouterr() - assert printed.out == "format=text\n" + # Windows writes \r\n to the terminal. + assert printed.out.replace("\r\n", "\n") == "format=text\n" assert printed.err == "note" def test_non_zero_exit_keeps_output_and_summarises_the_error( From 34ca95a06930e0a2bb32dd0eecaed25522b725ad Mon Sep 17 00:00:00 2001 From: "M.Bozorgmehr" Date: Wed, 14 Oct 2026 21:27:44 +0330 Subject: [PATCH 3/9] fix(workflows): start the Windows CLI suspended until it is in the job The CLI was assigned to its job object only after it had started running, so a process it started in that window was outside the job and survived the kill. The CLI is now created suspended, assigned to the job and then resumed. If any of that fails, the CLI is killed before it runs and the error propagates, instead of running without a way to kill its descendants. Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 --- src/specify_cli/integrations/_capture.py | 167 +++++++++++++++++------ tests/integrations/test_capture.py | 30 ++++ 2 files changed, 155 insertions(+), 42 deletions(-) diff --git a/src/specify_cli/integrations/_capture.py b/src/specify_cli/integrations/_capture.py index 50c3332ec8..6d58cc6884 100644 --- a/src/specify_cli/integrations/_capture.py +++ b/src/specify_cli/integrations/_capture.py @@ -119,20 +119,24 @@ def _join(readers: Sequence[threading.Thread], deadline: float | None) -> bool: return not any(reader.is_alive() for reader in readers) +# Windows process creation flag; not exported by ``subprocess``. +_CREATE_SUSPENDED = 0x00000004 + + class _ProcessTree: """Run a process so that it and every descendant can be killed together. Killing only the CLI would leave the tools it started running, still changing the project and holding the pipes open after the step has failed. On POSIX the CLI leads a new session and the whole process group - is killed; on Windows it is assigned to a job object that is terminated. + is killed; on Windows it runs inside a job object that is terminated. A separate session also keeps the terminal's Ctrl+C from reaching the CLI directly: the interrupt is handled here, by killing the tree. """ def __init__(self, args: Sequence[str], cwd: str | None) -> None: isolation: dict[str, Any] = ( - {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} + {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP | _CREATE_SUSPENDED} if os.name == "nt" else {"start_new_session": True} ) @@ -143,13 +147,25 @@ def __init__(self, args: Sequence[str], cwd: str | None) -> None: cwd=cwd, **isolation, ) - self._job = _assign_job(self.proc) if os.name == "nt" else None + self._job: Any = None + if os.name == "nt": + try: + self._job = _start_in_job(self.proc) + except BaseException: + # Fail closed: a CLI whose descendants could not be killed + # never runs. + self.proc.kill() + self.proc.wait() + for pipe in (self.proc.stdout, self.proc.stderr): + if pipe is not None: + pipe.close() + raise def kill(self) -> None: """Kill the process and its descendants, then reap the process.""" if self._job is not None: - _terminate_job(self._job) - elif os.name != "nt": + _windows().TerminateJobObject(self._job, 1) + else: try: os.killpg(self.proc.pid, signal.SIGKILL) except OSError: @@ -162,57 +178,124 @@ def kill(self) -> None: def close(self) -> None: if self._job is not None: - _close_handle(self._job) + _windows().CloseHandle(self._job) self._job = None -def _kernel32() -> Any: - import ctypes - from ctypes import wintypes - - kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) - kernel32.CreateJobObjectW.argtypes = (wintypes.LPVOID, wintypes.LPCWSTR) - kernel32.CreateJobObjectW.restype = wintypes.HANDLE - kernel32.AssignProcessToJobObject.argtypes = (wintypes.HANDLE, wintypes.HANDLE) - kernel32.AssignProcessToJobObject.restype = wintypes.BOOL - kernel32.TerminateJobObject.argtypes = (wintypes.HANDLE, wintypes.UINT) - kernel32.TerminateJobObject.restype = wintypes.BOOL - kernel32.CloseHandle.argtypes = (wintypes.HANDLE,) - kernel32.CloseHandle.restype = wintypes.BOOL - return kernel32 +_kernel32: Any = None + + +def _windows() -> Any: + """Return kernel32 with the signatures the job-object helpers use.""" + global _kernel32 + if _kernel32 is None: + import ctypes + from ctypes import wintypes + + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + signatures = { + "CreateJobObjectW": ((wintypes.LPVOID, wintypes.LPCWSTR), wintypes.HANDLE), + "AssignProcessToJobObject": ( + (wintypes.HANDLE, wintypes.HANDLE), + wintypes.BOOL, + ), + "TerminateJobObject": ((wintypes.HANDLE, wintypes.UINT), wintypes.BOOL), + "CreateToolhelp32Snapshot": ( + (wintypes.DWORD, wintypes.DWORD), + wintypes.HANDLE, + ), + "Thread32First": ((wintypes.HANDLE, ctypes.c_void_p), wintypes.BOOL), + "Thread32Next": ((wintypes.HANDLE, ctypes.c_void_p), wintypes.BOOL), + "OpenThread": ( + (wintypes.DWORD, wintypes.BOOL, wintypes.DWORD), + wintypes.HANDLE, + ), + "ResumeThread": ((wintypes.HANDLE,), wintypes.DWORD), + "CloseHandle": ((wintypes.HANDLE,), wintypes.BOOL), + } + for name, (argtypes, restype) in signatures.items(): + function = getattr(kernel32, name) + function.argtypes = argtypes + function.restype = restype + _kernel32 = kernel32 + return _kernel32 -def _assign_job(proc: subprocess.Popen) -> Any: - """Put *proc* in a new job object; return the job, or None on failure. +def _start_in_job(proc: subprocess.Popen) -> Any: + """Put the suspended *proc* in a new job object, then resume it. - Processes the CLI starts later join the job automatically. Without a job - only the CLI itself can be killed. + *proc* was created suspended, so it cannot start a descendant before it + is in the job; every process it starts afterwards joins the job too. + Returns the job handle. Raises OSError if any step fails. """ + import ctypes + + kernel32 = _windows() + job = kernel32.CreateJobObjectW(None, None) + if not job: + raise ctypes.WinError(ctypes.get_last_error()) try: - kernel32 = _kernel32() - job = kernel32.CreateJobObjectW(None, None) - if not job: - return None if not kernel32.AssignProcessToJobObject(job, int(proc._handle)): - kernel32.CloseHandle(job) - return None - return job - except (AttributeError, OSError): - return None + raise ctypes.WinError(ctypes.get_last_error()) + _resume_threads(proc.pid) + except BaseException: + kernel32.CloseHandle(job) + raise + return job -def _terminate_job(job: Any) -> None: - try: - _kernel32().TerminateJobObject(job, 1) - except OSError: - pass +def _resume_threads(pid: int) -> None: + """Resume the threads of the suspended process *pid*. + ``subprocess`` closes the main thread's handle, so the thread is found + through a thread snapshot. + """ + import ctypes + from ctypes import wintypes -def _close_handle(handle: Any) -> None: + class THREADENTRY32(ctypes.Structure): + _fields_ = [ + ("dwSize", wintypes.DWORD), + ("cntUsage", wintypes.DWORD), + ("th32ThreadID", wintypes.DWORD), + ("th32OwnerProcessID", wintypes.DWORD), + ("tpBasePri", wintypes.LONG), + ("tpDeltaPri", wintypes.LONG), + ("dwFlags", wintypes.DWORD), + ] + + th32cs_snapthread = 0x00000004 + thread_suspend_resume = 0x0002 + resume_failed = 0xFFFFFFFF + invalid_handle = wintypes.HANDLE(-1).value + + kernel32 = _windows() + snapshot = kernel32.CreateToolhelp32Snapshot(th32cs_snapthread, 0) + if not snapshot or snapshot == invalid_handle: + raise ctypes.WinError(ctypes.get_last_error()) + resumed = 0 try: - _kernel32().CloseHandle(handle) - except OSError: - pass + entry = THREADENTRY32() + entry.dwSize = ctypes.sizeof(THREADENTRY32) + more = kernel32.Thread32First(snapshot, ctypes.byref(entry)) + while more: + if entry.th32OwnerProcessID == pid: + thread = kernel32.OpenThread( + thread_suspend_resume, False, entry.th32ThreadID + ) + if not thread: + raise ctypes.WinError(ctypes.get_last_error()) + try: + if kernel32.ResumeThread(thread) == resume_failed: + raise ctypes.WinError(ctypes.get_last_error()) + finally: + kernel32.CloseHandle(thread) + resumed += 1 + more = kernel32.Thread32Next(snapshot, ctypes.byref(entry)) + finally: + kernel32.CloseHandle(snapshot) + if not resumed: + raise OSError(f"Could not find the threads of process {pid} to resume.") def _decode(data: bytes, *, complete: bool) -> str: diff --git a/tests/integrations/test_capture.py b/tests/integrations/test_capture.py index f9748cb180..280989ed29 100644 --- a/tests/integrations/test_capture.py +++ b/tests/integrations/test_capture.py @@ -218,6 +218,36 @@ def fail(self): result = run_captured(_python("print('done')"), timeout=30) assert result == CapturedProcess(exit_code=0, stdout="done\n", stderr="") + @pytest.mark.skipif(sys.platform != "win32", reason="Windows job objects") + def test_windows_cli_is_in_the_job_before_its_first_instruction(self): + """The CLI starts suspended, so it cannot spawn outside the job.""" + result = run_captured( + _python( + "import ctypes\n" + "in_job = ctypes.c_int()\n" + "ctypes.windll.kernel32.IsProcessInJob(\n" + " ctypes.windll.kernel32.GetCurrentProcess(), None,\n" + " ctypes.byref(in_job))\n" + "print(bool(in_job.value))\n" + ), + timeout=60, + ) + assert result.stdout == "True\n" + + @pytest.mark.skipif(sys.platform != "win32", reason="Windows job objects") + def test_windows_cli_never_runs_without_a_job(self, tmp_path, monkeypatch): + def fail(proc): + raise OSError("job assignment failed") + + marker = tmp_path / "ran" + monkeypatch.setattr(_capture, "_start_in_job", fail) + with pytest.raises(OSError, match="job assignment failed"): + run_captured( + _python("import sys; open(sys.argv[1], 'w').close()", str(marker)) + ) + time.sleep(1) + assert not marker.exists() + @pytest.mark.parametrize("max_bytes", [0, -2, True, False, 1.5, "10", None]) def test_invalid_limit_is_rejected_before_starting_a_process( self, max_bytes, monkeypatch From e0346e02373bda406376233ace27ed9670494603 Mon Sep 17 00:00:00 2001 From: "M.Bozorgmehr" Date: Wed, 14 Oct 2026 22:48:12 +0330 Subject: [PATCH 4/9] fix(workflows): state the group/job kill boundary and fix the Windows test The timeout docs and the capture helper's docstrings claimed every process the CLI started is killed. The kill boundary is the CLI's process group (POSIX) or job object (Windows); a helper that detaches from it survives. Say so in the reference and in the helper. The Windows test that checked job membership from inside the CLI called GetCurrentProcess() without a declared return type, so the 64-bit pseudo handle was truncated and IsProcessInJob always reported False. A runner may also already run inside a job, so the check could not show the race was closed. Replace it with a test that the CLI has not run before it is assigned to the job, and runs normally afterwards. Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 --- docs/reference/workflows.md | 2 +- src/specify_cli/integrations/_capture.py | 26 +++++++++++++---------- tests/integrations/test_capture.py | 27 +++++++++++++++++------- 3 files changed, 35 insertions(+), 20 deletions(-) diff --git a/docs/reference/workflows.md b/docs/reference/workflows.md index b68844a6d0..8b95c6918e 100644 --- a/docs/reference/workflows.md +++ b/docs/reference/workflows.md @@ -1081,7 +1081,7 @@ where later steps can read it: | ------------------- | ------------------- | --------------------------------------------------------------------------------------------------- | | `capture` | `command`, `prompt` | `true` captures stdout and stderr instead of streaming them. Must be a literal boolean; default `false`. | | `capture_max_bytes` | `command`, `prompt` | Limit for each captured stream: a positive integer, or `-1` for no limit. Default `1048576` (1 MiB). Requires `capture: true`. | -| `timeout` | `command`, `prompt` | Seconds before the CLI is killed; with `capture: true`, every process it started is killed too. For `command` it applies only with `capture: true` (default `600`); a streaming `command` step has no timeout. For `prompt` it always applies (default `300`). | +| `timeout` | `command`, `prompt` | Seconds before the CLI is killed; with `capture: true`, its process group (POSIX) or job object (Windows) is killed, which includes the processes it started unless they detached from it, as a daemonising helper can. For `command` it applies only with `capture: true` (default `600`); a streaming `command` step has no timeout. For `prompt` it always applies (default `300`). | A captured step prints nothing while it runs. Its output gains these fields: diff --git a/src/specify_cli/integrations/_capture.py b/src/specify_cli/integrations/_capture.py index 6d58cc6884..464f6fba94 100644 --- a/src/specify_cli/integrations/_capture.py +++ b/src/specify_cli/integrations/_capture.py @@ -124,14 +124,17 @@ def _join(readers: Sequence[threading.Thread], deadline: float | None) -> bool: class _ProcessTree: - """Run a process so that it and every descendant can be killed together. + """Run a process so that it and its process group or job can be killed. Killing only the CLI would leave the tools it started running, still changing the project and holding the pipes open after the step has - failed. On POSIX the CLI leads a new session and the whole process group - is killed; on Windows it runs inside a job object that is terminated. + failed. On POSIX the CLI leads a new session and its process group is + killed; on Windows it runs inside a job object that is terminated. The + boundary is that group or job, not every descendant: a process that + moves to another session or process group (POSIX) or breaks away from + the job (Windows), as a daemonising helper can, survives the kill. A separate session also keeps the terminal's Ctrl+C from reaching the CLI - directly: the interrupt is handled here, by killing the tree. + directly: the interrupt is handled here, by killing the group or job. """ def __init__(self, args: Sequence[str], cwd: str | None) -> None: @@ -162,7 +165,7 @@ def __init__(self, args: Sequence[str], cwd: str | None) -> None: raise def kill(self) -> None: - """Kill the process and its descendants, then reap the process.""" + """Kill the process group or job, then reap the process.""" if self._job is not None: _windows().TerminateJobObject(self._job, 1) else: @@ -323,8 +326,9 @@ def run_captured( *max_bytes* counts raw bytes before decoding; ``-1`` disables the cap. A timeout or ``KeyboardInterrupt`` kills the process and is reported on the result (exit code ``-1`` or ``130``) together with the output read so far, - instead of being raised, after the process and its descendants are killed. - ``OSError`` from starting the process propagates. + instead of being raised, after the process's group (POSIX) or job + (Windows) is killed; a descendant that left it is not killed. ``OSError`` + from starting the process propagates. """ if not capture_max_bytes_is_valid(max_bytes): raise ValueError( @@ -358,10 +362,10 @@ def run_captured( interrupted = True if timed_out or interrupted: - # Killing the whole tree closes every write end of the pipes, so - # the readers reach end-of-file. Only a descendant that left the - # session or job could keep them open; the grace period bounds - # that wait. + # Killing the group or job closes the write ends of the pipes its + # processes hold, so the readers reach end-of-file. Only a + # descendant that left the group or job could keep them open; the + # grace period bounds that wait. tree.kill() _join(readers, time.monotonic() + _DRAIN_GRACE_SECONDS) finally: diff --git a/tests/integrations/test_capture.py b/tests/integrations/test_capture.py index 280989ed29..093d1b201a 100644 --- a/tests/integrations/test_capture.py +++ b/tests/integrations/test_capture.py @@ -219,20 +219,31 @@ def fail(self): assert result == CapturedProcess(exit_code=0, stdout="done\n", stderr="") @pytest.mark.skipif(sys.platform != "win32", reason="Windows job objects") - def test_windows_cli_is_in_the_job_before_its_first_instruction(self): + def test_windows_cli_does_not_run_before_it_is_in_the_job( + self, tmp_path, monkeypatch + ): """The CLI starts suspended, so it cannot spawn outside the job.""" + marker = tmp_path / "ran" + start_in_job = _capture._start_in_job + ran_before_job: list[bool] = [] + + def observe(proc): + # A running CLI would create the marker well within this delay. + time.sleep(1) + ran_before_job.append(marker.exists()) + return start_in_job(proc) + + monkeypatch.setattr(_capture, "_start_in_job", observe) result = run_captured( _python( - "import ctypes\n" - "in_job = ctypes.c_int()\n" - "ctypes.windll.kernel32.IsProcessInJob(\n" - " ctypes.windll.kernel32.GetCurrentProcess(), None,\n" - " ctypes.byref(in_job))\n" - "print(bool(in_job.value))\n" + "import sys; open(sys.argv[1], 'w').close(); print('ran')", + str(marker), ), timeout=60, ) - assert result.stdout == "True\n" + assert ran_before_job == [False] + assert result.stdout == "ran\n" + assert marker.exists() @pytest.mark.skipif(sys.platform != "win32", reason="Windows job objects") def test_windows_cli_never_runs_without_a_job(self, tmp_path, monkeypatch): From baa78b466e1fe11ab9f65d02bbfa47e8661d6cbc Mon Sep 17 00:00:00 2001 From: "M.Bozorgmehr" Date: Wed, 14 Oct 2026 23:23:11 +0330 Subject: [PATCH 5/9] fix(workflows): fail closed when captured setup is interrupted Ctrl+C during setup could leave processes running. On Windows, an interruption after the CLI was resumed but before setup returned closed the job without terminating it, so a process the CLI had started survived the CLI's kill. In run_captured, an interruption while the reader threads were starting escaped before the try block that kills the process group or job, leaving the whole tree running. Terminate the job before closing it on that path, and start the readers inside the try so the interruption is handled like any other: the group or job is killed and the step reports interrupted. Pipes whose reader never started are closed. Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 --- src/specify_cli/integrations/_capture.py | 23 ++++++++++----- tests/integrations/test_capture.py | 36 ++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 7 deletions(-) diff --git a/src/specify_cli/integrations/_capture.py b/src/specify_cli/integrations/_capture.py index 464f6fba94..f985f071a0 100644 --- a/src/specify_cli/integrations/_capture.py +++ b/src/specify_cli/integrations/_capture.py @@ -242,6 +242,10 @@ def _start_in_job(proc: subprocess.Popen) -> Any: raise ctypes.WinError(ctypes.get_last_error()) _resume_threads(proc.pid) except BaseException: + # The CLI may already be running if the failure or interruption came + # after a thread was resumed: kill everything in the job before the + # handle goes, because closing it does not. + kernel32.TerminateJobObject(job, 1) kernel32.CloseHandle(job) raise return job @@ -338,17 +342,19 @@ def run_captured( tree = _ProcessTree(args, cwd) proc = tree.proc sinks = (_Sink(max_bytes), _Sink(max_bytes)) - readers = [ - threading.Thread(target=_drain, args=(pipe, sink), daemon=True) - for pipe, sink in zip((proc.stdout, proc.stderr), sinks) - ] - for reader in readers: - reader.start() - + readers: list[threading.Thread] = [] deadline = None if timeout is None else time.monotonic() + timeout timed_out = interrupted = False try: try: + # Starting the readers is inside the try, so an interruption here + # still kills the process group or job below. + for pipe, sink in zip((proc.stdout, proc.stderr), sinks): + reader = threading.Thread( + target=_drain, args=(pipe, sink), daemon=True + ) + reader.start() + readers.append(reader) try: proc.wait(timeout=timeout) except subprocess.TimeoutExpired: @@ -368,6 +374,9 @@ def run_captured( # grace period bounds that wait. tree.kill() _join(readers, time.monotonic() + _DRAIN_GRACE_SECONDS) + # A pipe whose reader never started is closed here instead. + for pipe in (proc.stdout, proc.stderr)[len(readers) :]: + pipe.close() finally: tree.close() diff --git a/tests/integrations/test_capture.py b/tests/integrations/test_capture.py index 093d1b201a..e5c22e725d 100644 --- a/tests/integrations/test_capture.py +++ b/tests/integrations/test_capture.py @@ -210,6 +210,23 @@ def wait(self, timeout=None): assert result.exit_code == 130 _assert_stopped(heartbeat) + def test_interrupt_while_starting_the_readers_kills_descendants( + self, tmp_path, monkeypatch + ): + class InterruptedStart(threading.Thread): + def start(self): + # Interrupt once the CLI is running and has a descendant. + while not heartbeat.exists(): + time.sleep(0.05) + raise KeyboardInterrupt + + heartbeat = tmp_path / "heartbeat" + monkeypatch.setattr(_capture.threading, "Thread", InterruptedStart) + result = run_captured(_spawn_descendant(heartbeat, "time.sleep(60)")) + assert result.interrupted is True + assert result.exit_code == 130 + _assert_stopped(heartbeat) + def test_completed_process_is_not_killed(self, monkeypatch): def fail(self): raise AssertionError("a completed process must not be killed") @@ -245,6 +262,25 @@ def observe(proc): assert result.stdout == "ran\n" assert marker.exists() + @pytest.mark.skipif(sys.platform != "win32", reason="Windows job objects") + def test_windows_interrupt_after_resume_kills_the_job( + self, tmp_path, monkeypatch + ): + """Setup interrupted after the CLI was resumed still fails closed.""" + resume_threads = _capture._resume_threads + + def resume_then_interrupt(pid): + resume_threads(pid) + while not heartbeat.exists(): + time.sleep(0.05) + raise KeyboardInterrupt + + heartbeat = tmp_path / "heartbeat" + monkeypatch.setattr(_capture, "_resume_threads", resume_then_interrupt) + with pytest.raises(KeyboardInterrupt): + run_captured(_spawn_descendant(heartbeat, "time.sleep(60)")) + _assert_stopped(heartbeat) + @pytest.mark.skipif(sys.platform != "win32", reason="Windows job objects") def test_windows_cli_never_runs_without_a_job(self, tmp_path, monkeypatch): def fail(proc): From 44f0a1a2ec023cbebdcbf145830ce685bc19e73d Mon Sep 17 00:00:00 2001 From: "M.Bozorgmehr" Date: Wed, 14 Oct 2026 23:54:03 +0330 Subject: [PATCH 6/9] fix(workflows): report an interrupt during capture setup as interrupted An interrupt while the process tree was being set up escaped run_captured as KeyboardInterrupt, so the workflow paused instead of failing the step with exit_code 130 and interrupted: true as the capture contract says. The constructor has already killed what it started, so run_captured now returns an interrupted result with nothing captured. Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 --- src/specify_cli/integrations/_capture.py | 11 +++++++---- tests/integrations/test_capture.py | 18 ++++++++++++++++-- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/src/specify_cli/integrations/_capture.py b/src/specify_cli/integrations/_capture.py index f985f071a0..ce44ad3905 100644 --- a/src/specify_cli/integrations/_capture.py +++ b/src/specify_cli/integrations/_capture.py @@ -339,7 +339,12 @@ def run_captured( f"max_bytes must be a positive integer or -1, got {max_bytes!r}." ) - tree = _ProcessTree(args, cwd) + try: + tree = _ProcessTree(args, cwd) + except KeyboardInterrupt: + # The constructor has already killed the process it started, so this + # is reported like any other interruption, with nothing captured. + return CapturedProcess(exit_code=130, stdout="", stderr="", interrupted=True) proc = tree.proc sinks = (_Sink(max_bytes), _Sink(max_bytes)) readers: list[threading.Thread] = [] @@ -350,9 +355,7 @@ def run_captured( # Starting the readers is inside the try, so an interruption here # still kills the process group or job below. for pipe, sink in zip((proc.stdout, proc.stderr), sinks): - reader = threading.Thread( - target=_drain, args=(pipe, sink), daemon=True - ) + reader = threading.Thread(target=_drain, args=(pipe, sink), daemon=True) reader.start() readers.append(reader) try: diff --git a/tests/integrations/test_capture.py b/tests/integrations/test_capture.py index e5c22e725d..c0c22018ad 100644 --- a/tests/integrations/test_capture.py +++ b/tests/integrations/test_capture.py @@ -227,6 +227,19 @@ def start(self): assert result.exit_code == 130 _assert_stopped(heartbeat) + def test_interrupt_during_process_setup_is_reported_not_raised( + self, monkeypatch + ): + class InterruptedSetup: + def __init__(self, args, cwd): + raise KeyboardInterrupt + + monkeypatch.setattr(_capture, "_ProcessTree", InterruptedSetup) + result = run_captured(_python("pass")) + assert result == CapturedProcess( + exit_code=130, stdout="", stderr="", interrupted=True + ) + def test_completed_process_is_not_killed(self, monkeypatch): def fail(self): raise AssertionError("a completed process must not be killed") @@ -277,8 +290,9 @@ def resume_then_interrupt(pid): heartbeat = tmp_path / "heartbeat" monkeypatch.setattr(_capture, "_resume_threads", resume_then_interrupt) - with pytest.raises(KeyboardInterrupt): - run_captured(_spawn_descendant(heartbeat, "time.sleep(60)")) + result = run_captured(_spawn_descendant(heartbeat, "time.sleep(60)")) + assert result.interrupted is True + assert result.exit_code == 130 _assert_stopped(heartbeat) @pytest.mark.skipif(sys.platform != "win32", reason="Windows job objects") From fa7ea6584d4f43c5ef5d13ff80bc9e8dff87aea7 Mon Sep 17 00:00:00 2001 From: "M.Bozorgmehr" Date: Thu, 15 Oct 2026 18:21:23 +0330 Subject: [PATCH 7/9] fix(workflows): stop captures on Ctrl+C after spawn and in fan-out workers Two interruption paths could leave a captured CLI running. A Ctrl+C right after Popen returned, before the code that kills the process on interruption was in place, was reported as interrupted while the CLI kept running in its own session. SIGINT is now recorded instead of raised while the process is set up, and acted on once cleanup is in place. Ctrl+C raises KeyboardInterrupt in the main thread only, and a captured CLI does not see the terminal's Ctrl+C, so captures in concurrent fan-out workers ran on until they finished or timed out while the pool waited for them. Running captures are now registered, and on Ctrl+C the fan-out kills them before waiting for its workers; they report interrupted. A capture that starts during that wait is killed as soon as it is registered. Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 --- src/specify_cli/integrations/_capture.py | 155 +++++++++++++++++++---- src/specify_cli/workflows/_execution.py | 10 +- tests/integrations/test_capture.py | 69 ++++++++++ tests/workflows/test_step_capture.py | 51 ++++++++ 4 files changed, 260 insertions(+), 25 deletions(-) diff --git a/src/specify_cli/integrations/_capture.py b/src/specify_cli/integrations/_capture.py index ce44ad3905..d7dd1ef64f 100644 --- a/src/specify_cli/integrations/_capture.py +++ b/src/specify_cli/integrations/_capture.py @@ -14,7 +14,8 @@ import subprocess import threading import time -from collections.abc import Sequence +from collections.abc import Iterator, Sequence +from contextlib import contextmanager from dataclasses import dataclass from typing import Any @@ -164,11 +165,14 @@ def __init__(self, args: Sequence[str], cwd: str | None) -> None: pipe.close() raise - def kill(self) -> None: - """Kill the process group or job, then reap the process.""" + self._lock = threading.Lock() + # Set when another thread kills the tree through interrupt(). + self.interrupted = False + + def _send_kill(self) -> None: if self._job is not None: _windows().TerminateJobObject(self._job, 1) - else: + elif os.name != "nt": try: os.killpg(self.proc.pid, signal.SIGKILL) except OSError: @@ -177,12 +181,100 @@ def kill(self) -> None: self.proc.kill() except OSError: pass + + def interrupt(self) -> None: + """Kill the process group or job from another thread. + + The thread running the capture sees the process exit and reports + the capture as interrupted. + """ + with self._lock: + self.interrupted = True + self._send_kill() + + def kill(self) -> None: + """Kill the process group or job, then reap the process.""" + with self._lock: + self._send_kill() self.proc.wait() def close(self) -> None: - if self._job is not None: - _windows().CloseHandle(self._job) - self._job = None + with self._lock: + if self._job is not None: + _windows().CloseHandle(self._job) + self._job = None + + +# Captures running in any thread, so an interrupt of the main thread can +# reach the ones running in worker threads; see cancel_captures(). +_active_lock = threading.Lock() +_active: set[_ProcessTree] = set() +_cancelling = 0 + + +def _register(tree: _ProcessTree) -> None: + with _active_lock: + _active.add(tree) + cancelling = _cancelling > 0 + if cancelling: + tree.interrupt() + + +def _unregister(tree: _ProcessTree) -> None: + with _active_lock: + _active.discard(tree) + + +@contextmanager +def cancel_captures() -> Iterator[None]: + """Kill every running capture, and any that starts, until the block ends. + + Ctrl+C raises ``KeyboardInterrupt`` in the main thread only, and a + captured CLI runs in its own session or job, so it does not receive the + terminal's Ctrl+C either. When the main thread is interrupted while + worker threads run captures, wait for those workers inside this block: + their captures are killed and report ``interrupted``, and a capture that + starts meanwhile is killed as soon as it is registered. + """ + global _cancelling + with _active_lock: + _cancelling += 1 + trees = list(_active) + try: + for tree in trees: + tree.interrupt() + yield + finally: + with _active_lock: + _cancelling -= 1 + + +@contextmanager +def _deferred_interrupt() -> Iterator[list[int]]: + """Record Ctrl+C instead of raising it until cleanup is in place. + + A ``KeyboardInterrupt`` raised after ``Popen`` returned but before the + code that kills the process on interruption would leave the CLI running + in its own session (POSIX) or suspended outside its job (Windows). + Within this block a SIGINT is appended to the yielded list; the caller + treats a non-empty list as an interruption. Only the main thread + receives SIGINT, and only Python's default handler is replaced, so a + custom or ignoring handler keeps its behaviour. + """ + pending: list[int] = [] + if ( + threading.current_thread() is not threading.main_thread() + or signal.getsignal(signal.SIGINT) is not signal.default_int_handler + ): + yield pending + return + previous = signal.signal( + signal.SIGINT, lambda signum, frame: pending.append(signum) + ) + try: + yield pending + finally: + signal.signal(signal.SIGINT, previous) _kernel32: Any = None @@ -339,27 +431,27 @@ def run_captured( f"max_bytes must be a positive integer or -1, got {max_bytes!r}." ) - try: - tree = _ProcessTree(args, cwd) - except KeyboardInterrupt: - # The constructor has already killed the process it started, so this - # is reported like any other interruption, with nothing captured. - return CapturedProcess(exit_code=130, stdout="", stderr="", interrupted=True) - proc = tree.proc + tree: _ProcessTree | None = None sinks = (_Sink(max_bytes), _Sink(max_bytes)) readers: list[threading.Thread] = [] deadline = None if timeout is None else time.monotonic() + timeout timed_out = interrupted = False try: try: - # Starting the readers is inside the try, so an interruption here - # still kills the process group or job below. - for pipe, sink in zip((proc.stdout, proc.stderr), sinks): - reader = threading.Thread(target=_drain, args=(pipe, sink), daemon=True) - reader.start() - readers.append(reader) + with _deferred_interrupt() as pending_interrupt: + tree = _ProcessTree(args, cwd) + _register(tree) + for pipe, sink in zip((tree.proc.stdout, tree.proc.stderr), sinks): + reader = threading.Thread( + target=_drain, args=(pipe, sink), daemon=True + ) + reader.start() + readers.append(reader) + # From here a Ctrl+C raises KeyboardInterrupt inside this try. + if pending_interrupt: + raise KeyboardInterrupt try: - proc.wait(timeout=timeout) + tree.proc.wait(timeout=timeout) except subprocess.TimeoutExpired: timed_out = True else: @@ -369,6 +461,19 @@ def run_captured( timed_out = not _join(readers, deadline) except KeyboardInterrupt: interrupted = True + if tree is None: + # Setup was interrupted, and the constructor has already + # killed the process it started. + return CapturedProcess( + exit_code=130, stdout="", stderr="", interrupted=True + ) + except BaseException: + if tree is not None: + tree.kill() + raise + if tree.interrupted: + # Killed through cancel_captures() while this thread waited. + interrupted, timed_out = True, False if timed_out or interrupted: # Killing the group or job closes the write ends of the pipes its @@ -378,10 +483,12 @@ def run_captured( tree.kill() _join(readers, time.monotonic() + _DRAIN_GRACE_SECONDS) # A pipe whose reader never started is closed here instead. - for pipe in (proc.stdout, proc.stderr)[len(readers) :]: + for pipe in (tree.proc.stdout, tree.proc.stderr)[len(readers) :]: pipe.close() finally: - tree.close() + if tree is not None: + _unregister(tree) + tree.close() (stdout, stdout_truncated), (stderr, stderr_truncated) = ( sink.snapshot() for sink in sinks @@ -392,7 +499,7 @@ def run_captured( elif timed_out: exit_code = -1 else: - exit_code = proc.returncode + exit_code = tree.proc.returncode return CapturedProcess( exit_code=exit_code, stdout=_decode(stdout, complete=not (stopped or stdout_truncated)), diff --git a/src/specify_cli/workflows/_execution.py b/src/specify_cli/workflows/_execution.py index 6d2a2e89df..cec955d3ed 100644 --- a/src/specify_cli/workflows/_execution.py +++ b/src/specify_cli/workflows/_execution.py @@ -14,6 +14,8 @@ import yaml +from specify_cli.integrations._capture import cancel_captures + from .base import StepContext, StepResult, StepStatus from .composition import ( CallError, @@ -949,9 +951,15 @@ def run_item_guarded(index): for index in range(len(items)): try: outcome, error, value = futures.pop(index).result() - except BaseException: + except BaseException as exc: for future in futures.values(): future.cancel() + if isinstance(exc, KeyboardInterrupt): + # Ctrl+C reaches only this thread, and captured CLIs + # run in their own session or job: kill the workers' + # captures before the pool waits for the workers. + with cancel_captures(): + pool.shutdown(wait=True) raise results.append(value) if outcome in HALTING: diff --git a/tests/integrations/test_capture.py b/tests/integrations/test_capture.py index c0c22018ad..8d95e2da18 100644 --- a/tests/integrations/test_capture.py +++ b/tests/integrations/test_capture.py @@ -2,6 +2,7 @@ from __future__ import annotations +import signal import subprocess import sys import threading @@ -14,6 +15,7 @@ from specify_cli.integrations._capture import ( DEFAULT_CAPTURE_MAX_BYTES, CapturedProcess, + cancel_captures, run_captured, ) @@ -240,6 +242,26 @@ def __init__(self, args, cwd): exit_code=130, stdout="", stderr="", interrupted=True ) + def test_interrupt_right_after_the_process_starts_kills_it(self, monkeypatch): + """Ctrl+C after Popen returns, before cleanup is set up, still kills.""" + spawned: list[subprocess.Popen] = [] + + class InterruptedAfterSpawn(subprocess.Popen): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + spawned.append(self) + signal.raise_signal(signal.SIGINT) + + monkeypatch.setattr(_capture.subprocess, "Popen", InterruptedAfterSpawn) + started = time.monotonic() + result = run_captured(_python("import time; time.sleep(60)")) + assert time.monotonic() - started < 30 + assert result.interrupted is True + assert result.exit_code == 130 + (proc,) = spawned + assert proc.returncode is not None, "the CLI was left running" + assert signal.getsignal(signal.SIGINT) is signal.default_int_handler + def test_completed_process_is_not_killed(self, monkeypatch): def fail(self): raise AssertionError("a completed process must not be killed") @@ -463,3 +485,50 @@ def test_capture_limit_with_streaming_is_rejected(self, recorded): "speckit.plan", capture_max_bytes=1024 ) assert recorded == [] + + +class TestCancelCaptures: + """Interrupting captures that run in worker threads.""" + + @staticmethod + def _in_thread(args): + results: list[CapturedProcess] = [] + worker = threading.Thread( + target=lambda: results.append(run_captured(args, timeout=60)) + ) + worker.start() + return worker, results + + def test_running_capture_is_killed_and_reports_interrupted(self, tmp_path): + heartbeat = tmp_path / "heartbeat" + worker, results = self._in_thread( + _spawn_descendant(heartbeat, "time.sleep(60)") + ) + while not heartbeat.exists(): + time.sleep(0.05) + started = time.monotonic() + with cancel_captures(): + worker.join(30) + assert not worker.is_alive() + assert time.monotonic() - started < 30 + (result,) = results + assert result.interrupted is True + assert result.timed_out is False + assert result.exit_code == 130 + _assert_stopped(heartbeat) + + def test_capture_started_while_cancelling_is_killed(self, tmp_path): + with cancel_captures(): + worker, results = self._in_thread( + _python("import time; time.sleep(60)") + ) + worker.join(30) + assert not worker.is_alive() + (result,) = results + assert result.interrupted is True + + def test_captures_after_the_block_run_normally(self): + with cancel_captures(): + pass + result = run_captured(_python("print('ok')"), timeout=30) + assert result == CapturedProcess(exit_code=0, stdout="ok\n", stderr="") diff --git a/tests/workflows/test_step_capture.py b/tests/workflows/test_step_capture.py index 2fca03a769..77e5d6b57f 100644 --- a/tests/workflows/test_step_capture.py +++ b/tests/workflows/test_step_capture.py @@ -3,7 +3,10 @@ from __future__ import annotations import json +import signal import sys +import threading +import time import pytest import yaml @@ -456,3 +459,51 @@ def test_continue_on_error_lets_downstream_inspect_a_failed_capture( assert state.step_results["ask"]["output"]["exit_code"] == -1 report = state.step_results["report"]["output"]["stdout"] assert report.strip() == "partial=partial" + + +@pytest.mark.skipif( + not hasattr(signal, "pthread_kill"), reason="needs a SIGINT to the main thread" +) +def test_ctrl_c_during_concurrent_fan_out_kills_the_captures(project_dir, fake_cli): + """Ctrl+C reaches only the main thread; worker captures must still stop.""" + fake_cli( + "import os\n" + f"beat = os.path.join({str(project_dir)!r}, f'beat-{{os.getpid()}}')\n" + "while True:\n" + " open(beat, 'a').write('.')\n" + " time.sleep(0.1)\n" + ) + engine, definition = _install_workflow( + project_dir, + [ + { + "id": "fan", + "type": "fan-out", + "items": [1, 2], + "max_concurrency": 2, + "step": {**_config("prompt", capture=True, timeout=20)}, + } + ], + ) + main = threading.main_thread().ident + + def interrupt_once_both_run(): + while len(list(project_dir.glob("beat-*"))) < 2: + time.sleep(0.05) + signal.pthread_kill(main, signal.SIGINT) + + threading.Thread(target=interrupt_once_both_run, daemon=True).start() + started = time.monotonic() + try: + engine.execute(definition, {}) + except KeyboardInterrupt: + pass + # Without cancellation the pool waits for both captures' 20s timeout. + assert time.monotonic() - started < 15 + + beats = list(project_dir.glob("beat-*")) + assert len(beats) == 2 + time.sleep(0.5) + sizes = [beat.stat().st_size for beat in beats] + time.sleep(1) + assert [beat.stat().st_size for beat in beats] == sizes, "a capture kept running" From 46a455525e265ce0a954a1a4b81841c22b09a4a2 Mon Sep 17 00:00:00 2001 From: "M.Bozorgmehr" Date: Thu, 15 Oct 2026 19:06:19 +0330 Subject: [PATCH 8/9] fix(workflows): cancel fan-out captures during shutdown and reader waits Two paths still let a Ctrl+C miss running captures. When a fan-out item failed or paused, the early return waited for the other workers in the executor's context-manager exit, where a Ctrl+C was not handled, so their captures kept running. The pool is now shut down by _shutdown_workers(), which cancels pending items and, after an interruption before or during the wait, kills the workers' captures and waits again. A capture whose CLI had exited while a detached helper held its pipes was waiting on its readers until the step deadline, ignoring cancel_captures(). The reader join now polls for cancellation and stops early, so cleanup uses the two-second drain grace instead of the full timeout. Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 --- src/specify_cli/integrations/_capture.py | 41 ++++++++++++--- src/specify_cli/workflows/_execution.py | 48 ++++++++++++----- tests/integrations/test_capture.py | 41 +++++++++++++++ tests/workflows/test_step_capture.py | 65 ++++++++++++++++++++++++ 4 files changed, 174 insertions(+), 21 deletions(-) diff --git a/src/specify_cli/integrations/_capture.py b/src/specify_cli/integrations/_capture.py index d7dd1ef64f..93d8a10fb7 100644 --- a/src/specify_cli/integrations/_capture.py +++ b/src/specify_cli/integrations/_capture.py @@ -110,14 +110,37 @@ def _drain(pipe: Any, sink: _Sink) -> None: pass -def _join(readers: Sequence[threading.Thread], deadline: float | None) -> bool: - """Join *readers* until *deadline*; return whether all reached end-of-file.""" +# How often a reader join checks whether the capture was cancelled. +_JOIN_POLL_SECONDS = 0.1 + + +def _join( + readers: Sequence[threading.Thread], + deadline: float | None, + cancelled: Any = None, +) -> bool: + """Join *readers* until *deadline*; return whether all reached end-of-file. + + When *cancelled* is given, it is polled while waiting, and the join stops + early, returning False, once it returns true. + """ for reader in readers: - if deadline is None: - reader.join() - else: - reader.join(max(0.0, deadline - time.monotonic())) - return not any(reader.is_alive() for reader in readers) + while reader.is_alive(): + if cancelled is not None and cancelled(): + return False + if deadline is None: + wait = None if cancelled is None else _JOIN_POLL_SECONDS + else: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + wait = ( + remaining + if cancelled is None + else min(remaining, _JOIN_POLL_SECONDS) + ) + reader.join(wait) + return True # Windows process creation flag; not exported by ``subprocess``. @@ -458,7 +481,9 @@ def run_captured( # The process can exit while a descendant still holds the # pipes open; the same deadline bounds the wait for # end-of-file. - timed_out = not _join(readers, deadline) + # cancel_captures() kills the CLI, but a detached helper can + # still hold the pipes, so this wait also stops on it. + timed_out = not _join(readers, deadline, lambda: tree.interrupted) except KeyboardInterrupt: interrupted = True if tree is None: diff --git a/src/specify_cli/workflows/_execution.py b/src/specify_cli/workflows/_execution.py index cec955d3ed..785c806300 100644 --- a/src/specify_cli/workflows/_execution.py +++ b/src/specify_cli/workflows/_execution.py @@ -398,6 +398,30 @@ def result_view(node): return result +def _shutdown_workers( + pool: ThreadPoolExecutor, pending: Any, *, interrupted: bool +) -> None: + """Cancel *pending* fan-out futures and wait for the running workers. + + Ctrl+C raises ``KeyboardInterrupt`` in this thread only, and captured CLIs + run in their own session or job, so a worker's capture would run on while + the pool waits for it. After an interruption, before the wait or during + it, the workers' captures are killed and the wait is repeated. + """ + for future in list(pending): + future.cancel() + if not interrupted: + try: + pool.shutdown(wait=True) + return + except KeyboardInterrupt: + with cancel_captures(): + pool.shutdown(wait=True) + raise + with cancel_captures(): + pool.shutdown(wait=True) + + @dataclass(frozen=True) class Occurrence: """One invocation, never stored on a shared step implementation.""" @@ -946,27 +970,25 @@ def run_item_guarded(index): halted.set() raise - with ThreadPoolExecutor(max_workers=workers) as pool: - futures = {i: pool.submit(run_item_guarded, i) for i in range(workers)} + pool = ThreadPoolExecutor(max_workers=workers) + futures = {} + interrupted = False + try: + futures.update( + (i, pool.submit(run_item_guarded, i)) for i in range(workers) + ) for index in range(len(items)): try: outcome, error, value = futures.pop(index).result() - except BaseException as exc: - for future in futures.values(): - future.cancel() - if isinstance(exc, KeyboardInterrupt): - # Ctrl+C reaches only this thread, and captured CLIs - # run in their own session or job: kill the workers' - # captures before the pool waits for the workers. - with cancel_captures(): - pool.shutdown(wait=True) + except KeyboardInterrupt: + interrupted = True raise results.append(value) if outcome in HALTING: - for future in futures.values(): - future.cancel() return outcome, error, results following = index + workers if following < len(items) and not halted.is_set(): futures[following] = pool.submit(run_item_guarded, following) + finally: + _shutdown_workers(pool, futures.values(), interrupted=interrupted) return "completed", None, results diff --git a/tests/integrations/test_capture.py b/tests/integrations/test_capture.py index 8d95e2da18..5c0bc2af8d 100644 --- a/tests/integrations/test_capture.py +++ b/tests/integrations/test_capture.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os import signal import subprocess import sys @@ -517,6 +518,46 @@ def test_running_capture_is_killed_and_reports_interrupted(self, tmp_path): assert result.exit_code == 130 _assert_stopped(heartbeat) + @pytest.mark.skipif( + sys.platform == "win32", reason="leaving the job needs breakaway rights" + ) + def test_cancel_stops_waiting_for_a_detached_helper(self, tmp_path): + """The CLI exited, but a helper outside its group holds the pipes.""" + pid_file = tmp_path / "helper.pid" + cli = _python( + "import os, subprocess, sys\n" + "helper = subprocess.Popen(\n" + " [sys.executable, '-c', 'import time; time.sleep(60)'],\n" + " start_new_session=True)\n" + "open(sys.argv[1] + '.tmp', 'w').write(str(helper.pid))\n" + "os.replace(sys.argv[1] + '.tmp', sys.argv[1])\n", + str(pid_file), + ) + results: list[CapturedProcess] = [] + worker = threading.Thread( + target=lambda: results.append(run_captured(cli, timeout=20)) + ) + worker.start() + try: + while not pid_file.exists(): + time.sleep(0.05) + # Let the CLI exit, so the capture is waiting on its readers. + time.sleep(0.5) + started = time.monotonic() + with cancel_captures(): + worker.join(30) + assert time.monotonic() - started < 10, "the join ignored cancellation" + (result,) = results + assert result.interrupted is True + assert result.timed_out is False + finally: + if pid_file.exists(): + try: + os.kill(int(pid_file.read_text()), signal.SIGKILL) + except ProcessLookupError: + pass + worker.join(30) + def test_capture_started_while_cancelling_is_killed(self, tmp_path): with cancel_captures(): worker, results = self._in_thread( diff --git a/tests/workflows/test_step_capture.py b/tests/workflows/test_step_capture.py index 77e5d6b57f..b2b7420b38 100644 --- a/tests/workflows/test_step_capture.py +++ b/tests/workflows/test_step_capture.py @@ -507,3 +507,68 @@ def interrupt_once_both_run(): sizes = [beat.stat().st_size for beat in beats] time.sleep(1) assert [beat.stat().st_size for beat in beats] == sizes, "a capture kept running" + + +@pytest.mark.skipif( + not hasattr(signal, "pthread_kill"), reason="needs a SIGINT to the main thread" +) +def test_ctrl_c_while_fan_out_shuts_down_after_a_failure_kills_the_captures( + project_dir, monkeypatch +): + """Item 1 fails at once; Ctrl+C arrives while the pool waits for item 2.""" + script = project_dir / "cli.py" + script.write_text( + "import os, sys, time\n" + "if 'item=1' in sys.argv[1]:\n" + " sys.exit(1)\n" + f"beat = os.path.join({str(project_dir)!r}, f'beat-{{os.getpid()}}')\n" + "while True:\n" + " open(beat, 'a').write('.')\n" + " time.sleep(0.1)\n", + encoding="utf-8", + ) + monkeypatch.setattr( + type(get_integration("claude")), + "build_exec_args", + lambda self, prompt, **kwargs: [sys.executable, "-u", str(script), prompt], + ) + engine, definition = _install_workflow( + project_dir, + [ + { + "id": "fan", + "type": "fan-out", + "items": [1, 2], + "max_concurrency": 2, + "step": { + "id": "ask", + "type": "prompt", + "prompt": "item={{ item }}", + "capture": True, + "timeout": 20, + }, + } + ], + ) + main = threading.main_thread().ident + + def interrupt_during_shutdown(): + while not list(project_dir.glob("beat-*")): + time.sleep(0.05) + # Item 1 has failed by now and the fan-out is waiting for item 2. + time.sleep(1) + signal.pthread_kill(main, signal.SIGINT) + + threading.Thread(target=interrupt_during_shutdown, daemon=True).start() + started = time.monotonic() + try: + engine.execute(definition, {}) + except KeyboardInterrupt: + pass + assert time.monotonic() - started < 15 + + (beat,) = project_dir.glob("beat-*") + time.sleep(0.5) + size = beat.stat().st_size + time.sleep(1) + assert beat.stat().st_size == size, "the capture kept running" From 44871f0344651eece000e9ec856a671511f83610 Mon Sep 17 00:00:00 2001 From: "M.Bozorgmehr" Date: Thu, 15 Oct 2026 19:36:20 +0330 Subject: [PATCH 9/9] fix(workflows): treat Ctrl+C anywhere in fan-out collection as interrupt Only a KeyboardInterrupt raised while waiting on a future's result marked the fan-out as interrupted. One raised during the initial or a replacement submit(), or between collections, reached the shutdown with interrupted=False, so the pool waited for running captures without cancelling them. The handler now covers the whole submission and collection block. Assisted-by: Claude Code (model: Claude Opus 5.5, autonomous) Co-Authored-By: Claude Opus 5.5 --- src/specify_cli/workflows/_execution.py | 10 ++-- tests/workflows/test_step_capture.py | 72 +++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 5 deletions(-) diff --git a/src/specify_cli/workflows/_execution.py b/src/specify_cli/workflows/_execution.py index 785c806300..7a880632b1 100644 --- a/src/specify_cli/workflows/_execution.py +++ b/src/specify_cli/workflows/_execution.py @@ -978,17 +978,17 @@ def run_item_guarded(index): (i, pool.submit(run_item_guarded, i)) for i in range(workers) ) for index in range(len(items)): - try: - outcome, error, value = futures.pop(index).result() - except KeyboardInterrupt: - interrupted = True - raise + outcome, error, value = futures.pop(index).result() results.append(value) if outcome in HALTING: return outcome, error, results following = index + workers if following < len(items) and not halted.is_set(): futures[following] = pool.submit(run_item_guarded, following) + except KeyboardInterrupt: + # Anywhere in submission or collection, not only in result(). + interrupted = True + raise finally: _shutdown_workers(pool, futures.values(), interrupted=interrupted) return "completed", None, results diff --git a/tests/workflows/test_step_capture.py b/tests/workflows/test_step_capture.py index b2b7420b38..101039da5e 100644 --- a/tests/workflows/test_step_capture.py +++ b/tests/workflows/test_step_capture.py @@ -572,3 +572,75 @@ def interrupt_during_shutdown(): size = beat.stat().st_size time.sleep(1) assert beat.stat().st_size == size, "the capture kept running" + + +def test_ctrl_c_while_fan_out_submits_an_item_kills_the_running_captures( + project_dir, monkeypatch +): + """Ctrl+C lands in submit(), not result(), while item 2's capture runs.""" + from concurrent.futures import ThreadPoolExecutor + + from specify_cli.workflows import _execution + + script = project_dir / "cli.py" + script.write_text( + "import os, sys, time\n" + "if 'item=1' in sys.argv[1]:\n" + " sys.exit(0)\n" + f"beat = os.path.join({str(project_dir)!r}, f'beat-{{os.getpid()}}')\n" + "while True:\n" + " open(beat, 'a').write('.')\n" + " time.sleep(0.1)\n", + encoding="utf-8", + ) + monkeypatch.setattr( + type(get_integration("claude")), + "build_exec_args", + lambda self, prompt, **kwargs: [sys.executable, "-u", str(script), prompt], + ) + + class InterruptedThirdSubmit(ThreadPoolExecutor): + submitted = 0 + + def submit(self, *args, **kwargs): + type(self).submitted += 1 + if type(self).submitted == 3: + # Item 1 has completed and item 2's capture is running. + while not list(project_dir.glob("beat-*")): + time.sleep(0.05) + raise KeyboardInterrupt + return super().submit(*args, **kwargs) + + monkeypatch.setattr(_execution, "ThreadPoolExecutor", InterruptedThirdSubmit) + engine, definition = _install_workflow( + project_dir, + [ + { + "id": "fan", + "type": "fan-out", + "items": [1, 2, 3], + "max_concurrency": 2, + "step": { + "id": "ask", + "type": "prompt", + "prompt": "item={{ item }}", + "capture": True, + "timeout": 20, + }, + } + ], + ) + started = time.monotonic() + try: + engine.execute(definition, {}) + except KeyboardInterrupt: + pass + # Without cancellation the pool waits for item 2's 20s timeout. + assert time.monotonic() - started < 15 + assert InterruptedThirdSubmit.submitted == 3 + + (beat,) = project_dir.glob("beat-*") + time.sleep(0.5) + size = beat.stat().st_size + time.sleep(1) + assert beat.stat().st_size == size, "the capture kept running"