Mint priming before per-request state; cover old-pv replay path

Hoists _mint_priming_event to the top of the SSE arm so a user EventStore
raising on the priming row returns a 500 with no per-request state allocated
(previously _request_streams[id] and _sse_stream_writers[id] leaked for the
session). The shared _request_streams registration is pushed into each branch.

Adds an old-pv-reconnect test in test_hosting_resume.py covering the
priming_event-is-None replay arm; drops the no-branch pragma. The new
priming-failure test covers the outer except handler, so its pragmas and the
dead 'if writer:' check are removed.

Author: maxisbey

src/mcp/server/streamable_http.py

@@ -546,15 +546,13 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 else request.headers.get(MCP_PROTOCOL_VERSION_HEADER, DEFAULT_NEGOTIATED_VERSION)
             )
 
-            # Extract the request ID outside the try block for proper scope
             request_id = str(message.id)
-            # Register this stream for the request ID
-            self._request_streams[request_id] = anyio.create_memory_object_stream[EventMessage](
-                REQUEST_STREAM_BUFFER_SIZE
-            )
-            request_stream_reader = self._request_streams[request_id][1]
 
             if self.is_json_response_enabled:
+                self._request_streams[request_id] = anyio.create_memory_object_stream[EventMessage](
+                    REQUEST_STREAM_BUFFER_SIZE
+                )
+                request_stream_reader = self._request_streams[request_id][1]
                 # Process the message
                 metadata = ServerMessageMetadata(request_context=request)
                 session_message = SessionMessage(message, metadata=metadata)
@@ -598,16 +596,18 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 finally:
                     await self._clean_up_memory_streams(request_id)
             else:
-                # Create SSE stream
-                sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0)
+                # Mint the priming event before any per-request state exists:
+                # `EventStore.store_event` is user code and may raise, in which
+                # case the outer handler returns a 500 with nothing to clean up.
+                # Still strictly precedes dispatch, so storage order == wire order.
+                priming_event = await self._mint_priming_event(request_id, protocol_version)
 
-                # Store writer reference so close_sse_stream() can close it
+                sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0)
                 self._sse_stream_writers[request_id] = sse_stream_writer
