fix(acp): retain live subprocess stderr

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))

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

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,
         )

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:

src/benchflow/sandbox/process/docker.py

@@ -161,14 +161,15 @@ 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,
                 stderr=asyncio.subprocess.PIPE,
                 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})"
         )

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(

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 = []