Sever a superseded POST instead of guarding one resolution path

Re-issuing a request id (legal once its waiter was popped) now cancels the
superseded POST's scope at registration, so a zombie stream cannot answer,
fail, or resolve the successor's request through any exit: a late real
response, a status-derived error, a parse error, a reconnection replay, or
the synthesized stream-death resolution. This replaces the narrower identity
guard inside _resolve_abandoned_request, which covered only its own callers;
the guard, the RequestContext.in_flight_post threading, and the ctx-taking
signature all revert.

Also give the synthesized-error helper a code parameter: a 202 to a request
is a server-shape violation, not a connection loss, so it now resolves with
INVALID_REQUEST, matching the unexpected-content-type precedent and the
CONNECTION_CLOSED constant's own contract.

Author: maxisbey

src/mcp/client/streamable_http.py

@@ -71,8 +71,6 @@ class RequestContext:
     session_message: SessionMessage
     metadata: ClientMessageMetadata | None
     read_stream_writer: StreamWriter
-    in_flight_post: _InFlightPost | None = None
-    """This request's `_in_flight_posts` registration; None for notifications."""
 
 
 @dataclass(slots=True)
@@ -334,7 +332,10 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
                     # A request's response arrives on this POST's body; 202 says
                     # none will follow. Resolve rather than park the caller forever.
                     await self._resolve_abandoned_request(
-                        ctx, message.id, "server answered a request with 202 Accepted"
+                        ctx.read_stream_writer,
+                        message.id,
+                        "server answered a request with 202 Accepted",
+                        code=INVALID_REQUEST,
                     )
                 return
 
@@ -453,24 +454,21 @@ async def _handle_sse_response(
         else:
             # Not resumable: resolve the waiter, else a listen stream's consumer
             # would hang forever instead of learning the subscription is lost.
-            await self._resolve_abandoned_request(ctx, original_request_id, "SSE stream ended without a response")
+            await self._resolve_abandoned_request(
+                ctx.read_stream_writer, original_request_id, "SSE stream ended without a response"
+            )
 
-    async def _resolve_abandoned_request(self, ctx: RequestContext, request_id: RequestId, message: str) -> None:
+    async def _resolve_abandoned_request(
+        self, read_stream_writer: StreamWriter, request_id: RequestId, message: str, *, code: int = CONNECTION_CLOSED
+    ) -> None:
         """Resolve a request whose response can never arrive with a synthesized error.
 
-        Skipped when a same-id successor has taken over the `_in_flight_posts`
-        registration: ids are reusable once their waiter is popped, so the
-        pending waiter now belongs to the successor and a stale stream's death
-        must not fail it. Best-effort: a closed read stream means the session
-        is tearing down.
+        Best-effort: a closed read stream means the session is tearing down.
         """
-        if ctx.in_flight_post is not None and self._in_flight_posts.get(request_id) is not ctx.in_flight_post:
-            logger.debug("request %r was re-issued; leaving its waiter to the successor", request_id)
-            return
-        error_data = ErrorData(code=CONNECTION_CLOSED, message=message)
+        error_data = ErrorData(code=code, message=message)
         error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data))
         try:
-            await ctx.read_stream_writer.send(error_msg)
+            await read_stream_writer.send(error_msg)
         except (anyio.BrokenResourceError, anyio.ClosedResourceError):
             logger.debug("read stream closed before request %r could be resolved", request_id)
 
