fix(sandbox): keep timeout teardown from erasing the timeout

src/benchflow/acp/transport.py

@@ -1,6 +1,7 @@
 """ACP transports — stdio and SSE."""
 
 import asyncio
+import contextlib
 import json
 import logging
 from abc import ABC, abstractmethod
@@ -133,13 +134,28 @@ async def receive(self) -> dict[str, Any]:
             logger.debug(f"Non-JSON-RPC line from agent: {text[:200]}")
 
     async def close(self) -> None:
+        """Stop the agent process. Safe to call after it has already exited.
+
+        Teardown usually runs while another exception is in flight, so this
+        must not raise one of its own. An agent that already exited — crashed,
+        or finished on its own — has been reaped by asyncio, and signalling a
+        reaped child raises ``ProcessLookupError`` with no args, which would
+        both mask the real failure and print as an empty message (#1065). The
+        equivalent ``SubprocessLiveProcess.close()`` makes the same promise.
+        """
         if self._process:
             if self._process.stdin:
-                self._process.stdin.close()
-            self._process.terminate()
-            try:
-                await asyncio.wait_for(self._process.wait(), timeout=5)
-            except TimeoutError:
-                self._process.kill()
-                await self._process.wait()
+                with contextlib.suppress(OSError):
+                    self._process.stdin.close()
+            if self._process.returncode is None:
+                with contextlib.suppress(ProcessLookupError):
+                    self._process.terminate()
+                try:
+                    await asyncio.wait_for(self._process.wait(), timeout=5)
+                except TimeoutError:
+                    # The grace period is an await, so the child may exit
+                    # between the terminate above and this escalation.
+                    with contextlib.suppress(ProcessLookupError):
+                        self._process.kill()
+                    await self._process.wait()
             logger.info("Agent process terminated")

src/benchflow/rollout/__init__.py

@@ -1903,7 +1903,9 @@ async def soft_verify(self) -> tuple[dict | None, str | None, str | None]:
                 user="root",
             )
         except Exception as e:
-            verifier_error = f"soft verifier crashed: {e}"
+            # describe_exception for the same reason as the funnel below: an
+            # argument-less exception stringifies to nothing (#1065).
+            verifier_error = f"soft verifier crashed: {describe_exception(e)}"
             logger.error(verifier_error)
             return None, None, verifier_error
 
@@ -1938,7 +1940,7 @@ async def soft_verify(self) -> tuple[dict | None, str | None, str | None]:
             )
             logger.error(verifier_error)
         except Exception as e:
-            verifier_error = f"soft verifier crashed: {e}"
+            verifier_error = f"soft verifier crashed: {describe_exception(e)}"
             logger.error(verifier_error)
         return rewards, verifier_output, verifier_error
 

src/benchflow/rollout/_setup.py

@@ -33,6 +33,7 @@
 from pathlib import Path, PurePosixPath
 from typing import Any
 
+from benchflow._utils.text import describe_exception
 from benchflow.contracts import RolloutPlanes, default_rollout_planes
 from benchflow.diagnostics import VerifierTimeoutDiagnostic
 from benchflow.environment.manifest import EnvironmentManifest
