Pin initialize-abandon suppression and tidy the interaction suite

- Add an interaction test proving a timed-out initialize sends no
  notifications/cancelled, and drop the stale deferred rationale for the
  initialize-not-cancellable requirement
- Record null-id error-response drops in the requirements manifest and
  describe the cancelled-request error response as applying to both seats
- Record wire messages only after the inner send succeeds, so a failed or
  cancelled write is not logged as sent
- Bound the remaining indefinite waits, refresh comments stale since the
  courtesy-cancel change, and assert the issue-88 timeout by error code
- Rewrite the suite README's concurrency section for the dispatcher model

Author: maxisbey

tests/interaction/README.md

@@ -193,11 +193,13 @@ many requirements at once; if the assertions would be separate, write separate t
 
 ### Notifications and concurrency
 
-The client's receive loop dispatches each incoming message to completion before reading the next,
-and the in-memory transport delivers everything on one ordered stream. Together these guarantee
-that every notification a server handler emits before its response reaches the client callback
-before the originating request returns — so tests collect notifications into a plain list and
-assert after the call, with no synchronisation. The exceptions:
+The client's dispatcher starts a task per incoming notification in arrival order but does not
+await it before reading the next message, so completion order is not structural. What still
+holds: the in-memory transport delivers everything on one ordered stream, and a callback that
+records synchronously (no `await` before the append) finishes its scheduling slice before the
+awaited request's waiter — woken strictly later — resumes. So tests whose callbacks are plain
+appends may still collect into a list and assert after the call. A callback that awaits before
+recording loses that ordering and must synchronise. The other exceptions:
 
 - a notification not triggered by a request the test is awaiting needs an `anyio.Event` set in
   the receiving handler and awaited under `anyio.fail_after(5)`;
@@ -220,9 +222,8 @@ but still inside an outer `async with`, and no restructure can avoid it.
 
 A handful of `# pragma: lax no cover` markers in `src/` cover teardown exception handlers whose
 execution is timing-dependent under the in-process HTTP bridge — the POST-stream and
-stateless-session `except Exception` handlers in `server/streamable_http*.py`, the `_terminated`
-check in `message_router`, and the response-stream double-close guard in
-`BaseSession._receive_loop`. `strict-no-cover` does not check `lax` lines; do not promote them to
-strict `no cover` without first making the teardown ordering deterministic. The suite also relies
-on a one-line `src/mcp/server/sse.py` fix (`sse_stream_reader.aclose()`) that closes a stream the
-SSE leg would otherwise leak.
+stateless-session `except Exception` handlers in `server/streamable_http*.py` and the
+`_terminated` check in `message_router`. `strict-no-cover` does not check `lax` lines; do not
+promote them to strict `no cover` without first making the teardown ordering deterministic. The
+suite also relies on a one-line `src/mcp/server/sse.py` fix (`sse_stream_reader.aclose()`) that
+closes a stream the SSE leg would otherwise leak.

tests/interaction/_helpers.py

@@ -67,8 +67,11 @@ def __init__(self, inner: WriteStream[SessionMessage], log: list[SessionMessage]
         self._log = log
 
     async def send(self, item: SessionMessage, /) -> None:
-        self._log.append(item)
+        # Record only after the inner send returns: a send that raises (or is cancelled
+        # mid-write) never reached the transport, so logging it first would fabricate a
+        # "sent" message in wire-level assertions.
         await self._inner.send(item)
+        self._log.append(item)
 
     async def aclose(self) -> None:
         await self._inner.aclose()

tests/interaction/_requirements.py

@@ -268,18 +268,15 @@ def __post_init__(self) -> None:
         divergence=Divergence(
             note=(
                 "The spec says receivers of a cancellation SHOULD NOT send a response for the cancelled "
-                "request; the server sends an error response (code 0, 'Request cancelled'), which is what "
-                "unblocks the SDK client's pending call."
+                "request; both seats send an error response (code 0, 'Request cancelled') instead — the "
+                "server for cancelled client requests, and the client for cancelled server-initiated "
+                "requests — which is what unblocks the sender's pending call."
             ),
         ),
     ),
     "protocol:cancel:initialize-not-cancellable": Requirement(
         source=f"{SPEC_BASE_URL}/basic/utilities/cancellation#behavior-requirements",
         behavior="The client never sends notifications/cancelled for the initialize request.",
-        deferred=(
-            "Not implemented in the SDK: the client has no public cancellation API at all, so no pathway "
-            "exists that could cancel initialize; there is no distinct behaviour to pin beyond that absence."
-        ),
     ),
     "protocol:cancel:late-response-ignored": Requirement(
         source=f"{SPEC_BASE_URL}/basic/utilities/cancellation#behavior-requirements",
@@ -342,6 +339,26 @@ def __post_init__(self) -> None:
         source=f"{SPEC_BASE_URL}/basic#responses",
         behavior="A request whose method has no registered handler is answered with a METHOD_NOT_FOUND error.",
     ),
