replay_sender finally + replace _mint_priming_event unit tests with end-to-end

Adds a finally to replay_sender (mirroring _run_sse_writer) so resumed
connections clean up _sse_stream_writers[stream_id] and _request_streams[stream_id]
on disconnect. Nested inside the stream_id-is-set block so no edge-case
None-handling is needed.

Drops the four unit tests that poked _mint_priming_event directly and adds an
end-to-end test in test_hosting_resume.py asserting the event store records
[(S, priming), (S, msg1), ..., (S, response)] for a real POST through a real
MCPServer + transport. The dropped tests' branches remain covered by the
existing high-level tests in the same file.

Author: maxisbey

src/mcp/server/streamable_http.py

@@ -889,28 +889,32 @@ async def send_event(event_message: EventMessage) -> None:
 
                         # If stream ID not in mapping, create it
                         if stream_id and stream_id not in self._request_streams:  # pragma: no branch
-                            # Register SSE writer so close_sse_stream() can close it
-                            self._sse_stream_writers[stream_id] = sse_stream_writer
-
-                            # Prime the resumed connection so the client sees the stream
-                            # 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:
-                                await sse_stream_writer.send(priming_event)
-
-                            # Create new request streams for this connection
-                            self._request_streams[stream_id] = anyio.create_memory_object_stream[EventMessage](
-                                REQUEST_STREAM_BUFFER_SIZE
-                            )
-                            msg_reader = self._request_streams[stream_id][1]
-
-                            # Forward messages to SSE
-                            async with msg_reader:
-                                async for event_message in msg_reader:
-                                    event_data = self._create_event_data(event_message)
-
-                                    await sse_stream_writer.send(event_data)
+                            try:
+                                # Register SSE writer so close_sse_stream() can close it
+                                self._sse_stream_writers[stream_id] = sse_stream_writer
+
+                                # Prime the resumed connection so the client sees the stream
+                                # 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:
+                                    await sse_stream_writer.send(priming_event)
+
+                                # Create new request streams for this connection
+                                self._request_streams[stream_id] = anyio.create_memory_object_stream[EventMessage](
+                                    REQUEST_STREAM_BUFFER_SIZE
+                                )
+                                msg_reader = self._request_streams[stream_id][1]
+
+                                # Forward messages to SSE
+                                async with msg_reader:
+                                    async for event_message in msg_reader:
+                                        event_data = self._create_event_data(event_message)
+
+                                        await sse_stream_writer.send(event_data)
+                            finally:
+                                self._sse_stream_writers.pop(stream_id, None)
+                                await self._clean_up_memory_streams(stream_id)
                 except anyio.ClosedResourceError:  # pragma: lax no cover
                     # Expected when close_sse_stream() is called
                     logger.debug("Replay SSE stream closed by close_sse_stream()")

tests/interaction/transports/test_hosting_resume.py

@@ -113,6 +113,43 @@ async def test_a_post_sse_stream_begins_with_a_priming_event_and_stamps_every_ev
     )
 
 
+@requirement("hosting:resume:priming")
+async def test_the_priming_row_is_stored_before_any_handler_output_for_that_stream() -> None:
+    """The priming cursor is the first row the event store records for a request's stream.
+
+    The POST handler stores the priming row before dispatching the request, so by construction
+    it precedes anything `message_router` can store for that stream id.
+    """
+    store = SequencedEventStore()
+    mcp = MCPServer("resumable")
+
+    @mcp.tool()
+    async def burst(ctx: Context) -> str:
+        await ctx.info("a")  # pyright: ignore[reportDeprecated]
+        await ctx.info("b")  # pyright: ignore[reportDeprecated]
+        await ctx.info("c")  # pyright: ignore[reportDeprecated]
+        return "done"
+
+    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(  # pragma: no branch
+                "POST", "/mcp", content=_tools_call(2, "burst", {}), headers=base_headers(session_id=session_id)
+            ) as response:
+                await _read_events(response, 5)
+
+    # initialize wrote two rows (its own priming + response); everything after is this call.
+    call_rows = store._events[2:]
+    stream_id = call_rows[0][0]
+    assert [(s, None if m is None else type(m).__name__) for s, m in call_rows] == [
+        (stream_id, None),
+        (stream_id, "JSONRPCNotification"),
+        (stream_id, "JSONRPCNotification"),
+        (stream_id, "JSONRPCNotification"),
+        (stream_id, "JSONRPCResponse"),
+    ]
+
+
 @requirement("hosting:resume:replay")
 @requirement("hosting:resume:stream-scoped")
 @requirement("hosting:resume:buffered-replay")

tests/shared/test_streamable_http.py

@@ -1634,34 +1634,6 @@ async def test_handle_sse_event_skips_empty_data() -> None:
         await read_stream.aclose()
 
 
-@pytest.mark.anyio
-async def test_priming_event_not_sent_for_old_protocol_version() -> None:
-    """`_mint_priming_event` skips for old protocol versions (backwards compat)."""
-    transport = StreamableHTTPServerTransport("/mcp", event_store=SimpleEventStore())
-    assert await transport._mint_priming_event("test-request-id", "2025-06-18") is None
-    assert await transport._mint_priming_event("test-request-id-2", "2025-11-25") is not None
-
-
-@pytest.mark.anyio
-async def test_priming_event_not_sent_without_event_store() -> None:
-    """`_mint_priming_event` returns `None` when no event_store is configured."""
-    transport = StreamableHTTPServerTransport("/mcp")
-    assert await transport._mint_priming_event("test-request-id", "2025-11-25") is None
-
-
-@pytest.mark.anyio
-async def test_priming_event_includes_retry_interval() -> None:
-    """`_mint_priming_event` includes the retry field when `retry_interval` is set."""
-    transport = StreamableHTTPServerTransport(
-        "/mcp",
-        event_store=SimpleEventStore(),
-        retry_interval=5000,
-    )
-    event = await transport._mint_priming_event("test-request-id", "2025-11-25")
-    assert event is not None
-    assert event["retry"] == 5000
-
-
 @pytest.mark.anyio
 async def test_close_sse_stream_callback_not_provided_for_old_protocol_version() -> None:
     """close_sse_stream callbacks are only provided for protocol versions that support polling."""
@@ -1694,17 +1666,6 @@ async def test_close_sse_stream_callback_not_provided_for_old_protocol_version()
     assert session_msg_new.metadata.close_standalone_sse_stream is not None
 
 
-@pytest.mark.anyio
-async def test_priming_event_not_sent_for_unknown_protocol_version() -> None:
-    """`_mint_priming_event` treats unrecognized version strings conservatively.
-
-    A garbage version must not be mistaken for a future one (lexicographically
-    "zzz" sorts after every date-shaped revision).
-    """
-    transport = StreamableHTTPServerTransport("/mcp", event_store=SimpleEventStore())
-    assert await transport._mint_priming_event("test-request-id", "zzz") is None
-
-
 @pytest.mark.anyio
 async def test_close_sse_stream_callback_not_provided_for_unknown_protocol_version() -> None:
     """close_sse_stream callbacks are withheld when the client's version is unrecognized."""