Fix session-id parsing, stranded-stream replay, priming, and restore races

Rhys Sullivan committed 2 months ago

Commit 803b0871 parent d2e97bc15 files · +521 −90Browse files

Unified · Split

15 files

apps/cloud/src/auth/handlers.ts +6 −0
apps/cloud/src/mcp/agent-handler.ts +9 −8
apps/cloud/src/mcp/session-durable-object.ts +5 −6
apps/host-cloudflare/src/mcp/agent-handler.ts +9 −8
apps/host-cloudflare/src/mcp/index.ts +6 −2
apps/host-cloudflare/src/mcp/session-durable-object.ts +5 −6
packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts +133 −4
packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts +145 −42
packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts +3 −5
packages/hosts/cloudflare/src/mcp/do-event-store.test.ts +32 −0
packages/hosts/cloudflare/src/mcp/do-event-store.ts +79 −0
packages/hosts/cloudflare/src/mcp/modern-request-router.test.ts +36 −0
packages/hosts/cloudflare/src/mcp/modern-request-router.ts +8 −2
packages/hosts/cloudflare/src/mcp/session-stub.test.ts +30 −0
packages/hosts/cloudflare/src/mcp/session-stub.ts +15 −7

@@ -624,6 +624,9 @@ export const CloudSessionAuthHandlers = HttpApiBuilder.group( · 623 unmodified lines
624	624	Effect.gen(function* () {
625	625	const owner = yield* requireSelectedOrganization;
626	626	const stub = getMcpSessionStub(params.mcpSessionId);
	627	+ if (!stub) {
	628	+ return yield* new McpExecutionNotFoundError({ executionId: params.executionId });
	629	+ }
627	630	const result = yield* Effect.promise(() =>
628	631	stub.getPausedExecutionForApproval(params.executionId, {
629	632	accountId: owner.accountId,
@@ -645,6 +648,9 @@ export const CloudSessionAuthHandlers = HttpApiBuilder.group( · 15 unmodified lines
645	648	Effect.gen(function* () {
646	649	const owner = yield* requireSelectedOrganization;
647	650	const stub = getMcpSessionStub(params.mcpSessionId);
	651	+ if (!stub) {
	652	+ return yield* new McpExecutionNotFoundError({ executionId: params.executionId });
	653	+ }
648	654	const result = yield* Effect.promise(() =>
649	655	stub.resumeExecutionForApproval(
650	656	params.executionId,

apps/cloud/src/mcp/agent-handler.ts+9−8

@@ -161,11 +161,10 @@ export const makeCloudMcpAgentHandler = () => { · 160 unmodified lines
161	161	// / JWKS failure) and `Unauthorized` (retry with a fresh token) must leave
162	162	// the session intact, so the condemn path is gated on `Forbidden` alone.
163	163	if (Predicate.isTagged(outcome, "Forbidden") && sessionId) {
	164	+ const session = mcpSessionStub(env.MCP_SESSION, sessionId);
164	165	await Effect.runPromise(
165	166	Effect.ignore(
166		− Effect.tryPromise(() =>
167		− mcpSessionStub(env.MCP_SESSION, sessionId)._cf_scheduleDestroy(),
168		− ),
	167	+ session ? Effect.tryPromise(() => session._cf_scheduleDestroy()) : Effect.void,
169	168	),
170	169	);
171	170	}
@@ -212,8 +211,12 @@ export const makeCloudMcpAgentHandler = () => { · 40 unmodified lines
212	211	});
213	212	}
214	213
215		− if (sessionId) {
216		− const owner = await mcpSessionStub(env.MCP_SESSION, sessionId).validateMcpSessionOwner({
	214	+ const existingSession = sessionId ? mcpSessionStub(env.MCP_SESSION, sessionId) : null;
	215	+ if (sessionId && !existingSession) {
	216	+ return jsonRpcResponse(404, -32001, "Session not found");
	217	+ }
	218	+ if (existingSession) {
	219	+ const owner = await existingSession.validateMcpSessionOwner({
217	220	accountId: outcome.principal.accountId,
218	221	organizationId: outcome.principal.organizationId,
219	222	});
@@ -244,9 +247,7 @@ export const makeCloudMcpAgentHandler = () => { · 24 unmodified lines
244	247	),
245	248	propagation,
246	249	);
247		− const target = sessionId
248		− ? mcpSessionStub(env.MCP_SESSION, sessionId)
249		− : createMcpSessionStub(env.MCP_SESSION).stub;
	250	+ const target = existingSession ?? createMcpSessionStub(env.MCP_SESSION).stub;
250	251	let response: Response;
251	252	// oxlint-disable-next-line executor/no-try-catch-or-throw -- adapter boundary: a condemned DO abort can reject its direct fetch
252	253	try {

apps/cloud/src/mcp/session-durable-object.ts+5−6

@@ -325,13 +325,12 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase<Env, CloudSessionD · 324 unmodified lines
325	325	executionId: string,
326	326	response: ResumeResponse,
327	327	): Effect.Effect<McpSessionModelResumeResult, unknown> {
	328	+ const ownerSession = mcpSessionStubForOwner(env.MCP_SESSION, owner);
	329	+ if (!ownerSession) {
	330	+ return Effect.succeed({ status: "execution_expired", ttlMs: PAUSED_APPROVAL_TIMEOUT_MS });
	331	+ }
328	332	return Effect.tryPromise({
329		− try: () =>
330		− mcpSessionStubForOwner(env.MCP_SESSION, owner).resumeExecutionForModel(
331		− executionId,
332		− identity,
333		− response,
334		− ),
	333	+ try: () => ownerSession.resumeExecutionForModel(executionId, identity, response),
335	334	catch: (cause) => new McpModelResumeForwardError({ cause }),
336	335	});
337	336	}

apps/host-cloudflare/src/mcp/agent-handler.ts+9−8

@@ -99,11 +99,10 @@ export const makeCloudflareMcpAgentHandler = (config: CloudflareConfig) => { · 98 unmodified lines
99	99	const { auth, outcome } = await Effect.runPromise(authenticate(request, config));
100	100	if (!Predicate.isTagged(outcome, "Authenticated")) {
101	101	if (Predicate.isTagged(outcome, "Forbidden") && sessionId) {
	102	+ const session = mcpSessionStub(env.MCP_SESSION, sessionId);
102	103	await Effect.runPromise(
103	104	Effect.ignore(
104		− Effect.tryPromise(() =>
105		− mcpSessionStub(env.MCP_SESSION, sessionId)._cf_scheduleDestroy(),
106		− ),
	105	+ session ? Effect.tryPromise(() => session._cf_scheduleDestroy()) : Effect.void,
107	106	),
108	107	);
109	108	}
@@ -140,8 +139,12 @@ export const makeCloudflareMcpAgentHandler = (config: CloudflareConfig) => { · 30 unmodified lines
140	139	return new Response(null, { status: 204, headers: { "access-control-allow-origin": "*" } });
141	140	}
142	141
143		− if (sessionId) {
144		− const owner = await mcpSessionStub(env.MCP_SESSION, sessionId).validateMcpSessionOwner({
	142	+ const existingSession = sessionId ? mcpSessionStub(env.MCP_SESSION, sessionId) : null;
	143	+ if (sessionId && !existingSession) {
	144	+ return jsonRpcResponse(404, -32001, "Session not found");
	145	+ }
	146	+ if (existingSession) {
	147	+ const owner = await existingSession.validateMcpSessionOwner({
145	148	accountId: outcome.principal.accountId,
146	149	organizationId: outcome.principal.organizationId,
147	150	});
@@ -170,9 +173,7 @@ export const makeCloudflareMcpAgentHandler = (config: CloudflareConfig) => { · 22 unmodified lines
170	173	),
171	174	propagation,
172	175	);
173		− const target = sessionId
174		− ? mcpSessionStub(env.MCP_SESSION, sessionId)
175		− : createMcpSessionStub(env.MCP_SESSION).stub;
	176	+ const target = existingSession ?? createMcpSessionStub(env.MCP_SESSION).stub;
176	177	return withMcpResponseHeaders(await target.fetch(forwarded));
177	178	};
178	179	};

apps/host-cloudflare/src/mcp/index.ts+6−2

@@ -35,7 +35,9 @@ export const makeCloudflareApprovalHandler = ( · 34 unmodified lines
35	35
36	36	const paused = PAUSED_PATH.exec(pathname);
37	37	if (paused && request.method === "GET") {
38		− const result = await stubFor(decodeURIComponent(paused[1]!)).getPausedExecutionForApproval(
	38	+ const stub = stubFor(decodeURIComponent(paused[1]!));
	39	+ if (!stub) return jsonResponse({ error: "Paused execution not found" }, 404);
	40	+ const result = await stub.getPausedExecutionForApproval(
39	41	decodeURIComponent(paused[2]!),
40	42	owner,
41	43	);
@@ -53,7 +55,9 @@ export const makeCloudflareApprovalHandler = ( · 11 unmodified lines
53	55	const response = raw === null ? null : decodeResumeResponse(raw);
54	56	if (!response) return jsonResponse({ error: "Invalid approval response" }, 400);
55	57
56		− const result = await stubFor(decodeURIComponent(resume[1]!)).resumeExecutionForApproval(
	58	+ const stub = stubFor(decodeURIComponent(resume[1]!));
	59	+ if (!stub) return jsonResponse({ error: "Paused execution not found" }, 404);
	60	+ const result = await stub.resumeExecutionForApproval(
57	61	decodeURIComponent(resume[2]!),
58	62	owner,
59	63	response,

apps/host-cloudflare/src/mcp/session-durable-object.ts+5−6

@@ -210,13 +210,12 @@ export class McpSessionDO extends McpAgentSessionDOBase<CloudflareEnv, CfSession · 209 unmodified lines
210	210	executionId: string,
211	211	response: ResumeResponse,
212	212	): Effect.Effect<McpSessionModelResumeResult, unknown> {
	213	+ const ownerSession = mcpSessionStubForOwner(this.cfEnv.MCP_SESSION, owner);
	214	+ if (!ownerSession) {
	215	+ return Effect.succeed({ status: "execution_expired", ttlMs: PAUSED_APPROVAL_TIMEOUT_MS });
	216	+ }
213	217	return Effect.tryPromise({
214		− try: () =>
215		− mcpSessionStubForOwner(this.cfEnv.MCP_SESSION, owner).resumeExecutionForModel(
216		− executionId,
217		− identity,
218		− response,
219		− ),
	218	+ try: () => ownerSession.resumeExecutionForModel(executionId, identity, response),
220	219	catch: (cause) => new McpModelResumeForwardError({ cause }),
221	220	});
222	221	}

packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts+133−4

@@ -1,4 +1,4 @@
1		−import { describe, expect, it } from "@effect/vitest";
	1	+import { afterEach, describe, expect, it, vi } from "@effect/vitest";
2	2	import { Cause, Effect } from "effect";
3	3	import { Client } from "@modelcontextprotocol/sdk/client/index.js";
4	4	import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
@@ -181,17 +181,28 @@ const engine: ExecutionEngine<Cause.YieldableError> = { · 176 unmodified lines
181	181	getDescription: Effect.succeed("test Durable Object executor"),
182	182	};
183	183
184		−class HarnessSession extends McpAgentSessionDOBase<Cloudflare.Env, { readonly end: () => void }> {
	184	+class HarnessSession extends McpAgentSessionDOBase<
	185	+ Cloudflare.Env,
	186	+ { readonly end: () => void \| Promise<void> }
	187	+> {
185	188	constructor(
186	189	ctx: DurableObjectState,
187	190	env: Cloudflare.Env,
188	191	private readonly sessionEngine: ExecutionEngine<Cause.YieldableError> = engine,
	192	+ private readonly runtimeOptions: {
	193	+ readonly end?: () => void \| Promise<void>;
	194	+ readonly sessionTimeoutMs?: number;
	195	+ } = {},
189	196	) {
190	197	super(ctx, env);
191	198	}
192	199
193		− protected override openSessionDb(): { readonly end: () => void } {
194		− return { end: () => undefined };
	200	+ protected override openSessionDb(): { readonly end: () => void \| Promise<void> } {
	201	+ return { end: this.runtimeOptions.end ?? (() => undefined) };
	202	+ }
	203	+
	204	+ protected override sessionTimeoutMs(): number {
	205	+ return this.runtimeOptions.sessionTimeoutMs ?? super.sessionTimeoutMs();
195	206	}
196	207
197	208	protected override resolveSessionMeta(token: McpSessionInit): Effect.Effect<SessionMeta> {
@@ -257,6 +268,10 @@ const makeClientHarness = (state = new MemoryDurableObjectState()) => { · 59 unmodified lines
257	268	};
258	269
259	270	describe("McpAgentSessionDOBase session serving", () => {
	271	+ afterEach(() => {
	272	+ vi.restoreAllMocks();
	273	+ });
	274	+
260	275	it("serves and reuses a legacy v1 SDK client through the Durable Object", async () => {
261	276	const harness = makeClientHarness();
262	277	await harness.client.connect(harness.transport);
@@ -390,6 +405,120 @@ describe("McpAgentSessionDOBase session serving", () => { · 127 unmodified lines
390	405	const replayBody = await replay.text();
391	406	expect(replayBody).toContain("slow result");
392	407	expect(replayBody).toContain(`id: ${replayEventId.slice(0, replayEventId.lastIndexOf(":"))}:`);
	408	+
	409	+ const standaloneReplay = await session.fetch(
	410	+ verifiedRequest(
	411	+ new Request("https://executor.test/mcp", {
	412	+ method: "GET",
	413	+ headers: {
	414	+ accept: "text/event-stream",
	415	+ "mcp-session-id": SESSION_ID,
	416	+ "mcp-protocol-version": "2025-06-18",
	417	+ },
	418	+ }),
	419	+ ),
	420	+ );
	421	+ const standaloneReplayBody = await standaloneReplay.text();
	422	+ expect(standaloneReplayBody).toContain("slow result");
	423	+ expect(standaloneReplayBody).toContain("event: message");
	424	+ await state.flushWaitUntil();
	425	+ });
	426	+
	427	+ it("restores once while an idle runtime generation is still closing", async () => {
	428	+ let now = 1_000;
	429	+ vi.spyOn(Date, "now").mockImplementation(() => now);
	430	+ let closeStarted = (): void => undefined;
	431	+ const closing = new Promise<void>((resolve) => {
	432	+ closeStarted = resolve;
	433	+ });
	434	+ let finishClose = (): void => undefined;
	435	+ const closeGate = new Promise<void>((resolve) => {
	436	+ finishClose = resolve;
	437	+ });
	438	+ let closeCount = 0;
	439	+ const state = new MemoryDurableObjectState();
	440	+ const session = new HarnessSession(state, {} as Cloudflare.Env, engine, {
	441	+ sessionTimeoutMs: 10,
	442	+ end: () => {
	443	+ closeCount += 1;
	444	+ if (closeCount !== 1) return;
	445	+ closeStarted();
	446	+ return closeGate;
	447	+ },
	448	+ });
	449	+ const post = (body: unknown): Request =>
	450	+ verifiedRequest(
	451	+ new Request("https://executor.test/mcp", {
	452	+ method: "POST",
	453	+ headers: {
	454	+ accept: "application/json, text/event-stream",
	455	+ "content-type": "application/json",
	456	+ "mcp-session-id": SESSION_ID,
	457	+ "mcp-protocol-version": "2025-06-18",
	458	+ },
	459	+ body: JSON.stringify(body),
	460	+ }),
	461	+ );
	462	+
	463	+ const initialize = await session.fetch(
	464	+ verifiedRequest(
	465	+ new Request("https://executor.test/mcp", {
	466	+ method: "POST",
	467	+ headers: {
	468	+ accept: "application/json, text/event-stream",
	469	+ "content-type": "application/json",
	470	+ },
	471	+ body: JSON.stringify({
	472	+ jsonrpc: "2.0",
	473	+ id: "initialize",
	474	+ method: "initialize",
	475	+ params: {
	476	+ protocolVersion: "2025-06-18",
	477	+ capabilities: {},
	478	+ clientInfo: { name: "restore-race", version: "1.0.0" },
	479	+ },
	480	+ }),
	481	+ }),
	482	+ ),
	483	+ );
	484	+ await initialize.text();
	485	+ await session.fetch(post({ jsonrpc: "2.0", method: "notifications/initialized", params: {} }));
	486	+
	487	+ now += 100;
	488	+ const alarm = session.alarm();
	489	+ await closing;
	490	+
	491	+ const get = session.fetch(
	492	+ verifiedRequest(
	493	+ new Request("https://executor.test/mcp", {
	494	+ method: "GET",
	495	+ headers: {
	496	+ accept: "text/event-stream",
	497	+ "mcp-session-id": SESSION_ID,
	498	+ "mcp-protocol-version": "2025-06-18",
	499	+ },
	500	+ }),
	501	+ ),
	502	+ );
	503	+ const list = session.fetch(
	504	+ post({ jsonrpc: "2.0", id: "concurrent-list", method: "tools/list", params: {} }),
	505	+ );
	506	+
	507	+ finishClose();
	508	+ const [getResponse, listResponse] = await Promise.all([get, list]);
	509	+ expect(getResponse.status).toBe(200);
	510	+ await getResponse.body?.cancel();
	511	+ expect(listResponse.status).toBe(200);
	512	+ expect(await listResponse.text()).toContain("execute");
	513	+ await alarm;
	514	+ await expect(state.storage.get("executor:mcp:v2:last-activity-ms")).resolves.toBe(now);
	515	+ await expect(state.storage.getAlarm()).resolves.toBe(now + 10);
	516	+
	517	+ const followUp = await session.fetch(
	518	+ post({ jsonrpc: "2.0", id: "follow-up-list", method: "tools/list", params: {} }),
	519	+ );
	520	+ expect(followUp.status).toBe(200);
	521	+ expect(await followUp.text()).toContain("execute");
393	522	});
394	523
395	524	it("persists session metadata and rejects a different principal", async () => {

packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts+145−42

@@ -256,6 +256,21 @@ const legacyPrimingFrame = (eventId: string): Uint8Array => · 255 unmodified lines
256	256	`event: mcp-priming\nid: ${eventId}\ndata: ${JSON.stringify(LEGACY_PRIMING_MESSAGE)}\n\n`,
257	257	);
258	258
	259	+const replayFrame = (eventId: string, message: JSONRPCMessage): Uint8Array =>
	260	+ new TextEncoder().encode(`event: message\nid: ${eventId}\ndata: ${JSON.stringify(message)}\n\n`);
	261	+
	262	+const combineFrames = (frames: readonly Uint8Array[]): ArrayBuffer => {
	263	+ const byteLength = frames.reduce((total, frame) => total + frame.byteLength, 0);
	264	+ const buffer = new ArrayBuffer(byteLength);
	265	+ const combined = new Uint8Array(buffer);
	266	+ let offset = 0;
	267	+ for (const frame of frames) {
	268	+ combined.set(frame, offset);
	269	+ offset += frame.byteLength;
	270	+ }
	271	+ return buffer;
	272	+};
	273	+
259	274	const mcpResourceFromKey = (resourceKey: string): McpResource =>
260	275	resourceKey.startsWith("toolkit:") && resourceKey.length > "toolkit:".length
261	276	? { kind: "toolkit", slug: resourceKey.slice("toolkit:".length) }
@@ -552,6 +567,7 @@ export abstract class McpAgentSessionDOBase< · 290 unmodified lines
552	567
553	568	private async disposeIdleRuntime(input: {
554	569	readonly idleMs: number;
	570	+ readonly lastActivityMs: number;
555	571	readonly pausedExecutionCount: number;
556	572	}): Promise<void> {
557	573	console.info(
@@ -563,6 +579,7 @@ export abstract class McpAgentSessionDOBase< · 5 unmodified lines
563	579	}),
564	580	);
565	581	await Effect.runPromise(this.closeRuntime());
	582	+<<<<<<< HEAD
566	583	await Effect.runPromise(
567	584	Effect.all([
568	585	Effect.ignore(Effect.tryPromise(() => this.ctx.storage.deleteAlarm())),
@@ -573,6 +590,18 @@ export abstract class McpAgentSessionDOBase< · 4 unmodified lines
573	590	),
574	591	]),
575	592	);
	593	+=======
	594	+ const activityKey =
	595	+ this.runtimeKind === "modern" ? MODERN_LAST_ACTIVITY_KEY : LEGACY_V2_LAST_ACTIVITY_KEY;
	596	+ const cleared = await this.ctx.storage.transaction(async (transaction) => {
	597	+ const current = await transaction.get<number>(activityKey);
	598	+ if (current !== input.lastActivityMs) return false;
	599	+ await transaction.delete([LEGACY_V2_LAST_ACTIVITY_KEY, MODERN_LAST_ACTIVITY_KEY]);
	600	+ await transaction.deleteAlarm();
	601	+ return true;
	602	+ });
	603	+ if (cleared) this.lastActivityMs = 0;
	604	+>>>>>>> d62e666df (Fix session-id parsing, stranded-stream replay, priming, and restore races)
576	605	}
577	606
578	607	private resolveAndStoreSessionMeta(token: McpSessionInit) {
@@ -823,34 +852,39 @@ export abstract class McpAgentSessionDOBase< · 244 unmodified lines
823	852	private closeRuntime(): Effect.Effect<void> {
824	853	const self = this;
825	854	return Effect.gen(function* () {
826		− yield* self.releaseAllPendingApprovalLeases();
827		− if (self.transport) {
828		− const transport = self.transport;
829		− self.transport = null;
830		− yield* Effect.promise(() => transport.close()).pipe(Effect.ignore);
831		− }
832		− if (self.server) {
833		− const server = self.server;
834		− delete (self as { server?: McpServer }).server;
835		− yield* Effect.promise(() => server.close()).pipe(Effect.ignore);
836		− }
837		− if (self.modernHandler) {
838		− const handler = self.modernHandler;
839		− self.modernHandler = null;
840		− yield* Effect.promise(() => handler.close()).pipe(Effect.ignore);
841		− }
	855	+ // Detach the complete generation before awaiting cleanup. A request that
	856	+ // interleaves with a slow server/DB close must build a fresh generation,
	857	+ // never observe initialized=true with a closing/null transport, and the
	858	+ // old cleanup must never clear fields belonging to that fresh runtime.
	859	+ const transport = self.transport;
	860	+ const server = self.server;
	861	+ const modernHandler = self.modernHandler;
	862	+ const dbHandle = self.dbHandle;
	863	+ self.transport = null;
	864	+ delete (self as { server?: McpServer }).server;
	865	+ self.modernHandler = null;
	866	+ self.dbHandle = null;
842	867	self.engine = null;
843	868	self.modernRuntime = null;
844	869	self.activeLegacyStreamCount = 0;
845	870	self.legacyRunningRequestCount = 0;
846	871	self.modernRequestBodies = new WeakMap<Request, unknown>();
847	872	self.modernRequestPropagation = new WeakMap<Request, IncomingTraceHeaders \| undefined>();
848		− if (self.dbHandle) {
849		− const dbHandle = self.dbHandle;
850		− self.dbHandle = null;
	873	+ self.initialized = false;
	874	+
	875	+ yield* self.releaseAllPendingApprovalLeases();
	876	+ if (transport) {
	877	+ yield* Effect.promise(() => transport.close()).pipe(Effect.ignore);
	878	+ }
	879	+ if (server) {
	880	+ yield* Effect.promise(() => server.close()).pipe(Effect.ignore);
	881	+ }
	882	+ if (modernHandler) {
	883	+ yield* Effect.promise(() => modernHandler.close()).pipe(Effect.ignore);
	884	+ }
	885	+ if (dbHandle) {
851	886	yield* Effect.promise(() => Promise.resolve(dbHandle.end())).pipe(Effect.ignore);
852	887	}
853		− self.initialized = false;
854	888	});
855	889	}
856	890
@@ -1000,6 +1034,69 @@ export abstract class McpAgentSessionDOBase< · 143 unmodified lines
1000	1034	return typeof streamId === "string" ? streamId : null;
1001	1035	}
1002	1036
	1037	+ private async supersedeReplayStream(
	1038	+ transport: WebStandardStreamableHTTPServerTransport,
	1039	+ lastEventId: string,
	1040	+ ): Promise<void> {
	1041	+ const streamId = await this.eventStore.getStreamIdForEventId(lastEventId);
	1042	+ if (!streamId) return;
	1043	+ // SAFETY: the installed SDK exposes closeSSEStream(requestId), but not the
	1044	+ // reverse request-id map needed to supersede a stale POST connection before
	1045	+ // replay. This is the same pinned map used by requestStreamId above.
	1046	+ const mapping: unknown = Reflect.get(transport, "_requestToStreamMapping");
	1047	+ if (!(mapping instanceof Map)) return;
	1048	+ for (const [requestId, mappedStreamId] of mapping) {
	1049	+ if (
	1050	+ mappedStreamId === streamId &&
	1051	+ (typeof requestId === "string" \|\| typeof requestId === "number")
	1052	+ ) {
	1053	+ transport.closeSSEStream(requestId);
	1054	+ return;
	1055	+ }
	1056	+ }
	1057	+ }
	1058	+
	1059	+ private trackedLegacyResponse = (
	1060	+ response: Response,
	1061	+ options: { readonly initialFrame?: Uint8Array; readonly acknowledge?: readonly string[] } = {},
	1062	+ ): Response =>
	1063	+ rotateSseResponse(response, {
	1064	+ ...(options.initialFrame ? { initialFrame: options.initialFrame } : {}),
	1065	+ onOpen: () => {
	1066	+ this.activeLegacyStreamCount += 1;
	1067	+ },
	1068	+ onClose: (reason) => {
	1069	+ this.activeLegacyStreamCount = Math.max(0, this.activeLegacyStreamCount - 1);
	1070	+ if (reason === "complete" && options.acknowledge && options.acknowledge.length > 0) {
	1071	+ this.ctx.waitUntil(this.eventStore.acknowledgeUndeliveredStreams(options.acknowledge));
	1072	+ }
	1073	+ },
	1074	+ });
	1075	+
	1076	+ private async replayUndeliveredOnStandaloneGet(request: Request): Promise<Response \| null> {
	1077	+ if (request.method !== "GET" \|\| request.headers.has("last-event-id")) return null;
	1078	+ const frames: Uint8Array[] = [];
	1079	+ const streamIds = await this.eventStore.replayUndeliveredStreams({
	1080	+ send: (eventId, message) => {
	1081	+ frames.push(replayFrame(eventId, message));
	1082	+ return Promise.resolve();
	1083	+ },
	1084	+ });
	1085	+ if (frames.length === 0) return null;
	1086	+ return this.trackedLegacyResponse(
	1087	+ new Response(combineFrames(frames), {
	1088	+ headers: {
	1089	+ "content-type": "text/event-stream",
	1090	+ "cache-control": "no-cache, no-transform",
	1091	+ connection: "keep-alive",
	1092	+ "x-accel-buffering": "no",
	1093	+ "mcp-session-id": this.sessionId,
	1094	+ },
	1095	+ }),
	1096	+ { acknowledge: streamIds },
	1097	+ );
	1098	+ }
	1099	+
1003	1100	private async serializedTransportRequest<A>(run: () => Promise<A>): Promise<A> {
1004	1101	const previous = this.transportRequestTail;
1005	1102	let release = (): void => undefined;
@@ -1019,25 +1116,34 @@ export abstract class McpAgentSessionDOBase< · 13 unmodified lines
1019	1116	request: Request,
1020	1117	parsedBody: unknown,
1021	1118	): Promise<Response> {
1022		− const transport = this.transport;
1023		− if (!transport) {
1024		− return jsonRpcErrorBody(404, -32001, "Session not found", { cors: false });
1025		− }
1026	1119	return this.serializedTransportRequest(async () => {
	1120	+ const transport = this.transport;
	1121	+ if (!transport) {
	1122	+ return jsonRpcErrorBody(404, -32001, "Session not found", { cors: false });
	1123	+ }
	1124	+ if (request.method === "GET") {
	1125	+ const lastEventId = request.headers.get("last-event-id");
	1126	+ if (lastEventId) {
	1127	+ await this.supersedeReplayStream(transport, lastEventId);
	1128	+ } else {
	1129	+ // Latest-listener-wins. Client cancellation is not reliably relayed
	1130	+ // through every workerd/Vite streaming hop, so explicitly retire a
	1131	+ // stale standalone mapping before opening its replacement.
	1132	+ transport.closeStandaloneSSEStream();
	1133	+ const replay = await this.replayUndeliveredOnStandaloneGet(request);
	1134	+ if (replay) return replay;
	1135	+ }
	1136	+ }
1027	1137	const toolCallIds = legacyToolCallRequestIds(parsedBody);
1028	1138	const protocolVersion =
1029	1139	request.headers.get("mcp-protocol-version") ?? DEFAULT_NEGOTIATED_PROTOCOL_VERSION;
1030	1140	const needsLegacyPrime =
1031	1141	toolCallIds.length > 0 && protocolVersion < LEGACY_PRIMING_PROTOCOL_VERSION;
1032	1142	if (!needsLegacyPrime) {
1033		− return rotateSseResponse(await transport.handleRequest(request), {
1034		− onOpen: () => {
1035		− this.activeLegacyStreamCount += 1;
1036		− },
1037		− onClose: () => {
1038		− this.activeLegacyStreamCount = Math.max(0, this.activeLegacyStreamCount - 1);
1039		− },
1040		− });
	1143	+ const response = await transport.handleRequest(request);
	1144	+ const streamId = toolCallIds[0] ? this.requestStreamId(transport, toolCallIds[0]) : null;
	1145	+ if (streamId) await this.eventStore.markStreamUndelivered(streamId);
	1146	+ return this.trackedLegacyResponse(response);
1041	1147	}
1042	1148
1043	1149	const originalOnMessage = transport.onmessage;
@@ -1057,17 +1163,14 @@ export abstract class McpAgentSessionDOBase< · 13 unmodified lines
1057	1163	const eventId = streamId
1058	1164	? await this.eventStore.storeEvent(streamId, LEGACY_PRIMING_MESSAGE)
1059	1165	: null;
1060		− for (const item of queued) originalOnMessage?.(item.message, item.extra);
1061		−
1062		− return rotateSseResponse(response, {
	1166	+ if (streamId) await this.eventStore.markStreamUndelivered(streamId);
	1167	+ const rotated = this.trackedLegacyResponse(response, {
1063	1168	...(eventId ? { initialFrame: legacyPrimingFrame(eventId) } : {}),
1064		− onOpen: () => {
1065		− this.activeLegacyStreamCount += 1;
1066		− },
1067		− onClose: () => {
1068		− this.activeLegacyStreamCount = Math.max(0, this.activeLegacyStreamCount - 1);
1069		− },
1070	1169	});
	1170	+ // ReadableStream.start enqueues the priming frame synchronously while
	1171	+ // building `rotated`; only then may the server see the tools/call.
	1172	+ for (const item of queued) originalOnMessage?.(item.message, item.extra);
	1173	+ return rotated;
1071	1174	});
1072	1175	}
1073	1176
@@ -1444,7 +1547,7 @@ export abstract class McpAgentSessionDOBase< · 370 unmodified lines
1444	1547	return;
1445	1548	}
1446	1549
1447		− await this.disposeIdleRuntime({ idleMs, pausedExecutionCount });
	1550	+ await this.disposeIdleRuntime({ idleMs, lastActivityMs, pausedExecutionCount });
1448	1551	}
1449	1552
1450	1553	private validateApprovalIdentity(

packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts+3−5

@@ -425,11 +425,9 @@ describe("McpAgentSessionDOBase cross-session model resume", () => { · 424 unmodified lines
425	425	response: ResumeResponse,
426	426	) =>
427	427	Effect.promise(async () => {
428		− return mcpSessionStub(sessionNamespace, owner.sessionId).resumeExecutionForModel(
429		− executionId,
430		− identity,
431		− response,
432		− );
	428	+ const ownerSession = mcpSessionStub(sessionNamespace, owner.sessionId);
	429	+ if (!ownerSession) return { status: "execution_expired" as const, ttlMs: 0 };
	430	+ return ownerSession.resumeExecutionForModel(executionId, identity, response);
433	431	}),
434	432	);
435	433

packages/hosts/cloudflare/src/mcp/do-event-store.test.ts+32−0

@@ -113,6 +113,38 @@ describe("DurableObjectMcpEventStore", () => { · 112 unmodified lines
113	113	expect(replayed).toEqual([second]);
114	114	});
115	115
	116	+ it("replays a marked POST stream on standalone recovery and clears it after acknowledgement", async () => {
	117	+ const { storage, store } = makeStore();
	118	+ await store.markStreamUndelivered("post-stream");
	119	+ const prime = await store.storeEvent("post-stream", {
	120	+ jsonrpc: "2.0",
	121	+ method: "notifications/message",
	122	+ params: { level: "debug", data: "prime" },
	123	+ });
	124	+ const result = await store.storeEvent("post-stream", {
	125	+ jsonrpc: "2.0",
	126	+ id: "call-1",
	127	+ result: { content: [{ type: "text", text: "completed" }] },
	128	+ });
	129	+ const replayed: string[] = [];
	130	+
	131	+ const streamIds = await store.replayUndeliveredStreams({
	132	+ send: (eventId) => {
	133	+ replayed.push(eventId);
	134	+ return Promise.resolve();
	135	+ },
	136	+ });
	137	+
	138	+ expect(streamIds).toEqual(["post-stream"]);
	139	+ expect(replayed).toEqual([prime, result]);
	140	+
	141	+ await store.acknowledgeUndeliveredStreams(streamIds);
	142	+ await expect(
	143	+ store.replayUndeliveredStreams({ send: () => Promise.resolve() }),
	144	+ ).resolves.toEqual([]);
	145	+ expect(eventKeys(storage)).toEqual([]);
	146	+ });
	147	+
116	148	it("evicts oldest events at the byte cap but never the newest", async () => {
117	149	const { storage, store } = makeStore();
118	150	const bigMessage = (marker: string) => ({

packages/hosts/cloudflare/src/mcp/do-event-store.ts+79−0

@@ -1,6 +1,7 @@
1	1	import type { EventId, EventStore, JSONRPCMessage, StreamId } from "@modelcontextprotocol/server";
2	2
3	3	const EVENT_KEY_PREFIX = "executor:mcp:v2:event:";
	4	+const UNDELIVERED_STREAM_KEY_PREFIX = "executor:mcp:v2:undelivered-stream:";
4	5	const SEQUENCE_WIDTH = 16;
5	6	const REPLAY_LIMIT = 1_000;
6	7	const DELETE_CHUNK_SIZE = 128;
@@ -26,6 +27,9 @@ type StoredEntry = { · 19 unmodified lines
26	27
27	28	const eventPrefix = (streamId: StreamId): string => `${EVENT_KEY_PREFIX}${streamId}:`;
28	29
	30	+const undeliveredStreamKey = (streamId: StreamId): string =>
	31	+ `${UNDELIVERED_STREAM_KEY_PREFIX}${streamId}`;
	32	+
29	33	const eventIdFromKey = (key: string): EventId => key.slice(EVENT_KEY_PREFIX.length);
30	34
31	35	const streamIdFromEventId = (eventId: EventId): StreamId \| undefined => {
@@ -165,6 +169,81 @@ export class DurableObjectMcpEventStore implements EventStore { · 133 unmodified lines
165	169	return Promise.resolve(streamIdFromEventId(eventId));
166	170	}
167	171
	172	+ /**
	173	+ * Mark a tool-call stream as requiring at-least-once delivery confirmation.
	174	+ *
	175	+ * Direct workerd streaming cannot prove that a completed POST body reached
	176	+ * the remote client, so the marker remains until a later standalone GET
	177	+ * drains the stream and its response body completes.
	178	+ */
	179	+ async markStreamUndelivered(streamId: StreamId): Promise<void> {
	180	+ // oxlint-disable-next-line executor/no-try-catch-or-throw -- storage boundary: a marker failure cannot block the live tool request
	181	+ try {
	182	+ await this.storage.put(undeliveredStreamKey(streamId), true);
	183	+ } catch {
	184	+ logStoreWarning("mcp_event_store_put_failed", {
	185	+ operation: "mark_undelivered",
	186	+ streamId,
	187	+ });
	188	+ }
	189	+ }
	190	+
	191	+ /** Replay every marked POST stream onto a standalone recovery response. */
	192	+ async replayUndeliveredStreams({
	193	+ send,
	194	+ }: {
	195	+ readonly send: (eventId: EventId, message: JSONRPCMessage) => Promise<void>;
	196	+ }): Promise<readonly StreamId[]> {
	197	+ const replayedStreamIds: StreamId[] = [];
	198	+ // oxlint-disable-next-line executor/no-try-catch-or-throw -- storage/replay boundary: recovery is best-effort and leaves markers intact for a later GET
	199	+ try {
	200	+ const markers = await this.storage.list<boolean>({
	201	+ prefix: UNDELIVERED_STREAM_KEY_PREFIX,
	202	+ limit: REPLAY_LIMIT,
	203	+ });
	204	+ for (const key of markers.keys()) {
	205	+ const streamId = key.slice(UNDELIVERED_STREAM_KEY_PREFIX.length);
	206	+ let replayed = false;
	207	+ const rows = await this.storage.list<JSONRPCMessage>({
	208	+ prefix: eventPrefix(streamId),
	209	+ limit: REPLAY_LIMIT,
	210	+ });
	211	+ for (const [eventKey, message] of rows) {
	212	+ await send(eventIdFromKey(eventKey), message);
	213	+ replayed = true;
	214	+ }
	215	+ if (replayed) replayedStreamIds.push(streamId);
	216	+ }
	217	+ } catch {
	218	+ logStoreWarning("mcp_event_store_list_failed", {
	219	+ operation: "replay_undelivered",
	220	+ });
	221	+ }
	222	+ return replayedStreamIds;
	223	+ }
	224	+
	225	+ /** Clear successfully drained recovery streams and their delivery markers. */
	226	+ async acknowledgeUndeliveredStreams(streamIds: readonly StreamId[]): Promise<void> {
	227	+ // oxlint-disable-next-line executor/no-try-catch-or-throw -- storage boundary: acknowledgement cleanup is best-effort and duplicate replay is safe
	228	+ try {
	229	+ for (const streamId of streamIds) {
	230	+ const rows = await this.storage.list<JSONRPCMessage>({
	231	+ prefix: eventPrefix(streamId),
	232	+ limit: REPLAY_LIMIT,
	233	+ });
	234	+ const keys = [undeliveredStreamKey(streamId), ...rows.keys()];
	235	+ for (let index = 0; index < keys.length; index += DELETE_CHUNK_SIZE) {
	236	+ await this.storage.delete(keys.slice(index, index + DELETE_CHUNK_SIZE));
	237	+ }
	238	+ }
	239	+ } catch {
	240	+ logStoreWarning("mcp_event_store_delete_failed", {
	241	+ operation: "acknowledge_undelivered",
	242	+ streamCount: streamIds.length,
	243	+ });
	244	+ }
	245	+ }
	246	+
168	247	/** Replay persisted events after the supplied ID, in storage-key order. */
169	248	async replayEventsAfter(
170	249	lastEventId: EventId,

packages/hosts/cloudflare/src/mcp/modern-request-router.test.ts+36−0

@@ -120,6 +120,8 @@ class MemorySessions implements McpModernSessionNamespace<string> { · 119 unmodified lines
120	120	readonly forwarded: ForwardedRequest[] = [];
121	121	uniqueIds = 0;
122	122
	123	+ constructor(private readonly rejectStringIds = false) {}
	124	+
123	125	newUniqueId(): string {
124	126	this.uniqueIds += 1;
125	127	return `unique-${this.uniqueIds}`;
@@ -130,6 +132,10 @@ class MemorySessions implements McpModernSessionNamespace<string> { · 4 unmodified lines
130	132	}
131	133
132	134	idFromString(id: string): string {
	135	+ if (this.rejectStringIds) {
	136	+ // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test boundary: model Cloudflare rejecting a foreign/checksum-invalid owner id
	137	+ throw new Error("invalid Durable Object id");
	138	+ }
133	139	return `id:${id}`;
134	140	}
135	141
@@ -352,6 +358,36 @@ describe("modern Cloudflare MCP worker routing", () => { · 216 unmodified lines
352	358	expect(sessions.forwarded.map(({ id }) => id)).toEqual(["id:legacy-session"]);
353	359	});
354	360
	361	+ it("falls back to the worker when a persisted modern owner id is invalid", async () => {
	362	+ const executionId = "exec-invalid-owner";
	363	+ const directory = new MemoryDirectory();
	364	+ directory.records.set(executionId, {
	365	+ executionId,
	366	+ owner: modernMcpExecutionOwnerRoute("foreign-do-id"),
	367	+ accountId: principal.accountId,
	368	+ organizationId: principal.organizationId,
	369	+ expiresAt: new Date(Date.now() + 60_000).toISOString(),
	370	+ ttlMs: 60_000,
	371	+ });
	372	+ const sessions = new MemorySessions(true);
	373	+ const builds = { count: 0 };
	374	+
	375	+ const response = await dispatch({
	376	+ body: modernBody({
	377	+ method: "tools/call",
	378	+ name: "resume",
	379	+ arguments: { executionId, action: "accept" },
	380	+ }),
	381	+ sessions,
	382	+ directory,
	383	+ builder: makeBuilder(builds),
	384	+ });
	385	+
	386	+ expect(await response.json()).toMatchObject({ error: { code: -32602 } });
	387	+ expect(builds.count).toBe(1);
	388	+ expect(sessions.forwarded).toEqual([]);
	389	+ });
	390	+
355	391	it("rejects tampered and expired continuation state without touching a DO", async () => {
356	392	const valid = await mintRequestState("exec-invalid");
357	393	const middle = Math.floor(valid.length / 2);

packages/hosts/cloudflare/src/mcp/modern-request-router.ts+8−2

@@ -134,7 +134,10 @@ const toModernSessionStub = (stub: unknown): McpModernSessionStub => · 133 unmodified lines
134	134	const stubForOwner = <Id>(
135	135	sessions: McpModernSessionNamespace<Id>,
136	136	owner: { readonly sessionId: string },
137		−): McpModernSessionStub => toModernSessionStub(mcpSessionStubForOwner(sessions, owner));
	137	+): McpModernSessionStub \| null => {
	138	+ const stub = mcpSessionStubForOwner(sessions, owner);
	139	+ return stub ? toModernSessionStub(stub) : null;
	140	+};
138	141
139	142	const freshStub = <Id>(sessions: McpModernSessionNamespace<Id>): McpModernSessionStub =>
140	143	toModernSessionStub(sessions.get(sessions.newUniqueId()));
@@ -240,7 +243,10 @@ export const makeMcpModernRequestRouter = (): McpModernRequestRouter => { · 99 unmodified lines
240	243	jsonRpcErrorBody(403, -32003, "MCP execution does not belong to the current bearer"),
241	244	);
242	245	}
243		− return withModernMcpCors(await serveDo(stubForOwner(input.sessions, owner.owner), input));
	246	+ const ownerStub = stubForOwner(input.sessions, owner.owner);
	247	+ return withModernMcpCors(
	248	+ ownerStub ? await serveDo(ownerStub, input) : await serveWorker(input),
	249	+ );
244	250	},
245	251	close: () =>
246	252	Promise.all(Array.from(handlers.values(), (handler) => handler.close())).then(

packages/hosts/cloudflare/src/mcp/session-stub.test.ts+30−0

@@ -0,0 +1,30 @@
	1	+import { describe, expect, it, vi } from "@effect/vitest";
	2	+
	3	+import { mcpSessionStub } from "./session-stub";
	4	+
	5	+describe("mcpSessionStub", () => {
	6	+ it("returns null when Cloudflare rejects a client-supplied Durable Object id", () => {
	7	+ const get = vi.fn();
	8	+ const namespace = {
	9	+ idFromString: (_id: string): string => {
	10	+ // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test boundary: model Cloudflare's throwing namespace checksum parser
	11	+ throw new Error("Durable Object ID is not valid for this namespace");
	12	+ },
	13	+ get,
	14	+ };
	15	+
	16	+ expect(mcpSessionStub(namespace, "0".repeat(64))).toBeNull();
	17	+ expect(get).not.toHaveBeenCalled();
	18	+ });
	19	+
	20	+ it("resolves a namespace-validated id to its generated RPC stub", () => {
	21	+ const stub = { fetch: vi.fn() };
	22	+ const namespace = {
	23	+ idFromString: (id: string): string => `parsed:${id}`,
	24	+ get: vi.fn(() => stub),
	25	+ };
	26	+
	27	+ expect(mcpSessionStub(namespace, "issued-id")).toBe(stub);
	28	+ expect(namespace.get).toHaveBeenCalledWith("parsed:issued-id");
	29	+ });
	30	+});

packages/hosts/cloudflare/src/mcp/session-stub.ts+15−7

@@ -49,9 +49,19 @@ export interface McpSessionStub { · 48 unmodified lines
49	49	export const mcpSessionStub = <Id>(
50	50	namespace: McpSessionNamespace<Id>,
51	51	sessionId: string,
52		−): McpSessionStub =>
53		− // oxlint-disable-next-line executor/no-double-cast -- boundary: Workers types expose only DurableObjectStub, but fetch and RPC methods are generated from the bound DO class.
54		− namespace.get(namespace.idFromString(sessionId)) as unknown as McpSessionStub;
	52	+): McpSessionStub \| null => {
	53	+ let id: Id;
	54	+ // oxlint-disable-next-line executor/no-try-catch-or-throw -- platform boundary: Cloudflare validates the namespace checksum only through throwing idFromString
	55	+ try {
	56	+ id = namespace.idFromString(sessionId);
	57	+ } catch {
	58	+ return null;
	59	+ }
	60	+ return (
	61	+ // oxlint-disable-next-line executor/no-double-cast -- boundary: Workers types expose only DurableObjectStub, but fetch and RPC methods are generated from the bound DO class.
	62	+ namespace.get(id) as unknown as McpSessionStub
	63	+ );
	64	+};
55	65
56	66	/** Allocate one unique session DO and return its client-visible ID and stub. */
57	67	export const createMcpSessionStub = <Id>(
@@ -69,9 +79,7 @@ export const createMcpSessionStub = <Id>( · 11 unmodified lines
69	79	export const mcpSessionStubForOwner = <Id>(
70	80	namespace: McpOwnerSessionNamespace<Id>,
71	81	owner: McpExecutionOwnerRoute,
72		−): McpSessionStub => {
	82	+): McpSessionStub \| null => {
73	83	const modernId = modernMcpDurableObjectId(owner);
74		− const id = namespace.idFromString(modernId ?? owner.sessionId);
75		− // oxlint-disable-next-line executor/no-double-cast -- boundary: Workers generates this RPC surface from the bound DO class.
76		− return namespace.get(id) as unknown as McpSessionStub;
	84	+ return mcpSessionStub(namespace, modernId ?? owner.sessionId);
77	85	};