History

Guard the legacy-SSE bridge forward loop against unknown frames

The worker<->DO unknown-frame tolerance was applied to the streamable-HTTP POST and GET forward loops and the DO ingress, but the legacy-SSE forward loop still dropped unparseable frames silently. Add the same mcpBridgeIsUnknownFrame guard there so a frame carrying an unrecognized bridge type during a staged deploy is warned and skipped while the stream survives, and cover it in the test.

Rhys Sullivan committed 4 months ago

Commit dda3dad1 parent 5d11ea52 files · +116 −14Browse files

Unified · Split

2 files
  1. packages/hosts/cloudflare/src/mcp/agents-bridge-unknown-frame.test.ts +101 −0
  2. patches/agents@0.17.3.patch +15 −14
packages/hosts/cloudflare/src/mcp/agents-bridge-unknown-frame.test.ts+101−0
@@ -91,6 +91,56 @@ const openPostSse = async (): Promise<{ response: Response; ws: FakeWebSocket }> · 90 unmodified lines
9191 return { response, ws };
9292 };
9393
94+// The legacy-SSE GET stream never closes on its own (no final-frame handshake),
95+// so open it, keep a live reader, and let the caller pull whatever bytes have
96+// been forwarded so far without waiting for a stream end that never comes.
97+const openLegacySse = async (): Promise<{
98+ response: Response;
99+ ws: FakeWebSocket;
100+ readSoFar: () => Promise<string>;
101+}> => {
102+ const ws = makeWebSocket();
103+ const agent = makeAgentStub(ws);
104+ const namespace = makeNamespace(agent);
105+ const handler = McpAgent.serve("/mcp", { binding: "MCP_SESSION", transport: "sse" });
106+ const response = await handler.fetch(
107+ new Request("https://executor.sh/mcp?sessionId=session-1", {
108+ headers: { accept: "text/event-stream" },
109+ method: "GET",
110+ }),
111+ { MCP_SESSION: namespace } as never,
112+ makeExecutionContext(),
113+ );
114+ expect(response.status).toBe(200);
115+ expect(ws.accepted).toBe(true);
116+ expect(response.body).toBeDefined();
117+
118+ const reader = response.body?.getReader();
119+ const decoder = new TextDecoder();
120+ let text = "";
121+ // A single outstanding read at a time (a default reader forbids concurrent
122+ // reads). Keep the pending read across calls: race it against a microtask
123+ // flush so a still-open stream resolves with only the bytes queued so far.
124+ let pending: ReturnType<NonNullable<typeof reader>["read"]> | undefined;
125+ const IDLE = { idle: true } as const;
126+ const readSoFar = async (): Promise<string> => {
127+ if (!reader) return text;
128+ for (;;) {
129+ pending ??= reader.read();
130+ const raced = await Promise.race([
131+ pending.then((r) => ({ idle: false as const, r })),
132+ flushMicrotasks().then(() => IDLE),
133+ ]);
134+ if (raced.idle) return text; // nothing more queued; keep `pending`
135+ pending = undefined;
136+ if (raced.r.done) break;
137+ text += decoder.decode(raced.r.value, { stream: true });
138+ }
139+ return text;
140+ };
141+ return { response, readSoFar, ws };
142+};
143+
94144 const drainResponse = (response: Response): Promise<string> => {
95145 const reader = response.body?.getReader();
96146 if (!reader) return Promise.resolve("");
@@ -202,4 +252,55 @@ describe("worker<->DO bridge unknown-frame tolerance", () => { · 105 unmodified lines
202252 expect(parseWarnings(warnLogs)).toEqual([]);
203253 expect(errorLogs).toEqual([]);
204254 });
255+
256+ // The legacy-SSE DO->worker forward loop is a separate code path from the
257+ // streamable-HTTP handlers above: it validates each frame against
258+ // JSONRPCMessageSchema and silently drops anything that fails. Without the
259+ // guard an unknown bridge frame (worker/DO version skew) would be dropped with
260+ // no signal; with it, the frame is warned + skipped and the stream survives.
261+ it("ignores an unknown bridge frame on the legacy-SSE stream and keeps it alive", async () => {
262+ const { readSoFar, ws } = await openLegacySse();
263+ // The initial `event: endpoint` frame is written on open.
264+ expect(await readSoFar()).toContain("event: endpoint");
265+
266+ // A newer DO emits a frame type this (older) worker has never seen.
267+ emitFrame(ws, { type: "cf_mcp_future_control", nonce: "abc" });
268+ await flushMicrotasks();
269+
270+ // The stream survives: no close, no error forwarded.
271+ expect(ws.closeCode).toBeUndefined();
272+ expect(ws.closeReason).toBeUndefined();
273+ expect(errorLogs).toEqual([]);
274+
275+ // Logged once, structured, with the frame type + direction.
276+ const warnings = parseWarnings(warnLogs);
277+ expect(warnings).toHaveLength(1);
278+ expect(warnings[0]).toMatchObject({
279+ event: "mcp_bridge_unknown_frame",
280+ frameType: "cf_mcp_future_control",
281+ direction: "do->worker",
282+ });
283+
284+ // A subsequent RECOGNIZED bare JSON-RPC frame still flows through as an SSE
285+ // message, proving the unknown frame did not wedge the legacy bridge.
286+ emitFrame(ws, { id: 1, jsonrpc: "2.0", result: { ok: true } });
287+ await flushMicrotasks();
288+ const body = await readSoFar();
289+ expect(body).toContain("event: message");
290+ expect(body).toContain('"ok":true');
291+ expect(ws.closeCode).toBeUndefined();
292+ });
293+
294+ it("does not warn on a recognized JSON-RPC frame over legacy-SSE", async () => {
295+ const { readSoFar, ws } = await openLegacySse();
296+ await readSoFar();
297+
298+ emitFrame(ws, { id: 1, jsonrpc: "2.0", result: { ok: true } });
299+ await flushMicrotasks();
300+ const body = await readSoFar();
301+
302+ expect(body).toContain("event: message");
303+ expect(parseWarnings(warnLogs)).toEqual([]);
304+ expect(errorLogs).toEqual([]);
305+ });
205306 });
patches/agents@0.17.3.patch+15−14
@@ -19,7 +19,7 @@ index c8fad448e8797b89690a99d93490d1363851b225..77f9fe3f6f2375eadc9f7a2974f0d202 · 18 unmodified lines
1919 McpAgent,
2020 type McpAuthContext,
2121 diff --git a/dist/mcp/index.js b/dist/mcp/index.js
22−index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd925f5d07dd 100644
22+index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..53917745f80c9d2960f2fa14f4f3bc8bfa7a320f 100644
2323 --- a/dist/mcp/index.js
2424 +++ b/dist/mcp/index.js
2525 @@ -7,6 +7,32 @@ import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/
@@ -397,7 +397,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 371 unmodified lines
397397 endpointUrl.pathname = encodeURI(`${basePath}/message`);
398398 endpointUrl.searchParams.set("sessionId", sessionId);
399399 const endpointMessage = `event: endpoint\ndata: ${endpointUrl.pathname + endpointUrl.search + endpointUrl.hash}\n\n`;
400−@@ -414,35 +606,94 @@ const createLegacySseHandler = (basePath, namespace, options = {}) => {
400+@@ -414,35 +606,95 @@ const createLegacySseHandler = (basePath, namespace, options = {}) => {
401401 console.error("Failed to establish WebSocket connection");
402402 await writer.close();
403403 return new Response("Failed to establish WebSocket connection", { status: 500 });
@@ -470,6 +470,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 66 unmodified lines
470470 - const messageText = `event: message\ndata: ${JSON.stringify(result.data)}\n\n`;
471471 - await writer.write(encoder.encode(messageText));
472472 - } catch (error) {
473++ if (mcpBridgeIsUnknownFrame(message, "do->worker")) return;
473474 + const result = JSONRPCMessageSchema.safeParse(message);
474475 + if (!result.success) return;
475476 + const messageText = `event: message\ndata: ${JSON.stringify(result.data)}\n\n`;
@@ -510,7 +511,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 34 unmodified lines
510511 console.error("Error closing SSE connection:", error);
511512 }
512513 }
513−@@ -634,7 +885,23 @@ var StreamableHTTPServerTransport = class {
514+@@ -634,7 +886,23 @@ var StreamableHTTPServerTransport = class {
514515 }
515516 this.supersedePriorStreamConnections(agent, connection.id, resumedStreamId);
516517 connection.setState(resumeState);
@@ -535,7 +536,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 18 unmodified lines
535536 return;
536537 }
537538 }
538−@@ -644,6 +911,26 @@ var StreamableHTTPServerTransport = class {
539+@@ -644,6 +912,26 @@ var StreamableHTTPServerTransport = class {
539540 _standaloneSse: true
540541 };
541542 connection.setState(standaloneState);
@@ -562,7 +563,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 20 unmodified lines
562563 }
563564 /**
564565 * Close any connection (other than `selfId`) currently bound to
565−@@ -664,12 +951,14 @@ var StreamableHTTPServerTransport = class {
566+@@ -664,12 +952,14 @@ var StreamableHTTPServerTransport = class {
566567 * Only used when resumability is enabled
567568 */
568569 async replayEvents(lastEventId) {
@@ -578,7 +579,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 9 unmodified lines
578579 this.writeSSEEvent(connection, message, eventId);
579580 } catch (error) {
580581 this.onerror?.(error);
581−@@ -678,6 +967,33 @@ var StreamableHTTPServerTransport = class {
582+@@ -678,6 +968,33 @@ var StreamableHTTPServerTransport = class {
582583 } catch (error) {
583584 this.onerror?.(error);
584585 }
@@ -612,7 +613,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 27 unmodified lines
612613 }
613614 /**
614615 * Writes an event to the SSE stream with proper formatting
615−@@ -689,6 +1005,8 @@ var StreamableHTTPServerTransport = class {
616+@@ -689,6 +1006,8 @@ var StreamableHTTPServerTransport = class {
616617 return connection.send(JSON.stringify({
617618 type: "cf_mcp_agent_event",
618619 event: eventData,
@@ -621,7 +622,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 2 unmodified lines
621622 close
622623 }));
623624 }
624−@@ -777,9 +1095,11 @@ var StreamableHTTPServerTransport = class {
625+@@ -777,9 +1096,11 @@ var StreamableHTTPServerTransport = class {
625626 } catch (error) {
626627 this.onerror?.(error);
627628 }
@@ -635,7 +636,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 7 unmodified lines
635636 }
636637 }
637638 async send(message, options) {
638−@@ -861,12 +1181,10 @@ var StreamableHTTPServerTransport = class {
639+@@ -861,12 +1182,10 @@ var StreamableHTTPServerTransport = class {
639640 *
640641 * ## Lifecycle
641642 *
@@ -652,7 +653,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 10 unmodified lines
652653 *
653654 * Standalone GET stream events (`_GET_stream`) are *not* cleared
654655 * automatically; they accumulate for the lifetime of the DO. Bounded
655−@@ -899,6 +1217,7 @@ var DurableObjectEventStore = class DurableObjectEventStore {
656+@@ -899,6 +1218,7 @@ var DurableObjectEventStore = class DurableObjectEventStore {
656657 const eventId = `${streamId}:${seq.toString(16).padStart(DurableObjectEventStore.SEQ_PAD, "0")}`;
657658 const eventKey = `${DurableObjectEventStore.EVENT_KEY_PREFIX}${eventId}`;
658659 await this.storage.put(eventKey, message);
@@ -660,7 +661,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 1 unmodified line
660661 return eventId;
661662 }
662663 async getStreamIdForEventId(eventId) {
663−@@ -915,9 +1234,59 @@ var DurableObjectEventStore = class DurableObjectEventStore {
664+@@ -915,9 +1235,59 @@ var DurableObjectEventStore = class DurableObjectEventStore {
664665 start: startKey,
665666 limit: DurableObjectEventStore.REPLAY_LIMIT
666667 });
@@ -721,7 +722,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 54 unmodified lines
721722 /**
722723 * Drop the event log for a single stream. Called by the transport
723724 * immediately after a POST's final response has been written to the
724−@@ -973,6 +1342,9 @@ DurableObjectEventStore.EVENT_KEY_PREFIX = "__mcp_event__:";
725+@@ -973,6 +1343,9 @@ DurableObjectEventStore.EVENT_KEY_PREFIX = "__mcp_event__:";
725726 DurableObjectEventStore.SEQ_PAD = 16;
726727 DurableObjectEventStore.DELETE_CHUNK = 128;
727728 DurableObjectEventStore.REPLAY_LIMIT = 1e3;
@@ -731,7 +732,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 3 unmodified lines
731732 //#endregion
732733 //#region src/mcp/client-transports.ts
733734 /**
734−@@ -1381,6 +1753,52 @@ var McpAgent = class McpAgent extends Agent {
735+@@ -1381,6 +1754,52 @@ var McpAgent = class McpAgent extends Agent {
735736 async deleteStreamRequestIds(streamId) {
736737 await this.ctx.storage.delete(`${McpAgent.STREAM_REQS_KEY_PREFIX}${streamId}`);
737738 }
@@ -784,7 +785,7 @@ index 1edcf0c8c9e67aa211ae515e7672cdf79912101e..e9cf6bd3cdec39b4e032f65b3deffd92 · 46 unmodified lines
784785 /**
785786 * Reverse lookup: find which POST stream a given `requestId` belongs
786787 * to, and return the stream's full `requestIds` list in the same
787−@@ -1697,7 +2115,8 @@ var McpAgent = class McpAgent extends Agent {
788+@@ -1697,7 +2116,8 @@ var McpAgent = class McpAgent extends Agent {
788789 }
789790 };
790791 McpAgent.STREAM_REQS_KEY_PREFIX = "__mcp_stream_reqs__:";