@@ -491,7 +489,7 @@ async def _handle_reconnection(
             # stream) would otherwise hang its caller forever.
             logger.debug(f"Max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
             await self._resolve_abandoned_request(
-                ctx, original_request_id, "SSE stream ended and reconnection attempts were exhausted"
+                ctx.read_stream_writer, original_request_id, "SSE stream ended and reconnection attempts were exhausted"
             )
             return
 
@@ -599,8 +597,13 @@ async def handle_request_async():
                             scope=anyio.CancelScope(),
                             modern=self._protocol_version_header in MODERN_PROTOCOL_VERSIONS,
                         )
+                        superseded = self._in_flight_posts.get(message.id)
+                        if superseded is not None:
+                            # A reused id means the waiter belongs to this attempt now:
+                            # sever the old POST so its zombie stream cannot answer,
+                            # fail, or resolve the successor's request.
+                            superseded.scope.cancel()
                         self._in_flight_posts[message.id] = post
-                        ctx.in_flight_post = post
                         tg.start_soon(self._run_request_post, handle_request_async, post, message.id)
                     else:
                         await handle_request_async()

tests/client/test_streamable_http.py

@@ -19,6 +19,7 @@
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
     CONNECTION_CLOSED,
+    INVALID_REQUEST,
     METHOD_NOT_FOUND,
     PROTOCOL_VERSION_META_KEY,
     JSONRPCError,
@@ -631,23 +632,6 @@ def handler(request: httpx.Request) -> httpx.Response:
     assert reply.message.error.code == CONNECTION_CLOSED
 
 
-class _DoomedSSEStream(httpx.AsyncByteStream):
-    """Parks after opening, then breaks on command without ever carrying an event id."""
-
-    def __init__(self) -> None:
-        self.opened = anyio.Event()
-        self.die = anyio.Event()
-
-    async def __aiter__(self) -> AsyncIterator[bytes]:
-        self.opened.set()
-        yield b": stalling\n\n"
-        await self.die.wait()
-        raise httpx.ReadError("connection reset")
-
-    async def aclose(self) -> None:
-        pass
-
-
 class _DeliverOnCommandSSEStream(httpx.AsyncByteStream):
     """Parks after opening, then delivers one JSON-RPC response when told."""
 
@@ -666,13 +650,13 @@ async def aclose(self) -> None:
 
 
 @pytest.mark.anyio
-async def test_a_stale_streams_death_does_not_resolve_a_reused_ids_successor() -> None:
-    """Once a same-id successor is registered, an abandoned earlier stream ending
-    responselessly must not synthesize an error for the id: the pending waiter
-    belongs to the successor, whose real response must still arrive."""
-    doomed = _DoomedSSEStream()
-    succeeding = _DeliverOnCommandSSEStream({"jsonrpc": "2.0", "id": "dup-1", "result": {}})
-    streams: list[httpx.AsyncByteStream] = [doomed, succeeding]
+async def test_a_superseded_posts_late_real_response_cannot_answer_the_successor() -> None:
+    """SDK-defined: re-issuing an id severs the superseded POST, so nothing from its
+    stream (a late real response, or a synthesized error for its death) can resolve
+    the reused id's waiter; only the successor's own response arrives."""
+    stale = _DeliverOnCommandSSEStream({"jsonrpc": "2.0", "id": "dup-1", "result": {"origin": "stale"}})
+    succeeding = _DeliverOnCommandSSEStream({"jsonrpc": "2.0", "id": "dup-1", "result": {"origin": "fresh"}})
+    streams: list[httpx.AsyncByteStream] = [stale, succeeding]
 
     def handler(request: httpx.Request) -> httpx.Response:
         return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=streams.pop(0))
@@ -683,24 +667,23 @@ def handler(request: httpx.Request) -> httpx.Response:
             streamable_http_client("http://test/mcp", http_client=http) as (read, write),
         ):
             await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="dup-1", method="tools/call", params={})))
-            await doomed.opened.wait()
-            # The caller abandoned the first attempt and re-issued the id while
-            # the first stream lingers server-side.
+            await stale.opened.wait()
             await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="dup-1", method="tools/call", params={})))
             await succeeding.opened.wait()
-            doomed.die.set()
+            stale.deliver.set()
             await anyio.wait_all_tasks_blocked()
             succeeding.deliver.set()
             reply = await read.receive()
     assert isinstance(reply, SessionMessage)
     assert isinstance(reply.message, JSONRPCResponse), reply.message
-    assert reply.message.id == "dup-1"
+    assert reply.message.result == {"origin": "fresh"}
 
 
 @pytest.mark.anyio
 async def test_a_202_to_a_request_resolves_the_waiter_with_an_error() -> None:
-    """A server that answers a request with 202 Accepted has declared no response
-    will follow; the transport resolves the waiter instead of parking the caller forever."""
+    """SDK-defined: a server that answers a request with 202 Accepted has declared no
+    response will follow (the spec requires SSE or JSON for requests); the transport
+    resolves the waiter with INVALID_REQUEST instead of parking the caller forever."""
 
     def handler(request: httpx.Request) -> httpx.Response:
         return httpx.Response(202)
@@ -717,7 +700,7 @@ def handler(request: httpx.Request) -> httpx.Response:
     assert isinstance(reply, SessionMessage)
     assert isinstance(reply.message, JSONRPCError)
     assert reply.message.id == "listen-1"
-    assert reply.message.error.code == CONNECTION_CLOSED
+    assert reply.message.error.code == INVALID_REQUEST
 
 
 def _abandoned_request_context(