fix(v1): seal the session at unregister — no late mutation of a concluded trace

Caught live by the test_user flake hunt: a provider call that outlives the
harness window (208s observed against a 180s deadline) kept its aiohttp handler
running after the rollout was scored and persisted — aiohttp (3.9+) does not
cancel handlers on client death — and the late turn then committed onto the
concluded trace. The test saw num_turns=1 on an in-memory trace whose persisted
form has zero nodes; the straggler went on to dial the already-torn-down user
sim. Same seal gap for record_call, refused()'s trace.stop, and state PUTs.

unregister() now releases the session: in-flight handlers are cancelled, and
every mutation site (turn commit on both paths, call records, the drive loop,
state writes) refuses on a released session (409) — the in-memory trace stays
exactly what the run concluded with.

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

Author: hallerite

tests/v1/test_interception.py

@@ -0,0 +1,102 @@
+"""The v1 interception server's session seal: a rollout that concludes while an
+exchange is still upstream must not have its trace mutated by the straggler."""
+
+import asyncio
+from types import SimpleNamespace
+
+import httpx
+
+import verifiers.v1 as vf
+from verifiers.v1.interception.server import InterceptionServer
+from verifiers.v1.session import RolloutSession
+from verifiers.v1.trace import Trace, TraceTask
+
+
+class HangingClient:
+    """A model client whose call parks until cancelled — the slow-provider tail."""
+
+    def __init__(self):
+        self.entered = asyncio.Event()
+        self.cancelled = False
+
+    async def get_response(self, *args, **kwargs):
+        self.entered.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            self.cancelled = True
+            raise
+
+
+async def test_unregister_seals_the_trace_and_cancels_stragglers():
+    """The deadline path: the harness window closes (rollout unregisters) while the
+    model call is still upstream. The handler must be cancelled, and the sealed
+    trace must keep zero turns and zero call records — what was persisted is what
+    the in-memory trace still says."""
+    client = HangingClient()
+    trace = Trace(task=TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="hi")))
+    session = RolloutSession(
+        ctx=SimpleNamespace(client=client, model="m", sampling=vf.SamplingConfig()),
+        trace=trace,
+    )
+    server = InterceptionServer()
+    await server.start()
+    try:
+        secret = server.register(session)
+        async with httpx.AsyncClient() as http:
+            request = asyncio.ensure_future(
+                http.post(
+                    f"{server.base_url}/v1/chat/completions",
+                    json={
+                        "model": "m",
+                        "messages": [{"role": "user", "content": "hi"}],
+                    },
+                    headers={"Authorization": f"Bearer {secret}"},
+                    timeout=30,
+                )
+            )
+            await asyncio.wait_for(client.entered.wait(), 10)
+            assert session.tasks  # the handler adopted itself
+            server.unregister(secret)
+            # The straggler dies instead of finishing the exchange: either the fence
+            # answered 409 or the cancellation dropped the connection.
+            try:
+                response = await asyncio.wait_for(request, 10)
+                assert response.status_code == 409
+            except httpx.HTTPError:
+                pass
+        await asyncio.sleep(0)  # let cancellation unwind
+        assert client.cancelled
+        assert session.released
+        assert trace.nodes == [] and trace.calls == []  # the seal held
+        assert trace.stop_condition is None  # refused() never ran post-seal
+    finally:
+        await server.stack.aclose()
+
+
+async def test_released_session_refuses_new_exchanges():
+    """A request arriving for a concluded rollout is refused outright (409), it
+    never reaches the model client."""
+    client = HangingClient()
+    trace = Trace(task=TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="hi")))
+    session = RolloutSession(
+        ctx=SimpleNamespace(client=client, model="m", sampling=vf.SamplingConfig()),
+        trace=trace,
+    )
+    session.released = True
+    server = InterceptionServer()
+    await server.start()
+    try:
+        server.sessions["still-routed"] = session  # released but not yet unregistered
+        async with httpx.AsyncClient() as http:
+            response = await http.post(
+                f"{server.base_url}/v1/chat/completions",
+                json={"model": "m", "messages": [{"role": "user", "content": "hi"}]},
+                headers={"Authorization": "Bearer still-routed"},
+                timeout=10,
+            )
+        assert response.status_code == 409
+        assert not client.entered.is_set()
+        assert trace.nodes == [] and trace.calls == []
+    finally:
+        await server.stack.aclose()

verifiers/v1/interception/server.py

@@ -156,7 +156,12 @@ def register(self, session: RolloutSession) -> str:
         return secret
 
     def unregister(self, secret: str) -> None:
-        self.sessions.pop(secret, None)
+        session = self.sessions.pop(secret, None)
+        if session is not None:
+            # The rollout concluded; its trace is sealed. Cancel straggler handlers
+            # (aiohttp keeps them alive past client death) so a slow upstream call
+            # can't commit a late turn onto the concluded trace.
+            session.release()
 
     @asynccontextmanager
     async def acquire(self, session: RolloutSession) -> AsyncIterator[Slot]:
