test: deterministically cover message_router closed-stream handler

Replace the two remaining `lax no cover` pragmas with a unit test that
deterministically triggers the race the handler defends against.

The scenario: sse_writer receives a JSONRPCResponse and breaks its
async-for loop; its finally block pops _request_streams. If
message_router is concurrently routing a trailing message to the same
request_id, the membership check passes but send() hits a closed
receiver. Under xdist the scheduler contention made this fire ~15% of
the time — enough to break strict-no-cover but not enough to rely on.

The new test injects a closed-receiver stream directly into
_request_streams, routes a message, and asserts the handler popped the
stale entry. No scheduler timing involved.

Also removed `# pragma: no cover` from `except Exception` in
standalone_sse_writer (line 717) — the thread-based GET stream tests
reliably cover it (both sequential and xdist). It was only unhit on
main because subprocess tests couldn't see into the subprocess.

Net: 0 lax no cover added, 2 more no cover removed.

Author: maxisbey

src/mcp/server/streamable_http.py

@@ -715,11 +715,7 @@ async def standalone_sse_writer():
                         event_data = self._create_event_data(event_message)
                         await sse_stream_writer.send(event_data)
             except Exception:
-                # Timing race: if task cancellation reaches this writer before it
-                # hits the closed stream, we get CancelledError (not caught here).
-                # If the closed-stream error fires first, this logs. Neither order is
-                # a bug — both are valid shutdown sequences.
-                logger.exception("Error in standalone SSE writer")  # pragma: lax no cover
+                logger.exception("Error in standalone SSE writer")
             finally:
                 logger.debug("Closing standalone SSE writer")
                 await self._clean_up_memory_streams(GET_STREAM_KEY)
@@ -1019,13 +1015,7 @@ async def message_router():
                             try:
                                 # Send both the message and the event ID
                                 await self._request_streams[request_stream_id][0].send(EventMessage(message, event_id))
-                            except (  # pragma: lax no cover
-                                # Timing race: if cancellation reaches this coroutine
-                                # before send() hits the closed stream, CancelledError
-                                # propagates instead. Either shutdown sequence is valid.
-                                anyio.BrokenResourceError,
-                                anyio.ClosedResourceError,
-                            ):
+                            except (anyio.BrokenResourceError, anyio.ClosedResourceError):
                                 # Stream might be closed, remove from registry
                                 self._request_streams.pop(request_stream_id, None)
                         else:

tests/shared/test_streamable_http.py

@@ -1705,6 +1705,46 @@ async def test_priming_event_not_sent_for_old_protocol_version():
         await read_stream.aclose()
 
 
+@pytest.mark.anyio
+async def test_message_router_handles_closed_request_stream():
+    """message_router must survive a stream closing between membership check and send.
+
+    Real scenario: sse_writer receives a JSONRPCResponse, breaks its async-for,
+    and runs cleanup — while message_router is concurrently routing a trailing
+    notification to the same request_id. The router's `if id in _request_streams`
+    check passes, but by the time send() tries to deliver, the receiver is gone.
+    The except (BrokenResourceError, ClosedResourceError) handler must catch
+    this and pop the stale entry so the router keeps routing other messages.
+
+    This test reproduces the race deterministically rather than relying on
+    scheduler timing (which only triggers under xdist contention).
+    """
+    transport = StreamableHTTPServerTransport(mcp_session_id="test-session")
+
+    async with transport.connect() as (_, write_stream):
+        # Register a request stream pair as if a POST request had opened one,
+        # then close the receiver to simulate the sse_writer breaking its loop
+        # before cleanup pops the dict entry.
+        request_id = "req-1"
+        send_side, recv_side = anyio.create_memory_object_stream[EventMessage](0)
+        transport._request_streams[request_id] = (send_side, recv_side)
+        await recv_side.aclose()  # receiver gone — paired sender will now raise
+
+        # Route a response targeting that request_id. message_router's membership
+        # check passes (still in dict), send() raises BrokenResourceError, the
+        # except handler pops the entry.
+        response = types.JSONRPCResponse(jsonrpc="2.0", id=request_id, result={})
+        await write_stream.send(SessionMessage(response))
+
+        # Give the router a tick to process
+        with anyio.fail_after(1):
+            while request_id in transport._request_streams:
+                await anyio.sleep(0)
+
+        assert request_id not in transport._request_streams
+        await send_side.aclose()
+
+
 @pytest.mark.anyio
 async def test_priming_event_not_sent_without_event_store():
     """Test that _maybe_send_priming_event returns early when no event_store is configured."""