diff --git a/argus_skill/agent_cli/_prompt_delivery.py b/argus_skill/agent_cli/_prompt_delivery.py index 65904acf9..9206dc2bd 100644 --- a/argus_skill/agent_cli/_prompt_delivery.py +++ b/argus_skill/agent_cli/_prompt_delivery.py @@ -5,7 +5,9 @@ import os import subprocess import tempfile +from contextlib import contextmanager from pathlib import Path +from typing import Iterator, TextIO from ..core.sandbox import sandboxed_child_env from ._sandbox_commands import ( @@ -114,6 +116,28 @@ def _opencode_full_access_env() -> dict[str, str]: class PromptDeliveryMixin: """Deliver large role prompts without exposing them in process arguments.""" + @staticmethod + @contextmanager + def _prompt_stdin(prompt: str | None) -> Iterator[TextIO | int]: + """Provide finite stdin without writing into a live child's pipe. + + CLIs may emit startup output before reading their prompt. A synchronous + pipe write before starting stdout/stderr readers deadlocks under that + backpressure. A temporary file preserves stdin delivery and EOF without + a writer thread, prompt argv exposure, or a pipe-capacity dependency. + The child inherits its own handle; the parent closes its copy at spawn. + """ + if prompt is None: + yield subprocess.DEVNULL + return + with tempfile.TemporaryFile(mode="w+", encoding="utf-8", errors="replace") as stream: + stream.write(prompt) + if not prompt.endswith("\n"): + stream.write("\n") + stream.seek(0) + yield stream + + @staticmethod def _write_prompt(*, process: subprocess.Popen[str], prompt: str) -> None: if process.stdin is None: diff --git a/argus_skill/agent_cli/_run_exec.py b/argus_skill/agent_cli/_run_exec.py index 459feffce..758f91bab 100644 --- a/argus_skill/agent_cli/_run_exec.py +++ b/argus_skill/agent_cli/_run_exec.py @@ -234,35 +234,29 @@ def _spawn_turn_process( prompt_path, ) try: - process = subprocess.Popen( - command, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - # Pin UTF-8 explicitly: without this, text mode uses the OS locale - # encoding, which is cp1252 on Windows and raises UnicodeEncodeError - # when the prompt or streamed model output contains non-Latin-1 - # characters (e.g. "\u2192", CJK, emoji). errors="replace" keeps the - # reader from crashing on malformed bytes mid-stream. - encoding="utf-8", - errors="replace", - bufsize=1, - cwd=options.working_dir or None, - env=self._child_env(options, executable=command[0]), - **background_subprocess_kwargs(), - ) + with self._prompt_stdin(stdin_prompt) as child_stdin: + process = subprocess.Popen( + command, + stdin=child_stdin, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + # Pin UTF-8 explicitly: without this, text mode uses the OS locale + # encoding, which is cp1252 on Windows and raises UnicodeEncodeError + # when the prompt or streamed model output contains non-Latin-1 + # characters (e.g. "\u2192", CJK, emoji). errors="replace" keeps the + # reader from crashing on malformed bytes mid-stream. + encoding="utf-8", + errors="replace", + bufsize=1, + cwd=options.working_dir or None, + env=self._child_env(options, executable=command[0]), + **background_subprocess_kwargs(), + ) except BaseException: if prompt_path is not None: prompt_path.unlink(missing_ok=True) raise - if stdin_prompt is not None: - self._write_prompt( - process=process, - prompt=stdin_prompt, - ) - else: - self._close_stdin(process) return command, process, None, prompt_path def _stream_turn_output( diff --git a/tests/agent_cli/test_prompt_stdin_backpressure.py b/tests/agent_cli/test_prompt_stdin_backpressure.py new file mode 100644 index 000000000..b71f7ff8c --- /dev/null +++ b/tests/agent_cli/test_prompt_stdin_backpressure.py @@ -0,0 +1,104 @@ +"""Prompt delivery under real input/output backpressure.""" + +import hashlib +import json +import sys +import time +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from argus_skill.agent_cli import _run_exec +from argus_skill.agent_cli.agent_cli_runner import AgentCliRunner, RunnerOptions + + +def run_fixture(monkeypatch, tmp_path, body, prompt, *, options=None, callback=None): + runner = AgentCliRunner(agent_bin=sys.executable, backend="codex", event_callback=callback) + monkeypatch.setattr( + runner, "_build_command", lambda **kw: [sys.executable, "-X", "utf8", "-u", "-c", body] + ) + processes = [] + original = _run_exec.subprocess.Popen + + def capture(*args, **kwargs): + proc = original(*args, **kwargs) + processes.append(proc) + return proc + + monkeypatch.setattr(_run_exec.subprocess, "Popen", capture) + start = time.monotonic() + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit( + runner.run_exec, + prompt=prompt, + resume_thread_id=None, + options=options or RunnerOptions(working_dir=str(tmp_path)), + run_label="engineer-r1", + ) + try: + result = future.result(timeout=5) + return result, time.monotonic() - start, processes + finally: + for proc in processes: + runner._terminate_process(proc) + # If an assertion or timeout fired, terminating the identified root + # releases a blocked stdin write before joining the executor. + if not future.done(): + future.result(timeout=5) + + +@pytest.mark.parametrize("stream", [1, 2]) +@pytest.mark.parametrize("prompt", ["p" * 262144, "δΈ­ζ–‡πŸ™‚\n" * 65536], ids=["ascii", "unicode"]) +def test_large_prompt_and_startup_output_cannot_deadlock(monkeypatch, tmp_path, stream, prompt): + received = tmp_path / "received.json" + body = ( + "import os,sys,json,hashlib;from pathlib import Path;" + f"os.write({stream},b'x'*262144+b'\\n');data=sys.stdin.read();" + f"Path({str(received)!r}).write_text(json.dumps({{'hash':hashlib.sha256(data.encode()).hexdigest(),'length':len(data)}}));" + "print(json.dumps({'type':'turn.completed'}),flush=True)" + ) + result, elapsed, processes = run_fixture(monkeypatch, tmp_path, body, prompt) + expected = prompt if prompt.endswith("\n") else prompt + "\n" + assert json.loads(received.read_text()) == { + "hash": hashlib.sha256(expected.encode()).hexdigest(), + "length": len(expected), + } + assert result.turn_completed and result.exit_code == 0 and elapsed < 3 + assert all(p.stdin is None for p in processes) + + +@pytest.mark.parametrize("reason", ["stop", "hard_idle"]) +def test_child_not_reading_input_can_be_stopped(monkeypatch, tmp_path, reason): + start = time.monotonic() + options = RunnerOptions(working_dir=str(tmp_path), watchdog_hard_idle_seconds=1) + if reason == "stop": + options.external_interrupt_reason_provider = lambda: ( + "daemon stop requested" if time.monotonic() - start > 0.6 else None + ) + result, elapsed, processes = run_fixture( + monkeypatch, tmp_path, "import time;time.sleep(20)", "p" * 262144, options=options + ) + assert elapsed < 3 and result.turn_failed + assert ( + "daemon stop requested" if reason == "stop" else "hard idle timeout" + ) in result.fatal_error + assert all(p.poll() is not None for p in processes) + + +def test_spawn_failure_closes_temporary_stdin(monkeypatch, tmp_path): + inputs = [] + + def fail(*args, **kwargs): + inputs.append(kwargs["stdin"]) + assert kwargs["stdin"].read() == "fixture\n" + raise OSError("injected spawn error") + + monkeypatch.setattr(_run_exec.subprocess, "Popen", fail) + runner = AgentCliRunner(agent_bin=sys.executable, backend="codex") + with pytest.raises(OSError, match="injected spawn"): + runner.run_exec( + prompt="fixture", + resume_thread_id=None, + options=RunnerOptions(working_dir=str(tmp_path)), + ) + assert inputs and inputs[0].closed