diff --git a/src/benchflow/rollout/_setup.py b/src/benchflow/rollout/_setup.py index 33c95c7e8..29f4c1b11 100644 --- a/src/benchflow/rollout/_setup.py +++ b/src/benchflow/rollout/_setup.py @@ -437,7 +437,14 @@ async def _publish_trajectory_for_verifier( payload = redact_acp_trajectory_jsonl(trajectory) + "\n" agent_dir.mkdir(parents=True, exist_ok=True) (agent_dir / "acp_trajectory.jsonl").write_text(payload) - await env.exec("mkdir -p /logs/agent", user="root", timeout_sec=10) + # Only defensive: mounted backends already have /logs/agent from the bind + # mount, and remote ones fail loudly on the upload below if it is missing. + # Publishing runs after the agent finished, so a slow or failing exec here + # must not discard an otherwise complete rollout. + try: + await env.exec("mkdir -p /logs/agent", user="root", timeout_sec=10) + except Exception as exc: + logger.warning(f"Could not pre-create /logs/agent, publishing anyway: {exc}") with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f: f.write(payload) tmp_path = f.name diff --git a/src/benchflow/trajectories/_capture.py b/src/benchflow/trajectories/_capture.py index 0f4ea02a6..f4af8b1a4 100644 --- a/src/benchflow/trajectories/_capture.py +++ b/src/benchflow/trajectories/_capture.py @@ -230,10 +230,17 @@ async def _scrape_agent_trajectory( # Gemini CLI: writes ~/.gemini/sessions/*/gemini-cli.trajectory.json if "gemini" in agent: - result = await env.exec( - f"cat $(find {home}/.gemini -name 'gemini-cli.trajectory.json' 2>/dev/null | head -1) 2>/dev/null", - timeout_sec=10, - ) + # A non-zero return code and unparseable JSON both degrade to "no + # scraped trajectory" below, so a slow container must not be the one + # case that propagates and aborts verification instead (#948). + try: + result = await env.exec( + f"cat $(find {home}/.gemini -name 'gemini-cli.trajectory.json' 2>/dev/null | head -1) 2>/dev/null", + timeout_sec=10, + ) + except Exception as e: + logger.warning(f"Could not scrape gemini trajectory: {e}") + return [] if result.return_code == 0 and result.stdout and result.stdout.strip(): try: return _parse_gemini_trajectory(json.loads(result.stdout)) diff --git a/tests/test_capture_trajectory.py b/tests/test_capture_trajectory.py index eac6622f6..857861c3e 100644 --- a/tests/test_capture_trajectory.py +++ b/tests/test_capture_trajectory.py @@ -6,12 +6,15 @@ import json +import pytest + from benchflow.acp.session import ACPSession from benchflow.acp.types import ToolCallStatus from benchflow.trajectories._capture import ( _capture_session_trajectory, _parse_provider_tool_evidence, _reconcile_tool_evidence, + _scrape_agent_trajectory, ) from benchflow.trajectories.types import LLMExchange, LLMRequest, LLMResponse @@ -897,3 +900,20 @@ def test_pending_cleared_between_flushes(self) -> None: ] assert result[1]["text"] == "before tool" assert result[3]["text"] == "after tool" + + +@pytest.mark.asyncio +async def test_scrape_agent_trajectory_survives_exec_timeout(): + """Guards issue #948 at the sibling call site that runs first. + + ``verify()`` awaits the scrape before publishing, so a container too slow + for the 10 second budget aborted the rollout here before the publish-side + guard could apply. Every other failure in this fallback already degrades to + "no scraped trajectory"; a timeout must do the same. + """ + + class TimingOutEnv: + async def exec(self, command, user=None, timeout_sec=None): + raise RuntimeError(f"Command timed out after {timeout_sec} seconds") + + assert await _scrape_agent_trajectory(TimingOutEnv(), "gemini", None) == [] diff --git a/tests/test_rollout_upload.py b/tests/test_rollout_upload.py index 67be7a659..62f7b2e1f 100644 --- a/tests/test_rollout_upload.py +++ b/tests/test_rollout_upload.py @@ -381,6 +381,63 @@ async def test_publish_trajectory_for_verifier_uploads_acp_jsonl( ] +class FakeTimingOutMkdirEnv(FakeUploadEnv): + """Loaded backend whose bookkeeping ``exec`` blows its client-side timeout.""" + + async def exec( + self, command: str, user: str | None = None, timeout_sec: int | None = None + ) -> None: + await super().exec(command, user, timeout_sec) + raise RuntimeError(f"Command timed out after {timeout_sec} seconds") + + +@pytest.mark.asyncio +async def test_publish_trajectory_survives_mkdir_timeout(tmp_path: Path) -> None: + """Guards issue #948: a timed-out ``mkdir -p /logs/agent`` discarded rollouts. + + Publishing runs after the agent already finished, so a client-side timeout + on that bookkeeping exec used to throw away a completed rollout — a ~40 + minute run lost to a 10 second budget on a step that does no work on + mounted backends. Both trajectory copies must still be published. + """ + env = FakeTimingOutMkdirEnv() + agent_dir = tmp_path / "agent" + + await _publish_trajectory_for_verifier( + env, [{"type": "agent_message", "text": "ok"}], agent_dir + ) + + expected = '{"type": "agent_message", "text": "ok"}\n' + assert ("mkdir -p /logs/agent", "root", 10) in env.exec_calls + assert env.uploaded_file_contents == [ + (expected, "/logs/agent/acp_trajectory.jsonl") + ] + assert (agent_dir / "acp_trajectory.jsonl").read_text() == expected + + +@pytest.mark.asyncio +async def test_publish_trajectory_still_raises_on_upload_failure( + tmp_path: Path, +) -> None: + """Guards issue #948 against over-suppression: only the mkdir is defensive. + + A genuinely missing ``/logs/agent`` surfaces through the upload, which must + stay fatal — otherwise the rollout would claim a trajectory the verifier + never received. + """ + + class FailingUploadEnv(FakeTimingOutMkdirEnv): + async def upload_file(self, source: Path | str, target: str) -> None: + raise RuntimeError("upload boom") + + with pytest.raises(RuntimeError, match="upload boom"): + await _publish_trajectory_for_verifier( + FailingUploadEnv(), + [{"type": "agent_message", "text": "ok"}], + tmp_path / "agent", + ) + + class FakeMountedUploadEnv(FakeUploadEnv): """Docker-like backend: ``/logs/agent`` is bind-mounted to the host agent dir."""