diff --git a/CLAUDE.md b/CLAUDE.md index 5350ae7..4318bf6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -80,7 +80,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use. - **`workspace`**: path resolve under primary root **or** temporary allowlist roots; path substring denylist; command basename denylist; clearer deny messages. - **`file`**: `read_file` (`with_lineno`), `write_file`, `listdir`, `tree` (VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace`, `edit_lineno` (1-based range), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind). -- **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep). +- **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); `run_command_batch` (serial steps, `pipe_out` on provider, `mix_stderr`, `stop_on_error`); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep). - PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); session limit/idle TTL/output budget; close terminate→kill. - CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget. - Destructive confirms: `classify_danger` + `ToolRegistry(on_confirm=…)`; CLI default deny; config `confirm_destructive` / `path_denylist`. Soft confirm also for shells/REPLs and `python|bash -c` one-liners (argv + code preview); deny may include a user comment for the model. Session YOLO: `/yolo on|off|once` and `--yes` (skip soft confirms; hard denylists unchanged; `once` expires after the next user turn). diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index 86e746b..1858009 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -20,6 +20,7 @@ from .process import close_pty as close_pty from .process import open_pty as open_pty from .process import read_pty as read_pty from .process import run_command as run_command +from .process import run_command_batch as run_command_batch from .process import write_pty as write_pty from .process import write_pty_keys as write_pty_keys from .temp_workspace import cleanup_temporary_workspaces as cleanup_temporary_workspaces diff --git a/src/plyngent/tools/danger.py b/src/plyngent/tools/danger.py index 0aa7fd4..010c6f4 100644 --- a/src/plyngent/tools/danger.py +++ b/src/plyngent/tools/danger.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from typing import TYPE_CHECKING, cast from plyngent.tools.workspace import WorkspaceError, resolve_path @@ -161,6 +162,44 @@ def _run_command_reason(args: Mapping[str, object]) -> str | None: return _shell_or_dash_c_reason(argv, via="run_command") +def _batch_step_argv(item: object) -> list[str] | None: + if not isinstance(item, dict): + return None + step = cast("dict[str, object]", item) + command = step.get("command") + if not isinstance(command, list) or not command: + return None + argv: list[str] = [] + for part in command: + if not isinstance(part, str): + return None + argv.append(part) + return argv or None + + +def _run_command_batch_reason(args: Mapping[str, object]) -> str | None: + """One confirm for the whole batch if any step is shell/REPL/-c.""" + raw = args.get("commands") + if isinstance(raw, str): + try: + loaded: object = json.loads(raw) + except json.JSONDecodeError: + return 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 := _shell_or_dash_c_reason(argv, via=f"run_command_batch[{index}]")) is not None + ] + if not risky: + return None + header = f"run_command_batch: {len(risky)} risky step(s) (review before allow)" + return header + "\n" + "\n".join(risky) + + def _open_pty_reason(args: Mapping[str, object]) -> str | None: argv = _as_argv(args) if argv is None: @@ -189,6 +228,8 @@ def classify_danger(name: str, args: Mapping[str, object]) -> str | None: # noq return _edit_lineno_reason(args) if name == "run_command": return _run_command_reason(args) + if name == "run_command_batch": + return _run_command_batch_reason(args) if name == "open_pty": return _open_pty_reason(args) return None diff --git a/src/plyngent/tools/process/__init__.py b/src/plyngent/tools/process/__init__.py index be0e8f8..35988a7 100644 --- a/src/plyngent/tools/process/__init__.py +++ b/src/plyngent/tools/process/__init__.py @@ -4,7 +4,16 @@ from .pty_terminal import decode_write_data as decode_write_data from .pty_terminal import sanitize_pty_output_for_tool as sanitize_pty_output_for_tool from .read_pty import read_pty as read_pty from .run_command import run_command as run_command +from .run_command_batch import run_command_batch as run_command_batch from .write_pty import write_pty as write_pty from .write_pty_keys import write_pty_keys as write_pty_keys -PROCESS_TOOLS = [run_command, open_pty, read_pty, write_pty, write_pty_keys, close_pty] +PROCESS_TOOLS = [ + run_command, + run_command_batch, + open_pty, + read_pty, + write_pty, + write_pty_keys, + close_pty, +] diff --git a/src/plyngent/tools/process/command_exec.py b/src/plyngent/tools/process/command_exec.py new file mode 100644 index 0000000..3b4f726 --- /dev/null +++ b/src/plyngent/tools/process/command_exec.py @@ -0,0 +1,143 @@ +from __future__ import annotations + +import asyncio +import os +import shlex +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from pathlib import Path + +from plyngent.tools.workspace import ( + WorkspaceError, + check_command_allowed, + get_workspace_root, + resolve_path, +) + +DEFAULT_TIMEOUT_SECONDS = 30 +DEFAULT_MAX_OUTPUT_CHARS = 32_000 +DEFAULT_MAX_BATCH_STEPS = 20 + + +def truncate_output(text: str, label: str, max_chars: int = DEFAULT_MAX_OUTPUT_CHARS) -> str: + if len(text) <= max_chars: + return text + return text[:max_chars] + f"\n...[{label} truncated]" + + +def format_command_result( + *, + returncode: int | None, + workdir_display: str, + command: list[str], + stdout: str, + stderr: str, + timed_out: bool = False, +) -> str: + parts = [ + f"exit_code={returncode if returncode is not None else ''}", + f"timed_out={'true' if timed_out else 'false'}", + f"cwd={workdir_display}", + f"cmd={shlex.join(command)}", + "--- stdout ---", + truncate_output(stdout, "stdout"), + "--- stderr ---", + truncate_output(stderr, "stderr"), + ] + return "\n".join(parts) + + +def resolve_workdir(cwd: str) -> tuple["Path", str]: + """Return (absolute workdir, display path relative to workspace when possible).""" + workdir = resolve_path(cwd) + if not workdir.is_dir(): + msg = f"not a directory: {cwd}" + raise WorkspaceError(msg) + try: + workdir_display = str(workdir.relative_to(get_workspace_root())) + except ValueError: + workdir_display = str(workdir) + return workdir, workdir_display + + +@dataclass +class CommandStepResult: + command: list[str] + cwd_display: str + exit_code: int | None + timed_out: bool + stdout: str + stderr: str + mix_stderr: bool + captured: str # piped to the next step when provider sets pipe_out + + +async def execute_argv( + command: list[str], + *, + cwd: str = ".", + env: dict[str, str] | None = None, + stdin: str | None = None, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + mix_stderr: bool = False, +) -> CommandStepResult: + """Run one argv command; no shell. Used by run_command and run_command_batch.""" + if not command: + msg = "command argv must not be empty" + raise WorkspaceError(msg) + check_command_allowed(command) + workdir, workdir_display = resolve_workdir(cwd) + + stdin_data = None if stdin is None else stdin.encode() + run_env: dict[str, str] | None = None + if env is not None: + run_env = {str(k): str(v) for k, v in {**os.environ, **env}.items()} + + try: + stderr_arg = asyncio.subprocess.STDOUT if mix_stderr else asyncio.subprocess.PIPE + proc = await asyncio.create_subprocess_exec( + *command, + cwd=str(workdir), + env=run_env, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=stderr_arg, + ) + except FileNotFoundError: + msg = f"executable not found: {command[0]!r}" + raise WorkspaceError(msg) from None + except OSError as exc: + msg = f"failed to start command: {exc}" + raise WorkspaceError(msg) from exc + + timed_out = False + try: + stdout_b, stderr_b = await asyncio.wait_for( + proc.communicate(input=stdin_data), + timeout=timeout_seconds, + ) + except TimeoutError: + timed_out = True + proc.kill() + stdout_b, stderr_b = await proc.communicate() + + stdout = (stdout_b or b"").decode(errors="replace") + if mix_stderr: + stderr = "" + captured = stdout + else: + stderr = (stderr_b or b"").decode(errors="replace") + captured = stdout + + return CommandStepResult( + command=list(command), + cwd_display=workdir_display, + exit_code=proc.returncode, + timed_out=timed_out, + stdout=stdout, + stderr=stderr, + mix_stderr=mix_stderr, + captured=captured, + ) diff --git a/src/plyngent/tools/process/run_command.py b/src/plyngent/tools/process/run_command.py index 01ade75..b61ff46 100644 --- a/src/plyngent/tools/process/run_command.py +++ b/src/plyngent/tools/process/run_command.py @@ -1,133 +1,14 @@ from __future__ import annotations -import asyncio -import os -import shlex -from typing import TYPE_CHECKING - from plyngent.agent import tool -from plyngent.tools.workspace import ( - WorkspaceError, - check_command_allowed, - get_workspace_root, - resolve_path, +from plyngent.tools.workspace import WorkspaceError + +from .command_exec import ( + DEFAULT_TIMEOUT_SECONDS, + execute_argv, + format_command_result, ) -if TYPE_CHECKING: - from pathlib import Path - -DEFAULT_TIMEOUT_SECONDS = 30 -DEFAULT_MAX_OUTPUT_CHARS = 32_000 - - -def _truncate(text: str, label: str) -> str: - if len(text) <= DEFAULT_MAX_OUTPUT_CHARS: - return text - return text[:DEFAULT_MAX_OUTPUT_CHARS] + f"\n...[{label} truncated]" - - -def _format_result( - *, - returncode: int | None, - workdir_display: str, - command: list[str], - stdout: str, - stderr: str, - timed_out: bool = False, -) -> str: - parts = [ - f"exit_code={'' if returncode is None else returncode}", - f"timed_out={'true' if timed_out else 'false'}", - f"cwd={workdir_display}", - f"cmd={shlex.join(command)}", - ] - if stdout: - parts.append("--- stdout ---\n" + stdout.rstrip("\n")) - if stderr: - parts.append("--- stderr ---\n" + stderr.rstrip("\n")) - return "\n".join(parts) - - -def _validate_env(env: object) -> str | None: - """Runtime guard for tool-JSON args (may not match static typing at the boundary).""" - if env is None: - return None - if type(env) is not dict: - return "error: env must be an object of string keys and values" - # Tool schema should already enforce dict[str, str]; keep a light check. - return None - - -def _merge_env(overrides: dict[str, str] | None) -> dict[str, str] | None: - if overrides is None: - return None - return {**os.environ, **overrides} - - -async def _run_exec( - command: list[str], - *, - workdir: str, - timeout_seconds: float, - stdin_data: bytes | None, - env: dict[str, str] | None, -) -> tuple[int | None, str, str, bool] | str: - """Return ``(returncode, stdout, stderr, timed_out)`` or an error string.""" - stdin = asyncio.subprocess.PIPE if stdin_data is not None else None - try: - proc = await asyncio.create_subprocess_exec( - *command, - cwd=workdir, - env=env, - stdin=stdin, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - except FileNotFoundError: - return f"error: executable not found: {command[0]}" - except OSError as exc: - return f"error: failed to start command: {exc}" - - timed_out = False - try: - stdout_b, stderr_b = await asyncio.wait_for( - proc.communicate(input=stdin_data), - timeout=timeout_seconds, - ) - except TimeoutError: - timed_out = True - proc.kill() - stdout_b, stderr_b = await proc.communicate() - - stdout = _truncate(stdout_b.decode(errors="replace"), "stdout") - stderr = _truncate(stderr_b.decode(errors="replace"), "stderr") - return proc.returncode, stdout, stderr, timed_out - - -def _validate_run_args( - command: list[str], - *, - cwd: str, - timeout_seconds: float, - env: dict[str, str] | None, -) -> Path | str: - """Return resolved workdir, or an error string.""" - if not command: - return "error: command must not be empty" - try: - check_command_allowed(command) - workdir = resolve_path(cwd) - except WorkspaceError as exc: - return f"error: {exc}" - if not workdir.is_dir(): - return f"error: cwd is not a directory: {cwd}" - if timeout_seconds <= 0: - return "error: timeout_seconds must be > 0" - env_error = _validate_env(env) - if env_error is not None: - return env_error - return workdir - @tool async def run_command( @@ -138,35 +19,24 @@ async def run_command( stdin: str | None = None, env: dict[str, str] | None = None, ) -> str: - """Run a command without a shell (argv list) under the workspace. + """Run ``command`` (argv, no shell) under the workspace; capture stdout/stderr.""" + try: + result = await execute_argv( + command, + cwd=cwd, + env=env, + stdin=stdin, + timeout_seconds=timeout_seconds, + mix_stderr=False, + ) + except WorkspaceError as exc: + return f"error: {exc}" - ``cwd`` is relative to or under the workspace root. Optional ``stdin`` is - written to the process stdin. Optional ``env`` overlays process environment - variables (merged with the current environment). Output is truncated. - - On timeout the process is killed and any partial stdout/stderr is still - returned with ``timed_out=true``. - """ - workdir = _validate_run_args(command, cwd=cwd, timeout_seconds=timeout_seconds, env=env) - if isinstance(workdir, str): - return workdir - - stdin_data = None if stdin is None else stdin.encode() - result = await _run_exec( - command, - workdir=str(workdir), - timeout_seconds=timeout_seconds, - stdin_data=stdin_data, - env=_merge_env(env), - ) - if isinstance(result, str): - return result - returncode, stdout, stderr, timed_out = result - return _format_result( - returncode=returncode, - workdir_display=str(workdir.relative_to(get_workspace_root())), - command=command, - stdout=stdout, - stderr=stderr, - timed_out=timed_out, + return format_command_result( + returncode=result.exit_code, + workdir_display=result.cwd_display, + command=result.command, + stdout=result.stdout, + stderr=result.stderr, + timed_out=result.timed_out, ) diff --git a/src/plyngent/tools/process/run_command_batch.py b/src/plyngent/tools/process/run_command_batch.py new file mode 100644 index 0000000..8585594 --- /dev/null +++ b/src/plyngent/tools/process/run_command_batch.py @@ -0,0 +1,213 @@ +from __future__ import annotations + +import json +import shlex +from typing import Any, cast + +from plyngent.agent import tool +from plyngent.tools.workspace import WorkspaceError + +from .command_exec import ( + DEFAULT_MAX_BATCH_STEPS, + DEFAULT_MAX_OUTPUT_CHARS, + DEFAULT_TIMEOUT_SECONDS, + CommandStepResult, + execute_argv, + truncate_output, +) + +BATCH_OUTPUT_SOFT_CAP_FACTOR = 4 + + +def _as_bool(value: object, *, default: bool = False) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + if isinstance(value, str): + return value.strip().lower() in {"1", "true", "yes", "on"} + return bool(value) + + +def _as_float(value: object, *, default: float) -> float: + if value is None: + return default + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, str) and value.strip(): + return float(value) + return default + + +def _as_str_dict(value: object) -> dict[str, str] | None: + if value is None: + return None + if not isinstance(value, dict): + msg = "env must be an object of string keys/values" + raise WorkspaceError(msg) + 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]: + if not isinstance(raw, dict): + msg = f"commands[{index}] must be an object" + raise WorkspaceError(msg) + data = cast("dict[str, object]", raw) + command = data.get("command") + if not isinstance(command, list) or not command: + msg = f"commands[{index}].command must be a non-empty argv list" + raise WorkspaceError(msg) + argv = [part for part in command if isinstance(part, str)] + if len(argv) != len(command): + msg = f"commands[{index}].command must be a list of strings" + raise WorkspaceError(msg) + cwd = data.get("cwd") + stdin = data.get("stdin") + stop = data.get("stop_on_error", None) + return { + "command": argv, + "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), + } + + +def _normalize_commands(commands: list[dict[str, object]] | str) -> list[object]: + if isinstance(commands, str): + try: + loaded: object = json.loads(commands) + except json.JSONDecodeError as exc: + msg = f"commands must be a JSON array: {exc}" + raise WorkspaceError(msg) from exc + if not isinstance(loaded, list): + msg = "commands must be a JSON array of step objects" + raise WorkspaceError(msg) + return cast("list[object]", loaded) + return cast("list[object]", commands) + + +def _format_step(index: int, result: CommandStepResult, *, pipe_out: bool) -> str: + return "\n".join( + [ + 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"mix_stderr={'true' if result.mix_stderr else 'false'}", + "--- stdout ---", + truncate_output(result.stdout, "stdout"), + "--- stderr ---", + truncate_output(result.stderr, "stderr"), + ] + ) + + +def _merge_env(base_env: dict[str, str] | None, step_env: dict[str, str] | None) -> dict[str, str] | None: + if base_env is None and step_env is None: + return 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 + + for index, step in enumerate(steps): + stdin_text = prev_capture if prev_piped else step["stdin"] + result = await execute_argv( + step["command"], + 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"], + ) + results.append((step, result)) + prev_capture = result.captured + prev_piped = bool(step["pipe_out"]) + total_chars += len(result.stdout) + len(result.stderr) + + 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 > DEFAULT_MAX_OUTPUT_CHARS * BATCH_OUTPUT_SOFT_CAP_FACTOR: + return results, True + del index # used for clarity in loop only + + return results, stopped_early + + +@tool +async def run_command_batch( + commands: list[dict[str, object]] | str, + *, + cwd: str = ".", + env: dict[str, str] | None = None, + stop_on_error: bool = True, +) -> str: + """Run a serial batch of argv commands (no shell). + + Each element of ``commands`` is an object:: + + { + "command": ["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 + "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; total output budget 32k chars. + """ + try: + raw_steps = _normalize_commands(commands) + if not raw_steps: + return "error: commands must not be empty" + if len(raw_steps) > DEFAULT_MAX_BATCH_STEPS: + return f"error: at most {DEFAULT_MAX_BATCH_STEPS} commands per batch" + + steps = [_parse_step(item, i) for i, item in enumerate(raw_steps)] + if steps[-1]["pipe_out"]: + return "error: last command has pipe_out=true but there is no next step" + + 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) + ) + if len(body) > DEFAULT_MAX_OUTPUT_CHARS: + body = truncate_output(body, "batch") + last_exit = results[-1][1].exit_code if results else "" + return "\n".join( + [ + f"steps={len(steps)} ran={len(results)} " + f"stop_on_error={'true' if stop_on_error else 'false'} " + f"stopped_early={'true' if stopped_early else 'false'}", + body, + "--- summary ---", + f"last_exit={last_exit if last_exit is not None else ''}", + ] + ) + except WorkspaceError as exc: + return f"error: {exc}" + except (TypeError, ValueError) as exc: + return f"error: invalid batch arguments: {exc}" diff --git a/tests/test_tools/test_danger.py b/tests/test_tools/test_danger.py index 43ce308..cc5cf8e 100644 --- a/tests/test_tools/test_danger.py +++ b/tests/test_tools/test_danger.py @@ -36,6 +36,28 @@ def test_classify_safe_tools() -> None: assert classify_danger("run_command", {"command": ["ls", "-la"]}) is None +def test_classify_run_command_batch_risky() -> None: + reason = classify_danger( + "run_command_batch", + { + "commands": [ + {"command": ["echo", "ok"]}, + {"command": ["bash", "-c", "echo risky"]}, + ] + }, + ) + assert reason is not None + assert "run_command_batch" in reason + assert "risky" in reason or "bash" in reason + assert ( + classify_danger( + "run_command_batch", + {"commands": [{"command": ["echo", "ok"]}]}, + ) + is None + ) + + def test_classify_shell_and_dash_c() -> None: r = classify_danger("run_command", {"command": ["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 167e9ec..005de38 100644 --- a/tests/test_tools/test_process.py +++ b/tests/test_tools/test_process.py @@ -9,6 +9,7 @@ from plyngent.tools.process import ( open_pty, read_pty, run_command, + run_command_batch, write_pty, write_pty_keys, ) @@ -46,6 +47,85 @@ async def test_run_command_echo(workspace: object) -> None: assert "hi" in out +async def test_run_command_batch_serial(workspace: object) -> None: + del workspace + out = await call_async( + run_command_batch, + [ + {"command": _py("print('a')")}, + {"command": _py("print('b')")}, + ], + ) + assert "steps=2" in out + assert "ran=2" in out + assert "stopped_early=false" in out + assert "--- step 0 ---" in out + assert "--- step 1 ---" in out + assert "a" in out and "b" in out + + +async def test_run_command_batch_stop_on_error(workspace: object) -> None: + del workspace + out = await call_async( + run_command_batch, + [ + {"command": _py("import sys; sys.exit(2)")}, + {"command": _py("print('should-not-run')")}, + ], + stop_on_error=True, + ) + assert "ran=1" in out + assert "stopped_early=true" in out + assert "should-not-run" not in out + + +async def test_run_command_batch_pipe_out(workspace: object) -> None: + del workspace + out = await call_async( + run_command_batch, + [ + { + "command": _py("print('piped-payload', end='')"), + "pipe_out": True, + }, + { + "command": _py("import sys; print(sys.stdin.read())"), + }, + ], + ) + assert "ran=2" in out + assert "piped-payload" in out + assert "pipe_out=true" in out + + +async def test_run_command_batch_last_pipe_out_error(workspace: object) -> None: + del workspace + out = await call_async( + run_command_batch, + [{"command": _py("print(1)"), "pipe_out": True}], + ) + assert "error:" in out + assert "pipe_out" in out + + +async def test_run_command_batch_mix_stderr(workspace: object) -> None: + del workspace + out = await call_async( + run_command_batch, + [ + { + "command": _py("import sys; sys.stdout.write('out'); sys.stderr.write('err')"), + "mix_stderr": True, + "pipe_out": True, + }, + {"command": _py("import sys; print(repr(sys.stdin.read()))")}, + ], + ) + assert "ran=2" in out + # Merged capture should include both (order depends on OS scheduling). + assert "out" in out and "err" in out + + async def test_run_command_denied(workspace: object) -> None: del workspace out = await call_async(run_command, ["rm", "-rf", "."])