@@ -249,6 +254,10 @@ def record_call(
         the model + effective settings that went upstream, timing, and — when the call
         committed no turn — the error, coupled to the exchange that raised it. Called
         once per real exchange; replayed/coalesced SDK retries never reach it."""
+        if (
+            session.released
+        ):  # the trace is sealed — a straggler exchange isn't recorded
+            return
         sampling = None
         if request is not None:
             try:
@@ -291,6 +300,7 @@ async def handle_request(
         if session is None:
             logger.warning("interception: unauthorized request")
             return web.json_response(dialect.error_body("unauthorized"), status=401)
+        session.adopt(asyncio.current_task())
         raw = await request.read()
         try:
             body = from_json(raw)
@@ -386,6 +396,12 @@ def serve(response: Response) -> web.Response:
         )
         try:
             while True:
+                # The rollout may conclude (deadline, teardown) while this exchange was
+                # upstream: the trace is sealed, so drop the turn instead of mutating it.
+                if session.released:
+                    return web.json_response(
+                        dialect.error_body("rollout concluded"), status=409
+                    )
                 try:
                     refused = await session.refused()
                 except RolloutError as e:
@@ -433,6 +449,10 @@ def serve(response: Response) -> web.Response:
                             session.trace.id,
                             len(call_response.message.tool_calls or []),
                         )
+                        if session.released:  # concluded while sampling — seal holds
+                            return web.json_response(
+                                dialect.error_body("rollout concluded"), status=409
+                            )
                         # One node per new message; branches fall out of walking the
                         # graph (see Trace.branches / verifiers.v1.graph).
                         node = turn.commit(call_response, tools)
@@ -543,6 +563,10 @@ async def _stream(
         """A streamed (SSE) model turn: relay the provider's stream through to the program,
         incrementally assembling the response to record on the trace. Single-shot — a streamed
         turn never drives a user simulator (the only client that streams is the eval relay)."""
+        if session.released:  # concluded while this request queued — seal holds
+            return web.json_response(
+                dialect.error_body("rollout concluded"), status=409
+            )
         try:
             refused = await session.refused()
         except RolloutError as e:
@@ -667,8 +691,9 @@ async def _stream(
                 if parser_error is not None:
                     raise parser_error
                 response = parser.finish()
-                node = turn.commit(response, tools)
-                logger.debug("intercept stream turn: id=%s", session.trace.id)
+                if not session.released:  # concluded mid-stream — seal holds
+                    node = turn.commit(response, tools)
+                    logger.debug("intercept stream turn: id=%s", session.trace.id)
             finally:
                 # Release the withheld events only now — after the commit — then close.
                 with contextlib.suppress(ConnectionResetError):
@@ -705,6 +730,7 @@ async def handle_aux(
         session = self.sessions.get(dialect.secret(request.headers))
         if session is None:
             return web.json_response(dialect.error_body("unauthorized"), status=401)
+        session.adopt(asyncio.current_task())
         logger.debug("intercept aux %s: id=%s", route, session.trace.id)
         try:
             result = await session.ctx.client.relay_aux(
@@ -732,7 +758,10 @@ def _session_for(self, request: web.Request) -> RolloutSession | None:
         same per-rollout secret the model routes use (dialect-independent, so parsed directly)."""
         auth = request.headers.get("Authorization", "")
         secret = auth[len("Bearer ") :] if auth.startswith("Bearer ") else ""
-        return self.sessions.get(secret)
+        session = self.sessions.get(secret)
+        if session is not None:  # state writes must not land on a sealed trace either
+            session.adopt(asyncio.current_task())
+        return session
 
     async def handle_state_get(self, request: web.Request) -> web.Response:
         """Hand a rollout's tool/user server the current shared `trace.state` (it pulls before each
@@ -783,5 +812,7 @@ async def handle_state_put(self, request: web.Request) -> web.Response:
                 {"error": f"invalid state PUT for {state_cls.__name__}: {e}"},
                 status=400,
             )
+        if session.released:  # the trace is sealed — a straggler write must not land
+            return web.json_response({"error": "rollout concluded"}, status=409)
         session.trace.state = new_state
         return web.json_response({"ok": True})

verifiers/v1/session.py

@@ -99,6 +99,26 @@ class RolloutSession:
     covers a retry after the attempt finished). Because a slow turn is coalesced rather than
     re-sampled, retries stay safe without an inflated client timeout. The future resolves to the
     served response, or to None if the attempt produced no servable response (error/refuse)."""
+    released: bool = False
+    """Set when the rollout unregisters the session: the trace is sealed (its conclusion is
+    what scored and persisted), so a handler still in flight must not commit turns, record
+    calls, or write state onto it — the in-memory trace must stay what the run produced."""
+    tasks: set["asyncio.Task"] = field(default_factory=set)
+    """Handler tasks currently serving this session. aiohttp does not cancel a handler when
+    its client disconnects, so a request whose program died at teardown would keep driving
+    the exchange (upstream call, simulator turn) — unregistering cancels these instead."""
+
+    def adopt(self, task: "asyncio.Task | None") -> None:
+        """Track a handler task serving this session, for cancellation at release."""
+        if task is not None:
+            self.tasks.add(task)
+            task.add_done_callback(self.tasks.discard)
+
+    def release(self) -> None:
+        """Seal the session: no further trace mutation, and in-flight handlers cancel."""
+        self.released = True
+        for task in list(self.tasks):
+            task.cancel()
 
     async def refused(self) -> str | None:
         """The framework's limits (turns / token budget) and `@stop` checks, run before each