From 5c13d10989535fbcdfcd25a5e767dce7333253ee Mon Sep 17 00:00:00 2001 From: worldmozara Date: Mon, 2 Nov 2026 14:21:26 +0800 Subject: [PATCH] tools: model batch steps as nested executions --- src/plyngent/cli/display.py | 59 ++-- src/plyngent/config/models.py | 2 +- src/plyngent/tools/danger.py | 39 ++- src/plyngent/tools/process/command_exec.py | 4 +- src/plyngent/tools/process/run_argv_batch.py | 271 ++++++++++++++----- tests/test_cli/test_display.py | 21 +- tests/test_tools/test_danger.py | 28 ++ tests/test_tools/test_process.py | 160 ++++++++++- 8 files changed, 474 insertions(+), 110 deletions(-) diff --git a/src/plyngent/cli/display.py b/src/plyngent/cli/display.py index abdfd48..24ae70e 100644 --- a/src/plyngent/cli/display.py +++ b/src/plyngent/cli/display.py @@ -36,7 +36,8 @@ _TOOL_ARGS_PREVIEW = 80 _ASK_SUBJECT_PREVIEW = 60 _PTY_PAYLOAD_PREVIEW = 40 # ``run_argv_batch`` lists the commands it runs above its ``* Run batch `` row — one -# line per command, a ``pipe_out`` chain joined into one — elided past this width. +# line per command, a pipe chain joined into one, a ``then`` / ``else`` body +# indented under the node it follows — elided past this width. _RUN_BATCH_CMD_LIMIT = 80 _MCP_TOOL_PREFIX = "mcp__" _SKILL_MAIN_FILE = "SKILL.md" @@ -417,32 +418,54 @@ def _batch_steps(args_json: str) -> list[dict[str, object]]: value = json.loads(value) except ValueError: return [] + return _batch_nodes(value) + + +def _batch_nodes(value: object) -> list[dict[str, object]]: + """Nodes of one continuation: a single node, or a list of them.""" + if isinstance(value, dict): + return [cast("dict[str, object]", value)] if not isinstance(value, list): return [] - raw_steps = cast("list[object]", value) - return [cast("dict[str, object]", step) for step in raw_steps if isinstance(step, dict)] + return [cast("dict[str, object]", item) for item in cast("list[object]", value) if isinstance(item, dict)] -def _batch_command_lines(args_json: str) -> list[str]: - """One command per line, a ``pipe_out`` chain joined into one line with ``|``.""" - lines: list[str] = [] - piped_from_previous = False - for step in _batch_steps(args_json): - argv = _str_list(step.get("argv")) - command = shlex.join(argv) if argv else "?" - if piped_from_previous and lines: - lines[-1] = f"{lines[-1]} | {command}" - else: - lines.append(command) - # Same flag spellings the tool itself accepts (bool, or "true"/"1"/…). - piped_from_previous = str(step.get("pipe_out") or "").strip().lower() in {"1", "true", "yes", "on"} +def _batch_chain(node: dict[str, object]) -> str: + """``a | b | c`` for a node and the commands its pipe feeds.""" + argv = _str_list(node.get("argv")) + parts = [shlex.join(argv) if argv else "?"] + parts.extend(_batch_chain(child) for child in _batch_nodes(node.get("pipe"))) + return " | ".join(parts) + + +def _batch_node_lines( + nodes: list[dict[str, object]], + depth: int, + prefix: str = "", +) -> list[tuple[str, bool]]: + """One line per command, and whether it is a command (``$``) or a branch.""" + lines: list[tuple[str, bool]] = [] + for node in nodes: + lines.append((f"{' ' * depth}{prefix}{_batch_chain(node)}", not prefix)) + for branch in ("then", "else"): + children = _batch_nodes(node.get(branch)) + if children: + lines.extend(_batch_node_lines(children, depth + 1, f"{branch} ")) return lines +def _batch_command_lines(args_json: str) -> list[tuple[str, bool]]: + """One line per command, a pipe chain joined into one line with ``|``.""" + return _batch_node_lines(_batch_steps(args_json), 0) + + def _run_argv_batch_prefix(args_json: str) -> str: """``* Run batch `` with the commands it runs listed above it.""" - commands = _batch_command_lines(args_json) - body = "\n".join(_pretty_prefix(f"$ {_preview(command, _RUN_BATCH_CMD_LIMIT)}") for command in commands) + lines = [ + _pretty_prefix(f"$ {_preview(text, _RUN_BATCH_CMD_LIMIT)}" if is_command else f"{text}") + for text, is_command in _batch_command_lines(args_json) + ] + body = "\n".join(lines) header = _pretty_prefix("* Run batch ") return f"{body}\n{header}" if body else header diff --git a/src/plyngent/config/models.py b/src/plyngent/config/models.py index 09a4d54..e23add1 100644 --- a/src/plyngent/config/models.py +++ b/src/plyngent/config/models.py @@ -76,7 +76,7 @@ confirm blocks the turn, so it costs time. Invoke the target command directly (a of wrapping it in `bash -c` / `python -c` / `node`, and reach for a dedicated tool \ (`read_file` / `edit_*` / `vcs_*` / `fetch` / `fetch_local` / …) when one already does the job. - Several `run_argv` calls in one step may run in parallel; use `run_argv_batch` \ -for ordered pipelines (`pipe_out` / `mix_stderr` as needed). +for ordered commands, with `then` / `else` / `pipe` continuations as needed. - Prefer `vcs_*` for status/diff/log/branch when enough. ### Network diff --git a/src/plyngent/tools/danger.py b/src/plyngent/tools/danger.py index 3f5f17e..2ec5b84 100644 --- a/src/plyngent/tools/danger.py +++ b/src/plyngent/tools/danger.py @@ -10,7 +10,7 @@ from plyngent.tools.trash import under_system_temp from plyngent.tools.workspace import AccessMode, WorkspaceError, get_trash_enabled, resolve_path if TYPE_CHECKING: - from collections.abc import Mapping, Sequence + from collections.abc import Iterator, Mapping, Sequence def _as_argv(args: Mapping[str, object], *, key: str = "command") -> list[str] | None: @@ -188,6 +188,32 @@ def _batch_step_argv(item: object) -> list[str] | None: return parts or None +def _batch_nodes(value: object) -> list[dict[str, object]]: + """Nodes of one continuation: a single node, or a list of them.""" + if isinstance(value, dict): + return [cast("dict[str, object]", value)] + if not isinstance(value, list): + return [] + return [cast("dict[str, object]", item) for item in cast("list[object]", value) if isinstance(item, dict)] + + +def _walk_batch_argv(value: object, where: str) -> Iterator[tuple[str, list[str]]]: + """Every command of a batch tree, in the order it can run, with its path. + + ``then`` / ``else`` / ``pipe`` hold the continuations, so a command hiding in + one is reviewed exactly like a top-level step. + """ + for index, node in enumerate(_batch_nodes(value)): + label = f"{where}[{index}]" + argv = _batch_step_argv(node) + if argv is not None: + yield label, argv + for branch in ("then", "else", "pipe"): + child = node.get(branch) + if child is not None: + yield from _walk_batch_argv(child, f"{label}.{branch}") + + def _run_argv_batch_reason(args: Mapping[str, object]) -> DangerReason | None: """One confirm for the whole batch if any step runs code or detaches. @@ -203,12 +229,11 @@ def _run_argv_batch_reason(args: Mapping[str, object]) -> DangerReason | None: raw = loaded if not isinstance(raw, list): return None - risky = [ - reason - for index, item in enumerate(cast("list[object]", raw)) - if (argv := _batch_step_argv(item)) is not None - and (reason := _command_reason(argv, via=f"run_argv_batch[{index}]")) is not None - ] + risky: list[DangerReason] = [] + for label, argv in _walk_batch_argv(cast("list[object]", raw), "run_argv_batch"): + reason = _command_reason(argv, via=label) + if reason is not None: + risky.append(reason) if not risky: return None header = f"run_argv_batch: {len(risky)} risky step(s) (review before allow)" diff --git a/src/plyngent/tools/process/command_exec.py b/src/plyngent/tools/process/command_exec.py index 50d96fc..f150897 100644 --- a/src/plyngent/tools/process/command_exec.py +++ b/src/plyngent/tools/process/command_exec.py @@ -19,6 +19,8 @@ from plyngent.tools.workspace import ( DEFAULT_TIMEOUT_SECONDS = 30 DEFAULT_MAX_BATCH_STEPS = 20 +# A batch node's continuations nest, so the tree needs its own ceiling. +MAX_BATCH_DEPTH = 4 def format_command_result( @@ -65,7 +67,7 @@ class CommandStepResult: stdout: str stderr: str mix_stderr: bool - captured: str # piped to the next step when provider sets pipe_out + captured: str # stdout, fed to the next command of a pipe async def execute_argv( diff --git a/src/plyngent/tools/process/run_argv_batch.py b/src/plyngent/tools/process/run_argv_batch.py index 0078b11..afbe1af 100644 --- a/src/plyngent/tools/process/run_argv_batch.py +++ b/src/plyngent/tools/process/run_argv_batch.py @@ -2,6 +2,7 @@ from __future__ import annotations import json import shlex +from dataclasses import dataclass, field from typing import Any, cast from plyngent.agent import ToolTag, tool @@ -10,6 +11,7 @@ from plyngent.tools.workspace import WorkspaceError from .command_exec import ( DEFAULT_MAX_BATCH_STEPS, DEFAULT_TIMEOUT_SECONDS, + MAX_BATCH_DEPTH, CommandStepResult, execute_argv, ) @@ -19,6 +21,14 @@ from .command_exec import ( BATCH_OUTPUT_BUDGET_CHARS = 32_000 BATCH_OUTPUT_SOFT_CAP_FACTOR = 4 +# A node is one command plus up to three continuations: ``then`` runs when it +# succeeds, ``else`` when it fails, ``pipe`` takes its stdout as stdin. Only a +# pipe hands data to its child, so a pipe stage may not branch: the stage's +# result belongs to the node that owns the pipe. +_CONTINUATIONS = ("then", "else", "pipe") +_COMMAND_KEYS = frozenset({"argv", "cwd", "env", "stdin", "mix_stderr", "stop_on_error", "timeout_seconds"}) +_ACCEPTED_KEYS = "argv, cwd, env, stdin, mix_stderr, stop_on_error, timeout_seconds, then, else, pipe" + def _as_bool(value: object, *, default: bool = False) -> bool: if value is None: @@ -49,33 +59,86 @@ def _as_str_dict(value: object) -> dict[str, str] | None: return {str(key): str(val) for key, val in cast("dict[object, object]", value).items()} -def _parse_step(raw: object, index: int) -> dict[str, Any]: +@dataclass +class _Budget: + """Parsed node count: every node costs one, however deeply it nests.""" + + nodes: int = 0 + + def take(self, where: str) -> None: + self.nodes += 1 + if self.nodes > DEFAULT_MAX_BATCH_STEPS: + msg = f"more than {DEFAULT_MAX_BATCH_STEPS} nodes in one batch (at {where})" + raise WorkspaceError(msg) + + +def _parse_nodes( + value: object, + *, + where: str, + depth: int, + budget: _Budget, + fed: bool, +) -> list[dict[str, Any]]: + """A continuation: one node or a list of them, in the order they run.""" + if value is None: + return [] + items = cast("list[object]", value) if isinstance(value, list) else [value] + parsed: list[dict[str, Any]] = [] + for index, item in enumerate(items): + at = f"{where}[{index}]" if isinstance(value, list) else where + parsed.append(_parse_node(item, where=at, depth=depth, budget=budget, fed=fed)) + return parsed + + +def _parse_node(raw: object, *, where: str, depth: int, budget: _Budget, fed: bool) -> dict[str, Any]: if not isinstance(raw, dict): - msg = f"steps[{index}] must be an object" + msg = f"{where} must be an object" + raise WorkspaceError(msg) + if depth > MAX_BATCH_DEPTH: + msg = f"{where}: nodes nest at most {MAX_BATCH_DEPTH} levels deep" raise WorkspaceError(msg) data = cast("dict[str, object]", raw) + unknown = sorted(set(data) - _COMMAND_KEYS - set(_CONTINUATIONS)) + if unknown: + hint = '; write a pipeline as {"pipe": [...]}, not "pipe_out"' if "pipe_out" in unknown else "" + msg = f"{where} has unknown key(s) {', '.join(unknown)}; accepted keys are {_ACCEPTED_KEYS}{hint}" + raise WorkspaceError(msg) + budget.take(where) argv = data.get("argv") if not isinstance(argv, list) or not argv: - msg = f"steps[{index}].argv must be a non-empty argv list" + msg = f"{where}.argv must be a non-empty argv list" raise WorkspaceError(msg) argv_parts = cast("list[object]", argv) argv_str = [part for part in argv_parts if isinstance(part, str)] if len(argv_str) != len(argv_parts): - msg = f"steps[{index}].argv must be a list of strings" + msg = f"{where}.argv must be a list of strings" + raise WorkspaceError(msg) + stdin = data.get("stdin") + if fed and stdin is not None: + msg = f"{where}: stdin comes from the pipe; drop stdin or split the command" raise WorkspaceError(msg) cwd = data.get("cwd") - stdin = data.get("stdin") stop = data.get("stop_on_error", None) - return { + node: dict[str, Any] = { "argv": argv_str, "cwd": str(cwd) if isinstance(cwd, str) and cwd else None, "env": _as_str_dict(data.get("env")), "stdin": None if stdin is None else str(stdin), - "pipe_out": _as_bool(data.get("pipe_out"), default=False), "mix_stderr": _as_bool(data.get("mix_stderr"), default=False), "stop_on_error": None if stop is None else _as_bool(stop, default=True), "timeout_seconds": _as_float(data.get("timeout_seconds"), default=DEFAULT_TIMEOUT_SECONDS), } + node["then"] = _parse_nodes(data.get("then"), where=f"{where}.then", depth=depth + 1, budget=budget, fed=False) + node["else"] = _parse_nodes(data.get("else"), where=f"{where}.else", depth=depth + 1, budget=budget, fed=False) + node["pipe"] = _parse_nodes(data.get("pipe"), where=f"{where}.pipe", depth=depth + 1, budget=budget, fed=True) + if data.get("pipe") is not None and not node["pipe"]: + msg = f"{where}.pipe must not be empty" + raise WorkspaceError(msg) + if fed and (node["then"] or node["else"]): + msg = f"{where}: a pipe step takes no then/else; branch on the node that owns the pipe" + raise WorkspaceError(msg) + return node def _normalize_steps(steps: list[dict[str, object]] | str) -> list[object]: @@ -92,15 +155,39 @@ def _normalize_steps(steps: list[dict[str, object]] | str) -> list[object]: return cast("list[object]", steps) -def _format_step(index: int, result: CommandStepResult, *, pipe_out: bool) -> str: - return "\n".join( +@dataclass(frozen=True) +class _Step: + """One executed command, and whether its stdout fed another one.""" + + result: CommandStepResult + depth: int + via: tuple[str, int] | None + feeds: bool + + +@dataclass(frozen=True) +class _Stage: + """Exit code of a node's command stage — its pipe chain, for a piped node.""" + + code: int | None + failed: bool + captured: str + + +def _format_step(index: int, step: _Step) -> str: + result = step.result + lines = [f"--- step {index} ---"] + if step.depth > 0: + lines.append(f"depth={step.depth}") + if step.via is not None: + lines.append(f"via={step.via[0]} of step {step.via[1]}") + lines.extend( [ - f"--- step {index} ---", f"exit_code={result.exit_code if result.exit_code is not None else ''}", f"timed_out={'true' if result.timed_out else 'false'}", f"cwd={result.cwd_display}", f"cmd={shlex.join(result.command)}", - f"pipe_out={'true' if pipe_out else 'false'}", + f"pipe={'true' if step.feeds else 'false'}", f"mix_stderr={'true' if result.mix_stderr else 'false'}", "--- stdout ---", result.stdout, @@ -108,6 +195,7 @@ def _format_step(index: int, result: CommandStepResult, *, pipe_out: bool) -> st result.stderr, ] ) + return "\n".join(lines) def _merge_env(base_env: dict[str, str] | None, step_env: dict[str, str] | None) -> dict[str, str] | None: @@ -116,43 +204,88 @@ def _merge_env(base_env: dict[str, str] | None, step_env: dict[str, str] | None) return {**(base_env or {}), **(step_env or {})} -async def _run_batch_steps( - steps: list[dict[str, Any]], - *, - cwd: str, - env: dict[str, str] | None, - stop_on_error: bool, -) -> tuple[list[tuple[dict[str, Any], CommandStepResult]], bool]: - results: list[tuple[dict[str, Any], CommandStepResult]] = [] - prev_capture: str | None = None - prev_piped = False - stopped_early = False - total_chars = 0 +@dataclass +class _Runner: + """Runs the node tree, remembering each command and when to stop.""" - for index, step in enumerate(steps): - stdin_text = prev_capture if prev_piped else step["stdin"] + cwd: str + env: dict[str, str] | None + stop_on_error: bool + steps: list[_Step] = field(default_factory=list) + chars: int = 0 + stopped: bool = False + + async def run_list( + self, + nodes: list[dict[str, Any]], + *, + depth: int, + branch: str | None = None, + parent: int | None = None, + ) -> None: + via = (branch, parent) if branch is not None and parent is not None else None + for node in nodes: + if self.stopped: + return + _ = await self.run_node(node, depth=depth, via=via) + + async def run_node( + self, + node: dict[str, Any], + *, + depth: int, + via: tuple[str, int] | None, + stdin: str | None = None, + also_feeds: bool = False, + stage_only: bool = False, + ) -> _Stage: result = await execute_argv( - step["argv"], - cwd=step["cwd"] if step["cwd"] is not None else cwd, - env=_merge_env(env, step["env"]), - stdin=stdin_text, - timeout_seconds=step["timeout_seconds"], - mix_stderr=step["mix_stderr"], + node["argv"], + cwd=node["cwd"] if node["cwd"] is not None else self.cwd, + env=_merge_env(self.env, node["env"]), + stdin=stdin if stdin is not None else node["stdin"], + timeout_seconds=node["timeout_seconds"], + mix_stderr=node["mix_stderr"], ) - results.append((step, result)) - prev_capture = result.captured - prev_piped = bool(step["pipe_out"]) - total_chars += len(result.stdout) + len(result.stderr) + index = len(self.steps) + self.steps.append(_Step(result, depth, via, feeds=bool(node["pipe"]) or also_feeds)) + self.chars += len(result.stdout) + len(result.stderr) + stage = _Stage(result.exit_code, bool(result.timed_out) or result.exit_code != 0, result.captured) - failed = bool(result.timed_out) or result.exit_code != 0 - effective_stop = stop_on_error if step["stop_on_error"] is None else bool(step["stop_on_error"]) - if failed and effective_stop: - return results, True - if total_chars > BATCH_OUTPUT_BUDGET_CHARS * BATCH_OUTPUT_SOFT_CAP_FACTOR: - return results, True - del index # used for clarity in loop only + chain: list[dict[str, Any]] = node["pipe"] + piped = result.captured + for position, child in enumerate(chain): + if self.stopped: + break + stage = await self.run_node( + child, + depth=depth + 1, + via=("pipe", index), + stdin=piped, + also_feeds=position + 1 < len(chain), + stage_only=True, + ) + piped = stage.captured - return results, stopped_early + if self.chars > BATCH_OUTPUT_BUDGET_CHARS * BATCH_OUTPUT_SOFT_CAP_FACTOR: + self.stopped = True + return stage + # A pipe stage's failure is the pipe owner's to answer: the chain still + # ran to its last command, so only that result decides the branch. + if stage_only: + return stage + if stage.failed: + if node["else"]: + await self.run_list(node["else"], depth=depth + 1, branch="else", parent=index) + elif self._effective_stop(node): + self.stopped = True + return stage + if node["then"]: + await self.run_list(node["then"], depth=depth + 1, branch="then", parent=index) + return stage + + def _effective_stop(self, node: dict[str, Any]) -> bool: + return self.stop_on_error if node["stop_on_error"] is None else bool(node["stop_on_error"]) @tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE | ToolTag.YOLO) @@ -163,47 +296,49 @@ async def run_argv_batch( env: dict[str, str] | None = None, stop_on_error: bool = True, ) -> str: - """Run a serial batch of argv commands (no shell). + """Run a batch of argv commands (no shell), each with optional continuations. - Each element of ``steps`` is an object:: + A node is an object:: { - "argv": ["git", "status"], # required argv - "cwd": ".", # optional, else batch cwd - "env": {"FOO": "1"}, # optional overlay - "stdin": "...", # optional; unused if previous pipe_out - "pipe_out": false, # if true, feed capture into *next* stdin - "mix_stderr": false, # OS-level merge stderr into capture - "stop_on_error": null, # override batch stop_on_error + "argv": ["pytest", "-q"], # required argv + "then": {"argv": [...]}, # runs when argv succeeds (node or list) + "else": {"argv": [...]}, # runs when argv fails (node or list) + "pipe": {"argv": [...]}, # takes argv's stdout as its stdin + "cwd": ".", # optional, else batch cwd + "env": {"FOO": "1"}, # optional overlay + "stdin": "...", # optional; a pipe step takes it from the pipe + "mix_stderr": false, # OS-level merge stderr into capture + "stop_on_error": null, # override batch stop_on_error "timeout_seconds": 30 } - ``pipe_out`` is set on the **provider** step. Last-step ``pipe_out`` is an error. - Default ``stop_on_error`` is true. Max 20 steps; the batch stops early once - accumulated output passes the run budget (128k chars). Each step's output is - kept whole for the model; a long result carries a ``document_id`` to read on. + ``steps`` is a list of nodes, and ``then`` / ``else`` / ``pipe`` each take one + node or a list of them, so the tree nests. A failure with no ``else`` stops + the batch, unless that node sets ``stop_on_error: false``: it then skips its + own continuations and the next sibling runs. A ``pipe`` step takes no + ``then`` / ``else`` — branch on the node that owns the pipe. Max 20 nodes and + 4 nesting levels; the batch stops early once accumulated output passes the + run budget (128k chars). Each command's output is kept whole for the model; a + long result carries a ``document_id`` to read on. """ try: raw_steps = _normalize_steps(steps) if not raw_steps: return "error: steps must not be empty" - if len(raw_steps) > DEFAULT_MAX_BATCH_STEPS: - return f"error: at most {DEFAULT_MAX_BATCH_STEPS} steps per batch" - steps = [_parse_step(item, i) for i, item in enumerate(raw_steps)] - if steps[-1]["pipe_out"]: - return "error: last step has pipe_out=true but there is no next step" + budget = _Budget() + nodes = _parse_nodes(raw_steps, where="steps", depth=0, budget=budget, fed=False) + runner = _Runner(cwd=cwd, env=env, stop_on_error=stop_on_error) + await runner.run_list(nodes, depth=0) - results, stopped_early = await _run_batch_steps(steps, cwd=cwd, env=env, stop_on_error=stop_on_error) - body = "\n".join( - _format_step(i, result, pipe_out=bool(step["pipe_out"])) for i, (step, result) in enumerate(results) - ) - last_exit = results[-1][1].exit_code if results else "" + body = "\n".join(_format_step(i, step) for i, step in enumerate(runner.steps)) + last_exit = runner.steps[-1].result.exit_code if runner.steps else "" return "\n".join( [ - f"steps={len(steps)} ran={len(results)} " + f"steps={budget.nodes} ran={len(runner.steps)} " f"stop_on_error={'true' if stop_on_error else 'false'} " - f"stopped_early={'true' if stopped_early else 'false'}", + f"stopped_early={'true' if runner.stopped else 'false'}", body, "--- summary ---", f"last_exit={last_exit if last_exit is not None else ''}", diff --git a/tests/test_cli/test_display.py b/tests/test_cli/test_display.py index 5be6685..8d29711 100644 --- a/tests/test_cli/test_display.py +++ b/tests/test_cli/test_display.py @@ -456,9 +456,9 @@ async def test_pretty_run_argv_batch_stopped(capsys: pytest.CaptureFixture[str]) result = ( "steps=3 ran=2 stop_on_error=true stopped_early=true\n" "--- step 1 ---\nexit_code=1\ntimed_out=false\ncwd=.\ncmd=git status\n" - "pipe_out=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n" + "pipe=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n" "--- step 2 ---\nexit_code=0\ntimed_out=false\ncwd=.\ncmd=git diff\n" - "pipe_out=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n" + "pipe=false\nmix_stderr=false\n--- stdout ---\n--- stderr ---\n" "--- summary ---\nlast_exit=1\n" ) await render_events(_aiter([_pretty_call("run_argv_batch", '{"steps": []}'), _result(result)])) @@ -495,16 +495,27 @@ async def test_pretty_run_argv_batch_lists_commands(capsys: pytest.CaptureFixtur async def test_pretty_run_argv_batch_joins_pipe_chain(capsys: pytest.CaptureFixture[str]) -> None: - """``pipe_out`` steps share one line; a step they do not feed keeps its own.""" + """A pipe chain shares one line; a node it does not feed keeps its own.""" args = ( - '{"steps": [{"argv": ["grep", "-n", "foo", "a.py"], "pipe_out": true},' - ' {"argv": ["wc", "-l"]}, {"argv": ["git", "status"]}]}' + '{"steps": [{"argv": ["grep", "-n", "foo", "a.py"], "pipe": {"argv": ["wc", "-l"]}},' + ' {"argv": ["git", "status"]}]}' ) await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)])) out = capsys.readouterr().out assert "$ grep -n foo a.py | wc -l\n$ git status\n* Run batch (done)" in out +async def test_pretty_run_argv_batch_indents_branches(capsys: pytest.CaptureFixture[str]) -> None: + """A ``then`` / ``else`` body sits below the node it follows.""" + args = ( + '{"steps": [{"argv": ["pytest", "-q"], "then": {"argv": ["ruff", "format", "."]},' + ' "else": [{"argv": ["git", "status"]}]}]}' + ) + await render_events(_aiter([_pretty_call("run_argv_batch", args), _result(_BATCH_DONE_RESULT)])) + out = capsys.readouterr().out + assert "$ pytest -q\n then ruff format .\n else git status\n* Run batch (done)" in out + + async def test_pretty_run_argv_batch_elides_a_long_command(capsys: pytest.CaptureFixture[str]) -> None: long_arg = "x" * 200 args = f'{{"steps": [{{"argv": ["echo", "{long_arg}"]}}]}}' diff --git a/tests/test_tools/test_danger.py b/tests/test_tools/test_danger.py index f06619f..7ca47ff 100644 --- a/tests/test_tools/test_danger.py +++ b/tests/test_tools/test_danger.py @@ -108,6 +108,34 @@ def test_classify_run_argv_batch_risky() -> None: ) +def test_classify_run_argv_batch_sees_nested_steps() -> None: + """A command hiding in a continuation is reviewed like a top-level step.""" + reason = _danger( + "run_argv_batch", + { + "steps": [ + { + "argv": ["echo", "ok"], + "then": {"argv": ["bash", "-c", "echo risky"]}, + "else": [{"argv": ["sh", "-c", "echo risky"]}], + } + ] + }, + ) + assert "run_argv_batch[0].then[0]" in reason.text + assert "run_argv_batch[0].else[0]" in reason.text + # A plain interpreter run is YOLO mode's to drop, wherever it sits. + assert reason.yolo_skip is True + + +def test_a_nested_step_that_escalates_keeps_the_batch_prompt() -> None: + reason = _danger( + "run_argv_batch", + {"steps": [{"argv": ["echo", "ok"], "then": {"argv": ["sudo", "python", "-c", "print(1)"]}}]}, + ) + assert reason.yolo_skip is False + + def test_classify_shell_and_dash_c() -> None: r = _reason("run_argv", {"argv": ["bash", "-c", "rm -rf /"]}) assert r is not None diff --git a/tests/test_tools/test_process.py b/tests/test_tools/test_process.py index 7c6ca21..d51e6aa 100644 --- a/tests/test_tools/test_process.py +++ b/tests/test_tools/test_process.py @@ -112,33 +112,31 @@ async def test_run_argv_batch_stop_on_error(workspace: object) -> None: assert "should-not-run" not in out -async def test_run_argv_batch_pipe_out(workspace: object) -> None: +async def test_run_argv_batch_pipe(workspace: object) -> None: del workspace out = await call_async( run_argv_batch, [ { "argv": _py("print('piped-payload', end='')"), - "pipe_out": True, - }, - { - "argv": _py("import sys; print(sys.stdin.read())"), + "pipe": {"argv": _py("import sys; print(sys.stdin.read())")}, }, ], ) assert "ran=2" in out assert "piped-payload" in out - assert "pipe_out=true" in out + assert "pipe=true" in out -async def test_run_argv_batch_last_pipe_out_error(workspace: object) -> None: +async def test_run_argv_batch_pipe_out_is_gone(workspace: object) -> None: + """The old provider-side flag is refused, pointing at the pipe node.""" del workspace out = await call_async( run_argv_batch, [{"argv": _py("print(1)"), "pipe_out": True}], ) assert "error:" in out - assert "pipe_out" in out + assert '{"pipe": [...]}' in out async def test_run_argv_batch_mix_stderr(workspace: object) -> None: @@ -149,9 +147,8 @@ async def test_run_argv_batch_mix_stderr(workspace: object) -> None: { "argv": _py("import sys; sys.stdout.write('out'); sys.stderr.write('err')"), "mix_stderr": True, - "pipe_out": True, + "pipe": {"argv": _py("import sys; print(repr(sys.stdin.read()))")}, }, - {"argv": _py("import sys; print(repr(sys.stdin.read()))")}, ], ) assert "ran=2" in out @@ -159,6 +156,149 @@ async def test_run_argv_batch_mix_stderr(workspace: object) -> None: assert "out" in out and "err" in out +async def test_run_argv_batch_then_runs_on_success(workspace: object) -> None: + del workspace + out = await call_async( + run_argv_batch, + [{"argv": _py("print('first')"), "then": {"argv": _py("print('then-body')")}}], + ) + assert "then-body" in out + assert "ran=2" in out + assert "depth=1" in out + assert "via=then of step 0" in out + + +async def test_run_argv_batch_else_handles_the_failure(workspace: object) -> None: + """A failure with an ``else`` is answered, so the batch keeps going.""" + del workspace + out = await call_async( + run_argv_batch, + [ + { + "argv": _py("import sys; sys.exit(2)"), + "else": {"argv": _py("print('else-body')")}, + }, + {"argv": _py("print('sibling-ran')")}, + ], + ) + assert "else-body" in out + assert "sibling-ran" in out + assert "via=else of step 0" in out + assert "stopped_early=false" in out + + +async def test_run_argv_batch_runs_a_then_list_in_order(workspace: object) -> None: + del workspace + out = await call_async( + run_argv_batch, + [ + { + "argv": _py("print('first')"), + "then": [{"argv": _py("print('second')")}, {"argv": _py("print('third')")}], + } + ], + ) + assert "ran=3" in out + assert out.index("first") < out.index("second") < out.index("third") + + +async def test_run_argv_batch_a_failing_branch_stops_the_batch(workspace: object) -> None: + """Nobody answered a failure inside a continuation, so the batch stops.""" + del workspace + out = await call_async( + run_argv_batch, + [ + {"argv": _py("print('first')"), "then": {"argv": _py("import sys; sys.exit(4)")}}, + {"argv": _py("print('sibling-ran')")}, + ], + ) + assert "stopped_early=true" in out + assert "sibling-ran" not in out + + +async def test_run_argv_batch_stop_on_error_false_skips_only_its_continuations(workspace: object) -> None: + del workspace + out = await call_async( + run_argv_batch, + [ + { + "argv": _py("import sys; sys.exit(2)"), + "stop_on_error": False, + "then": {"argv": _py("print('then-ran')")}, + }, + {"argv": _py("print('sibling-ran')")}, + ], + ) + assert "then-ran" not in out + assert "sibling-ran" in out + assert "stopped_early=false" in out + + +async def test_run_argv_batch_branches_on_the_pipe_result(workspace: object) -> None: + """A piped node branches on the last command of its chain.""" + del workspace + out = await call_async( + run_argv_batch, + [ + { + "argv": _py("print('payload', end='')"), + "pipe": {"argv": _py("import sys; sys.stdin.read(); sys.exit(3)")}, + "then": {"argv": _py("print('pipe-ok')")}, + "else": {"argv": _py("print('pipe-failed')")}, + } + ], + ) + assert "pipe-failed" in out + assert "pipe-ok" not in out + assert "ran=3" in out + + +async def test_run_argv_batch_pipe_step_takes_no_branch(workspace: object) -> None: + del workspace + out = await call_async( + run_argv_batch, + [{"argv": _py("print(1)"), "pipe": {"argv": _py("print(2)"), "then": {"argv": _py("print(3)")}}}], + ) + assert "error:" in out + assert "pipe step takes no then/else" in out + + +async def test_run_argv_batch_pipe_step_takes_no_stdin(workspace: object) -> None: + del workspace + out = await call_async( + run_argv_batch, + [{"argv": _py("print(1)"), "pipe": {"argv": _py("print(2)"), "stdin": "own"}}], + ) + assert "error:" in out + assert "stdin comes from the pipe" in out + + +async def test_run_argv_batch_rejects_a_deep_tree(workspace: object) -> None: + del workspace + node: dict[str, object] = {"argv": _py("print(1)")} + for _ in range(6): + node = {"argv": _py("print(1)"), "then": node} + out = await call_async(run_argv_batch, [node]) + assert "error:" in out + assert "nest at most 4 levels deep" in out + + +async def test_run_argv_batch_counts_nested_nodes(workspace: object) -> None: + """Every node of the tree costs one, so a nested tree hits the same cap.""" + del workspace + out = await call_async(run_argv_batch, [{"argv": _py("print(1)")} for _ in range(21)]) + assert "error:" in out + assert "more than 20 nodes" in out + + +async def test_run_argv_batch_names_the_node_grammar(workspace: object) -> None: + del workspace + out = await call_async(run_argv_batch, [{"argv": _py("print(1)"), "when": "always"}]) + assert "error:" in out + assert "unknown key(s) when" in out + assert "then, else, pipe" in out + + async def test_run_argv_denied(workspace: object) -> None: del workspace out = await call_async(run_argv, ["rm", "-rf", "."])