Conversation
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
📚 Documentation preview
|
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 3 files
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:933">
P2: When a POST mints its priming cursor during a replay, `_handle_post_request` bypasses this lock and can mutate the store while `replay_events_after` iterates. Guard the POST priming `store_event` with the same lock; the deque-backed store can otherwise abort resumable replay.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| async for event_message in msg_reader: | ||
| await sse_stream_writer.send(self._create_event_data(event_message)) | ||
|
|
||
| async with self._event_store_lock: |
There was a problem hiding this comment.
P2: When a POST mints its priming cursor during a replay, _handle_post_request bypasses this lock and can mutate the store while replay_events_after iterates. Guard the POST priming store_event with the same lock; the deque-backed store can otherwise abort resumable replay.
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 933:
<comment>When a POST mints its priming cursor during a replay, `_handle_post_request` bypasses this lock and can mutate the store while `replay_events_after` iterates. Guard the POST priming `store_event` with the same lock; the deque-backed store can otherwise abort resumable replay.</comment>
<file context>
@@ -915,25 +918,43 @@ async def _replay_events(self, last_event_id: str, request: Request, send: Send)
- async for event_message in msg_reader:
- await sse_stream_writer.send(self._create_event_data(event_message))
+
+ async with self._event_store_lock:
+ if self._terminated or not self._connected:
+ return
</file context>
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
Beyond the inline findings, I also checked whether the new anyio.SpooledTemporaryFile use in src/mcp/server/streamable_http.py:922 breaks the anyio>=4.9 floor for Python < 3.14 — the tempfile wrappers exist in 4.9, and the lockfile pins 4.10, so that is not a concern.
Extended reasoning...
Inline findings cover the replay/lock behaviour changes, so this note only records one additional candidate that was examined and ruled out: the compatibility of the new anyio.SpooledTemporaryFile call with the declared anyio minimum (>=4.9 for Python < 3.14 in pyproject.toml, 4.10.0 in uv.lock). The API is available at that floor, so no dependency bump is needed. With several verified findings posted inline (and further verified findings not shown), this is not a candidate for approval.
Findings marked 🟡 are optional suggestions and need no follow-up push.
| await replay_buffer.write( | ||
| (json.dumps(event_data, ensure_ascii=False) + "\n").encode("utf-8") | ||
| ) |
There was a problem hiding this comment.
🔴 Operators whose server has no writable temp directory lose Last-Event-ID resumption for any stream holding over 1 MiB of history; on the base branch that history streamed fine. The buffer write at src/mcp/server/streamable_http.py:929 goes into anyio.SpooledTemporaryFile, whose spill opens tempfile.TemporaryFile in gettempdir() from a worker thread. The OSError escapes send_event and aborts replay_events_after, so the client gets an empty SSE stream on every retry. Fix: keep replay deliverable without a writable temp dir, e.g. catch OSError from the spill and keep buffering in memory (or make the spill dir/threshold configurable).
Extended reasoning...
Condition: the process runs with no writable temp location (Docker --read-only without --tmpfs /tmp, Kubernetes readOnlyRootFilesystem with no emptyDir at /tmp, TMPDIR set to a missing path, or a full disk) and a resumed stream has more than 1 MiB of un-acknowledged history, e.g. a long tool call emitting many progress or log notifications while the client was disconnected. On the base branch replay_sender sent each replayed event straight to sse_stream_writer with no filesystem involvement. After the change, send_event at streamable_http.py:926-931 writes every record into replay_buffer created at line 922 with max_size=1024*1024. anyio's SpooledTemporaryFile checks tell() against max_size on write and calls rollover(), which runs tempfile.TemporaryFile(mode='w+b', dir=None, ...) via to_thread.run_sync; tempfile.gettempdir() raises FileNotFoundError('No usable temporary directory found') when TMPDIR, /tmp, /var/tmp, /usr/tmp and cwd are all unwritable, and a full disk raises OSError(ENOSPC) on the copy. That exception propagates out of send_event, through the store's…
Verification: normal — triggered when the server process has no writable temp location (read-only root filesystem without a /tmp tmpfs/emptyDir, TMPDIR pointing at a missing path, or ENOSPC) and a resumed stream carries more than 1 MiB of replayable history. Mechanism verified in /home/claude/python-sdk/src/mcp/server/streamable_http.py: line 922 creates `replay_buffer =… | normal — triggers when the process…
| # messages will be replayed on the re-connect | ||
| event_id = None | ||
| if self._event_store: | ||
| event_id = await self._event_store.store_event(request_stream_id, message) | ||
| logger.debug(f"Stored {event_id} from {request_stream_id}") | ||
| with anyio.CancelScope() as lock_scope: | ||
| acquire_scope = lock_scope | ||
| try: | ||
| try: | ||
| self._event_store_lock.acquire_nowait() | ||
| except anyio.WouldBlock: | ||
| if not self._connected: | ||
| return | ||
| await self._event_store_lock.acquire() | ||
| finally: | ||
| acquire_scope = None | ||
| if lock_scope.cancelled_caught: | ||
| return | ||
| try: | ||
| event_id = await self._event_store.store_event(request_stream_id, message) | ||
| logger.debug(f"Stored {event_id} from {request_stream_id}") | ||
| target = self._request_streams.get(request_stream_id) | ||
| finally: | ||
| self._event_store_lock.release() | ||
| else: | ||
| target = self._request_streams.get(request_stream_id) | ||
|
|
||
| target = self._request_streams.get(request_stream_id) | ||
| if target is not None: | ||
| try: | ||
| # Send both the message and the event ID |
There was a problem hiding this comment.
🟡 (optional) Every other request in a session stalls while one reconnecting client drains a large replay, which the base branch never did during history delivery. streamable_http.py:944 registers the resumed stream under the lock before the buffered history is sent at :947-:949; live events for that stream now queue into the 16-slot buffer during the drain, and on the 17th the router blocks at :1091, so no response or notification in the session moves until the slow client finishes reading history. Fix: keep the live stream unregistered (or drain it) until history delivery completes, so the router only blocks on a stream that is actually being consumed; the base registered at the same point but only after the history was already on the wire. [also at: src/mcp/server/streamable_http.py:944 - Every other request in a session can stall for as long as one resuming client takes to download its history, which the base branch never did during replay. streamable_http.py:944 registers the resumed stream in _request_streams before any buffered history is sent, so live messages for that stream…]
Extended reasoning...
On the base, replay_events_after streamed history straight to the network and only afterwards registered request_streams, so live events for that stream during history delivery hit target None at :1094 and were stored without queuing. The PR moves registration to :944, inside the lock and before the drain loop at :947-:949. A client reconnects with Last-Event-ID to a GET stream (or a long-running request stream) whose history is large, up to and above the 1 MiB spill, over a slow link. During the drain the server keeps emitting notifications for that stream (progress, logging, list_changed). Each goes through the router: store_event, then target[0].send at :1091 into the 16-slot buffer created at :940. The 17th send blocks the router. The router is the only consumer of write_stream for the whole session,…
Verification: nit — acknowledged in diff: docs/run/legacy-clients.md adds "Live-stream backpressure can still delay other messages in the same session", and that bound is accurate (a delay for the remainder of the drain, not a deadlock); but the PR description's claim that delivering history outside the lock means "a slow reader cannot block sibling responses" is only true for the lock, not for the…
| async with self._event_store_lock: | ||
| if self._terminated or not self._connected: | ||
| return | ||
| stream_id = await event_store.replay_events_after(last_event_id, send_event) |
There was a problem hiding this comment.
🟡 (optional) Every message in a session now waits while one client's replay reads the event store, which the base branch never did. streamable_http.py:936 calls replay_events_after while holding _event_store_lock; the router then blocks at streamable_http.py:1074 before it can store or route anything. write_stream has no buffer, so every server handler's send in that session waits for the whole store read. Fix: hold the lock only to fix the snapshot boundary and register the live stream, and read history unlocked (e.g. record the last stored id under the lock and replay up to it outside it); this cost follows from the PR's serialize-under-lock purpose. The docs note names only live-stream backpressure. [also at: src/mcp/server/streamable_http.py:948 - Sessions with histories over 1 MiB hold _event_store_lock, and so block every message in the session, for a worker-thread hop per replayed event plus a 1 MiB copy; the base held no lock and did no thread work. streamable_http.py:929 writes each record through anyio.SpooledTemporaryFile while…]
Extended reasoning...
The PR text says the lock lets a slow replay reader not block sibling responses; that covers the network reader, not the store read itself, which now runs entirely under the lock. On the base the replay called replay_events_after with no lock and the router only serialized on its own store_event call, so a long replay never delayed other messages. After the merge: a client reconnects with Last-Event-ID to a session backed by a database or remote store holding a long history. replay_sender enters async with self._event_store_lock at streamable_http.py:933 and awaits event_store.replay_events_after at line 936; that call awaits the store once per historical row, so the lock is held for the full read. Meanwhile a tool handler for another request in the same session finishes and sends its response on write_stream (a 0-buffer stream created…
Verification: nit; conflicts with stated purpose: the PR description says the lock is held so that "a slow reader cannot block sibling responses", but that only covers network delivery — the event-store read itself now runs under the lock and stalls every outbound message in the session for its full duration. Trigger: a client reconnects with Last-Event-ID to a session whose EventStore is slow… | nit —…
| await replay_buffer.seek(0) | ||
| while event_data := await replay_buffer.readline(): | ||
| await sse_stream_writer.send(json.loads(event_data)) | ||
| await replay_buffer.aclose() | ||
| if request_streams is None or self._terminated or not self._connected: | ||
| return | ||
| if priming_event is not None: | ||
| await sse_stream_writer.send(priming_event) | ||
| async with request_streams[1] as msg_reader: | ||
| async for event_message in msg_reader: | ||
| await sse_stream_writer.send(self._create_event_data(event_message)) | ||
| except anyio.ClosedResourceError: # pragma: lax no cover | ||
| # Expected when close_sse_stream() is called | ||
| logger.debug("Replay SSE stream closed by close_sse_stream()") |
There was a problem hiding this comment.
🟡 (optional) Clients resuming against an event store that fails partway through a long scan now receive nothing and can never advance, where the base delivered everything up to the failure. streamable_http.py:936 buffers all events under the lock and only sends at line 947 after replay_events_after returns; an exception there is caught at line 961 and the buffered records are discarded. Fix: on a mid-replay store error, still deliver the events already buffered (then end the stream) so the client's cursor advances and each retry scans less, or document that stores must complete the whole scan atomically.
Extended reasoning...
A store backed by a database or Redis scan with a query timeout trips only on long scans. On the base, send_event wrote each event straight to the SSE writer, so a timeout after N events still delivered N events; the client reconnected with the Nth id and the next scan was shorter and succeeded. Now send_event at lines 926-931 only writes to replay_buffer, and the buffer is read back at lines 947-949 only after replay_events_after returns normally. If the store raises, control jumps to line 961, logs, and the finally closes the buffer; the SSE response ends with zero events. The client reconnects with the same Last-Event-ID, the same long scan runs again under _event_store_lock (stalling the session's router each time), times out again, and the client never progresses. The dismissing finder read the code but judged only 'partial delivery differs'; it did not follow the retry loop where partial delivery is what made progress possible. Remedy: deliver the already-buffered records before ending the stream on a store error.
Verification: nit — triggers when a user-supplied EventStore.replay_events_after raises after having invoked the callback for some events (any backend error mid-scan; permanent loss only if the failure is deterministic for that scan length). Mechanism verified in /home/claude/python-sdk/src/mcp/server/streamable_http.py: send_event (lines 926-931) now only does await replay_buffer.write(...); the… | nit…
| self._sse_stream_writers[stream_id] = sse_stream_writer | ||
| self._request_streams[stream_id] = request_streams | ||
| priming_event = await self._mint_priming_event(stream_id, replay_protocol_version) | ||
|
|
||
| await replay_buffer.seek(0) | ||
| while event_data := await replay_buffer.readline(): | ||
| await sse_stream_writer.send(json.loads(event_data)) |
There was a problem hiding this comment.
🟡 (optional) Servers using the polling pattern (close_sse_stream from a tool) now cut a resuming client's history mid-delivery and force it to reconnect and re-scan; on the base the replay always completed. streamable_http.py:943 publishes the writer in _sse_stream_writers before the buffered history is drained at lines 947-949, so a close during the drain closes the writer and the next send raises. Fix: make a close request during history delivery take effect only after the buffered history and priming event are sent (or keep the writer unregistered until the drain completes, as the base did).
Extended reasoning...
Tools that implement polling call ctx.close_sse_stream() periodically. A client reconnects with Last-Event-ID and has a large history (the spilled case is >1 MiB). Under the lock at line 943 the new code stores the writer in _sse_stream_writers, then releases the lock and sends the history record by record at line 949. The tool's next close_sse_stream(request_id) at line 273 pops and closes that writer and pops the request streams at line 279. The next send at line 949 raises ClosedResourceError, caught at line 958; the stream ends after a partial history with no priming event. The client waits the retry interval, reconnects, and the remaining history is scanned again under _event_store_lock, stalling the session router again; with a close interval shorter than the drain time this repeats for every chunk. On the base the writer was registered only after replay_events_after finished (old line 928), so close during replay was a no-op and the history arrived in one pass. No event is lost, but each resume becomes many round trips and many locked store scans. Remedy: defer the close…
Verification: nit. Trigger: a server using the polling pattern (a tool calling ctx.close_sse_stream() periodically, e.g. examples/servers/sse-polling-demo/mcp_sse_polling_demo/server.py:104-106) while a client resumes with Last-Event-ID and its buffered history is still being drained. Mechanism verified: src/mcp/server/streamable_http.py:943-944 registers sse_stream_writer in… | nit. Trigger: a server tool…
| await replay_buffer.write( | ||
| (json.dumps(event_data, ensure_ascii=False) + "\n").encode("utf-8") |
There was a problem hiding this comment.
🟡 (optional) Servers whose event store builds EventMessage with a non-JSON event id (a uuid.UUID, Decimal, DB id object) lose replay after this merges; the base delivered those events. streamable_http.py:930 runs json.dumps on the event dict, whose id is EventMessage.event_id exactly as the store passed it. json.dumps raises TypeError for such objects; the handler at streamable_http.py:961 logs it and the client gets an empty 200 stream and retries the same cursor forever. Fix: coerce event_id to str in _create_event_data (streamable_http.py:437) or pass default=str, so the live and replay paths accept the same ids. Third-party stores cannot be enumerated from this checkout; the SDK's own stores all pass str.
Extended reasoning...
EventMessage is a plain dataclass (streamable_http.py:106-111) and its event_id annotation is not enforced; stores construct it themselves inside replay_events_after. On the base, send_event handed the dict straight to sse_starlette, which renders the id with string formatting, so a UUID object, a Decimal or a driver row id worked on the wire; the same objects still work on the live path after the merge because _create_event_data at line 428-439 only copies them. On the replay path after the merge: a store such as a Postgres-backed one whose driver returns uuid.UUID for the id column calls send_callback(EventMessage(message, row.id)). send_event at streamable_http.py:926-931 builds event_data with id set to that UUID object and calls json.dumps(event_data, ensure_ascii=False) at line 930, which raises TypeError: Object of type UUID is not JSON serializable. The exception leaves replay_events_after, exits the lock, and is caught by except Exception at line 961, which logs Error in replay sender; the finally closes the buffer. The async with sse_stream_writer block ends, so the…
Verification: nit. Trigger: a third-party EventStore whose replay_events_after builds EventMessage with a non-str event_id (uuid.UUID, Decimal, DB row id), violating the declared but runtime-unenforced type. Mechanism verified: src/mcp/server/streamable_http.py:106-111 declares event_id: str | None on a plain dataclass (no runtime check); _create_event_data (lines 428-439) copies event_message.event_id…
1 participant
Extract the replay-to-live handoff changes from #3511, on top of #3540. Serialize the replay snapshot and live-route registration against event storage, then deliver buffered history outside the lock so a slow reader cannot block sibling responses.
Keep shutdown handling with the lock: normal exit finishes active store writes but cancels a router waiting for replay. The buffer spills above 1 MiB, and the regressions run on both AnyIO backends with locked dependencies.
Validation
macOS, Python 3.14:
./scripts/testpasses with 6,013 passed, 9 skipped, 1 xfailed, 100% line/branch coverage, andstrict-no-coverpassing. Ruff lint/format and Pyright pass.Stack
Part 2 of 5. Depends on #3540; retarget to
mainafter that merges. No HTTPX2 update is needed for this change.AI Disclaimer
This PR was developed with the assistance of either Claude or Codex. I've reviewed and verified the changes.