fix(server): terminate streamable HTTP sessions on manager shutdown

[baiyuxi930826]

Summary

Fixes #2150.

Streamable HTTP sessions were not terminated during manager shutdown, and terminate() did not close active SSE writers. That left EventSourceResponse / ASGI callables incomplete on CTRL+C (uvicorn incomplete-response noise and resource leaks under multi-session hosts).

Changes

1. StreamableHTTPSessionManager.run() — call terminate() on active transports before cancelling the task group.
2. StreamableHTTPServerTransport.terminate() — close all _sse_stream_writers via close_sse_stream() first so SSE responses can finish.
3. Regression tests for writer close + manager shutdown terminate.

Relation to existing work

• Open PR fix: terminate active StreamableHTTP sessions during shutdown #2253 targets v1.x, is older, and does not include the regression suite. This PR targets main with tests.
• Happy to close/adjust if maintainers prefer landing via fix: terminate active StreamableHTTP sessions during shutdown #2253 or a different base.

Test plan

• uv run pytest tests/server/test_streamable_http_manager.py::test_terminate_closes_active_sse_stream_writers -q
• uv run pytest tests/server/test_streamable_http_manager.py::test_manager_shutdown_terminates_active_sessions -q
• uv run pytest tests/server/test_streamable_http_manager.py -q (32 passed locally)

[cubic-dev-ai]

All reported issues were addressed across 4 files

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[baiyuxi930826]

Addressed the cubic P2:

• Standalone GET SSE: register _sse_stream_writers[GET_STREAM_KEY] as soon as the writer is created; pop on writer exit / response finally so terminate() can close it.
• Replay SSE: register a provisional _replay:... writer key before EventSourceResponse starts (covers the window while replay_events_after runs), then re-key to stream_id when live-tail begins; always pop in finally.
• Tests: test_terminate_closes_standalone_get_sse_writer_when_registered, test_terminate_closes_provisional_replay_sse_writer (4 terminate/shutdown tests green locally).

Happy to iterate if maintainers want a different provisional-key scheme.

[cubic-dev-ai]

1 issue found across 2 files (changes from recent commits).

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="src/mcp/server/streamable_http.py">

<violation number="1" location="src/mcp/server/streamable_http.py:704">
P2: Concurrent standalone GETs can still both establish, then overwrite the shared SSE-writer entry; shutdown closes only the last writer and one response remains incomplete. Include `_sse_stream_writers` in the active-GET conflict check so registration itself reserves `GET_STREAM_KEY`.</violation>
</file>

<sub>Tip: Review your code locally with the cubic CLI to iterate faster.

Re-trigger cubic</sub>

[cubic-dev-ai]

P2: Concurrent standalone GETs can still both establish, then overwrite the shared SSE-writer entry; shutdown closes only the last writer and one response remains incomplete. Include _sse_stream_writers in the active-GET conflict check so registration itself reserves GET_STREAM_KEY.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At src/mcp/server/streamable_http.py, line 704:

<comment>Concurrent standalone GETs can still both establish, then overwrite the shared SSE-writer entry; shutdown closes only the last writer and one response remains incomplete. Include `_sse_stream_writers` in the active-GET conflict check so registration itself reserves `GET_STREAM_KEY`.</comment>

<file context>
@@ -700,6 +700,8 @@ async def _handle_get_request(self, request: Request, send: Send) -> None:
         # Create SSE stream
         sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[SSEEvent](0)
+        # Register immediately so session terminate() can close this ASGI response.
+        self._sse_stream_writers[GET_STREAM_KEY] = sse_stream_writer
 
         async def standalone_sse_writer():
</file context>

[baiyuxi930826]

Pushed a follow-up for CI coverage (fail-under=100):

• Extended test_terminate_closes_active_sse_stream_writers so terminate also cleans an orphaned _request_streams entry (covers the remaining-keys loop / line ~800).
• Marked the defensive registered_stream_id is not None guard as # pragma: no branch (always set before the guarded body on that path).

Local: the four #2150 terminate tests pass. Please re-run CI on the latest commit.