From faec70071188a5ed8bb519254745ec07595a259d Mon Sep 17 00:00:00 2001 From: tulerfeng <1042914391@qq.com> Date: Tue, 1 Sep 2026 03:55:21 +0800 Subject: [PATCH] fix(sandbox): keep timeout teardown from erasing the timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signalling a child that asyncio has already reaped raises ProcessLookupError, which escapes the `except TimeoutError` handler before the RuntimeError describing the timeout is ever raised. That exception carries no args, so `f"verifier crashed: {e}"` renders with nothing after the colon: a finished rollout is scored `rewards: null` under a message naming neither the failure nor the fact that it had no detail. The window is one loop tick wide — asyncio clears its process handle from a `call_soon` callback — so it is rare on an idle loop and routine under concurrency. `subprocess.Popen.send_signal` already absorbs this race (bpo-38630, bpo-40550), which is why only the asyncio call sites are exposed. Guard all five of them, and render verifier errors through the existing `describe_exception` helper that the agent-side funnel already uses. Fixes #1065 --- src/benchflow/acp/transport.py | 30 +- src/benchflow/rollout/__init__.py | 6 +- src/benchflow/rollout/_setup.py | 9 +- src/benchflow/sandbox/apple_container.py | 7 +- src/benchflow/sandbox/docker.py | 47 +-- src/benchflow/sandbox/process/_base.py | 9 +- src/benchflow/sandbox/process/apple.py | 6 +- tests/test_timeout_teardown_reaped_child.py | 299 ++++++++++++++++++++ 8 files changed, 383 insertions(+), 30 deletions(-) create mode 100644 tests/test_timeout_teardown_reaped_child.py diff --git a/src/benchflow/acp/transport.py b/src/benchflow/acp/transport.py index 99a3a8778..d51006c60 100644 --- a/src/benchflow/acp/transport.py +++ b/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") diff --git a/src/benchflow/rollout/__init__.py b/src/benchflow/rollout/__init__.py index 8e73b0f96..2a785c39c 100644 --- a/src/benchflow/rollout/__init__.py +++ b/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 diff --git a/src/benchflow/rollout/_setup.py b/src/benchflow/rollout/_setup.py index 33c95c7e8..8616275be 100644 --- a/src/benchflow/rollout/_setup.py +++ b/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 diff --git a/src/benchflow/sandbox/apple_container.py b/src/benchflow/sandbox/apple_container.py index 793e61db7..4139a1d7c 100644 --- a/src/benchflow/sandbox/apple_container.py +++ b/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( diff --git a/src/benchflow/sandbox/docker.py b/src/benchflow/sandbox/docker.py index 4ae9f362d..4f95f816b 100644 --- a/src/benchflow/sandbox/docker.py +++ b/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 " diff --git a/src/benchflow/sandbox/process/_base.py b/src/benchflow/sandbox/process/_base.py index 974102213..57b6dd430 100644 --- a/src/benchflow/sandbox/process/_base.py +++ b/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") diff --git a/src/benchflow/sandbox/process/apple.py b/src/benchflow/sandbox/process/apple.py index 19363edff..377ada722 100644 --- a/src/benchflow/sandbox/process/apple.py +++ b/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: diff --git a/tests/test_timeout_teardown_reaped_child.py b/tests/test_timeout_teardown_reaped_child.py new file mode 100644 index 000000000..20dbb09cd --- /dev/null +++ b/tests/test_timeout_teardown_reaped_child.py @@ -0,0 +1,299 @@ +"""Timeout teardown must survive a child that exits inside the timeout window (#1065). + +Every ``except TimeoutError`` teardown in the codebase signals the child before +reporting the timeout. ``terminate()``/``kill()`` raise ``ProcessLookupError`` +once the child has exited and been reaped, and asyncio reaps as soon as the +process ends — so the signal is a race against a child that may already be gone. + +Losing that race is not a cosmetic failure. ``ProcessLookupError`` escapes the +handler, the ``RuntimeError("Command timed out ...")`` underneath never runs, +and the caller sees an exception whose ``args`` are empty. ``_verify_rollout`` +renders it with ``f"verifier crashed: {e}"``, so a finished rollout is scored +``rewards: null`` under a message that ends at the colon, with the one fact that +would explain it — that this was a timeout — destroyed on the way out. + +The race is timing-dependent in production. These tests reproduce it +deterministically: the child is genuinely spawned, exited and reaped, and +``communicate`` raises ``TimeoutError`` instead of sleeping, so each test +exercises the real teardown sequence without waiting on a real clock or +gambling on a scheduling coin flip. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch + +import pytest + +# Captured before any monkeypatching so the helpers below can still spawn while +# the code under test sees a patched ``create_subprocess_exec``. +_spawn = asyncio.create_subprocess_exec + + +async def _reaped_child() -> asyncio.subprocess.Process: + """A real, already-exited, already-reaped asyncio child process. + + Not a mock: ``ProcessLookupError`` here is raised by the OS through + asyncio's own transport, so these tests keep pinning real behavior rather + than a fixture's idea of it. + """ + process = await _spawn( + "sh", + "-c", + "exit 0", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + await process.communicate() + return process + + +def _patch_spawn_with_timed_out_child(monkeypatch, *, grace_expires: bool = False): + """Make every spawn yield a reaped child whose ``communicate`` times out. + + ``grace_expires`` extends the timeout to the post-``terminate`` drain as + well, which is what pushes the teardown on to ``kill()`` — the second half + of the same race. + """ + timeouts = 2 if grace_expires else 1 + + async def spawn(*_args, **_kwargs): + process = await _reaped_child() + calls = {"n": 0} + + async def communicate(*_a, **_k): + calls["n"] += 1 + if calls["n"] <= timeouts: + raise TimeoutError + return (b"", b"") + + # wraps= keeps attribute access working; the process-control methods + # are bound to the real child so the signals hit a real reaped pid. + wrapper = MagicMock(wraps=process) + wrapper.communicate = communicate + wrapper.terminate = process.terminate + wrapper.kill = process.kill + wrapper.wait = process.wait + wrapper.returncode = process.returncode + return wrapper + + monkeypatch.setattr(asyncio, "create_subprocess_exec", spawn) + + +class TestReapedChildPremise: + """The premise the rest of the module depends on.""" + + @pytest.mark.asyncio + async def test_signalling_a_reaped_child_raises_an_empty_error(self) -> None: + process = await _reaped_child() + + with pytest.raises(ProcessLookupError) as excinfo: + process.terminate() + + # Empty args are what make this failure mode invisible downstream: the + # generic handler in _verify_rollout formats the exception, not its type. + assert excinfo.value.args == () + assert str(excinfo.value) == "" + assert f"verifier crashed: {excinfo.value}" == "verifier crashed: " + + +class TestDockerComposeExecTimeout: + """``DockerSandbox.exec`` — the path that carries the hardening execs.""" + + @staticmethod + def _sandbox(tmp_path: Path): + from benchflow.sandbox.docker import DockerSandbox + + sandbox = DockerSandbox.__new__(DockerSandbox) + sandbox.session_id = "teardown-race" + sandbox.environment_dir = tmp_path + return sandbox + + @staticmethod + def _stub_compose_config(): + from benchflow.sandbox.docker import DockerSandbox + + return ( + patch.object( + DockerSandbox, + "_docker_compose_paths", + new_callable=PropertyMock, + return_value=[], + ), + patch.object(DockerSandbox, "_docker_compose_env", return_value={}), + ) + + @pytest.mark.asyncio + async def test_timeout_is_reported_as_a_timeout(self, tmp_path, monkeypatch): + """The RuntimeError must survive a terminate() that finds no child.""" + _patch_spawn_with_timed_out_child(monkeypatch) + sandbox = self._sandbox(tmp_path) + paths, env = self._stub_compose_config() + + with paths, env, pytest.raises(RuntimeError, match="timed out after"): + await sandbox._run_docker_compose_command( + ["exec", "-T", "main", "sh", "-c", "printenv PATH"], + check=False, + timeout_sec=1, + ) + + @pytest.mark.asyncio + async def test_timeout_survives_a_racing_kill(self, tmp_path, monkeypatch): + """The escalation path — grace period elapses, then kill() — races too.""" + _patch_spawn_with_timed_out_child(monkeypatch, grace_expires=True) + sandbox = self._sandbox(tmp_path) + paths, env = self._stub_compose_config() + + with paths, env, pytest.raises(RuntimeError, match="timed out after"): + await sandbox._run_docker_compose_command( + ["exec", "-T", "main", "sh", "-c", "printenv PATH"], + check=False, + timeout_sec=1, + ) + + +class TestPreComposeHookTimeout: + """The pre-compose hook repeats the teardown, so it repeats the race.""" + + @pytest.mark.asyncio + @pytest.mark.parametrize("grace_expires", [False, True]) + async def test_hook_timeout_is_reported_as_a_timeout( + self, tmp_path, monkeypatch, grace_expires + ): + from benchflow.sandbox.docker import DockerSandbox + + hook = tmp_path / "pre_compose.sh" + hook.write_text("#!/bin/sh\nexit 0\n") + + sandbox = DockerSandbox.__new__(DockerSandbox) + sandbox.environment_dir = tmp_path + sandbox.environment_name = "teardown-race" + sandbox.task_env_config = MagicMock(build_timeout_sec=1) + + _patch_spawn_with_timed_out_child(monkeypatch, grace_expires=grace_expires) + + with ( + patch.object( + DockerSandbox, + "_pre_compose_hook_path", + new_callable=PropertyMock, + return_value=hook, + ), + patch.object(DockerSandbox, "_docker_compose_env", return_value={}), + pytest.raises(RuntimeError, match="timed out after"), + ): + await sandbox._run_pre_compose_hook() + + +class TestAppleContainerExecTimeout: + """The Apple backend re-raises the TimeoutError, so the race replaces it.""" + + @pytest.mark.asyncio + async def test_timeout_is_reraised_as_a_timeout(self, monkeypatch) -> None: + from benchflow.sandbox import apple_container + + _patch_spawn_with_timed_out_child(monkeypatch) + + with pytest.raises(TimeoutError): + await apple_container._run_cli("ls", timeout=1) + + @pytest.mark.asyncio + async def test_env_write_timeout_is_reraised_as_a_timeout(self, monkeypatch): + """The Apple LiveProcess env write repeats the same teardown.""" + from benchflow.sandbox.process.apple import AppleContainerProcess + + _patch_spawn_with_timed_out_child(monkeypatch) + + process = AppleContainerProcess.__new__(AppleContainerProcess) + process._container_name = "teardown-race" + process._env_path = "/tmp/agent-env" + + with pytest.raises(TimeoutError): + await process._write_env_to_container({"A": "b"}) + + +class TestVerifierErrorIsNeverDetailFree: + """A recorded verifier error must name something, whatever was raised. + + ``self._error = describe_exception(e)`` on the agent side already documents + why ``str(e)`` is not enough for a persisted artifact. The verifier side + used plain interpolation, so the exception that #1065 delivers — raised + with no args — was recorded as a bare prefix. + """ + + def test_describe_exception_names_an_argument_less_exception(self) -> None: + from benchflow._utils.text import describe_exception + + assert describe_exception(ProcessLookupError()) == ( + "ProcessLookupError (no message)" + ) + + @pytest.mark.asyncio + async def test_hardening_failure_is_recorded_with_its_type(self, tmp_path) -> None: + """The exact shape of #1065: hardening raises, scoring records it. + + ``harden_before_verify`` is where the timed-out execs live, so an + exception escaping it lands in the generic handler that writes + ``verifier_error``. It must not write a bare prefix. + """ + from benchflow.rollout._setup import _verify_rollout + + planes = MagicMock() + planes.harden_before_verify = AsyncMock(side_effect=ProcessLookupError()) + + rollout_paths = MagicMock() + rollout_paths.verifier_dir = tmp_path / "verifier" + + task = MagicMock() + task.config.verifier.timeout_sec = 60 + + rewards, verifier_error, verifier_timeout = await _verify_rollout( + env=MagicMock(), + task=task, + rollout_paths=rollout_paths, + timing={}, + planes=planes, + ) + + assert rewards is None + assert verifier_timeout is None + assert verifier_error == "verifier crashed: ProcessLookupError (no message)" + # The precise regression: a message that stops at the colon names + # neither the failure nor the fact that it carried no detail. + assert not verifier_error.endswith(": ") + + +class TestAcpTransportClose: + """Closing an ACP transport after the agent exits must not raise. + + ``close()`` runs in teardown, typically while another exception is in + flight, so raising here replaces the real failure with an empty + ``ProcessLookupError``. ``sandbox/process/_base.py`` already guards its + equivalent ``close()`` with a ``returncode`` check and documents it as + "safe to call after process death"; this asserts the ACP transport keeps + the same promise. + """ + + @pytest.mark.asyncio + async def test_close_after_the_agent_process_exits(self) -> None: + from benchflow.acp.transport import StdioTransport + + transport = StdioTransport.__new__(StdioTransport) + transport._process = await _reaped_child() + + await transport.close() + + @pytest.mark.asyncio + async def test_subprocess_live_process_close_after_death_stays_safe(self) -> None: + """Guard the neighbour that already behaves, so it cannot regress.""" + from benchflow.sandbox.process._base import SubprocessLiveProcess + + class _Concrete(SubprocessLiveProcess): + async def start(self, *args, **kwargs) -> None: ... + + process = _Concrete() + process._process = await _reaped_child() + + await process.close()