fix(v1): bound a user-sim respond attempt so its retry can actually fire

The intermittent test_user CI stall (and a 1-in-20 local repro): the drive
loop's `session.user(...)` rode the MCP transport's 600s read timeout, so one
wedged connection — accepted, then silent — silently absorbed the whole
harness window (stop_condition=harness_timeout, errors=[], turn never
arrives). The per-attempt retry on a fresh session, and the server's
(seq, message) replay cache built exactly for retried turns, never got the
chance to recover it.

Each respond attempt (connect + initialize + call) is now bounded by
UserConfig.timeout (default 60s, raisable for model-backed simulators whose
turns legitimately run long); a timed-out attempt retries on a fresh session.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: hallerite

tests/v1/test_retries.py

@@ -72,3 +72,35 @@ async def attempt() -> Episode:
     episode = await run_episode_with_retry(masked_then_good, retry)
     assert calls == 2
     assert episode.ok
+
+
+async def test_user_respond_attempt_is_time_bounded(monkeypatch):
+    """A wedged user-server connection (accepted, then silent) fails its attempt at
+    the configured bound instead of sitting in the transport's read timeout, so the
+    retry — on a fresh session — recovers the turn within the harness window."""
+    import asyncio
+    import contextlib
+    import json
+    from types import SimpleNamespace
+
+    from verifiers.v1.mcp import launch
+
+    attempts = 0
+
+    @contextlib.asynccontextmanager
+    async def wedged_then_good(url):
+        nonlocal attempts
+        attempts += 1
+
+        async def call_tool(name, args):
+            if attempts == 1:
+                await asyncio.Event().wait()  # the wedge: connected, never answers
+            payload = json.dumps({"messages": [{"role": "user", "content": "next"}]})
+            return SimpleNamespace(content=[SimpleNamespace(type="text", text=payload)])
+
+        yield SimpleNamespace(call_tool=call_tool)
+
+    monkeypatch.setattr(launch, "user_session", wedged_then_good)
+    messages = await launch.user_respond("http://sim", "hi", 1, timeout=0.05)
+    assert attempts == 2
+    assert [m.content for m in messages] == ["next"]

verifiers/v1/mcp/launch.py

@@ -465,10 +465,12 @@ async def user_session(url: str) -> AsyncIterator[ClientSession]:
             await stack.aclose()
 
 
-async def user_respond(url: str, message: str, seq: int) -> Messages:
+async def user_respond(url: str, message: str, seq: int, *, timeout: float) -> Messages:
     """One `respond` turn against the user server, on a fresh session per attempt. A retried turn
     whose response was lost would advance the simulator twice, so the server dedups on
     (`seq`, `message`) and replays the recorded turn — making the retry effectively exactly-once.
+    `timeout` bounds each attempt (connect + initialize + call): a wedged connection must fail
+    fast enough for a retry to land inside the harness window instead of silently absorbing it.
     The payload is parsed outside the retry so a parse failure fails once."""
     from verifiers.v1.dialects import parse_message
     from verifiers.v1.retries import retrying
@@ -481,10 +483,11 @@ async def user_respond(url: str, message: str, seq: int) -> Messages:
             label=f"user respond ({url})",
         ):
             with attempt:
-                async with user_session(url) as session:
-                    result = await session.call_tool(
-                        "respond", {"message": message, "seq": seq}
-                    )
+                async with asyncio.timeout(timeout):
+                    async with user_session(url) as session:
+                        result = await session.call_tool(
+                            "respond", {"message": message, "seq": seq}
+                        )
         assert result is not None
         texts = [b.text for b in result.content if getattr(b, "type", None) == "text"]
         data = json.loads("\n".join(texts))
@@ -520,4 +523,4 @@ async def serve_user(
         state_secret=state_secret,
         state_base=state_base,
     ) as url:
-        yield partial(user_respond, url)
+        yield partial(user_respond, url, timeout=user.config.timeout)

verifiers/v1/mcp/user.py

@@ -30,6 +30,11 @@ class UserConfig(BaseConfig):
 
     colocated: bool = False
     runtime: RuntimeConfig = SubprocessConfig()
+    timeout: float = 60.0
+    """Bound on one `respond()` attempt (connect + call). A wedged connection fails the attempt
+    so the host's retry can recover it within the harness window; the turn's worst case is
+    this times the retry budget. Raise it for simulators whose turns legitimately run long
+    (e.g. model-backed users)."""
 
 
 ConfigT = TypeVar("ConfigT", bound=UserConfig)