Fix two waiter-resolution gaps in the streamable HTTP client

- A stale POST whose SSE stream ends responselessly no longer synthesizes
  CONNECTION_CLOSED for a request id that a same-id successor has re-issued;
  the pending waiter belongs to the successor, whose real response still
  arrives. Resolution is now identity-guarded the same way the in-flight
  registry's cleanup already was.
- A server that answers a JSON-RPC request with 202 Accepted now resolves
  the waiter with CONNECTION_CLOSED instead of leaving a timeout-free caller
  parked forever, including client.listen() enter.
- Rewrite the subscriptions story README around the client.listen() context
  manager the example now uses; it still described the old escape-hatch flow.
- Extend the docs import note to cover the watch.py snippet.

Author: maxisbey

docs/handlers/subscriptions.md

@@ -83,7 +83,7 @@ Two more properties of the handle:
 
 Open the subscription first, then start the watcher and get on with your work.
 
-`app.py` imports `BOARD` and `read_board` from the previous example, which this repo stores as `tutorial003.py`. If you save the rendered files side by side as `client.py` and `app.py`, write `from client import BOARD, read_board` instead.
+`app.py` imports `BOARD` and `read_board` from the previous example, which this repo stores as `tutorial003.py`. If you save the rendered files side by side as `client.py` and `app.py`, write `from client import BOARD, read_board` instead. The `watch.py` example further down imports `read_board` the same way.
 
 === "asyncio"
 

examples/stories/subscriptions/README.md

@@ -7,16 +7,16 @@ server publishes with `ctx.notify_resource_updated(uri)` /
 per-stream filtering, subscription-id tagging). Replaces the handshake-era
 `resources/subscribe` + standalone-GET notification path.
 
-The client edits a note it did not subscribe to (silence), edits the one it
-did (a tagged `notifications/resources/updated`), registers a tool at runtime
-(`notifications/tools/list_changed`, then re-lists and calls it), and finally
-stops listening - cancelling the parked request releases the local task, and
-closing the connection ends the stream server-side.
+The client opens the stream with `client.listen(...)`, edits a note it did
+not subscribe to (silence), edits the one it did (a typed `ResourceUpdated`),
+registers a tool at runtime (a typed `ToolsListChanged`, then re-lists and
+calls it), and finally leaves the `async with` block, which ends the
+subscription while the connection lives on.
 
 ## Run it
 
 ```bash
-# HTTP — the client self-hosts the server on a free port, runs, then tears it
+# HTTP: the client self-hosts the server on a free port, runs, then tears it
 # down (subscriptions/listen is 2026-era only)
 uv run python -m stories.subscriptions.client --http
 # same, against the lowlevel-API server variant
@@ -25,17 +25,18 @@ uv run python -m stories.subscriptions.client --http --server server_lowlevel
 
 ## What to look at
 
-- `client.py` — stream frames arrive as ordinary server notifications via the
-  constructor-only `message_handler=`. There is no client-side listen API yet,
-  so opening the stream drops to the `client.session` escape hatch; the request
-  parks for the stream's lifetime. Cancelling it releases the local task; over
-  HTTP the server-side stream ends when the connection closes. Every frame's
-  `_meta["io.modelcontextprotocol/subscriptionId"]` is the listen request's
-  JSON-RPC id.
-- `server.py` — publishing is one `await ctx.notify_*()` line per change; the
+- `client.py`: the whole subscription is one context manager,
+  `async with client.listen(...) as sub`. Entering waits for the server's
+  acknowledgment, so `sub.honored` is already in hand on the first line of the
+  block. Events arrive as typed values from `anext(sub)`; the edit to the
+  unsubscribed note never shows up, because the filter is enforced
+  server-side. Leaving the block ends the subscription (over HTTP the SDK
+  closes that request's response stream) and the session carries on, which the
+  final `search` call proves.
+- `server.py`: publishing is one `await ctx.notify_*()` line per change; the
   filter, the tagging, and the ack ordering are the SDK's job. Publishing with
   no subscribers is a no-op.
-- `server_lowlevel.py` — the same machinery held by hand: an
+- `server_lowlevel.py`: the same machinery held by hand: an
   `InMemorySubscriptionBus`, handlers that `await bus.publish(...)`, and
   `ListenHandler(bus)` passed as `on_subscriptions_listen=`. A multi-replica
   deployment swaps the bus for one backed by its own pub/sub
@@ -51,7 +52,7 @@ uv run python -m stories.subscriptions.client --http --server server_lowlevel
 
 ## Spec
 
-[Subscriptions — basic utilities](https://modelcontextprotocol.io/specification/draft/basic/utilities/subscriptions)
+[Subscriptions, basic utilities](https://modelcontextprotocol.io/specification/draft/basic/utilities/subscriptions)
 
 ## See also
 

src/mcp/client/streamable_http.py

@@ -71,6 +71,8 @@ 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)
@@ -328,6 +330,12 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
         ) as response:
             if response.status_code == 202:
                 logger.debug("Received 202 Accepted")
+                if isinstance(message, JSONRPCRequest):
+                    # 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"
+                    )
                 return
 
             if response.status_code >= 400:
@@ -445,21 +453,24 @@ 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.read_stream_writer, original_request_id, "SSE stream ended without a response"
-            )
+            await self._resolve_abandoned_request(ctx, original_request_id, "SSE stream ended without a response")
 
-    async def _resolve_abandoned_request(
-        self, read_stream_writer: StreamWriter, request_id: RequestId, message: str
-    ) -> None:
+    async def _resolve_abandoned_request(self, ctx: RequestContext, request_id: RequestId, message: str) -> None:
         """Resolve a request whose response can never arrive with a synthesized error.
 
-        Best-effort: a closed read stream means the session is tearing down.
+        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.
         """
+        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_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data))
         try:
-            await read_stream_writer.send(error_msg)
+            await ctx.read_stream_writer.send(error_msg)
         except (anyio.BrokenResourceError, anyio.ClosedResourceError):
             logger.debug("read stream closed before request %r could be resolved", request_id)
 
@@ -480,9 +491,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.read_stream_writer,
-                original_request_id,
-                "SSE stream ended and reconnection attempts were exhausted",
+                ctx, original_request_id, "SSE stream ended and reconnection attempts were exhausted"
             )
             return
 
@@ -591,6 +600,7 @@ async def handle_request_async():
                             modern=self._protocol_version_header in MODERN_PROTOCOL_VERSIONS,
                         )
                         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

@@ -631,6 +631,95 @@ 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."""
+
+    def __init__(self, response_body: dict[str, Any]) -> None:
+        self._event = f"data: {json.dumps(response_body)}\n\n".encode()
+        self.opened = anyio.Event()
+        self.deliver = anyio.Event()
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        self.opened.set()
+        await self.deliver.wait()
+        yield self._event
+
+    async def aclose(self) -> None:
+        pass
+
+
+@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]
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=streams.pop(0))
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http,
+            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 write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="dup-1", method="tools/call", params={})))
+            await succeeding.opened.wait()
+            doomed.die.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"
+
+
+@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."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(202)
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http,
+            streamable_http_client("http://test/mcp", http_client=http) as (read, write),
+        ):
+            await write.send(
+                SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="listen-1", method="subscriptions/listen", params={}))
+            )
+            reply = await read.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.id == "listen-1"
+    assert reply.message.error.code == CONNECTION_CLOSED
+
+
 def _abandoned_request_context(
     http: httpx.AsyncClient, send: ContextSendStream[SessionMessage | Exception]
 ) -> RequestContext: