History

Deliver MCP responses larger than the DO storage value cap

The DO transport persisted every outbound message for reconnect replay BEFORE writing the live SSE frame, and storage.put of a value over the 128 KiB DO cap throws — so an oversize response (the ~5MB ui://executor/shell.html resource) was neither stored nor delivered and the client hung on keepalives forever. Found live in prod after #1502 made the shell resource actually serve its bytes. - storeEvent skips persistence for messages over a 120 KiB guard, logging mcp_event_store_skipped_oversize and returning undefined; the caller delivers the live frame without a replay id. - sendOnStream treats any storeEvent failure as best-effort: log, then still write the live SSE frame. - New e2e (cloud/mcp-oversize-response.test.ts) drives resources/read of the shell over real workerd and fails on a hang. - Event-store unit tests updated for the skip contract.

Rhys Sullivan committed 3 months ago

Commit 64a3faf1 parent 25270b15 files · +281 −28Browse files

Unified · Split

5 files
  1. .changeset/tidy-pianos-shake.md +7 −0
  2. e2e/cloud/mcp-oversize-response.test.ts +136 −0
  3. packages/hosts/cloudflare/src/mcp/agents-event-store.test.ts +41 −12
  4. packages/hosts/cloudflare/src/mcp/agents-priming-event.test.ts +11 −4
  5. patches/agents@0.17.3.patch +86 −12
.changeset/tidy-pianos-shake.md+7−0
@@ -0,0 +1,7 @@
1---
2"executor": patch
3---
4
5**Fix: MCP responses larger than Durable Object storage's 128 KiB per-value cap hung the client instead of being delivered**
6
7The DO transport persisted every outbound message for reconnect replay before writing the live SSE frame, and `storage.put` of an oversize value throws — so a large response (the `ui://executor/shell.html` resource is ~5 MB) was neither stored nor sent, and the client waited on keepalives forever. The transport now delivers the live frame first and treats persistence as best-effort: an oversize message skips the event store with a logged warning and arrives without a replay id, which only costs replayability if the connection drops mid-delivery.
e2e/cloud/mcp-oversize-response.test.ts+136−0
@@ -0,0 +1,136 @@
1// Cloud: an MCP response bigger than DO storage's 128 KiB per-value cap is
2// still delivered. The ui://executor/shell.html resource (~5 MB of inlined
3// shell document) is the production case: the transport used to persist every
4// outbound message with storage.put BEFORE writing the live SSE frame, so the
5// put's failure killed the send and the client saw keepalives forever — every
6// artifact widget in every remote MCP client hung, with no error anywhere
7// (found live in prod, 2026-07-29). The patched transport delivers live first
8// and treats persistence as best-effort, so this pins the whole path over the
9// real workerd topology: POST resources/read → McpSessionDO → ASSETS-backed
10// shell loader → event store skip (oversize) → live SSE frame → this client.
11import { expect } from "@effect/vitest";
12import { Effect } from "effect";
13
14import { scenario } from "../src/scenario";
15import { Mcp, Target } from "../src/services";
16import type { Identity } from "../src/target";
17
18const PROTOCOL_VERSION = "2025-03-26";
19const JSON_AND_SSE = "application/json, text/event-stream";
20
21const SHELL_RESOURCE_URI = "ui://executor/shell.html";
22
23const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label;
24// DO storage's per-value cap; anything beyond it exercises the skip path.
25const DO_VALUE_CAP_BYTES = 128 * 1024;
26
27const mcpHeaders = (bearer: string, sessionId?: string) => ({
28 accept: JSON_AND_SSE,
29 authorization: `Bearer ${bearer}`,
30 "content-type": "application/json",
31 "mcp-protocol-version": PROTOCOL_VERSION,
32 ...(sessionId ? { "mcp-session-id": sessionId } : {}),
33});
34
35const postJson = (mcpUrl: string, bearer: string, body: unknown, sessionId?: string) =>
36 fetch(mcpUrl, {
37 method: "POST",
38 headers: mcpHeaders(bearer, sessionId),
39 body: JSON.stringify(body),
40 });
41
42const openSession = async (mcpUrl: string, bearer: string): Promise<string> => {
43 const initialized = await postJson(mcpUrl, bearer, {
44 jsonrpc: "2.0" as const,
45 id: "initialize",
46 method: "initialize",
47 params: {
48 protocolVersion: PROTOCOL_VERSION,
49 capabilities: {},
50 clientInfo: { name: "executor-e2e-oversize-response", version: "0.0.1" },
51 },
52 });
53 const sessionId = initialized.headers.get("mcp-session-id");
54 await initialized.text();
55 if (initialized.status !== 200 || !sessionId) {
56 throw new Error(`openSession: initialize failed (${initialized.status})`);
57 }
58 const notification = await postJson(
59 mcpUrl,
60 bearer,
61 { jsonrpc: "2.0" as const, method: "notifications/initialized" },
62 sessionId,
63 );
64 await notification.text();
65 expect(notification.status, "notifications/initialized is accepted").toBe(202);
66 return sessionId;
67};
68
69/** Read the POST's SSE body to completion and return the JSON-RPC payloads. */
70const readSseMessages = async (
71 response: Response,
72): Promise<ReadonlyArray<{ readonly id?: unknown; readonly result?: unknown }>> => {
73 const text = await response.text();
74 const messages: Array<{ readonly id?: unknown; readonly result?: unknown }> = [];
75 for (const block of text.replace(/\r/g, "").split("\n\n")) {
76 const data = block
77 .split("\n")
78 .filter((line) => line.startsWith("data:"))
79 .map((line) => line.slice("data:".length).trimStart())
80 .join("\n")
81 .trim();
82 if (data) messages.push(JSON.parse(data) as { readonly id?: unknown });
83 }
84 return messages;
85};
86
87scenario(
88 "MCP streamable HTTP · a response beyond the DO storage value cap is delivered live (the ui:// shell resource)",
89 {},
90 Effect.gen(function* () {
91 const target = yield* Target;
92 const mcp = yield* Mcp;
93 const identity = yield* target.newIdentity();
94 const bearer = yield* mcp.mintBearer(emailOf(identity));
95 // Artifacts opt-in is what registers the shell resource on the session.
96 const mcpUrl = `${target.mcpUrl}?artifacts=true`;
97
98 const sessionId = yield* Effect.promise(() => openSession(mcpUrl, bearer));
99
100 const read = yield* Effect.promise(() =>
101 postJson(
102 mcpUrl,
103 bearer,
104 {
105 jsonrpc: "2.0" as const,
106 id: "read-shell",
107 method: "resources/read",
108 params: { uri: SHELL_RESOURCE_URI },
109 },
110 sessionId,
111 ),
112 );
113 expect(read.status, "resources/read answers").toBe(200);
114
115 // The whole regression is "the final frame never arrives": bound the read
116 // so a hang fails the test instead of wedging the suite.
117 const messages = yield* Effect.promise(() => readSseMessages(read)).pipe(
118 Effect.timeoutOrElse({
119 duration: "60 seconds",
120 orElse: () => Effect.dieMessage("shell read hung: the response frame never arrived"),
121 }),
122 );
123
124 const response = messages.find((message) => message.id === "read-shell");
125 expect(response, "the read's response frame arrived on the POST stream").toBeDefined();
126 const result = response?.result as
127 | { readonly contents?: ReadonlyArray<{ readonly text?: string }> }
128 | undefined;
129 const html = result?.contents?.[0]?.text ?? "";
130 expect(
131 html.length,
132 "the document is larger than the DO storage cap, so it took the live-only path",
133 ).toBeGreaterThan(DO_VALUE_CAP_BYTES);
134 expect(html, "and it is the real shell document").toContain('name="executor-mcp-apps-shell"');
135 }),
136);
packages/hosts/cloudflare/src/mcp/agents-event-store.test.ts+41−12
@@ -62,7 +62,13 @@ describe("DurableObjectEventStore trimStream", () => { · 61 unmodified lines
62 vi.restoreAllMocks();62 vi.restoreAllMocks();
63 });63 });
64 64
65 it("keeps an oversize final response instead of evicting the only event", async () => {65 it("skips storing a message beyond DO storage's per-value cap, returning no replay id", async () => {
66 // Real DO storage rejects any value over 128 KiB. This used to be
67 // discovered inside storage.put — thrown BEFORE the live SSE write, so an
68 // oversize response (the ~5MB ui:// shell document) was neither stored nor
69 // delivered and the client hung on keepalives. The pinned contract now:
70 // oversize messages skip persistence with a warning and resolve undefined,
71 // and the caller delivers them live without a replay id.
66 const { storage, store } = makeStore();72 const { storage, store } = makeStore();
67 const hugeResult = {73 const hugeResult = {
68 jsonrpc: "2.0" as const,74 jsonrpc: "2.0" as const,
@@ -72,33 +78,56 @@ describe("DurableObjectEventStore trimStream", () => { · 3 unmodified lines
72 78
73 const eventId = await store.storeEvent("post-stream", hugeResult);79 const eventId = await store.storeEvent("post-stream", hugeResult);
74 80
75 expect(eventId, "storeEvent returns the event id").toBe("post-stream:0000000000000001");81 expect(eventId, "no replay id for an unstorable message").toBeUndefined();
76 expect(eventKeys(storage), "the >2MB final response survives its own trim pass").toEqual([82 expect(eventKeys(storage), "nothing was persisted").toEqual([]);
77 "__mcp_event__:post-stream:0000000000000001",83 expect(warnings.length, "the skip is logged").toBeGreaterThan(0);
78 ]);84 expect(warnings.at(-1)).toContain("mcp_event_store_skipped_oversize");
79 expect(warnings, "no eviction happened, so no eviction warning").toEqual([]);85 expect(warnings.at(-1)).toContain("post-stream");
86 });
87
88 it("does not advance the stream sequence when an oversize message is skipped", async () => {
89 const { storage, store } = makeStore();
90 await store.storeEvent("post-stream", {
91 jsonrpc: "2.0" as const,
92 method: "notifications/progress",
93 params: { blob: "x".repeat(3 * 1024 * 1024) },
94 });
95
96 const eventId = await store.storeEvent("post-stream", {
97 jsonrpc: "2.0" as const,
98 id: 1,
99 result: { ok: true },
100 });
101
102 expect(eventId, "the next storable event takes the first sequence slot").toBe(
103 "post-stream:0000000000000001",
104 );
105 expect(eventKeys(storage)).toEqual(["__mcp_event__:post-stream:0000000000000001"]);
80 });106 });
81 107
82 it("evicts oldest events at the byte cap but never the newest", async () => {108 it("evicts oldest events at the byte cap but never the newest", async () => {
83 const { storage, store } = makeStore();109 const { storage, store } = makeStore();
110 // 100 KiB each: storable (under the 120 KiB per-value guard), but 25 of
111 // them exceed the 2 MB per-stream byte cap.
84 const bigMessage = (marker: string) => ({112 const bigMessage = (marker: string) => ({
85 jsonrpc: "2.0" as const,113 jsonrpc: "2.0" as const,
86 method: "notifications/progress",114 method: "notifications/progress",
87 params: { marker, blob: "y".repeat(1024 * 1024) },115 params: { marker, blob: "y".repeat(100 * 1024) },
88 });116 });
89 117
90 await store.storeEvent("post-stream", bigMessage("one"));118 const total = 25;
91 await store.storeEvent("post-stream", bigMessage("two"));119 for (let index = 0; index < total; index += 1) {
92 await store.storeEvent("post-stream", bigMessage("three"));120 await store.storeEvent("post-stream", bigMessage(`msg-${index}`));
121 }
93 122
94 const remaining = eventKeys(storage);123 const remaining = eventKeys(storage);
95 expect(remaining[remaining.length - 1], "the newest event is always retained").toBe(124 expect(remaining[remaining.length - 1], "the newest event is always retained").toBe(
96 "__mcp_event__:post-stream:0000000000000003",125 `__mcp_event__:post-stream:${total.toString(16).padStart(16, "0")}`,
97 );126 );
98 expect(127 expect(
99 remaining.length,128 remaining.length,
100 "older events were evicted to satisfy the 2MB stream cap",129 "older events were evicted to satisfy the 2MB stream cap",
101 ).toBeLessThan(3);130 ).toBeLessThan(total);
102 expect(warnings.length, "eviction logs a warning").toBeGreaterThan(0);131 expect(warnings.length, "eviction logs a warning").toBeGreaterThan(0);
103 expect(warnings.at(-1)).toContain("mcp_event_store_evicted");132 expect(warnings.at(-1)).toContain("mcp_event_store_evicted");
104 expect(warnings.at(-1)).toContain("post-stream");133 expect(warnings.at(-1)).toContain("post-stream");
packages/hosts/cloudflare/src/mcp/agents-priming-event.test.ts+11−4
@@ -74,7 +74,10 @@ describe("POST-stream priming event: store ordering and replay", () => { · 73 unmodified lines
74 const store = new DurableObjectEventStore(storage as never);74 const store = new DurableObjectEventStore(storage as never);
75 const streamId = "post-stream";75 const streamId = "post-stream";
76 76
77 // Transport order: priming event first, then the tool response.77 // Transport order: priming event first, then the tool response. Both are
78 // tiny, so storeEvent always persists them and returns an id — the
79 // `undefined` arm of its signature is the oversize skip, pinned in
80 // agents-event-store.test.ts.
78 const primingId = await store.storeEvent(streamId, PRIMING_MESSAGE);81 const primingId = await store.storeEvent(streamId, PRIMING_MESSAGE);
79 const response = {82 const response = {
80 jsonrpc: "2.0" as const,83 jsonrpc: "2.0" as const,
@@ -87,10 +90,13 @@ describe("POST-stream priming event: store ordering and replay", () => { · 6 unmodified lines
87 expect(responseId, "response is seq 2, after the priming id").toBe(90 expect(responseId, "response is seq 2, after the priming id").toBe(
88 `${streamId}:0000000000000002`,91 `${streamId}:0000000000000002`,
89 );92 );
90 expect(primingId < responseId, "priming id sorts strictly before the response id").toBe(true);93 expect(
94 (primingId ?? "") < (responseId ?? ""),
95 "priming id sorts strictly before the response id",
96 ).toBe(true);
91 97
92 const replayed: Array<{ readonly eventId: string; readonly message: unknown }> = [];98 const replayed: Array<{ readonly eventId: string; readonly message: unknown }> = [];
93 await store.replayEventsAfter(primingId, {99 await store.replayEventsAfter(primingId ?? "", {
94 send: async (eventId: string, message: unknown) => {100 send: async (eventId: string, message: unknown) => {
95 replayed.push({ eventId, message });101 replayed.push({ eventId, message });
96 },102 },
@@ -108,9 +114,10 @@ describe("POST-stream priming event: store ordering and replay", () => { · 11 unmodified lines
108 const store = new DurableObjectEventStore(storage as never);114 const store = new DurableObjectEventStore(storage as never);
109 const streamId = "post-stream";115 const streamId = "post-stream";
110 const primingId = await store.storeEvent(streamId, PRIMING_MESSAGE);116 const primingId = await store.storeEvent(streamId, PRIMING_MESSAGE);
117 expect(primingId, "a tiny priming message always persists and gets an id").toBeDefined();
111 118
112 const replayed: string[] = [];119 const replayed: string[] = [];
113 await store.replayEventsAfter(primingId, {120 await store.replayEventsAfter(primingId ?? "", {
114 send: async (eventId: string) => {121 send: async (eventId: string) => {
115 replayed.push(eventId);122 replayed.push(eventId);
116 },123 },
patches/agents@0.17.3.patch+86−12
@@ -1,3 +1,22 @@
1diff --git a/node_modules/agents/.bun-tag-c0c639aa2299e502 b/.bun-tag-c0c639aa2299e502
2new file mode 100644
3index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391
4diff --git a/dist/agent-tool-types-CNyE1iz_.d.ts b/dist/agent-tool-types-CNyE1iz_.d.ts
5index 571eececebd5a1eaf7f2fbf5278801c4d34728ba..317e526489fcd3fd477a50525da0df666253bf0b 100644
6--- a/dist/agent-tool-types-CNyE1iz_.d.ts
7+++ b/dist/agent-tool-types-CNyE1iz_.d.ts
8@@ -480,7 +480,10 @@ declare class DurableObjectEventStore implements EventStore {
9 private readonly seqByStream;
10 private readonly seqInit;
11 constructor(storage: DurableObjectStorage);
12- storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId>;
13+ /** Resolves `undefined` for a message too large for DO storage's 128 KiB
14+ * per-value cap: the event is delivered live without a replay id rather
15+ * than failing the send. */
16+ storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId | undefined>;
17 getStreamIdForEventId(eventId: EventId): Promise<StreamId | undefined>;
18 replayEventsAfter(
19 lastEventId: EventId,
1diff --git a/dist/mcp/index.d.ts b/dist/mcp/index.d.ts20diff --git a/dist/mcp/index.d.ts b/dist/mcp/index.d.ts
2index c8fad448e8797b89690a99d93490d1363851b225..77f9fe3f6f2375eadc9f7a2974f0d202fe75b3bd 10064421index c8fad448e8797b89690a99d93490d1363851b225..77f9fe3f6f2375eadc9f7a2974f0d202fe75b3bd 100644
3--- a/dist/mcp/index.d.ts22--- a/dist/mcp/index.d.ts
@@ -19,7 +38,7 @@ index c8fad448e8797b89690a99d93490d1363851b225..77f9fe3f6f2375eadc9f7a2974f0d202 · 15 unmodified lines
19 McpAgent,38 McpAgent,
20 type McpAuthContext,39 type McpAuthContext,
21diff --git a/dist/mcp/index.js b/dist/mcp/index.js40diff --git a/dist/mcp/index.js b/dist/mcp/index.js
22index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..32d52598b42a4062d09020f0a1abfd67dbf3c266 10064441index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..444312ff41fbcff60e2ea681a9d5c756cd72faef 100644
23--- a/dist/mcp/index.js42--- a/dist/mcp/index.js
24+++ b/dist/mcp/index.js43+++ b/dist/mcp/index.js
25@@ -28,13 +28,17 @@ import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/44@@ -28,13 +28,17 @@ import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/
@@ -666,7 +685,31 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..32d52598b42a4062d09020f0a1abfd67 · 640 unmodified lines
666 for (const message of messages) {685 for (const message of messages) {
667 if (this.messageInterceptor) {686 if (this.messageInterceptor) {
668 if (await this.messageInterceptor(message, {687 if (await this.messageInterceptor(message, {
669@@ -777,9 +1136,11 @@ var StreamableHTTPServerTransport = class {688@@ -760,7 +1119,22 @@ var StreamableHTTPServerTransport = class {
689 * when the originating WS has dropped.
690 */
691 async sendOnStream(agent, streamId, relatedIds, liveConnection, message, requestId) {
692- const eventId = await this._eventStore?.storeEvent(streamId, message);
693+ // Persistence is best-effort and must never block live delivery: a
694+ // storeEvent failure (storage cap, storage outage) used to throw here,
695+ // before writeSSEEvent, so the response was neither stored NOR sent and
696+ // the client hung on keepalives. Deliver-live-first; a message without
697+ // an eventId just isn't replayable after a drop.
698+ let eventId;
699+ try {
700+ eventId = await this._eventStore?.storeEvent(streamId, message);
701+ } catch (error) {
702+ console.warn(JSON.stringify({
703+ event: "mcp_event_store_put_failed",
704+ streamId,
705+ error: String(error)
706+ }));
707+ this.onerror?.(error);
708+ }
709 let shouldClose = false;
710 if (isJSONRPCResultResponse(message) || isJSONRPCErrorResponse(message)) {
711 let responseIds = this._streamResponseIds.get(streamId);
712@@ -777,9 +1151,11 @@ var StreamableHTTPServerTransport = class {
670 } catch (error) {713 } catch (error) {
671 this.onerror?.(error);714 this.onerror?.(error);
672 }715 }
@@ -680,7 +723,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..32d52598b42a4062d09020f0a1abfd67 · 7 unmodified lines
680 }723 }
681 }724 }
682 async send(message, options) {725 async send(message, options) {
683@@ -861,12 +1222,10 @@ var StreamableHTTPServerTransport = class {726@@ -861,12 +1237,10 @@ var StreamableHTTPServerTransport = class {
684 *727 *
685 * ## Lifecycle728 * ## Lifecycle
686 *729 *
@@ -697,7 +740,34 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..32d52598b42a4062d09020f0a1abfd67 · 10 unmodified lines
697 *740 *
698 * Standalone GET stream events (`_GET_stream`) are *not* cleared741 * Standalone GET stream events (`_GET_stream`) are *not* cleared
699 * automatically; they accumulate for the lifetime of the DO. Bounded742 * automatically; they accumulate for the lifetime of the DO. Bounded
700@@ -899,6 +1258,7 @@ var DurableObjectEventStore = class DurableObjectEventStore {743@@ -893,12 +1267,34 @@ var DurableObjectEventStore = class DurableObjectEventStore {
744 }
745 async storeEvent(streamId, message) {
746 if (streamId.includes(":")) throw new Error(`DurableObjectEventStore: streamId must not contain ':' (got ${JSON.stringify(streamId)})`);
747+ // DO storage caps each value at 128 KiB; storage.put of a larger
748+ // message throws, and before this guard that throw escaped through
749+ // sendOnStream BEFORE the live SSE write, so an oversize response
750+ // (e.g. the ~5MB ui:// shell document) was never delivered at all —
751+ // the client saw only keepalives. Skip persistence instead: the
752+ // event is delivered live without a replay id, which is strictly
753+ // better than never delivering it. Undeliverable-if-dropped is the
754+ // documented cost, logged so it is visible.
755+ let messageBytes = 0;
756+ try {
757+ messageBytes = new TextEncoder().encode(JSON.stringify(message)).byteLength;
758+ } catch {}
759+ if (messageBytes > DurableObjectEventStore.MAX_STORABLE_EVENT_BYTES) {
760+ console.warn(JSON.stringify({
761+ event: "mcp_event_store_skipped_oversize",
762+ streamId,
763+ messageBytes,
764+ limit: DurableObjectEventStore.MAX_STORABLE_EVENT_BYTES
765+ }));
766+ return void 0;
767+ }
768 await this.ensureSeqLoaded(streamId);
769 const seq = (this.seqByStream.get(streamId) ?? 0) + 1;
770 this.seqByStream.set(streamId, seq);
701 const eventId = `${streamId}:${seq.toString(16).padStart(DurableObjectEventStore.SEQ_PAD, "0")}`;771 const eventId = `${streamId}:${seq.toString(16).padStart(DurableObjectEventStore.SEQ_PAD, "0")}`;
702 const eventKey = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${eventId}`;772 const eventKey = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${eventId}`;
703 await this.storage.put(eventKey, message);773 await this.storage.put(eventKey, message);
@@ -705,7 +775,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..32d52598b42a4062d09020f0a1abfd67 · 1 unmodified line
705 return eventId;775 return eventId;
706 }776 }
707 async getStreamIdForEventId(eventId) {777 async getStreamIdForEventId(eventId) {
708@@ -915,9 +1275,59 @@ var DurableObjectEventStore = class DurableObjectEventStore {778@@ -915,9 +1311,59 @@ var DurableObjectEventStore = class DurableObjectEventStore {
709 start: startKey,779 start: startKey,
710 limit: DurableObjectEventStore.REPLAY_LIMIT780 limit: DurableObjectEventStore.REPLAY_LIMIT
711 });781 });
@@ -713,8 +783,8 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..32d52598b42a4062d09020f0a1abfd67 · 1 unmodified line
713+ for (const [key, message] of rows) try {783+ for (const [key, message] of rows) try {
714+ await send(key.slice(DurableObjectEventStore.EVENT_KEY_PREFIX.length), message);784+ await send(key.slice(DurableObjectEventStore.EVENT_KEY_PREFIX.length), message);
715+ } catch {}785+ } catch {}
716 return streamId;786+ return streamId;
717 }787+ }
718+ async replayEventsForStream(streamId, { send }) {788+ async replayEventsForStream(streamId, { send }) {
719+ const prefix = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${streamId}:`;789+ const prefix = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${streamId}:`;
720+ const rows = await this.storage.list({790+ const rows = await this.storage.list({
@@ -724,8 +794,8 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..32d52598b42a4062d09020f0a1abfd67 · 3 unmodified lines
724+ for (const [key, message] of rows) try {794+ for (const [key, message] of rows) try {
725+ await send(key.slice(DurableObjectEventStore.EVENT_KEY_PREFIX.length), message);795+ await send(key.slice(DurableObjectEventStore.EVENT_KEY_PREFIX.length), message);
726+ } catch {}796+ } catch {}
727+ return streamId;797 return streamId;
728+ }798 }
729+ async trimStream(streamId) {799+ async trimStream(streamId) {
730+ const prefix = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${streamId}:`;800+ const prefix = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${streamId}:`;
731+ const rows = await this.storage.list({801+ const rows = await this.storage.list({
@@ -766,17 +836,21 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..32d52598b42a4062d09020f0a1abfd67 · 34 unmodified lines
766 /**836 /**
767 * Drop the event log for a single stream. Called by the transport837 * Drop the event log for a single stream. Called by the transport
768 * immediately after a POST's final response has been written to the838 * immediately after a POST's final response has been written to the
769@@ -973,6 +1383,9 @@ DurableObjectEventStore.EVENT_KEY_PREFIX = "__mcp_event__:";839@@ -973,6 +1419,13 @@ DurableObjectEventStore.EVENT_KEY_PREFIX = "__mcp_event__:";
770 DurableObjectEventStore.SEQ_PAD = 16;840 DurableObjectEventStore.SEQ_PAD = 16;
771 DurableObjectEventStore.DELETE_CHUNK = 128;841 DurableObjectEventStore.DELETE_CHUNK = 128;
772 DurableObjectEventStore.REPLAY_LIMIT = 1e3;842 DurableObjectEventStore.REPLAY_LIMIT = 1e3;
773+DurableObjectEventStore.MAX_EVENTS_PER_STREAM = 64;843+DurableObjectEventStore.MAX_EVENTS_PER_STREAM = 64;
774+DurableObjectEventStore.MAX_BYTES_PER_STREAM = 2 * 1024 * 1024;844+DurableObjectEventStore.MAX_BYTES_PER_STREAM = 2 * 1024 * 1024;
775+DurableObjectEventStore.MAX_EVENT_BYTES = 2 * 1024 * 1024;845+DurableObjectEventStore.MAX_EVENT_BYTES = 2 * 1024 * 1024;
846+// DO storage's per-value hard cap is 128 KiB. The JSON byte length measured in
847+// storeEvent is a close proxy for the runtime's serialized size; the margin
848+// below it absorbs the difference. Anything larger is delivered live only.
849+DurableObjectEventStore.MAX_STORABLE_EVENT_BYTES = 120 * 1024;
776 //#endregion850 //#endregion
777 //#region src/mcp/client-transports.ts851 //#region src/mcp/client-transports.ts
778 /**852 /**
779@@ -1381,6 +1794,47 @@ var McpAgent = class McpAgent extends Agent {853@@ -1381,6 +1834,47 @@ var McpAgent = class McpAgent extends Agent {
780 async deleteStreamRequestIds(streamId) {854 async deleteStreamRequestIds(streamId) {
781 await this.ctx.storage.delete(`${McpAgent.STREAM_REQS_KEY_PREFIX}${streamId}`);855 await this.ctx.storage.delete(`${McpAgent.STREAM_REQS_KEY_PREFIX}${streamId}`);
782 }856 }
@@ -824,7 +898,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..32d52598b42a4062d09020f0a1abfd67 · 41 unmodified lines
824 /**898 /**
825 * Reverse lookup: find which POST stream a given `requestId` belongs899 * Reverse lookup: find which POST stream a given `requestId` belongs
826 * to, and return the stream's full `requestIds` list in the same900 * to, and return the stream's full `requestIds` list in the same
827@@ -1697,7 +2151,8 @@ var McpAgent = class McpAgent extends Agent {901@@ -1697,7 +2191,8 @@ var McpAgent = class McpAgent extends Agent {
828 }902 }
829 };903 };
830 McpAgent.STREAM_REQS_KEY_PREFIX = "__mcp_stream_reqs__:";904 McpAgent.STREAM_REQS_KEY_PREFIX = "__mcp_stream_reqs__:";