Rework channel liveness, stream ids, and init ordering after review

The second review round found several defects that traced back to a few
structural gaps rather than isolated bugs; this addresses the gaps.

- Channel delivery is now a value: write() returns whether the message
  reached the event store or an attached response, an EventStore failure is
  contained inside the write (resumability degrades, the stream survives,
  no store text on the wire), and attach() returns None on a dead channel
  so nothing can stream from one. A server-to-client request that cannot
  reach any client fails the caller with CONNECTION_CLOSED instead of
  parking it, and the JSON-response body derives from the channel's
  recorded outcome (result, terminated-404, or 500) rather than a
  "cannot happen" branch.
- Stream ids handed to the EventStore are minted by the transport in a
  session-scoped namespace, so a client-chosen request id can no longer
  name the standalone GET stream and two sessions on one store can no
  longer replay each other's frames.
- One SSE-response runner owns the pump, error containment, and cleanup
  for the POST, GET, and replay responses (the POST path had lost the
  guard its siblings kept).
- A session-level gate holds requests that arrive while an initialize is
  still being served until the handshake commits, restoring the ordering
  the stream-pair driver's parked read loop used to guarantee.
- The POSTed client message is delivered even when the 202 could not be
  written back, and the correlator marks the single site where the
  cancelled-request answer policy lives for every transport.

Adds regression tests for each of the above.

Author: maxisbey

docs/migration.md

@@ -744,10 +744,14 @@ transport = StreamableHTTPServerTransport(mcp_session_id=session_id, ...)
 async with transport.connect() as (read_stream, write_stream):
     await server.run(read_stream, write_stream, server.create_initialization_options())
 
-# After (v2): let the manager own transports (mount it or use streamable_http_app())
-session_manager = StreamableHTTPSessionManager(app=server, event_store=..., json_response=...)
+# After (v2): serve the app the SDK builds ...
+app = server.streamable_http_app(event_store=..., json_response=...)
 ```
 
+... or, when composing your own Starlette/FastAPI app, mount a `StreamableHTTPSessionManager` and
+enter `session_manager.run()` in the lifespan — see [Mounting the ASGI app](run/asgi.md) for the
+full wiring.
+
 Behaviour clarified in the same change:
 
 - In JSON-response mode a *request-scoped* server-to-client request (`ctx.elicit()`, or any
@@ -757,6 +761,18 @@ Behaviour clarified in the same change:
   `related_request_id`) are unchanged and still ride the standalone GET stream.
 - A GET carrying `Last-Event-ID` on a server without an `EventStore` opens the standalone stream
   as a plain GET would, since there is nothing to replay.
+- Two concurrent POSTs that share a JSON-RPC request id each keep their own response stream; the
+  second no longer silently takes over the first's queue.
+- Stream ids handed to your `EventStore` are minted by the transport in its own session-scoped
+  namespace (previously the raw `str(request_id)` and a single global GET-stream key), so two
+  sessions sharing one store no longer collide, and a `Last-Event-ID` replay only releases frames
+  of the requesting session's own streams. Treat the ids as opaque.
+- A failing `EventStore.store_event` degrades resumability for that message rather than taking
+  the stream down: the message is still delivered live (with no event id to resume from) and the
+  store's exception is logged, never sent to the client.
+- A server-to-client request that can reach no client at all (no attached stream and nothing
+  storing it, or a request-scoped one in JSON-response mode) fails the calling handler with
+  `CONNECTION_CLOSED` instead of parking it for an answer that cannot arrive.
 
 ### `MCPServer.get_context()` removed
 

src/mcp/server/runner.py

@@ -230,7 +230,12 @@ async def _inner(ctx: ServerRequestContext[LifespanT, Any]) -> HandlerResult:
         result = _dump_result(await call(ctx))
         if method == "initialize":
             # Commit only on chain success, so a middleware veto leaves no state.
-            # Race-free: the read loop is parked until this call returns.
+            # Race-free for the session's first handshake: the transport runs no
+            # other request until it returns (a stream driver's read loop is
+            # parked here; streamable HTTP holds later requests behind the
+            # in-progress initialize, and the session id only ships with its
+            # response). A repeated initialize on an established session (a
+            # recorded divergence) recommits alongside whatever is running.
             # TODO: this re-reads the wire `params`, so a middleware that rewrote
             # `ctx.params` (or `ctx.method`, or short-circuited without `call_next`)
             # can leave `connection.protocol_version` out of step with the

src/mcp/server/streamable_http.py

@@ -448,6 +448,11 @@ async def serve_inbound(
                 # result write above did not happen - no double response.
                 # TODO(L38): spec says SHOULD NOT respond after cancel;
                 # the existing server always has, so match that for now.
+                # This is the single site every transport shares for the
+                # cancelled-request answer policy; change it here for all
+                # of them (a transport whose response stream must still end
+                # sees this write, so a suppressed answer needs the stream
+                # terminated another way).
                 answer_write_started = True
                 await write_error(ErrorData(code=0, message="Request cancelled"))
         except anyio.get_cancelled_exc_class():

src/mcp/shared/_correlation.py

@@ -280,7 +280,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 `except Exception` arms
-around the standalone-GET and replay `response(...)` calls in `server/streamable_http.py`.
+around the SSE-response runner (`_run_sse_response`) and the replay entry path in
+`server/streamable_http.py`.
 `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

tests/interaction/README.md

@@ -423,7 +423,7 @@ async def mock_receive():
             assert transport._terminated, "Transport should be terminated after stateless request"
 
             # Verify internal state is cleaned up: no request streams left open.
-            assert not transport._channels, "Transport should have no active request channels"
+            assert not transport._streams, "Transport should have no active request streams"
             assert not transport._standalone.attached, "Transport should have no standalone stream attached"