Deliver MCP responses larger than the DO storage value cap

.changeset/tidy-pianos-shake.md

@@ -0,0 +1,7 @@
+---
+"executor": patch
+---
+
+**Fix: MCP responses larger than Durable Object storage's 128 KiB per-value cap hung the client instead of being delivered**
+
+The 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

@@ -0,0 +1,136 @@
+// Cloud: an MCP response bigger than DO storage's 128 KiB per-value cap is
+// still delivered. The ui://executor/shell.html resource (~5 MB of inlined
+// shell document) is the production case: the transport used to persist every
+// outbound message with storage.put BEFORE writing the live SSE frame, so the
+// put's failure killed the send and the client saw keepalives forever — every
+// artifact widget in every remote MCP client hung, with no error anywhere
+// (found live in prod, 2026-07-29). The patched transport delivers live first
+// and treats persistence as best-effort, so this pins the whole path over the
+// real workerd topology: POST resources/read → McpSessionDO → ASSETS-backed
+// shell loader → event store skip (oversize) → live SSE frame → this client.
+import { expect } from "@effect/vitest";
+import { Effect } from "effect";
+
+import { scenario } from "../src/scenario";
+import { Mcp, Target } from "../src/services";
+import type { Identity } from "../src/target";
+
+const PROTOCOL_VERSION = "2025-03-26";
+const JSON_AND_SSE = "application/json, text/event-stream";
+
+const SHELL_RESOURCE_URI = "ui://executor/shell.html";
+
+const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label;
+// DO storage's per-value cap; anything beyond it exercises the skip path.
+const DO_VALUE_CAP_BYTES = 128 * 1024;
+
+const mcpHeaders = (bearer: string, sessionId?: string) => ({
+  accept: JSON_AND_SSE,
+  authorization: `Bearer ${bearer}`,
+  "content-type": "application/json",
+  "mcp-protocol-version": PROTOCOL_VERSION,
+  ...(sessionId ? { "mcp-session-id": sessionId } : {}),
+});
+
+const postJson = (mcpUrl: string, bearer: string, body: unknown, sessionId?: string) =>
+  fetch(mcpUrl, {
+    method: "POST",
+    headers: mcpHeaders(bearer, sessionId),
+    body: JSON.stringify(body),
+  });
+
+const openSession = async (mcpUrl: string, bearer: string): Promise<string> => {
+  const initialized = await postJson(mcpUrl, bearer, {
+    jsonrpc: "2.0" as const,
+    id: "initialize",
+    method: "initialize",
+    params: {
+      protocolVersion: PROTOCOL_VERSION,
+      capabilities: {},
+      clientInfo: { name: "executor-e2e-oversize-response", version: "0.0.1" },
+    },
+  });
+  const sessionId = initialized.headers.get("mcp-session-id");
+  await initialized.text();
+  if (initialized.status !== 200 || !sessionId) {
+    throw new Error(`openSession: initialize failed (${initialized.status})`);
+  }
+  const notification = await postJson(
+    mcpUrl,
+    bearer,
+    { jsonrpc: "2.0" as const, method: "notifications/initialized" },
+    sessionId,
+  );
+  await notification.text();
+  expect(notification.status, "notifications/initialized is accepted").toBe(202);
+  return sessionId;
+};
+
+/** Read the POST's SSE body to completion and return the JSON-RPC payloads. */
+const readSseMessages = async (
+  response: Response,
+): Promise<ReadonlyArray<{ readonly id?: unknown; readonly result?: unknown }>> => {
+  const text = await response.text();
+  const messages: Array<{ readonly id?: unknown; readonly result?: unknown }> = [];
+  for (const block of text.replace(/\r/g, "").split("\n\n")) {
+    const data = block
+      .split("\n")
+      .filter((line) => line.startsWith("data:"))
+      .map((line) => line.slice("data:".length).trimStart())
+      .join("\n")
+      .trim();
+    if (data) messages.push(JSON.parse(data) as { readonly id?: unknown });
+  }
+  return messages;
+};
+
+scenario(
+  "MCP streamable HTTP · a response beyond the DO storage value cap is delivered live (the ui:// shell resource)",
+  {},
+  Effect.gen(function* () {
+    const target = yield* Target;
+    const mcp = yield* Mcp;
+    const identity = yield* target.newIdentity();
+    const bearer = yield* mcp.mintBearer(emailOf(identity));
+    // Artifacts opt-in is what registers the shell resource on the session.
+    const mcpUrl = `${target.mcpUrl}?artifacts=true`;
+
+    const sessionId = yield* Effect.promise(() => openSession(mcpUrl, bearer));
+
+    const read = yield* Effect.promise(() =>
+      postJson(
+        mcpUrl,
+        bearer,
+        {
+          jsonrpc: "2.0" as const,
+          id: "read-shell",
+          method: "resources/read",
+          params: { uri: SHELL_RESOURCE_URI },
+        },
+        sessionId,
+      ),
+    );
+    expect(read.status, "resources/read answers").toBe(200);
+
+    // The whole regression is "the final frame never arrives": bound the read
+    // so a hang fails the test instead of wedging the suite.
+    const messages = yield* Effect.promise(() => readSseMessages(read)).pipe(
+      Effect.timeoutOrElse({
+        duration: "60 seconds",
+        orElse: () => Effect.die(new Error("shell read hung: the response frame never arrived")),
+      }),
+    );
+
+    const response = messages.find((message) => message.id === "read-shell");
+    expect(response, "the read's response frame arrived on the POST stream").toBeDefined();
+    const result = response?.result as
+      | { readonly contents?: ReadonlyArray<{ readonly text?: string }> }
+      | undefined;
+    const html = result?.contents?.[0]?.text ?? "";
+    expect(
+      html.length,
+      "the document is larger than the DO storage cap, so it took the live-only path",
+    ).toBeGreaterThan(DO_VALUE_CAP_BYTES);
+    expect(html, "and it is the real shell document").toContain('name="executor-mcp-apps-shell"');
+  }),
+);