@@ -516,7 +517,13 @@ async def _verify_rollout(
         logger.error(verifier_error)
     except Exception as e:
         timing["verifier"] = (datetime.now() - t0).total_seconds()
-        verifier_error = f"verifier crashed: {e}"
+        # describe_exception, not str(e), for the reason the agent-side funnel
+        # already documents: an exception raised with no args stringifies to
+        # nothing, so the recorded error names neither the failure nor the fact
+        # that it had no detail. A teardown ProcessLookupError landing here is
+        # exactly that shape, and it arrives in place of the timeout it
+        # displaced (#1065).
+        verifier_error = f"verifier crashed: {describe_exception(e)}"
         rewards = None
         logger.error(verifier_error)
     return rewards, verifier_error, verifier_timeout

src/benchflow/sandbox/apple_container.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import os
 import platform
@@ -144,7 +145,11 @@ async def _run_cli(
             proc.communicate(input=stdin_data), timeout=timeout
         )
     except TimeoutError:
-        proc.kill()
+        # A child that finished inside the timeout window is already reaped,
+        # and killing it would raise an empty ProcessLookupError in place of
+        # the TimeoutError this re-raises (#1065).
+        with contextlib.suppress(ProcessLookupError):
+            proc.kill()
         await proc.wait()
         raise
     return ExecResult(

src/benchflow/sandbox/docker.py

@@ -85,6 +85,35 @@ def _is_compose_up_network_race_error(message: str) -> bool:
     return is_compose_up_network_race_error(message)
 
 
+_TIMEOUT_TEARDOWN_GRACE_SEC = 5
+
+
+async def _drain_timed_out_process(
+    process: asyncio.subprocess.Process,
+) -> tuple[bytes | None, bytes | None]:
+    """Stop a child that overran its timeout and return whatever it emitted.
+
+    Signalling races the child's own exit: ``asyncio`` reaps as soon as the
+    process ends, and ``terminate``/``kill`` on a reaped child raise
+    ``ProcessLookupError`` — unlike ``subprocess.Popen``, which polls first and
+    swallows the same race (CPython bpo-38630, bpo-40550). Callers here are in
+    ``except TimeoutError`` blocks about to raise a description of the timeout,
+    so an escaping ``ProcessLookupError`` would replace that description with an
+    exception carrying no args at all, and ``_verify_rollout`` would record the
+    rollout as ``verifier crashed:`` with nothing after the colon (#1065).
+    """
+    with contextlib.suppress(ProcessLookupError):
+        process.terminate()
+    try:
+        return await asyncio.wait_for(
+            process.communicate(), timeout=_TIMEOUT_TEARDOWN_GRACE_SEC
+        )
+    except TimeoutError:
+        with contextlib.suppress(ProcessLookupError):
+            process.kill()
+        return await process.communicate()
+
+
 class DockerSandboxEnvVars(BaseModel):
     main_image_name: str
     context_dir: str
@@ -365,14 +394,7 @@ async def _run_docker_compose_command(
             else:
                 stdout_bytes, stderr_bytes = await process.communicate()
         except TimeoutError:
-            process.terminate()
-            try:
-                stdout_bytes, stderr_bytes = await asyncio.wait_for(
-                    process.communicate(), timeout=5
-                )
-            except TimeoutError:
-                process.kill()
-                stdout_bytes, stderr_bytes = await process.communicate()
+            await _drain_timed_out_process(process)
             raise RuntimeError(
                 f"Command timed out after {timeout_sec} seconds"
             ) from None
@@ -417,14 +439,7 @@ async def _run_pre_compose_hook(self) -> None:
                 process.communicate(), timeout=timeout_sec
             )
         except TimeoutError:
-            process.terminate()
-            try:
-                stdout_bytes, _ = await asyncio.wait_for(
-                    process.communicate(), timeout=5
-                )
-            except TimeoutError:
-                process.kill()
-                stdout_bytes, _ = await process.communicate()
+            stdout_bytes, _ = await _drain_timed_out_process(process)
             output = stdout_bytes.decode(errors="replace") if stdout_bytes else ""
             raise RuntimeError(
                 f"Pre-compose hook timed out after {timeout_sec} seconds for "

src/benchflow/sandbox/process/_base.py

@@ -268,11 +268,16 @@ async def close(self) -> None:
                 with contextlib.suppress(OSError):  # already closed
                     self._process.stdin.close()
             if self._process.returncode is None:
-                self._process.terminate()
+                with contextlib.suppress(ProcessLookupError):
+                    self._process.terminate()
                 try:
                     await asyncio.wait_for(self._process.wait(), timeout=5)
                 except TimeoutError:
-                    self._process.kill()
+                    # The returncode check above cannot cover this branch: the
+                    # grace period is an await, so the child may exit and be
+                    # reaped before the escalation lands (#1065).
+                    with contextlib.suppress(ProcessLookupError):
+                        self._process.kill()
                     await self._process.wait()
             await self._finish_stderr_drain(cancel_on_timeout=True)
             logger.info("Process terminated")

src/benchflow/sandbox/process/apple.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import logging
 import shlex
 import uuid
@@ -63,7 +64,10 @@ async def _write_env_to_container(self, env: dict[str, str]) -> None:
                 proc.communicate(lines.encode()), timeout=30
             )
         except TimeoutError:
-            proc.kill()
+            # Killing a child that finished inside the timeout window raises an
+            # empty ProcessLookupError over the TimeoutError below (#1065).
+            with contextlib.suppress(ProcessLookupError):
+                proc.kill()
             await proc.wait()
             raise
         if proc.returncode != 0: