fix(rollout): route verifier timeout cleanup by service

src/benchflow/rollout/_setup.py

@@ -468,6 +468,7 @@ async def _verify_rollout(
     verifier_error = None
     verifier_timeout: VerifierTimeoutDiagnostic | None = None
     timeout_budget = task.config.verifier.timeout_sec
+    verifier_service = task.config.verifier.service
     try:
         await planes.harden_before_verify(env, task, sandbox_user, workspace=workspace)
         logger.info("Running verifier...")
@@ -491,12 +492,14 @@ async def _verify_rollout(
                 # retry is safe and turns a lost rollout into a real score. A
                 # timeout WITH output is a genuinely slow/hung verifier and is
                 # never retried.
-                if attempt == 1 and await _verifier_wedged_without_output(env):
+                if attempt == 1 and await _verifier_wedged_without_output(
+                    env, service=verifier_service
+                ):
                     logger.warning(
                         "Verifier timed out with no output — exec-layer wedge "
                         "suspected; retrying verifier once"
                     )
-                    await _kill_orphan_verifier(env)
+                    await _kill_orphan_verifier(env, service=verifier_service)
                     continue
                 raise
         assert verifier_result is not None
@@ -522,7 +525,7 @@ async def _verify_rollout(
     return rewards, verifier_error, verifier_timeout
 
 
-async def _verifier_wedged_without_output(env: Any) -> bool:
+async def _verifier_wedged_without_output(env: Any, *, service: str) -> bool:
     """True when the timed-out verifier attempt left no test output behind.
 
     Zero bytes in ``/logs/verifier/test-stdout.txt`` (or no file at all) means
@@ -535,6 +538,7 @@ async def _verifier_wedged_without_output(env: Any) -> bool:
             env.exec(
                 "wc -c < /logs/verifier/test-stdout.txt 2>/dev/null || echo 0",
                 timeout_sec=10,
+                service=service,
             ),
             timeout=30,
         )
@@ -543,7 +547,7 @@ async def _verifier_wedged_without_output(env: Any) -> bool:
         return True
 
 
-async def _kill_orphan_verifier(env: Any) -> None:
+async def _kill_orphan_verifier(env: Any, *, service: str) -> None:
     """Best-effort kill of a possibly-orphaned first verifier attempt.
 
     The host-side timeout cancels only our await; if the in-sandbox test.sh
@@ -555,6 +559,7 @@ async def _kill_orphan_verifier(env: Any) -> None:
                 "pkill -f '/verifier/test.sh' 2>/dev/null || true",
                 user="root",
                 timeout_sec=10,
+                service=service,
             ),
             timeout=30,
         )

tests/test_verifier_wedge_retry.py

@@ -20,12 +20,17 @@
 from benchflow.rollout._setup import _verify_rollout
 
 
-def _mk_task():
+def _mk_task(service: str = "main"):
     return SimpleNamespace(
         name="wedge-task",
         task_dir=None,
         config=SimpleNamespace(
-            verifier=SimpleNamespace(timeout_sec=0.2, reward_range=None, env={})
+            verifier=SimpleNamespace(
+                timeout_sec=0.2,
+                reward_range=None,
+                env={},
+                service=service,
+            )
         ),
     )
 
@@ -50,7 +55,9 @@ def _env_with_probe_output(stdout: str):
 
 
 @pytest.mark.asyncio
-async def test_zero_output_timeout_retries_once_and_scores(tmp_path):
+@pytest.mark.parametrize("verifier_service", ["main", "scorer"])
+async def test_zero_output_timeout_retries_once_and_scores(tmp_path, verifier_service):
+    """Retries a silent timeout on the selected verifier service."""
     attempts = []
 
     async def verify():
@@ -63,17 +70,26 @@ async def verify():
     env = _env_with_probe_output("0\n")  # probe: empty test-stdout
 
     rewards, verr, vtimeout = await _verify_rollout(
-        env, _mk_task(), _mk_paths(tmp_path), {}, _mk_planes(verifier)
+        env,
+        _mk_task(verifier_service),
+        _mk_paths(tmp_path),
+        {},
+        _mk_planes(verifier),
     )
 
     assert len(attempts) == 2
     assert rewards == {"reward": 1.0}
     assert verr is None
     assert vtimeout is None
+    assert [call.kwargs.get("service") for call in env.exec.await_args_list] == [
+        verifier_service,
+        verifier_service,
+    ]
 
 
 @pytest.mark.asyncio
 async def test_timeout_with_output_is_not_retried(tmp_path):
+    """Does not retry a timeout after the verifier produced scorer output."""
     attempts = []
 
     async def verify():
@@ -84,17 +100,19 @@ async def verify():
     env = _env_with_probe_output("4242\n")  # probe: real output was produced
 
     rewards, verr, vtimeout = await _verify_rollout(
-        env, _mk_task(), _mk_paths(tmp_path), {}, _mk_planes(verifier)
+        env, _mk_task("scorer"), _mk_paths(tmp_path), {}, _mk_planes(verifier)
     )
 
     assert len(attempts) == 1
     assert rewards is None
     assert verr is not None and "timed out" in verr
     assert vtimeout is not None
+    assert env.exec.await_args.kwargs.get("service") == "scorer"
 
 
 @pytest.mark.asyncio
 async def test_wedged_retry_that_also_times_out_reports_timeout(tmp_path):
+    """Reports a timeout when both verifier attempts time out."""
     attempts = []
 
     async def verify():