Version Packages#25
Merged
Merged
Conversation
mattzcarey
pushed a commit
to mattzcarey/agents
that referenced
this pull request
Jun 1, 2026
In the collapse-the-duplicate refactor (6b5990c), the cleanup sequence in sendOnStream ended up running BEFORE writeSSEEvent \u2014 the exact ordering the docstring and changeset warn against, and the exact bug Sunil's clanker review and the original cloudflare#1583 ticket are about. Failure mode: tool finishes, sendOnStream is called with a still-live liveConnection (e.g. resumed GET), storeEvent writes the final event, shouldClose=true triggers clearStream which wipes the entire stream INCLUDING the event we just stored, then writeSSEEvent emits a frame with an event id that no longer resolves. A client losing the WS pipe at that exact moment can reconnect with Last-Event-ID and find nothing to replay. Fix: write the SSE frame first, then run cleanup. Matches the docstring already in place. Also fixes bonk cloudflare#25: changeset referenced McpAgent.getStreamIdForRequestId; renamed to getStreamForRequestId in 880c013.
threepointone
pushed a commit
that referenced
this pull request
Jun 1, 2026
…llow-up) (#1607) * fix(mcp): minimal alarm-driven cleanup for SSE event store (#1583 followup) Reworking the resumability cleanup after a series of review rounds. Net result: -233 lines from the previous followup, no functional regression on the actual fixes. The Important Fixes (unchanged from earlier rounds) 1. Drop clearStream-on-shouldClose. Final tool response stays in the event store so a client that lost the connection right as it was in flight can reconnect with Last-Event-ID and recover it. 2. Unconditional storage on the request-bound send() path, matching the SDK. Falls back to McpAgent.getStreamIdForRequestId (a reverse lookup against __mcp_stream_reqs__:*) when the originating WS has dropped, so events are still recorded for replay. 3. Resumed GET registers under the source streamId, matching the SDK exactly. Active POST resumption inherits requestIds; standalone resumption takes over the listen role; completed POST resumption is a one-shot replay channel. The Storage Bounding (simplified) Earlier rounds layered cap-based eviction + a complex per-event-or- per-stream timestamped sweep + a recurring cron. After working through it: - Per-stream cap removed. Real streams never approach 256 events, and truncating replay history silently was arguably worse than letting the stream grow. DO storage is bounded structurally anyway. - Sweep is alarm-driven, not cron. storeEvent fires onStoreEvent; McpAgent uses it to idempotent-schedule a single cleanup at now + maxAgeMs. The callback sweeps streams whose lastWriteAt is past cutoff, then either reschedules at the next-earliest expiry or doesn't reschedule at all. Quiescent DOs do no periodic work. - Sweep cost is O(active streams), not O(total events): a per-stream metadata key (__mcp_stream_evt_meta__) tracks lastWriteAt, the sweep scans only that index and bulk-deletes events for streams whose meta is expired. - Stream-reqs cleanup folded into the same sweep. When sweep deletes a stream, the agent also deletes the matching __mcp_stream_reqs__ entry. No separate sweepStreamReqs, no timestamp on reqs entries. - All the knobs removed: getEventStoreSweepCron, getEventStoreSweepMaxIterations, batchSize, the cursor-pagination question. Just getEventStoreMaxAgeMs (default 24h). DurableObjectEventStore is exported so callers embedding WorkerTransport inside an Agent (e.g. the elicitation example) can opt into resumability with new DurableObjectEventStore(this.ctx.storage). Tests: 455/455 MCP tests pass. npm run check exit 0 across all 88 projects. event-store.ts is now 198 lines (down from 323); McpAgent additions for cleanup are ~50 lines instead of ~200. * fix(mcp): clear stream on close + address clanker review Rip the alarm-driven sweep added in the previous push of this PR. Storage cost is now bounded by in-flight POST streams (cleared the moment their close frame is written) plus the standalone GET stream. - event-store: drop sweep(), drop the __mcp_stream_evt_meta__ index, drop the onStoreEvent option. Assert streamId contains no ':' to rule out prefix-scan collisions. Chunk multi-key deletes at the DO 128-key cap. Cap replayEventsAfter at 1000 events to bound memory. - mcp/index: drop _cf_armEventStoreCleanup, _cf_runEventStoreCleanup, and getEventStoreMaxAgeMs. getEventStore() is now plain 'new DurableObjectEventStore(this.ctx.storage)'. - transport: writeSSEEvent FIRST, then deleteStreamRequestIds + clearStream. Trade-off: if the WS pipe is enqueued but the client TCP dies before the bytes arrive, that one final message is lost. Every earlier event is still replayable while the stream is open. - transport: when a GET resumes an active POST stream, strip requestIds off any other connection still claiming them, so stale POST bridges can't win the routing race against the resumed GET. - transport: fan standalone notifications out to every _standaloneSse connection instead of last-writer-wins. Matches the spec. - tests: replace the completed-POST 'replay' test with a real mid-flight POST disconnect test using a new deferredGreet tool that emits a progress notification then sleeps. Drop the misnamed 'resumed GET is registered as standalone' test (it was actually a POST-id resume). Drop event-store sweep tests; add chunked-delete and replay-cap tests. * refactor(mcp): clean up transport.ts per review Four structural cleanups, no behaviour change. All 453 mcp tests still pass. - Add ClearableEventStore interface (extends EventStore with clearStream). Replaces an inline cast + optional chaining dance with a real named contract and proper 'in' narrowing. - Supersede stale POST bridges by closing them (conn.close(1000, 'Superseded by resumed stream')) instead of reaching in and mutating their state. Matches the SDK's last-writer-wins _streamMapping mental model. Removes a 'spooky action at a distance' pattern. - Split send() into sendStandalone() and sendForRequest(). The two paths share nothing but the storeEvent call and were fused for no good reason. Removes the disjoint shouldClose-cleanup-after-write block as a side effect. - Drop the now-unused 'eventStore' getter on the transport. Its only caller was _cf_runEventStoreCleanup which was deleted in the previous commit. * refactor(mcp): act on bonk review Three legit findings, two no-ops: - clearStream: replace 'list-all-then-chunk-delete' with a paginated list+delete loop bounded by DELETE_CHUNK. Removes the deleteChunked helper as a side effect \u2014 the chunk size is now enforced by the list limit itself, not a separate pass. Net simpler. - Clearer error message in sendForRequest: 'No active stream found' instead of 'No connection established', since the failure is a missing stream mapping, not a missing connection. - resumability test: assert the progress notification is NOT re-delivered on the resumed GET. The Last-Event-ID *was* the progress event, so replayEventsAfter must skip it. Not acting on: - O(n) full scan in getStreamIdForRequestId: working set is in-flight POST streams per DO, which is small. - clearStream/replayEventsAfter race: window is between two awaits in the same microtask chain and the consequence (replaying about-to-be- deleted events) is harmless. - Double storage read in disconnected-client path: only on the slow path; threading data through would complicate the API. - Standalone GET events accumulate: already documented; bounded by session lifetime by design. * refactor(mcp): act on bonk pass 2 Five small wins from the second-pass review: - Replace the 'clearStream' in this._eventStore duck-type check with a proper isClearableEventStore type guard. Same runtime check, better readability, narrows the type through the inferred call. - Add an ordering note on the deleteStreamRequestIds -> clearStream sequence: the tiny window where a concurrent GET resume sees no requestIds and starts replaying soon-to-be-deleted events is benign, but the comment saves future readers from re-deriving it. - getStreamIdForRequestId: add an O(n) cost comment and a defensive limit: 1000 on the list call so an abandoned-POST leak can't unbounded-load the scan. - Standalone fan-out: wrap each writeSSEEvent in try/catch so one dead WS in the middle of the loop can't block delivery to the remaining standalone connections. - Mid-flight resumability test: assert postBuf does NOT contain '"result"' at the point of cancel, so a timing regression turns into a clear failure rather than a false pass. - Comment on the supersede-by-close loop noting the last-writer-wins semantic for rapid back-to-back GET resumes on the same stream. Skipped: - Collapsing the double storage read on the disconnected path (getStreamIdForRequestId + getStreamRequestIds): would require widening the agent API to return both in one pass. Slow-path-only, working set is single-digit streams. Not worth the API change. * refactor(mcp): act on bonk pass 3 Six fixes: - Collapse the double storage read on the disconnected-client send path. getStreamIdForRequestId is now getStreamForRequestId and returns { streamId, requestIds } in one pass; sendForRequest no longer issues a second getStreamRequestIds for the same key. - Re-export ClearableEventStore from mcp/index.ts so embedders who want to implement their own clearable store can import the type. - Fix the replayEventsAfter off-by-one: use start: <key>+'\x00' so the list result strictly excludes lastEventId rather than including it and post-filtering. Effective cap is now exactly 1000, not 999. Drops the (key <= lastKey) guard as a side effect. - Tighten the replay-cap test from <=1000 to ==1000 \u2014 catches the off-by-one if anyone reverts. - Comment the no-live-connection branch in sendForRequest: the close frame is intentionally dropped when there's nowhere to write it. The client's reconnect with Last-Event-ID gets the final event (until cleanup runs immediately after) and the spec lets them treat it as final. - Log a warning if getStreamForRequestId hits its 1000-key scan cap. Hitting it means abandoned __mcp_stream_reqs__ entries are accumulating, which would silently turn into 'No active stream found' errors otherwise. * fix(mcp): repair botched merge of #1607 with main The merge of main into this branch left two fused copies of sendForRequest \u2014 main's collision-routing version and this PR's no-live-connection fallback version \u2014 plus dangling references to the old _requestResponseMap (renamed to _streamResponseIds in main). The file built but the logic was incoherent. This commit restructures sendForRequest so both intents coexist cleanly: 1. Pick the live connection using main's collision-safe rule: prefer the originating connection; fall back to the unique matching connection; otherwise null. 2. If multiple live connections claim the request id and none is the originating, route an Internal Error to each \u2014 main's #1639 protocol-safety behaviour. 3. If a single live connection owns the request, delegate to sendOnStream (main's canonical store + close-detect + cleanup path). 4. Otherwise fall back to this PR's getStreamForRequestId reverse lookup so the event is still stored for replay when the originating WS has dropped, and run cleanup on shouldClose. The close frame is intentionally dropped when there's no live connection \u2014 documented trade-off in send()'s docstring. No behaviour change for the live-connection path (that's main's sendOnStream verbatim). The dropped-WS replay path now exists again, which is what this PR was originally about. * refactor(mcp): collapse duplicated send path The previous commit's repair of the botched merge left sendOnStream and sendForRequest's no-live-connection fallback as two parallel implementations of the same logical function. They had already drifted in this branch (live path used a cast, fallback used the isClearableEventStore type guard), and a 'void eventId;' silencing betrayed the fact that the structure wasn't right. sendOnStream now takes (streamId, relatedIds, liveConnection, message, requestId) directly. The caller resolves where the message goes; sendOnStream handles store + close-detect + cleanup + write. writeSSEEvent is gated on liveConnection being non-null \u2014 same behaviour, expressed as one branch instead of two. sendForRequest is now four phases readable top-to-bottom: 1. find matching live connections 2. ambiguous-multi \u2192 error route to each 3. resolve streamId + relatedIds (from connection state, or persisted fallback if the WS dropped) 4. one sendOnStream call No behaviour change. 457/457 mcp tests still pass. Net -10 lines, two cleanup paths collapsed to one, type-guard drift fixed, 'void eventId;' smell removed. * fix(mcp): write SSE frame before clearStream (bonk #24) In the collapse-the-duplicate refactor (6b5990c), the cleanup sequence in sendOnStream ended up running BEFORE writeSSEEvent \u2014 the exact ordering the docstring and changeset warn against, and the exact bug Sunil's clanker review and the original #1583 ticket are about. Failure mode: tool finishes, sendOnStream is called with a still-live liveConnection (e.g. resumed GET), storeEvent writes the final event, shouldClose=true triggers clearStream which wipes the entire stream INCLUDING the event we just stored, then writeSSEEvent emits a frame with an event id that no longer resolves. A client losing the WS pipe at that exact moment can reconnect with Last-Event-ID and find nothing to replay. Fix: write the SSE frame first, then run cleanup. Matches the docstring already in place. Also fixes bonk #25: changeset referenced McpAgent.getStreamIdForRequestId; renamed to getStreamForRequestId in 880c013. * fix(mcp): sendOnStream relatedIds is readonly Connection state arrays are ImmutableArray<RequestId>; sendOnStream only reads via .every() so the parameter type can accept readonly RequestId[]. Caught by CI typecheck. * refactor(mcp): act on bonk pass + trim comment noise - Wrap supersede-by-close in try/catch so a dead WS can't abort the loop (bonk). - Trim verbose comments in sendOnStream + send dispatcher. No PR / ticket references in code. * fix(mcp): try/catch around writeSSEEvent so a dead WS doesn't orphan cleanup If connection.send throws inside writeSSEEvent (WS torn down between iteration and write), the exception would skip the cleanup block below and leave stream-reqs + stored events permanently orphaned. Also document standalone events accumulating for the DO lifetime in the event-store class docstring. * refactor(mcp): drop cargo-culted try/catch around close() Reverting the try/catch I added around the supersede-by-close loop. Closing a WS in the Workers runtime is fire-and-forget and doesn't throw — no other close() call in the codebase is wrapped (including the transport's own close() method). Added it reactively to a review question; the answer was 'no, it doesn't throw'. Kept the two writeSSEEvent wraps: those guard connection.send (which the codebase consistently wraps) — one protects the cleanup block from being skipped, the other keeps one dead WS from aborting standalone fan-out to the rest. * test(mcp): cover standalone fan-out + resume supersession Two behavioural changes in this PR had no direct coverage: - sendStandalone fans out to ALL _standaloneSse connections (was last-writer-wins). Two new tests: fan-out reaches every standalone stream and skips POST bridges; one throwing send doesn't block the rest of the loop. - handleGetRequest closes a stale POST connection when a GET resumes its stream. New test asserts the stale bridge is close()d with 1000/'Superseded by resumed stream', the resuming conn isn't closed, and it claims the persisted requestIds. Mutation-checked: commenting out the supersede close() fails the new test. 460/460 mcp tests pass. * fix(mcp): standalone send goes to ONE stream, not fanned out (spec) MCP 2025-06-18 'Multiple Connections': the server MUST send each JSON-RPC message on only one of the connected streams and MUST NOT broadcast the same message across multiple streams. The prior commit fanned standalone notifications out to every _standaloneSse connection — a direct violation. (The SDK reference enforces the stronger 'only one standalone GET per session' with a 409.) Fix: - handleGetRequest now supersedes prior connections for the resumed streamId in ALL resumable cases (POST resume, standalone resume, and fresh standalone GET) via a shared supersedePriorStreamConnections helper. At most one live connection per stream. - sendStandalone reverts to single-send: find the one standalone connection and write to it, else store-only for replay. No more broadcast loop. - Tests updated: 'sends on exactly one standalone stream', 'stores but does not write when none live', and a new 'supersedes a prior standalone GET when a fresh GET opens'. 461/461 mcp tests pass. (5 unrelated typecheck failures in ai-chat/think predate this branch — from main's recovery work.) * test(mcp): lock global event-id uniqueness across streams (spec) MCP resumability rule: the SSE event id MUST be globally unique across all streams within a session. We had monotonic-within-stream and ignore-other-streams coverage but nothing pinning cross-stream uniqueness. A future move to a global seq counter could silently break it. New test stores the same seq across three streams and asserts no id collides. Mutation-checked: dropping streamId from the id format fails it. * docs(mcp): sync changeset with shipped behaviour - 'stripped requestIds' -> stale connections are closed (supersede). - Replace the fan-out bullet (reverted) with the one-stream-per-message spec rule. - Note standalone GET events accumulate for the DO lifetime. * docs(mcp): fix stale 'lastKey' reference in replayEventsAfter comment The variable was renamed to startKey (built inline); the comment still said 'appending \x00 to lastKey'. No code change. --------- Co-authored-by: Matt Carey <matt@cloudflare.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This PR was opened by the Changesets release GitHub action. When you're ready to do a release, you can merge this and the packages will be published to npm automatically. If you're not ready to do a release yet, that's fine, whenever you add more changesets to main, this PR will be updated.
Releases
@cloudflare/agents@0.0.13
Patch Changes
066c378Thanks @threepointone! - setState() doesn't take source anymore