+    "protocol:error:null-id": Requirement(
+        source="sdk",
+        behavior=(
+            "An error response carrying a null id — the JSON-RPC shape for a peer reporting a failure it "
+            "could not attribute to a request, such as a parse error — is surfaced to the application "
+            "rather than silently discarded."
+        ),
+        divergence=Divergence(
+            note=(
+                "The dispatcher drops null-id error responses with a debug log; v1 surfaced them to "
+                "message_handler as an MCPError. A typed fault channel restoring visibility is planned "
+                "before v2 stable."
+            ),
+        ),
+        deferred=(
+            "Not yet covered here: the current drop is pinned at the dispatcher level by "
+            "tests/shared/test_jsonrpc_dispatcher.py; an interaction-level test waits on the planned "
+            "fault channel."
+        ),
+    ),
     "protocol:meta:related-task": Requirement(
         source=f"{SPEC_BASE_URL}/basic/utilities/tasks#related-task-metadata",
         behavior="Messages may carry related-task _meta associating them with a task.",

tests/interaction/lowlevel/test_cancellation.py

@@ -16,6 +16,7 @@
 from mcp.shared.memory import MessageStream, create_client_server_memory_streams
 from mcp.shared.message import SessionMessage
 from mcp.types import (
+    REQUEST_TIMEOUT,
     CallToolResult,
     EmptyResult,
     ErrorData,
@@ -196,7 +197,8 @@ async def sample() -> None:
                 raise NotImplementedError  # unreachable: the scope is cancelled
 
             abandon_scope.start_soon(sample)
-            await callback_started.wait()
+            with anyio.fail_after(5):
+                await callback_started.wait()
             abandon_scope.cancel_scope.cancel()
         with anyio.fail_after(5):
             await callback_cancelled.wait()
@@ -284,3 +286,60 @@ async def message_handler(message: IncomingMessage) -> None:
         assert pong == snapshot(EmptyResult())
         # The fabricated response was dropped silently: the ping after it still
         # round-tripped, and the message handler (a tripwire) was never invoked.
+
+
+@requirement("protocol:cancel:initialize-not-cancellable")
+async def test_timed_out_initialize_sends_no_cancellation() -> None:
+    """An abandoned initialize is not followed by notifications/cancelled on the wire.
+
+    Spec-mandated: the initialize request MUST NOT be cancelled. Abandoning any other request
+    sends a courtesy notifications/cancelled (see protocol:timeout:sends-cancellation); this
+    test pins that initialize opts out. A real Server always answers initialize, so the test
+    plays a stalling server by hand: it never answers initialize, the client's read timeout
+    fires, and the ping the test sends next is the marker — the in-memory stream is ordered and
+    a courtesy cancel goes out before the timeout error reaches the caller, so a regression
+    would put notifications/cancelled ahead of the ping in the recorded sequence.
+    """
+    received_methods: list[str] = []
+
+    async def scripted_server(streams: MessageStream) -> None:
+        server_read, server_write = streams
+
+        # Hold the initialize request unanswered until the client's read timeout fires.
+        init = await server_read.receive()
+        assert isinstance(init, SessionMessage)
+        assert isinstance(init.message, JSONRPCRequest)
+        received_methods.append(init.message.method)
+
+        follow_up = await server_read.receive()
+        assert isinstance(follow_up, SessionMessage)
+        assert isinstance(follow_up.message, JSONRPCRequest)
+        received_methods.append(follow_up.message.method)
+        await server_write.send(
+            SessionMessage(
+                JSONRPCResponse(
+                    jsonrpc="2.0",
+                    id=follow_up.message.id,
+                    # Serialized exactly as a real server serializes results onto the wire.
+                    result=EmptyResult().model_dump(by_alias=True, mode="json", exclude_none=True),
+                )
+            )
+        )
+
+    async with (
+        create_client_server_memory_streams() as ((client_read, client_write), server_streams),
+        anyio.create_task_group() as task_group,
+        # The session-level read timeout is the only public pathway that abandons initialize;
+        # the response never arrives, so any positive value fires on the next event-loop pass.
+        ClientSession(client_read, client_write, read_timeout_seconds=0.000001) as session,
+    ):
+        task_group.start_soon(scripted_server, server_streams)
+        with anyio.fail_after(5):
+            with pytest.raises(MCPError) as exc_info:
+                await session.initialize()
+            assert exc_info.value.error.code == REQUEST_TIMEOUT
+            # Override the session-level timeout: this ping must round-trip normally.
+            pong = await session.send_request(PingRequest(), EmptyResult, request_read_timeout_seconds=5)
+
+        assert pong == snapshot(EmptyResult())
+        assert received_methods == snapshot(["initialize", "ping"])

tests/interaction/lowlevel/test_timeouts.py

@@ -107,7 +107,8 @@ async def sampling_callback(
         context: ClientRequestContext, params: types.CreateMessageRequestParams
     ) -> types.CreateMessageResult:
         callback_started.set()
-        await release.wait()
+        with anyio.fail_after(5):
+            await release.wait()
         return types.CreateMessageResult(role="assistant", content=TextContent(text="too late"), model="test-model")
 
     async with Client(recording, sampling_callback=sampling_callback) as client:
@@ -146,7 +147,7 @@ async def list_tools(
     async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
         if params.name == "echo":
             return CallToolResult(content=[TextContent(text="still alive")])
-        await anyio.Event().wait()  # blocks until the session is torn down
+        await anyio.Event().wait()  # blocks until the courtesy cancellation interrupts it
         raise NotImplementedError  # unreachable
 
     server = Server("blocker", on_list_tools=list_tools, on_call_tool=call_tool)
@@ -178,7 +179,7 @@ async def test_session_level_timeout_applies_to_every_request() -> None:
 
     async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
         assert params.name == "block"
-        await anyio.Event().wait()  # blocks until the session is torn down
+        await anyio.Event().wait()  # blocks until the courtesy cancellation interrupts it
         raise NotImplementedError  # unreachable
 
     server = Server("blocker", on_call_tool=call_tool)

tests/issues/test_88_random_error.py

@@ -12,7 +12,14 @@
 from mcp.server import Server, ServerRequestContext
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import SessionMessage
-from mcp.types import CallToolRequestParams, CallToolResult, ListToolsResult, PaginatedRequestParams, TextContent
+from mcp.types import (
+    REQUEST_TIMEOUT,
+    CallToolRequestParams,
+    CallToolResult,
+    ListToolsResult,
+    PaginatedRequestParams,
+    TextContent,
+)
 
 
 @pytest.mark.anyio
@@ -97,7 +104,7 @@ async def client(
             # Use very small timeout to trigger quickly without waiting
             with pytest.raises(MCPError) as exc_info:
                 await session.call_tool("slow", read_timeout_seconds=0.000001)  # artificial timeout that always fails
-            assert "timed out" in str(exc_info.value)
+            assert exc_info.value.error.code == REQUEST_TIMEOUT
 
             # No-op if the courtesy cancellation already interrupted the handler.
             slow_request_lock.set()