Loom memory: send a call again when the gateway did not send it

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

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):

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 ``<session-context>`` 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,15 +747,46 @@ 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:
                 raise self._refusal(op, result.structured, result.text)
             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

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()

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()