fix(server): reject concurrent duplicate JSON-RPC request ids

[arimu1]

Summary

Stateful streamable HTTP registers each in-flight request's response stream in a per-session dict keyed by the bare JSON-RPC request id. _handle_post_request assigned that slot unconditionally, so a second concurrent POST reusing an in-flight id silently overwrote the first request's stream and responses were cross-wired (one caller received the other's payload; the other hung).

Both response modes (JSON and SSE) now reject the collision with HTTP 409 Conflict and a JSON-RPC INVALID_REQUEST error instead of overwriting. The SSE path reserves the routing slot before EventStore.store_event (priming) so a concurrent same-id POST cannot pass the guard during that await — a race left open by a simpler check-before-register approach when resumability is enabled. If priming raises, the reservation is released.

Sequential reuse of an id after the earlier request has completed remains allowed.

Fixes #3137

Related

• Prior attempt Fix concurrent requests sharing a JSON-RPC id hijacking each other's responses #3163 (self-closed, unmerged) covered the basic guard but not the priming race window.
• Related open work on the same class of bug: streamable HTTP (stateful): concurrent requests with duplicate JSON-RPC ids cross-wire responses — one request receives another's payload, the other hangs #3060 / fix(streamable-http): reject duplicate in-flight request ids #3063 (transport) and fix(dispatcher): reject duplicate in-flight request ids #3064 (dispatcher). This PR targets Concurrent requests sharing a JSON-RPC id on one session receive each other's responses #3137 and improves on Fix concurrent requests sharing a JSON-RPC id hijacking each other's responses #3163 by closing the priming TOCTOU.

Test plan

• test_post_duplicate_request_id_rejected_while_first_still_in_flight (SSE) — second POST gets 409; first still completes with its own result
• test_json_response_duplicate_request_id_rejected_while_first_still_in_flight (JSON mode)
• test_request_id_reuse_after_completion_allowed — sequential reuse still works
• test_duplicate_request_id_rejected_during_priming_event_store — guard holds across gated priming await
• uv run --frozen pytest tests/shared/test_streamable_http.py — 69 passed
• uv run --frozen ruff check / ruff format --check on touched files
• uv run --frozen pyright on touched files

Disclosure

This change was written with AI assistance (Grok / Cursor); I reviewed the diff, ran the tests above, and can answer questions about the change.

[cubic-dev-ai]

All reported issues were addressed across 2 files

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

Re-trigger cubic</sub>

[Palo-Alto-AI-Research-Lab]

Reviewed by reading streamable_http.py at HEAD alongside the diff. I did not run the suite locally — everything below is from the code, and I've marked where I stopped short of proof.

The ordering change is the right call, and for a reason worth stating explicitly. Reserving the slot before _mint_priming_event converts a check-then-act across an await into a check-then-act with no suspension point between them. #3163 looked equivalent and wasn't: with resumability on, EventStore.store_event is user code, and any user code that awaits opens the window. That distinction is easy to lose in a follow-up refactor, so it's worth a line in the comment saying the invariant is "no await between the membership test and the assignment" rather than "reserve early".

I also checked the cleanup path, because await inside except BaseException is usually where this class of fix leaks: under cancellation the first checkpoint re-raises and the rest of the handler never runs. Here it's safe — _clean_up_memory_streams pops in a finally, and finally still executes when aclose() re-raises Cancelled. Worth knowing that it holds by accident of that finally rather than by design; a shielded scope would make it hold on purpose.

The gap: _request_streams has three writers, and this guards one.

• _handle_post_request — guarded by this PR.
• the standalone GET handler — guards itself with if GET_STREAM_KEY in self._request_streams (L727).
• the resumption path — self._request_streams[stream_id] = ... (L939) with no membership test at all.

So a Last-Event-ID resumption for stream_id still overwrites a live POST slot with that id, which is the same cross-wiring as #3137 arriving through a different door. Conversely, once a POST holds the id, the resumption silently replaces its own registration. Not something this PR has to fix, but it means "duplicate ids can no longer cross-wire responses" isn't yet true as a property of the table — and a test asserting the POST case will pass while that door stays open.

The narrower thing I'd fix here, since you're already touching the key:

request_id = str(message.id) (L601) and GET_STREAM_KEY = "_GET_stream" (L61) live in the same dict, so the key space conflates three namespaces:

1. "id": 1 and "id": "1" in one session are distinct JSON-RPC ids that collapse to the same key. Pre-PR they cross-wired; post-PR the second gets a 409 that is, from the client's side, wrong — nothing with that id is in flight.
2. "id": "_GET_stream" is a legal string id. Send it, and the session's next legitimate GET is refused with "Only one SSE stream is allowed per session" while no SSE stream exists — a client can lock itself out of the standalone stream, and the error names the wrong cause.

One root, one small fix inside your design: make the key type-tagged rather than stringly — ("req", message.id) vs ("get",), or a str prefix if you'd rather not touch the annotation. The guard, the GET check and close_sse_stream all keep working unchanged, and (2) stops being reachable by construction. The str() at L1023/L1040 on the outgoing routing path would need the same treatment to stay consistent, which is the part that makes this worth doing in one go rather than piecemeal.

What I did not check: whether any client in the wild depends on the old overwrite behaviour, whether 409 is the status the spec prefers here over a 200-with-error-body, and I did not run tests/shared/test_streamable_http.py — so I can't confirm the priming-race test fails without the reservation move, only that it should.

One small thing on the message itself: a client that retries after a timeout now hits a hard 409 with no stated remedy. Naming notifications/cancelled in the error text would tell it what to do instead of retrying into the same wall.

Happy to send the type-tagged-key change as a separate PR if you want it split out, or leave it as a note if you'd rather keep this one focused on #3137. @arimu1 it's your patch — say if you'd rather take it yourself.

Disclosure: I'm an AI agent (Claude) reviewing as Palo Alto AI Research Lab. Line numbers are from main as of today; the reasoning is mine and the two scenarios above are read off the code, not observed in a running server.

[cubic-dev-ai]

2 issues 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:623">
P2: Direct callers of `close_sse_stream()` with its documented JSON-RPC request id now get a silent no-op because request streams are stored under encoded keys. Normalize this public method's input to the routing key while preserving the standalone GET key, and keep the internal callback aligned with that contract.</violation>

<violation number="2" location="src/mcp/server/streamable_http.py:1065">
P3: The router regression test now times out because its manually registered streams use raw ids while responses route to `s:`-prefixed keys. Update its fixture keys to the same encoded representation (or route through the request registration path) so CI continues to cover the buffering behavior.</violation>
</file>

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

Re-trigger cubic</sub>

[cubic-dev-ai]

P2: Direct callers of close_sse_stream() with its documented JSON-RPC request id now get a silent no-op because request streams are stored under encoded keys. Normalize this public method's input to the routing key while preserving the standalone GET key, and keep the internal callback aligned with that contract.

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 623:

<comment>Direct callers of `close_sse_stream()` with its documented JSON-RPC request id now get a silent no-op because request streams are stored under encoded keys. Normalize this public method's input to the routing key while preserving the standalone GET key, and keep the internal callback aligned with that contract.</comment>

<file context>
@@ -598,16 +615,22 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
+            # the assignment below must not have an await between them: with
+            # resumability on, EventStore.store_event is user code and any await
+            # re-opens a check-then-act window (#3137 / #3163).
+            request_id = _request_stream_key(message.id)
+
+            # A second concurrent POST reusing an in-flight id would overwrite that
</file context>

[cubic-dev-ai]

P3: The router regression test now times out because its manually registered streams use raw ids while responses route to s:-prefixed keys. Update its fixture keys to the same encoded representation (or route through the request registration path) so CI continues to cover the buffering behavior.

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 1065:

<comment>The router regression test now times out because its manually registered streams use raw ids while responses route to `s:`-prefixed keys. Update its fixture keys to the same encoded representation (or route through the request registration path) so CI continues to cover the buffering behavior.</comment>

<file context>
@@ -1039,7 +1062,7 @@ async def message_router():
                         # the GET stream since they can't be correlated.
                         if isinstance(message, JSONRPCResponse | JSONRPCError) and message.id is not None:
-                            target_request_id = str(message.id)
+                            target_request_id = _request_stream_key(message.id)
                         # Extract related_request_id from meta if it exists
                         elif (
</file context>