fix: remove stale no-cover pragma and simplify regression test

The integration test now covers the Exception branch in _receive_loop,
so the pragma is no longer valid (caught by strict-no-cover on CI).

Also simplified the regression test: zero-buffer write stream + no
concurrent receiver makes it deterministic — send_log_message() would
block on send(), then raise ClosedResourceError when _receive_loop
exits and closes the stream. No concurrent task, no uncovered branches,
no sleeps.

Author: maxisbey

src/mcp/shared/session.py

@@ -334,7 +334,7 @@ async def _receive_loop(self) -> None:
         async with self._read_stream, self._write_stream:
             try:
                 async for message in self._read_stream:
-                    if isinstance(message, Exception):  # pragma: no cover
+                    if isinstance(message, Exception):
                         await self._handle_incoming(message)
                     elif isinstance(message.message, JSONRPCRequest):
                         try:

tests/server/test_lowlevel_exception_handling.py

@@ -82,20 +82,25 @@ async def test_server_run_exits_cleanly_when_transport_yields_exception_then_clo
     """
     server = Server("test-server")
 
-    read_send, read_recv = anyio.create_memory_object_stream[SessionMessage | Exception](10)
-    write_send, write_recv = anyio.create_memory_object_stream[SessionMessage](10)
-
-    async def transport_side() -> None:
-        # What the streamable HTTP transport does: push the exception, then close.
-        await read_send.send(RuntimeError("simulated transport error"))
-        await read_send.aclose()
-        # Drain the write side so it doesn't back-pressure; it must see no messages.
-        async with write_recv:
-            async for _ in write_recv:  # pragma: no cover
-                pytest.fail("server must not write back when the transport reported an error")
+    read_send, read_recv = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+    # Zero-buffer on the write stream forces send() to block until received.
+    # With no receiver, a send() sits blocked until _receive_loop exits its
+    # `async with self._read_stream, self._write_stream:` block and closes the
+    # stream, at which point the blocked send raises ClosedResourceError.
+    # This deterministically reproduces the race without sleeps.
+    write_send, write_recv = anyio.create_memory_object_stream[SessionMessage](0)
+
+    # What the streamable HTTP transport does: push the exception, then close.
+    read_send.send_nowait(RuntimeError("simulated transport error"))
+    read_send.close()
 
     with anyio.fail_after(5):
-        async with anyio.create_task_group() as tg:
-            tg.start_soon(transport_side)
-            # stateless=True so server.run doesn't wait for initialize handshake
-            await server.run(read_recv, write_send, server.create_initialization_options(), stateless=True)
+        # stateless=True so server.run doesn't wait for initialize handshake.
+        # Before this fix, this raised ExceptionGroup(ClosedResourceError).
+        await server.run(read_recv, write_send, server.create_initialization_options(), stateless=True)
+
+    # write_send was closed inside _receive_loop's `async with`; receive_nowait
+    # raises EndOfStream iff the buffer is empty (i.e., server wrote nothing).
+    with pytest.raises(anyio.EndOfStream):
+        write_recv.receive_nowait()
+    write_recv.close()