diff --git a/src/benchflow/acp/container_transport.py b/src/benchflow/acp/container_transport.py index 64701ec7a..f5a784c2d 100644 --- a/src/benchflow/acp/container_transport.py +++ b/src/benchflow/acp/container_transport.py @@ -7,6 +7,7 @@ from benchflow.sandbox.process import LiveProcess from benchflow.sandbox.process._base import _ANSI_CSI_RE, _ANSI_OSC_RE +from benchflow.trajectories.types import redact_trajectory_text from .transport import Transport, decode_json_rpc_message @@ -138,3 +139,7 @@ async def close(self) -> None: self._agent_log_file.close() self._agent_log_file = None await self._cp.close() + stderr = getattr(self._cp, "stderr_tail", "") + if isinstance(stderr, str) and stderr and self._agent_log_path: + with self._agent_log_path.open("a") as agent_log: + agent_log.write(redact_trajectory_text(stderr)) diff --git a/src/benchflow/sandbox/process/_base.py b/src/benchflow/sandbox/process/_base.py index a875c24be..86e94c595 100644 --- a/src/benchflow/sandbox/process/_base.py +++ b/src/benchflow/sandbox/process/_base.py @@ -31,6 +31,8 @@ class with ``_process = None # Not used`` plus a full override of _BUFFER_LIMIT = 10 * 1024 * 1024 # 10MB readline buffer _DIAG_TRUNCATE = 2000 # max chars for diagnostic stderr in error messages +_STDERR_TAIL_LIMIT = 64 * 1024 # bounded stderr retained for rollout diagnostics +_STDERR_DRAIN_TIMEOUT_SEC = 2 _BOOTSTRAP_DONE = "__BENCHFLOW_BOOTSTRAP_DONE__" _ENV_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") @@ -136,6 +138,29 @@ class SubprocessLiveProcess(LiveProcess): _process: asyncio.subprocess.Process | None = None + def _set_process(self, process: asyncio.subprocess.Process) -> None: + """Store a subprocess and drain stderr without blocking its stdout pipe.""" + self._process = process + self._stderr_tail = bytearray() + self._stderr_task = ( + asyncio.create_task(self._drain_stderr(process.stderr)) + if isinstance(process.stderr, asyncio.StreamReader) + else None + ) + + async def _drain_stderr(self, stderr: asyncio.StreamReader | None) -> None: + if stderr is None: + return + while chunk := await stderr.read(8192): + self._stderr_tail.extend(chunk) + if len(self._stderr_tail) > _STDERR_TAIL_LIMIT: + del self._stderr_tail[:-_STDERR_TAIL_LIMIT] + + @property + def stderr_tail(self) -> str: + """Bounded stderr captured while the subprocess was alive.""" + return bytes(getattr(self, "_stderr_tail", b"")).decode(errors="replace") + async def readline(self) -> bytes: """Read one line from stdout.""" if not self._process or not self._process.stdout: @@ -149,8 +174,16 @@ async def readline(self) -> bytes: # Return empty line — caller will retry readline return b"" if not line: - stderr_text = "" - if self._process and self._process.stderr: + stderr_task = getattr(self, "_stderr_task", None) + if stderr_task: + with contextlib.suppress(TimeoutError): + await asyncio.wait_for( + asyncio.shield(stderr_task), timeout=_STDERR_DRAIN_TIMEOUT_SEC + ) + stderr_text = self.stderr_tail.strip() + else: + stderr_text = "" + if not stderr_task and self._process and self._process.stderr: try: stderr_bytes = await asyncio.wait_for( self._process.stderr.read(8192), timeout=2 @@ -179,7 +212,9 @@ async def readline(self) -> bytes: msg = f"Process closed stdout (rc={rc}): {hint}" stderr_snippet: str | None = None if stderr_text: - stderr_snippet = stderr_text[:_DIAG_TRUNCATE] + from benchflow.trajectories.types import redact_trajectory_text + + stderr_snippet = redact_trajectory_text(stderr_text)[:_DIAG_TRUNCATE] msg += f"\nstderr: {stderr_snippet}" # Raise a structured TransportClosedError at the source so # downstream code (rollout._build_rollout_result) doesn't have @@ -222,6 +257,16 @@ async def close(self) -> None: except TimeoutError: self._process.kill() await self._process.wait() + stderr_task = getattr(self, "_stderr_task", None) + if stderr_task: + try: + await asyncio.wait_for( + asyncio.shield(stderr_task), timeout=_STDERR_DRAIN_TIMEOUT_SEC + ) + except TimeoutError: + stderr_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await stderr_task logger.info("Process terminated") @property diff --git a/src/benchflow/sandbox/process/apple.py b/src/benchflow/sandbox/process/apple.py index 959c5f434..19363edff 100644 --- a/src/benchflow/sandbox/process/apple.py +++ b/src/benchflow/sandbox/process/apple.py @@ -110,13 +110,14 @@ async def start( args.extend(["--workdir", cwd]) args.extend([self._container_name, "bash", "-c", command]) try: - self._process = await asyncio.create_subprocess_exec( + process = await asyncio.create_subprocess_exec( *args, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, limit=_BUFFER_LIMIT, ) + self._set_process(process) except BaseException: if env: try: @@ -129,6 +130,6 @@ async def start( raise logger.info( "Apple Container process started (pid=%s, container=%s)", - self._process.pid, + process.pid, self._container_name, ) diff --git a/src/benchflow/sandbox/process/daytona.py b/src/benchflow/sandbox/process/daytona.py index 4a504d086..216c96071 100644 --- a/src/benchflow/sandbox/process/daytona.py +++ b/src/benchflow/sandbox/process/daytona.py @@ -360,22 +360,23 @@ async def start( "DaytonaProcess: ssh benchflow-daytona %s...", remote_cmd[:100], ) - self._process = await asyncio.create_subprocess_exec( + process = await asyncio.create_subprocess_exec( *cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, limit=_BUFFER_LIMIT, ) + self._set_process(process) except Exception: if remote_env_path: await self._cleanup_remote_env_file(remote_env_path) await self.close() raise self._ssh_config_cleanup_task = asyncio.create_task( - self._cleanup_ssh_config_after_exit(self._process, ssh_config_path) + self._cleanup_ssh_config_after_exit(process, ssh_config_path) ) - logger.info(f"Daytona process started (pid={self._process.pid})") + logger.info(f"Daytona process started (pid={process.pid})") async def close(self) -> None: try: diff --git a/src/benchflow/sandbox/process/docker.py b/src/benchflow/sandbox/process/docker.py index cc1de92f0..64f321643 100644 --- a/src/benchflow/sandbox/process/docker.py +++ b/src/benchflow/sandbox/process/docker.py @@ -161,7 +161,7 @@ async def start( logger.debug(f"DockerProcess: {' '.join(cmd[:10])}...") try: - self._process = await asyncio.create_subprocess_exec( + process = await asyncio.create_subprocess_exec( *cmd, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, @@ -169,6 +169,7 @@ async def start( env=proc_env, limit=_BUFFER_LIMIT, ) + self._set_process(process) except BaseException: if env: try: @@ -180,6 +181,5 @@ async def start( ) raise logger.info( - f"Docker process started (pid={self._process.pid}, " - f"project={self._project_name})" + f"Docker process started (pid={process.pid}, project={self._project_name})" ) diff --git a/tests/test_acp.py b/tests/test_acp.py index 783838a4a..c20819c61 100644 --- a/tests/test_acp.py +++ b/tests/test_acp.py @@ -512,6 +512,72 @@ async def test_container_transport_does_not_create_empty_agent_log( assert msg == {"jsonrpc": "2.0", "id": 2, "result": {"ok": True}} assert not agent_log.exists() + @pytest.mark.asyncio + async def test_container_transport_keeps_live_subprocess_stderr( + self, tmp_path + ) -> None: + """Guards PR #980 against losing stderr after a valid empty ACP result.""" + from benchflow.sandbox.process import SubprocessLiveProcess + + class _LocalProcess(SubprocessLiveProcess): + async def start(self, command, env=None, cwd=None) -> None: + self._set_process( + await asyncio.create_subprocess_exec( + sys.executable, + "-c", + "import sys, time; " + "sys.stderr.write('api_key=AIzaSy12345678901234567890\\n'); " + "sys.stderr.flush(); " + 'print(\'{\\"jsonrpc\\": \\"2.0\\", \\"id\\": 2, \\"result\\": {}}\', flush=True); ' + "time.sleep(60)", + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + ) + + agent_log = tmp_path / "agent.log" + transport = ContainerTransport( + _LocalProcess(), "agent acp", agent_log_path=agent_log + ) + await transport.start() + assert await transport.receive() == { + "jsonrpc": "2.0", + "id": 2, + "result": {}, + } + await transport.close() + assert "***REDACTED***" in agent_log.read_text() + assert "AIzaSy12345678901234567890" not in agent_log.read_text() + + @pytest.mark.asyncio + async def test_container_transport_stderr_tail_is_bounded(self, tmp_path) -> None: + """Guards PR #980 against unbounded live ACP stderr retention.""" + from benchflow.sandbox.process import SubprocessLiveProcess + + class _LocalProcess(SubprocessLiveProcess): + async def start(self, command, env=None, cwd=None) -> None: + self._set_process( + await asyncio.create_subprocess_exec( + sys.executable, + "-c", + "import sys, time; sys.stderr.write('x' * 70000); " + 'sys.stderr.flush(); print(\'{\\"jsonrpc\\": \\"2.0\\", \\"id\\": 2, \\"result\\": {}}\', flush=True); time.sleep(60)', + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + ) + + process = _LocalProcess() + transport = ContainerTransport( + process, "agent acp", agent_log_path=tmp_path / "agent.log" + ) + await transport.start() + await transport.receive() + await transport.close() + assert len(process.stderr_tail.encode()) == 64 * 1024 + @pytest.mark.asyncio async def test_container_transport_clears_stale_log_on_retry( self, tmp_path @@ -1542,7 +1608,7 @@ async def readline(self) -> bytes: return b"" async def read(self, _n: int) -> bytes: - return b"Connection to sandbox lost" + return b"api_key=AIzaSy12345678901234567890" class _LP(SubprocessLiveProcess): async def start( @@ -1558,7 +1624,8 @@ async def start( assert diag.process_exit_code == 255 assert diag.transport_diagnosis == "process_exited" assert diag.stderr_snippet is not None - assert "Connection to sandbox lost" in diag.stderr_snippet + assert "***REDACTED***" in diag.stderr_snippet + assert "AIzaSy12345678901234567890" not in str(exc_info.value) @pytest.mark.asyncio async def test_live_process_raises_typed_transport_error_when_remote_killed( diff --git a/tests/test_process.py b/tests/test_process.py index e108b0a9e..f6bec4a0c 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -17,6 +17,37 @@ ) +@pytest.mark.asyncio +async def test_subprocess_close_cancels_stderr_drain_that_never_reaches_eof( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Guards PR #980 against waiting forever for inherited stderr FDs.""" + from benchflow.sandbox.process import SubprocessLiveProcess + + class _Process: + returncode = None + stdin = None + + def terminate(self) -> None: + self.returncode = 0 + + async def wait(self) -> None: + return None + + class _LiveProcess(SubprocessLiveProcess): + async def start(self, command, env=None, cwd=None) -> None: + pass + + process = _LiveProcess() + process._process = _Process() # type: ignore[assignment] + process._stderr_task = asyncio.create_task(asyncio.Event().wait()) + monkeypatch.setattr( + "benchflow.sandbox.process._base._STDERR_DRAIN_TIMEOUT_SEC", 0.01 + ) + await process.close() + assert process._stderr_task.cancelled() + + class _FakeStdin: def __init__(self): self.writes = []