diff --git a/docs/config.md b/docs/config.md index 8992bbabd..dd1a40b5d 100644 --- a/docs/config.md +++ b/docs/config.md @@ -2144,7 +2144,14 @@ also when the backend did not start at all, and nothing is frozen into it; the sweep keeps its watermark, so the window is offered again. Reads are tried twice, and wait out a `Retry-After` of up to 2 seconds once (the gateway's per-agent limit sends 1); a longer one holds that operation (at -most 60 seconds). When the gateway is not reached, or a call gets no answer +most 60 seconds). A call the gateway answers with +`MCP server session ended; the state of earlier calls is lost` +did not reach Loom: the session the gateway kept with Loom had ended. Nerve +sends such a call again at once, an extraction batch too, at most three +times in a row, without using up a read's second try. A fourth such answer +in a row is a failure like any other `-32002`: a read goes on to its second +try, and a batch waits for the sweep. When the gateway is not reached, or a +call gets no answer in time (a read gives up long before the gateway's call limit, extraction a little after it), calls fail at once for 15 seconds, doubling with each such failure in a row up to 5 minutes: the gateway's own circuit does not count diff --git a/nerve/mcp_gateway.py b/nerve/mcp_gateway.py index 8e208b7f0..7e7e91fd0 100644 --- a/nerve/mcp_gateway.py +++ b/nerve/mcp_gateway.py @@ -234,6 +234,10 @@ async def _fetch(self) -> GatewayCatalog: # limit (MCP_GATEWAY_CALL_TIMEOUT, two minutes by default); the upstream may # still finish the call. UPSTREAM_TIMEOUT_MESSAGE = "MCP server timed out" +# The gateway's fixed message for a call that it did not send: the session it +# kept with the upstream server had ended. It answers so once and sends the +# next call in a new session. +SESSION_ENDED_MESSAGE = "MCP server session ended; the state of earlier calls is lost" class GatewayCallError(Exception): diff --git a/nerve/memory/loom.py b/nerve/memory/loom.py index f1ced0298..e94fa3dc6 100644 --- a/nerve/memory/loom.py +++ b/nerve/memory/loom.py @@ -40,7 +40,9 @@ pass (``-32002``), a server fault, a timeout, a rate limit — raises :class:`~nerve.memory.backend.MemoryBackendUnavailable`, so a session starts without priors instead of crashing and the sweep keeps its watermark. A -request Loom refused as invalid raises :class:`LoomRequestError`. +request Loom refused as invalid raises :class:`LoomRequestError`. A call the +gateway did not send, because the session it kept with Loom had ended, is +sent again at once, a transcript for extraction too. Recalled text is data from anyone who writes to the namespaces the agent reads. It reaches the session inside the ```` block, so the @@ -68,8 +70,10 @@ POLICY_REFUSED, RETRY_LATER, SERVER_FAULT, + SESSION_ENDED_MESSAGE, UPSTREAM_TIMEOUT_MESSAGE, GatewayCallError, + GatewayToolResult, McpGatewayCatalog, McpGatewayServer, ) @@ -126,11 +130,19 @@ class LoomCall: ), } -# Ops safe to send again after a failure: reads only. A transcript queued for -# extraction is never resent, since the lost answer may have queued it. +# Ops safe to send again after a failure: reads only. A transcript is not +# sent again after a failure, since the lost answer may have queued it; the +# one exception is a call the gateway did not send (below). _RETRIED_OPS = frozenset({"context.pack", "search.query", "namespace.summary"}) _MAX_ATTEMPTS = 2 _BACKOFF_S = 0.5 +# A call the gateway answers with its session-ended message did not reach +# Loom: the session the gateway kept with Loom had ended, and the next call +# opens a new one. Such a call goes again at once, a transcript too, and +# does not use up a read's second try. The gateway can give the answer a few +# times in a row, so Nerve sends the call again at most this many times; +# after that the answer is a failure like any other. +_SESSION_ENDED_RESENDS = 3 # Seconds a lifecycle call may take through the gateway, which gives the # upstream call up to two minutes: the startup read is on a session's way in # and must be quick; extraction waits for Loom's model pass, so it gets a bit @@ -708,10 +720,8 @@ async def _call(self, op: str, arguments: dict[str, Any], *, timeout: float = _R ) attempts = _MAX_ATTEMPTS if op in _RETRIED_OPS else 1 for attempt in range(1, attempts + 1): - server = await self._loom_server() - self._inflight[server] = self._inflight.get(server, 0) + 1 try: - result = await server.call_tool(name, call_args, timeout=timeout) + result = await self._send(op, name, call_args, timeout) except GatewayCallError as e: failure = self._gateway_failure(op, e) no_answer = e.status is None and e.code is None @@ -737,8 +747,6 @@ async def _call(self, op: str, arguments: dict[str, Any], *, timeout: float = _R raise failure from None await asyncio.sleep(self._backoff_s * attempt) continue - finally: - await self._release(server) self._breaker_failures = 0 self._down_until = 0.0 if result.is_error: @@ -746,6 +754,39 @@ async def _call(self, op: str, arguments: dict[str, Any], *, timeout: float = _R return result.structured raise AssertionError("unreachable") # pragma: no cover + async def _send( + self, op: str, name: str, call_args: dict[str, Any], timeout: float, + ) -> GatewayToolResult: + """One tool call through the gateway, on the client of Loom's server. + + A call the gateway did not send because its session with Loom had + ended goes again at once (see ``_SESSION_ENDED_RESENDS``); every + other failure is the caller's. + """ + resends = 0 + while True: + server = await self._loom_server() + self._inflight[server] = self._inflight.get(server, 0) + 1 + try: + return await server.call_tool(name, call_args, timeout=timeout) + except GatewayCallError as e: + # The gateway's own answer, word for word and unpaced: any + # other failure may come after the call reached Loom. + session_ended = ( + e.code == RETRY_LATER + and e.rpc_message == SESSION_ENDED_MESSAGE + and e.retry_after is None + ) + if not session_ended or resends == _SESSION_ENDED_RESENDS: + raise + resends += 1 + logger.info( + "Loom %s: the gateway's session with Loom had ended; sending the call again (%d of %d)", + op, resends, _SESSION_ENDED_RESENDS, + ) + finally: + await self._release(server) + @staticmethod def _gateway_failure(op: str, e: GatewayCallError) -> Exception: """What a failed gateway call means to the caller: Loom unavailable diff --git a/tests/test_loom_contract.py b/tests/test_loom_contract.py index e42f1ed23..14fdee472 100644 --- a/tests/test_loom_contract.py +++ b/tests/test_loom_contract.py @@ -102,7 +102,9 @@ def test_agents_are_told_only_about_tools_loom_lists() -> None: def test_only_reads_are_retried() -> None: """A call Nerve sends again after a lost answer must be one Loom marks - idempotent: queueing a transcript twice extracts it twice.""" + idempotent: queueing a transcript twice extracts it twice. A call the + gateway did not send is another matter: Nerve sends it again whatever + the op (``tests/test_loom_memory.py``).""" from nerve.memory.loom import _RETRIED_OPS assert {op for op in _RETRIED_OPS if not OPS[op]["idempotent"]} == set() diff --git a/tests/test_loom_memory.py b/tests/test_loom_memory.py index 741b60a8a..3eee113c3 100644 --- a/tests/test_loom_memory.py +++ b/tests/test_loom_memory.py @@ -629,6 +629,75 @@ async def test_a_condition_that_may_pass_is_tried_twice_for_a_read(make_memory) await memory.startup_recall("plan") +SESSION_ENDED = "MCP server session ended; the state of earlier calls is lost" + + +def _session_ended() -> httpx.Response: + """The gateway's answer to a call that it did not send: its session with + Loom had ended.""" + return rpc_error(-32002, SESSION_ENDED) + + +@pytest.mark.asyncio +async def test_a_call_the_gateway_did_not_send_goes_again(make_memory) -> None: + gateway = FakeGateway() + gateway.add_row(HOME, "k1", "the plan is ready") + memory = await make_memory(gateway, namespace=HOME) + + # A transcript too: nothing reached Loom, so nothing is extracted twice. + gateway.fail("loom_memory_set_from_messages", _session_ended()) + assert await memory.memorize_conversation("s", _messages(2, datetime.now(UTC))) is True + assert len(gateway.args("loom_memory_set_from_messages")) == 2 + assert len(gateway.queued) == 1 + + # A read keeps its second try for another failure, before or after. + gateway.fail("loom_context_pack", _session_ended(), rpc_error(-32002, "MCP server unavailable")) + out = await memory.startup_recall("plan") + assert [m["id"] for m in out] == ["k1"] + assert len(gateway.args("loom_context_pack")) == 3 + gateway.fail("loom_context_pack", rpc_error(-32002, "MCP server unavailable"), _session_ended()) + out = await memory.startup_recall("plan") + assert [m["id"] for m in out] == ["k1"] + assert len(gateway.args("loom_context_pack")) == 6 + + assert memory._retired == set() and memory._inflight == {} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("answer", [ + lambda: rpc_error(-32002, "MCP server session ended"), + lambda: rpc_error(-32000, SESSION_ENDED), + lambda: rpc_error(-32002, SESSION_ENDED, status=429, **{"Retry-After": "30"}), +], ids=["other words", "another code", "a paced answer"]) +async def test_only_the_gateways_own_answer_sends_a_transcript_again(make_memory, answer) -> None: + """A transcript goes again only on the gateway's own answer, word for + word: any other failure may come after the call reached Loom.""" + gateway = FakeGateway() + memory = await make_memory(gateway) + gateway.fail("loom_memory_set_from_messages", answer()) + with pytest.raises(MemoryBackendUnavailable): + await memory.memorize_conversation("s", _messages(2, datetime.now(UTC))) + assert len(gateway.args("loom_memory_set_from_messages")) == 1 + assert gateway.queued == [] + + +@pytest.mark.asyncio +async def test_session_ended_answers_in_a_row_are_bounded(make_memory) -> None: + """The gateway can give the answer a few times in a row; a gateway that + keeps answering so is an outage.""" + gateway = FakeGateway() + memory = await make_memory(gateway) + gateway.fail("loom_memory_set_from_messages", *[_session_ended() for _ in range(3)]) + assert await memory.memorize_conversation("s", _messages(2, datetime.now(UTC))) is True + assert len(gateway.args("loom_memory_set_from_messages")) == 4 + + gateway.fail("loom_memory_set_from_messages", *[_session_ended() for _ in range(4)]) + with pytest.raises(MemoryBackendUnavailable, match="session ended"): + await memory.memorize_conversation("s2", _messages(2, datetime.now(UTC))) + assert len(gateway.args("loom_memory_set_from_messages")) == 8 + assert len(gateway.queued) == 1 + + @pytest.mark.asyncio async def test_no_answer_opens_the_breaker(make_memory) -> None: gateway = FakeGateway()