From fcdd403c6476410ea7834861d60beeb914a644f7 Mon Sep 17 00:00:00 2001 From: pufit Date: Tue, 10 Nov 2026 13:00:09 -0500 Subject: [PATCH] Loom memory: send a call again when the gateway did not send it When the session that the MCP gateway keeps with Loom has ended, the gateway answers the next call once with "MCP server session ended; the state of earlier calls is lost" and does not send it. A read used its second try on that answer, and a transcript for extraction, which is not sent again after a failure, failed and waited for the next sweep. Such a call now goes again at once, a transcript too, since it did not reach Loom. It does not use up a read's second try. The gateway can give the answer a few times in a row: Nerve sends the call again at most three times. Only the gateway's own answer counts: its code, its message word for word, and no Retry-After. --- docs/config.md | 9 ++++- nerve/mcp_gateway.py | 4 +++ nerve/memory/loom.py | 57 +++++++++++++++++++++++++----- tests/test_loom_contract.py | 4 ++- tests/test_loom_memory.py | 69 +++++++++++++++++++++++++++++++++++++ 5 files changed, 133 insertions(+), 10 deletions(-) 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()