packages/hosts/cloudflare/src/mcp/agents-event-store.test.ts

@@ -62,7 +62,13 @@ describe("DurableObjectEventStore trimStream", () => {
     vi.restoreAllMocks();
   });
 
-  it("keeps an oversize final response instead of evicting the only event", async () => {
+  it("skips storing a message beyond DO storage's per-value cap, returning no replay id", async () => {
+    // Real DO storage rejects any value over 128 KiB. This used to be
+    // discovered inside storage.put — thrown BEFORE the live SSE write, so an
+    // oversize response (the ~5MB ui:// shell document) was neither stored nor
+    // delivered and the client hung on keepalives. The pinned contract now:
+    // oversize messages skip persistence with a warning and resolve undefined,
+    // and the caller delivers them live without a replay id.
     const { storage, store } = makeStore();
     const hugeResult = {
       jsonrpc: "2.0" as const,
@@ -72,33 +78,56 @@ describe("DurableObjectEventStore trimStream", () => {
 
     const eventId = await store.storeEvent("post-stream", hugeResult);
 
-    expect(eventId, "storeEvent returns the event id").toBe("post-stream:0000000000000001");
-    expect(eventKeys(storage), "the >2MB final response survives its own trim pass").toEqual([
-      "__mcp_event__:post-stream:0000000000000001",
-    ]);
-    expect(warnings, "no eviction happened, so no eviction warning").toEqual([]);
+    expect(eventId, "no replay id for an unstorable message").toBeUndefined();
+    expect(eventKeys(storage), "nothing was persisted").toEqual([]);
+    expect(warnings.length, "the skip is logged").toBeGreaterThan(0);
+    expect(warnings.at(-1)).toContain("mcp_event_store_skipped_oversize");
+    expect(warnings.at(-1)).toContain("post-stream");
+  });
+
+  it("does not advance the stream sequence when an oversize message is skipped", async () => {
+    const { storage, store } = makeStore();
+    await store.storeEvent("post-stream", {
+      jsonrpc: "2.0" as const,
+      method: "notifications/progress",
+      params: { blob: "x".repeat(3 * 1024 * 1024) },
+    });
+
+    const eventId = await store.storeEvent("post-stream", {
+      jsonrpc: "2.0" as const,
+      id: 1,
+      result: { ok: true },
+    });
+
+    expect(eventId, "the next storable event takes the first sequence slot").toBe(
+      "post-stream:0000000000000001",
+    );
+    expect(eventKeys(storage)).toEqual(["__mcp_event__:post-stream:0000000000000001"]);
   });
 
   it("evicts oldest events at the byte cap but never the newest", async () => {
     const { storage, store } = makeStore();
+    // 100 KiB each: storable (under the 120 KiB per-value guard), but 25 of
+    // them exceed the 2 MB per-stream byte cap.
     const bigMessage = (marker: string) => ({
       jsonrpc: "2.0" as const,
       method: "notifications/progress",
-      params: { marker, blob: "y".repeat(1024 * 1024) },
+      params: { marker, blob: "y".repeat(100 * 1024) },
     });
 
-    await store.storeEvent("post-stream", bigMessage("one"));
-    await store.storeEvent("post-stream", bigMessage("two"));
-    await store.storeEvent("post-stream", bigMessage("three"));
+    const total = 25;
+    for (let index = 0; index < total; index += 1) {
+      await store.storeEvent("post-stream", bigMessage(`msg-${index}`));
+    }
 
     const remaining = eventKeys(storage);
     expect(remaining[remaining.length - 1], "the newest event is always retained").toBe(
-      "__mcp_event__:post-stream:0000000000000003",
+      `__mcp_event__:post-stream:${total.toString(16).padStart(16, "0")}`,
     );
     expect(
       remaining.length,
       "older events were evicted to satisfy the 2MB stream cap",
-    ).toBeLessThan(3);
+    ).toBeLessThan(total);
     expect(warnings.length, "eviction logs a warning").toBeGreaterThan(0);
     expect(warnings.at(-1)).toContain("mcp_event_store_evicted");
     expect(warnings.at(-1)).toContain("post-stream");

packages/hosts/cloudflare/src/mcp/agents-priming-event.test.ts

@@ -74,7 +74,10 @@ describe("POST-stream priming event: store ordering and replay", () => {
     const store = new DurableObjectEventStore(storage as never);
     const streamId = "post-stream";
 
-    // Transport order: priming event first, then the tool response.
+    // Transport order: priming event first, then the tool response. Both are
+    // tiny, so storeEvent always persists them and returns an id — the
+    // `undefined` arm of its signature is the oversize skip, pinned in
+    // agents-event-store.test.ts.
     const primingId = await store.storeEvent(streamId, PRIMING_MESSAGE);
     const response = {
       jsonrpc: "2.0" as const,
@@ -87,10 +90,13 @@ describe("POST-stream priming event: store ordering and replay", () => {
     expect(responseId, "response is seq 2, after the priming id").toBe(
       `${streamId}:0000000000000002`,
     );
-    expect(primingId < responseId, "priming id sorts strictly before the response id").toBe(true);
+    expect(
+      (primingId ?? "") < (responseId ?? ""),
+      "priming id sorts strictly before the response id",
+    ).toBe(true);
 
     const replayed: Array<{ readonly eventId: string; readonly message: unknown }> = [];
-    await store.replayEventsAfter(primingId, {
+    await store.replayEventsAfter(primingId ?? "", {
       send: async (eventId: string, message: unknown) => {
         replayed.push({ eventId, message });
       },
@@ -108,9 +114,10 @@ describe("POST-stream priming event: store ordering and replay", () => {
     const store = new DurableObjectEventStore(storage as never);
     const streamId = "post-stream";
     const primingId = await store.storeEvent(streamId, PRIMING_MESSAGE);
+    expect(primingId, "a tiny priming message always persists and gets an id").toBeDefined();
 
     const replayed: string[] = [];
-    await store.replayEventsAfter(primingId, {
+    await store.replayEventsAfter(primingId ?? "", {
       send: async (eventId: string) => {
         replayed.push(eventId);
       },