Honor the cancel-silence commitment in the shutdown drain

The dispatcher's shutdown arm wrote the CONNECTION_CLOSED drain answer for
every in-flight request it caught unwinding, including one whose peer had
already cancelled it and whose loop layer had committed it to 2026 cancel
semantics - where no frame of any kind may follow notifications/cancelled.
A silence-committed peer-cancelled request caught at loop shutdown now
stays silent.

The rule lives in one predicate on the dispatch context, must_stay_silent,
consulted by both frame-suppressing sites (_answer_error and the shutdown
drain) so it cannot drift between them; the interrupt arm remains its
complementary write side. Requests the peer did NOT cancel keep the drain
answer regardless of the commitment - the EOF-drain doctrine is unchanged.

Author: maxisbey

src/mcp/shared/jsonrpc_dispatcher.py

@@ -179,6 +179,20 @@ def request_id(self) -> RequestId | None:
     def can_send_request(self) -> bool:
         return self.transport.can_send_request and not self._closed
 
+    @property
+    def must_stay_silent(self) -> bool:
+        """The peer cancelled this silence-committed request: nothing may be written for it.
+
+        Consulted by both frame-suppressing sites - `_answer_error` and the
+        shutdown drain. The interrupt arm expresses the same fact from the
+        write side: within `scope.cancelled_caught`, `cancel_requested` is
+        definitionally set, so its `cancel_answer is not None` check is
+        exactly `not must_stay_silent`. Once true, result, error, and drain
+        frames for this id are all forbidden - the transport spec's MUST NOT
+        after `notifications/cancelled`.
+        """
+        return self.cancel_answer is None and self.cancel_requested.is_set()
+
     async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         if self._closed:
             logger.debug("dropped %s: dispatch context closed", method)
@@ -796,9 +810,13 @@ async def _handle_request(
         except anyio.get_cancelled_exc_class():
             # Shutdown: answer the request so the peer isn't left waiting - unless
             # an answer write already started (it may have reached the transport;
-            # prefer possibly-zero answers over possibly-two). The shielded helper
-            # is needed because bare awaits re-raise here.
-            if not answer_write_started:
+            # prefer possibly-zero answers over possibly-two), or the peer already
+            # cancelled a silence-committed request (the 2026 rule forbids ANY
+            # frame after the cancel, the shutdown drain included; requests the
+            # peer did NOT cancel keep their drain answer regardless of the
+            # commitment). The shielded helper is needed because bare awaits
+            # re-raise here.
+            if not answer_write_started and not dctx.must_stay_silent:
                 await self._final_write(
                     partial(self._write_error, req.id, ErrorData(code=CONNECTION_CLOSED, message="Connection closed")),
                     shield=True,
@@ -848,7 +866,7 @@ async def _answer_error(
         failure was computed before the cancellation interrupt could land (a
         synchronous raise after the cancel arrived, or a shielded handler).
         """
-        if dctx.cancel_answer is None and dctx.cancel_requested.is_set():
+        if dctx.must_stay_silent:
             logger.debug("dropped error for %r: peer cancelled a silence-committed request", request_id)
             return
         await self._write_error(request_id, error)

tests/shared/test_jsonrpc_dispatcher.py

@@ -308,6 +308,64 @@ async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -
             assert await client.send_raw_request("m", None) == {"ok": True}
 
 
+@pytest.mark.anyio
+async def test_shutdown_drain_stays_silent_for_a_peer_cancelled_silence_committed_request():
+    """Loop shutdown catching a silence-committed request whose peer already
+    cancelled writes NOTHING for it: the 2026 rule forbids any frame after
+    the cancel, the CONNECTION_CLOSED drain answer included. Requests the
+    peer did NOT cancel keep the drain answer regardless of the commitment
+    (see test_shutdown_answers_in_flight_request_with_connection_closed)."""
+    c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32)
+    server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send)
+    handler_started = anyio.Event()
+    release_shield = anyio.Event()
+
+    async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
+        suppress_cancel_answer(ctx)
+        handler_started.set()
+        # Survive the peer-cancel interrupt inside the shield, then hold the
+        # request in flight until the loop shutdown catches it unwinding.
+        with anyio.CancelScope(shield=True):
+            await ctx.cancel_requested.wait()
+            await release_shield.wait()
+        await anyio.sleep_forever()
+        raise NotImplementedError
+
+    async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None:
+        pass
+
+    late: list[SessionMessage | Exception] = []
+    try:
+        async with anyio.create_task_group() as tg:
+            await tg.start(server.run, on_request, on_notify)
+            with anyio.fail_after(5):
+                await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=7, method="slow")))
+                await handler_started.wait()
+                await c2s_send.send(
+                    SessionMessage(
+                        message=JSONRPCNotification(
+                            jsonrpc="2.0", method="notifications/cancelled", params={"requestId": 7}
+                        )
+                    )
+                )
+                await anyio.wait_all_tasks_blocked()
+                # The handler observed the cancel inside its shield; shut the
+                # loop down while the request is still in flight.
+                tg.cancel_scope.cancel()
+                release_shield.set()
+        while True:
+            try:
+                late.append(s2c_recv.receive_nowait())
+            except (anyio.WouldBlock, anyio.EndOfStream, anyio.ClosedResourceError):
+                break
+    finally:
+        for s in (c2s_send, c2s_recv, s2c_send, s2c_recv):
+            s.close()
+    # Total silence for the cancelled id: no drain frame at shutdown.
+    assert late == []
+
+
 @pytest.mark.anyio
 async def test_peer_cancel_landing_after_handlers_last_checkpoint_writes_only_the_result():
     """A peer cancel that fails to interrupt the handler writes only the result: one answer per