diff --git a/BUGFIX.md b/BUGFIX.md index 067ed62..7b740e2 100644 --- a/BUGFIX.md +++ b/BUGFIX.md @@ -1,5 +1,23 @@ # Bug fixes +## 2026-10-07 — A call right after a worker leaves its gateway session reaches the backend + +**Symptom.** `test_without_an_attached_stream_the_backend_is_never_called` failed now and then +under `mise run check`, the scripted backend answering a call nobody had queued a reply for. It +passed when run alone. +**Root cause.** `GatewaySession.__aexit__` closed its side of the `Attach` stream and cancelled +it at once. model-gateway detaches the run only when it reads that close, so for a moment after +the session was left the run's keys were still accepted, and a call then was sent to the backend +before failing closed for want of an ack. +**Fix.** Leaving the session waits, up to 10 s, for the gateway to end the stream, which it does +after detaching; a stream that ends while leaving is not an error. +`swarmeval/gateway/model/client.py`. +**Guard.** `test_gateway.py::test_leaving_a_session_returns_after_model_gateway_detached_the_run`, +with the gateway's detach delayed. +**Touches.** The fail-closed rule (no call goes unrecorded) held throughout: a call in that +window got no response. A worker taking over a run attaches at a higher epoch, which replaces +the old stream at once; that path does not wait on the old session. + ## 2026-10-05 — The first call after `docker compose up --wait` finds the control plane down **Symptom.** On a Linux host, `deploy/compose/smoke.sh` failed at its first CLI call: diff --git a/CHANGELOG.md b/CHANGELOG.md index 2330b50..3bc64cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed +- Leaving a `GatewaySession` returns once model-gateway has detached the run, so no call + made after it reaches the backend. + ### Added - `browser` and `computer`: agents drive a web browser and a virtual screen inside their sandbox (`spec/2026-10-07-browser-computer-use`). A profile's `display` (`env.yaml` schema version 2) diff --git a/docs/services/model-gateway.md b/docs/services/model-gateway.md index 5d73618..1a7f9eb 100644 --- a/docs/services/model-gateway.md +++ b/docs/services/model-gateway.md @@ -83,7 +83,9 @@ with `ABORTED`; an equal or lower epoch is refused with `FAILED_PRECONDITION`. T takeover moves the stream. **Worker client.** `swarmeval.gateway.model.client.GatewaySession` implements the runtime's -`ModelClient` for one run. It attaches on entry and detaches on exit. For each call it commits +`ModelClient` for one run. It attaches on entry and detaches on exit: it closes its side of the +stream and waits, up to 10 s, for the gateway to end the stream, which it does once it has +detached the run, so the run's keys are refused by the time the session is left. For each call it commits the record as a `ModelEvent` through the run's `RunWriter`, then acks. It refuses a record whose request hash differs from the body it sent, and a response that differs from the one recorded, because either would split the evidence from what the agent saw. It replaces NUL in model output, diff --git a/swarmeval/gateway/model/client.py b/swarmeval/gateway/model/client.py index 1bbdce6..38903bc 100644 --- a/swarmeval/gateway/model/client.py +++ b/swarmeval/gateway/model/client.py @@ -14,6 +14,7 @@ import base64 import hashlib import json +import logging import secrets from collections.abc import Mapping, Sequence from dataclasses import dataclass, field @@ -68,6 +69,12 @@ ) from swarmeval.runtime.writer import RunWriter +log = logging.getLogger(__name__) + +DETACH_TIMEOUT_S = 10.0 +"""How long leaving a session waits for model-gateway to confirm the detach by ending the +stream.""" + UPSTREAM_ERROR_STATUS = 502 """model-gateway's status for a call the model backend failed after retries.""" @@ -138,6 +145,7 @@ def __init__( self._images = images self._sent_images: dict[str, bytes] = {} """The last request's images. The next one mostly carries the same ones.""" + self._leaving = False async def __aenter__(self) -> Self: stream = self._stub.Attach() @@ -169,7 +177,20 @@ async def __aexit__( tb: TracebackType | None, ) -> None: if self._stream is not None: + # model-gateway detaches the run when it sees this side close, then ends the stream. + # Waiting for that end means that once this returns, the run's keys are refused. + self._leaving = True await self._stream.done_writing() + if self._reader is not None: + try: + await asyncio.wait_for(asyncio.shield(self._reader), DETACH_TIMEOUT_S) + except TimeoutError: + log.warning( + "run %s: model-gateway did not end the stream within %gs of the " + "detach; it may accept the run's keys a moment longer", + self._run_id, + DETACH_TIMEOUT_S, + ) self._stream.cancel() if self._reader is not None: self._reader.cancel() @@ -244,6 +265,8 @@ async def _read( "follow." ) await self._commit(stream, message.record) + if self._leaving: + return raise ModelGatewayError(f"run `{self._run_id}`: model-gateway closed the stream.") except asyncio.CancelledError: raise diff --git a/tests/gateway/test_gateway.py b/tests/gateway/test_gateway.py index e0e4e2d..68119e7 100644 --- a/tests/gateway/test_gateway.py +++ b/tests/gateway/test_gateway.py @@ -11,7 +11,7 @@ from swarmeval.gateway.model.app import create_app from swarmeval.gateway.model.client import GatewaySession, ModelGatewayError from swarmeval.gateway.model.config import GatewayConfig, ReasoningPassback -from swarmeval.gateway.model.recorder import Attachments, NotAttachedError, Recorder +from swarmeval.gateway.model.recorder import Attachment, Attachments, NotAttachedError, Recorder from swarmeval.gateway.model.upstream import Upstreams, passback from swarmeval.gateway.model.wire import ( CALL_ID_HEADER, @@ -271,6 +271,24 @@ async def test_without_an_attached_stream_the_backend_is_never_called(rig: Rig) assert rig.backend.requests == [] +async def test_leaving_a_session_returns_after_model_gateway_detached_the_run( + rig: Rig, monkeypatch: pytest.MonkeyPatch +) -> None: + detach = rig.attachments.detach + + def slow(attachment: Attachment, reason: str) -> None: + """A loaded gateway: it gets to the detach a while after the worker's side closed.""" + asyncio.get_running_loop().call_later(0.2, detach, attachment, reason) + + monkeypatch.setattr(rig.attachments, "detach", slow) + session = rig.session() + async with session: + key = session._keys["agent:dev"] # pyright: ignore[reportPrivateUsage] + assert rig.attachments.lookup(key) is not None + + assert rig.attachments.lookup(key) is None + + async def test_a_record_the_worker_cannot_commit_fails_the_call(rig: Rig) -> None: rig.backend.reply(completion("lost"))