// The same-id JSON-RPC request queue used by the MCP worker transport.// Pure logic with no Cloudflare/agents dependency, in its own module so it// is unit-testable outside workerd.import { Effect, Exit, Match, Option, Predicate } from "effect"; type JsonRpcLike = { readonly id?: unknown; readonly method?: unknown; readonly params?: unknown;}; type ToolCallParamsLike = { readonly name?: unknown; readonly arguments?: unknown;}; type ResumeArgumentsLike = { readonly executionId?: unknown;}; const jsonRpcRequestIdKey = (id: unknown): string | null => Match.value(id).pipe( Match.whenOr( Match.string, Match.number, Match.boolean, (value) => `${typeof value}:${String(value)}`, ), Match.option, Option.getOrNull, ); const jsonRpcRequestQueueKey = (message: JsonRpcLike): string | null => { const idKey = jsonRpcRequestIdKey(message.id); if (!idKey) return null; if (message.method !== "tools/call") return idKey; if (!message.params || typeof message.params !== "object") return idKey; const params = message.params as ToolCallParamsLike; if (params.name !== "resume") return idKey; if (!params.arguments || typeof params.arguments !== "object") return idKey; const args = params.arguments as ResumeArgumentsLike; if (typeof args.executionId !== "string" || args.executionId.length === 0) return idKey; return `${idKey}:tools/call:resume:${args.executionId}`;}; const extractJsonRpcRequestQueueKeys = async (request: Request): Promise<ReadonlyArray<string>> => { if (request.method !== "POST") return []; const contentType = request.headers.get("content-type") ?? ""; if (!contentType.includes("application/json")) return []; const parsed = await Effect.runPromiseExit(Effect.tryPromise(() => request.clone().json())); if (Exit.isFailure(parsed)) { return []; } const messages = Array.isArray(parsed.value) ? parsed.value : [parsed.value]; return messages.flatMap((message) => { if (!message || typeof message !== "object") return []; const rpc = message as JsonRpcLike; if (typeof rpc.method !== "string") return []; const key = jsonRpcRequestQueueKey(rpc); return key ? [key] : []; });}; // Hard ceiling on how long a same-id JSON-RPC request will wait for an// earlier in-flight one to finish. Stays well under the 180s upstream// client timeout that Claude / Cowork enforce, so a poisoned queue slot// can't block the next request long enough for the client to give up.// If a previous request hasn't released within the budget, we proceed// anyway — at worst the MCP SDK rejects the second reply for a duplicate// id, which is recoverable; a perma-stuck queue is not.export const PREVIOUS_REQUEST_TIMEOUT_MS = 60_000; export class JsonRpcRequestIdQueue { private readonly inFlight = new Map<string, Promise<void>>(); private readonly previousTimeoutMs: number; constructor(options: { readonly previousTimeoutMs?: number } = {}) { this.previousTimeoutMs = options.previousTimeoutMs ?? PREVIOUS_REQUEST_TIMEOUT_MS; } async run<A>(request: Request, run: () => Promise<A>): Promise<A> { const ids = [...new Set(await extractJsonRpcRequestQueueKeys(request))]; if (ids.length === 0) return await run(); const previous = ids.map((id) => this.inFlight.get(id)).filter(Predicate.isNotUndefined); let release!: () => void; const current = new Promise<void>((resolve) => { release = resolve; }); for (const id of ids) { this.inFlight.set(id, current); } // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: promise queue must release in-flight ids after callback completion try { if (previous.length > 0) { const settled = Promise.all( previous.map((p) => Effect.runPromise(Effect.ignore(Effect.tryPromise(() => p)))), ); const timeout = new Promise<"timeout">((resolve) => setTimeout(() => resolve("timeout"), this.previousTimeoutMs), ); const outcome = await Promise.race([settled.then(() => "settled" as const), timeout]); if (outcome === "timeout") { console.warn( `[mcp-worker-transport] previous in-flight request for ids=${JSON.stringify(ids)} did not release within ${this.previousTimeoutMs}ms; proceeding anyway`, ); } } return await run(); } finally { for (const id of ids) { if (this.inFlight.get(id) === current) { this.inFlight.delete(id); } } release(); } }}
// The same-id JSON-RPC request queue used by the MCP worker transport.
// Pure logic with no Cloudflare/agents dependency, in its own module so it
// is unit-testable outside workerd.
import { Effect, Exit, Match, Option, Predicate } from "effect";
type JsonRpcLike = {
readonly id?: unknown;
readonly method?: unknown;
readonly params?: unknown;
};
type ToolCallParamsLike = {
readonly name?: unknown;
readonly arguments?: unknown;
};
type ResumeArgumentsLike = {
readonly executionId?: unknown;
};
const jsonRpcRequestIdKey = (id: unknown): string | null =>
Match.value(id).pipe(
Match.whenOr(
Match.string,
Match.number,
Match.boolean,
(value) => `${typeof value}:${String(value)}`,
),
Match.option,
Option.getOrNull,
);
const jsonRpcRequestQueueKey = (message: JsonRpcLike): string | null => {
const idKey = jsonRpcRequestIdKey(message.id);
if (!idKey) return null;
if (message.method !== "tools/call") return idKey;
if (!message.params || typeof message.params !== "object") return idKey;
const params = message.params as ToolCallParamsLike;
if (params.name !== "resume") return idKey;
if (!params.arguments || typeof params.arguments !== "object") return idKey;
const args = params.arguments as ResumeArgumentsLike;
if (typeof args.executionId !== "string" || args.executionId.length === 0) return idKey;
return `${idKey}:tools/call:resume:${args.executionId}`;
};
const extractJsonRpcRequestQueueKeys = async (request: Request): Promise<ReadonlyArray<string>> => {
if (request.method !== "POST") return [];
const contentType = request.headers.get("content-type") ?? "";
if (!contentType.includes("application/json")) return [];
const parsed = await Effect.runPromiseExit(Effect.tryPromise(() => request.clone().json()));
if (Exit.isFailure(parsed)) {
return [];
}
const messages = Array.isArray(parsed.value) ? parsed.value : [parsed.value];
return messages.flatMap((message) => {
if (!message || typeof message !== "object") return [];
const rpc = message as JsonRpcLike;
if (typeof rpc.method !== "string") return [];
const key = jsonRpcRequestQueueKey(rpc);
return key ? [key] : [];
});
};
// Hard ceiling on how long a same-id JSON-RPC request will wait for an
// earlier in-flight one to finish. Stays well under the 180s upstream
// client timeout that Claude / Cowork enforce, so a poisoned queue slot
// can't block the next request long enough for the client to give up.
// If a previous request hasn't released within the budget, we proceed
// anyway — at worst the MCP SDK rejects the second reply for a duplicate
// id, which is recoverable; a perma-stuck queue is not.
export const PREVIOUS_REQUEST_TIMEOUT_MS = 60_000;
export class JsonRpcRequestIdQueue {
private readonly inFlight = new Map<string, Promise<void>>();
private readonly previousTimeoutMs: number;
constructor(options: { readonly previousTimeoutMs?: number } = {}) {
this.previousTimeoutMs = options.previousTimeoutMs ?? PREVIOUS_REQUEST_TIMEOUT_MS;
}
async run<A>(request: Request, run: () => Promise<A>): Promise<A> {
const ids = [...new Set(await extractJsonRpcRequestQueueKeys(request))];
if (ids.length === 0) return await run();
const previous = ids.map((id) => this.inFlight.get(id)).filter(Predicate.isNotUndefined);
let release!: () => void;
const current = new Promise<void>((resolve) => {
release = resolve;
});
for (const id of ids) {
this.inFlight.set(id, current);
}
// oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: promise queue must release in-flight ids after callback completion
try {
if (previous.length > 0) {
const settled = Promise.all(
previous.map((p) => Effect.runPromise(Effect.ignore(Effect.tryPromise(() => p)))),
);
const timeout = new Promise<"timeout">((resolve) =>
setTimeout(() => resolve("timeout"), this.previousTimeoutMs),
);
const outcome = await Promise.race([settled.then(() => "settled" as const), timeout]);
if (outcome === "timeout") {
console.warn(
`[mcp-worker-transport] previous in-flight request for ids=${JSON.stringify(ids)} did not release within ${this.previousTimeoutMs}ms; proceeding anyway`,
);
}
}
return await run();
} finally {
for (const id of ids) {
if (this.inFlight.get(id) === current) {
this.inFlight.delete(id);
}
}
release();
}
}
}