-
-                # Store the priming event before the request is dispatched so its
-                # event-store position precedes anything message_router can store
-                # for this id (storage order == wire order by construction).
-                priming_event = await self._mint_priming_event(request_id, protocol_version)
+                self._request_streams[request_id] = anyio.create_memory_object_stream[EventMessage](
+                    REQUEST_STREAM_BUFFER_SIZE
+                )
+                request_stream_reader = self._request_streams[request_id][1]
 
                 headers = {
                     "Cache-Control": "no-cache, no-transform",
@@ -638,20 +638,16 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 finally:
                     await sse_stream_reader.aclose()
 
-        except Exception as err:  # pragma: lax no cover
-            # Reached only when something raises during POST handling outside
-            # the per-SSE-stream guard above; whether tests reach this depends
-            # on client teardown timing.
+        except Exception as err:
             logger.exception("Error handling POST request")
             response = self._create_error_response(
                 f"Error handling POST request: {err}",
                 HTTPStatus.INTERNAL_SERVER_ERROR,
                 INTERNAL_ERROR,
             )
             await response(scope, receive, send)
-            if writer:
-                await writer.send(Exception(err))
-            return  # pragma: no cover
+            await writer.send(Exception(err))
+            return
 
     async def _handle_get_request(self, request: Request, send: Send) -> None:
         """Handle GET request to establish SSE.
@@ -900,7 +896,7 @@ async def send_event(event_message: EventMessage) -> None:
                             # is re-registered. The replay→live-tail ordering window here
                             # is pre-existing and tracked separately.
                             priming_event = await self._mint_priming_event(stream_id, replay_protocol_version)
-                            if priming_event is not None:  # pragma: no branch
+                            if priming_event is not None:
                                 await sse_stream_writer.send(priming_event)
 
                             # Create new request streams for this connection

tests/interaction/transports/test_hosting_resume.py

@@ -182,6 +182,46 @@ async def count(ctx: Context) -> str:
     )
 
 
+@requirement("hosting:resume:priming")
+async def test_a_pre_2025_11_25_reconnect_replays_without_minting_a_priming_event() -> None:
+    """A pre-2025-11-25 client reconnecting via Last-Event-ID gets the replay with no priming row.
+
+    The store-length assertion is the load-bearing proof that no priming cursor was minted.
+    """
+    release = anyio.Event()
+    store = SequencedEventStore()
+    mcp = MCPServer("resumable")
+
+    @mcp.tool()
+    async def count(ctx: Context) -> str:
+        await ctx.info("tick 1")  # pyright: ignore[reportDeprecated]
+        await release.wait()
+        await ctx.info("tick 2")  # pyright: ignore[reportDeprecated]
+        return "counted"
+
+    async with mounted_app(mcp, event_store=store, retry_interval=0) as (http, _):
+        session_id = await initialize_via_http(http)
+        with anyio.fail_after(5):
+            async with http.stream(
+                "POST", "/mcp", content=_tools_call(1, "count", {}), headers=base_headers(session_id=session_id)
+            ) as response:
+                _, first = await _read_events(response, 2)
+            release.set()
+            await store.wait_until_stored(6)
+            old_client_headers = base_headers(session_id=session_id) | {
+                "mcp-protocol-version": "2025-06-18",
+                "last-event-id": first.id,
+            }
+            async with http.stream("GET", "/mcp", headers=old_client_headers) as replay:  # pragma: no branch
+                assert replay.status_code == 200
+                missed = await _read_events(replay, 2)
+
+    assert [(event.id, bool(event.data)) for event in missed] == snapshot([("5", True), ("6", True)])
+    # No priming cursor was minted on reconnect: the store still holds only the six rows
+    # written before the GET (init priming+response, POST priming, tick 1, tick 2, result).
+    assert len(store._events) == 6
+
+
 @requirement("hosting:resume:bad-event-id")
 async def test_an_unknown_last_event_id_yields_an_empty_replay_stream() -> None:
     """A Last-Event-ID the event store cannot map produces an empty SSE stream rather than an error.

tests/server/test_streamable_http_router.py

@@ -2,6 +2,7 @@
 
 import anyio
 import pytest
+from starlette.types import Message, Scope
 
 from mcp.server.streamable_http import (
     REQUEST_STREAM_BUFFER_SIZE,
@@ -28,6 +29,14 @@ async def replay_events_after(self, last_event_id: EventId, send_callback: Event
         raise NotImplementedError
 
 
+class _PrimingFailingStore(EventStore):
+    async def store_event(self, stream_id: StreamId, message: JSONRPCMessage | None) -> EventId:
+        raise RuntimeError("backend unavailable")
+
+    async def replay_events_after(self, last_event_id: EventId, send_callback: EventCallback) -> StreamId | None:
+        raise NotImplementedError
+
+
 @pytest.mark.anyio
 async def test_router_unconsumed_request_stream_does_not_block_siblings() -> None:
     """A response whose `sse_writer` is not yet receiving must not park the router (#1764).
@@ -99,3 +108,51 @@ async def test_priming_event_is_stored_before_any_routed_message() -> None:
     assert store.stored[0] == ("A", None)
     assert [sid for sid, _ in store.stored] == ["A"] * 6
     assert all(msg is not None for _, msg in store.stored[1:])
+
+
+@pytest.mark.anyio
+async def test_priming_store_failure_leaves_no_per_request_state() -> None:
+    """`EventStore.store_event` raising on the priming row must not leak per-request entries."""
+    transport = StreamableHTTPServerTransport(
+        mcp_session_id=None,
+        is_json_response_enabled=False,
+        event_store=_PrimingFailingStore(),
+    )
+
+    body = b'{"jsonrpc":"2.0","id":"req-1","method":"tools/list","params":{}}'
+    scope: Scope = {
+        "type": "http",
+        "method": "POST",
+        "path": "/",
+        "query_string": b"",
+        "headers": [
+            (b"accept", b"application/json, text/event-stream"),
+            (b"content-type", b"application/json"),
+            (b"mcp-protocol-version", b"2025-11-25"),
+        ],
+    }
+    body_sent = False
+
+    async def receive() -> Message:
+        nonlocal body_sent
+        if not body_sent:
+            body_sent = True
+            return {"type": "http.request", "body": body, "more_body": False}
+        raise NotImplementedError
+
+    sent: list[Message] = []
+
+    async def asgi_send(message: Message) -> None:
+        sent.append(message)
+
+    async with transport.connect() as (read_stream, _write_stream):
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(transport.handle_request, scope, receive, asgi_send)
+            with anyio.fail_after(5):
+                forwarded = await read_stream.receive()
+            assert isinstance(forwarded, Exception)
+
+    assert transport._request_streams == {}
+    assert transport._sse_stream_writers == {}
+    assert sent[0]["type"] == "http.response.start"
+    assert sent[0]["status"] == 500