fix(rollout): keep completed rollouts when publish-path execs time out

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

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

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

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."""