diff --git a/.changeset/mcp-session-idle-runtime-reclaim.md b/.changeset/mcp-session-idle-runtime-reclaim.md new file mode 100644 index 000000000..ed48d8482 --- /dev/null +++ b/.changeset/mcp-session-idle-runtime-reclaim.md @@ -0,0 +1,15 @@ +--- +"executor": patch +--- + +**Idle MCP session runtimes are actually reclaimed** + +The MCP session Durable Object has an idle timeout that disposes a session's execution runtime — the execution engine and its executor closure, the built tool catalog, and a live database handle — once the session has gone quiet. That timeout never ran. + +The session arms an idle alarm on every request. The agents framework independently recomputes the Durable Object alarm from its own schedule table and keep-alive refcount, and when it finds neither it does not leave the alarm alone: it deletes it. It releases the last keep-alive reference at the end of every ordinary tool call, from a `waitUntil` that runs just after the response goes out — so the idle alarm the session had armed moments earlier was erased, and a session that had just served a request was left with no alarm at all. Its runtime then stayed resident until the platform evicted the whole object. + +Durable Objects are colocated many-to-one onto an isolate with a single heap, so runtimes that are never reclaimed accumulate there. When the heap is exhausted the allocation that fails is whichever comes next, anywhere in the isolate — which is why the failure tended to surface from storage rather than from the runtimes that had consumed the memory. + +The idle deadline belongs to the session, not to the framework's scheduler, so it is now re-asserted after the framework has arranged whatever it needs — and only while a runtime is actually resident, since once there is nothing left to reclaim the framework's answer is correct. + +Disposal also now emits a span carrying a per-isolate resident-runtime gauge, alongside the same gauge on runtime build, so the reclaim can be confirmed in production rather than inferred. diff --git a/e2e/cloud/mcp-session-idle-runtime-disposal.test.ts b/e2e/cloud/mcp-session-idle-runtime-disposal.test.ts new file mode 100644 index 000000000..84188f502 --- /dev/null +++ b/e2e/cloud/mcp-session-idle-runtime-disposal.test.ts @@ -0,0 +1,252 @@ +// Cloud: an idle MCP session gives its execution runtime back to the isolate +// even while the client is still holding a stream open. +// +// The defect this pins: the idle timeout never actually ran. The session armed +// an idle alarm on every request, but the agents framework recomputes the +// Durable Object alarm from its own schedule table and keep-alive refcount and +// DELETES it when it finds neither — which it does at the end of every ordinary +// tool call, from a `waitUntil` just after the response goes out. So a session +// that had just served a request was left with no alarm at all, its idle +// deadline silently dropped, and its execution runtime resident until the +// platform evicted the whole object. Durable Objects share one isolate heap, so +// enough never-reclaimed runtimes exhaust it and the next allocation anywhere +// in that isolate fails. +// +// The contract asserted here, in three parts: +// +// 1. an idle session's runtime IS released, and the exported span says so. +// Before the fix this span never appeared at all, for any session; +// 2. the release reports the isolate's resident-runtime gauge, which is what +// makes the mechanism confirmable in production rather than inferred; +// 3. the very next call on the SAME session works and returns the right +// answer, restoring the runtime underneath the client transparently. +// +// A client stream is deliberately held open across the idle window, because +// that is the shape real clients have. Note what it does NOT do: it does not +// keep the runtime alive. The session Durable Object holds no connection for a +// standalone GET stream (`ctx.getWebSockets()` is empty), so the stream is not +// visible to the idle policy at all, and reclaiming the runtime ends it — the +// client reconnects and replays, which is the behaviour `mcp-sse-replay` pins. +// +// The out-of-memory failure itself is not reachable from a black-box test — +// filling a shared isolate heap on demand is not something a client can ask +// for — so reclamation, not the allocation failure, is the testable contract. +import { randomUUID } from "node:crypto"; + +import { expect } from "@effect/vitest"; +import { Effect, Schedule } from "effect"; + +import { scenario } from "../src/scenario"; +import { Mcp, Target, Telemetry } from "../src/services"; +import type { Identity } from "../src/target"; +import { configuredMcpSessionTimeoutMs } from "../setup/mcp-session-timeouts"; + +const PROTOCOL_VERSION = "2025-03-26"; +const JSON_AND_SSE = "application/json, text/event-stream"; + +/** Past the idle window with room for the alarm to actually fire. */ +const IDLE_GAP_MS = configuredMcpSessionTimeoutMs() + 4_000; + +const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label; + +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 => { + const initialized = await postJson(mcpUrl, bearer, { + jsonrpc: "2.0" as const, + id: "initialize", + method: "initialize", + params: { + protocolVersion: PROTOCOL_VERSION, + capabilities: {}, + clientInfo: { name: "executor-e2e-idle-runtime-disposal", version: "0.0.1" }, + }, + }); + const sessionId = initialized.headers.get("mcp-session-id"); + await initialized.text(); + expect(initialized.status, "initialize opens a session").toBe(200); + if (!sessionId) { + // oxlint-disable-next-line executor/no-error-constructor -- boundary: e2e setup precondition. + throw new Error("openSession: no mcp-session-id header"); + } + const notification = await postJson( + mcpUrl, + bearer, + { jsonrpc: "2.0" as const, method: "notifications/initialized" }, + sessionId, + ); + await notification.text(); + expect(notification.status, "the client completes the handshake").toBe(202); + return sessionId; +}; + +/** + * The standalone GET stream a real client leaves open for the whole + * conversation. Held open (never aborted) across the idle window — that is the + * exact condition the old policy read as "busy". + */ +type OpenStream = { + readonly closed: () => boolean; + readonly abort: () => void; +}; + +const openGetStream = async ( + mcpUrl: string, + bearer: string, + sessionId: string, +): Promise => { + const abortController = new AbortController(); + const response = await fetch(mcpUrl, { + method: "GET", + headers: { + accept: "text/event-stream", + authorization: `Bearer ${bearer}`, + "mcp-protocol-version": PROTOCOL_VERSION, + "mcp-session-id": sessionId, + }, + signal: abortController.signal, + }); + expect(response.status, "the standalone GET stream opens").toBe(200); + + let closed = false; + const reader = response.body?.getReader(); + // Drain in the background. The stream ending — for any reason — flips + // `closed`, which is what part (2) of the contract inspects. + void (async () => { + if (!reader) { + closed = true; + return; + } + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: reading an aborted stream throws; either way the stream is over. + try { + for (;;) { + const { done } = await reader.read(); + if (done) break; + } + } catch { + // fall through + } + closed = true; + })(); + + return { + closed: () => closed, + abort: () => abortController.abort("scenario complete"), + }; +}; + +const executeBody = (id: string, code: string) => ({ + jsonrpc: "2.0" as const, + id, + method: "tools/call", + params: { name: "execute", arguments: { code } }, +}); + +/** Run `execute` and return the response text once the call has fully settled. */ +const execute = async ( + mcpUrl: string, + bearer: string, + sessionId: string, + id: string, + code: string, +): Promise => { + const response = await postJson(mcpUrl, bearer, executeBody(id, code), sessionId); + const body = await response.text(); + expect(response.status, `execute ${id} is served`).toBe(200); + return body; +}; + +scenario( + "MCP session · an idle session disposes its runtime while the client stream stays open", + { timeout: 180_000 }, + Effect.gen(function* () { + const target = yield* Target; + const mcp = yield* Mcp; + const telemetry = yield* Telemetry; + + const identity = yield* target.newIdentity(); + const bearer = yield* mcp.mintBearer(emailOf(identity)); + + const sessionId = yield* Effect.promise(() => openSession(target.mcpUrl, bearer)); + + // The long-lived client stream, opened before any work and never closed. + const stream = yield* Effect.promise(() => openGetStream(target.mcpUrl, bearer, sessionId)); + + // ---- one call, then silence ------------------------------------------ + const marker = `before-${randomUUID().slice(0, 8)}`; + const first = yield* Effect.promise(() => + execute( + target.mcpUrl, + bearer, + sessionId, + "execute-before-idle", + `return ${JSON.stringify(marker)};`, + ), + ); + expect(first, "the session executes before going idle").toContain(marker); + + // ---- go idle, with the stream still connected ------------------------- + yield* Effect.sleep(`${IDLE_GAP_MS} millis`); + + // ---- 1. the runtime was released -------------------------------------- + // The alarm carries no client trace context, so this is found by operation + // and matched to this session by attribute rather than by trace id. + const disposals = yield* telemetry + .searchSpans({ operation: "mcp.session.idle_runtime_dispose" }) + .pipe( + Effect.map((spans) => + spans.filter((span) => (span.span.tags["mcp.session.id"] ?? "").includes(sessionId)), + ), + Effect.filterOrFail( + (spans) => spans.length > 0, + () => `no idle-runtime-disposal span exported for session ${sessionId}`, + ), + // The span is exported off the alarm's own flush, which is not on any + // response path — same polling grace `expectSpan` uses. + Effect.retry(Schedule.both(Schedule.spaced("500 millis"), Schedule.recurs(40))), + ); + + expect(disposals.length, "the idle session released its execution runtime").toBeGreaterThan(0); + + const disposal = disposals[0]!; + expect( + Number(disposal.span.tags["mcp.session.idle_ms"]), + "the runtime was released because the session went idle, not for some other reason", + ).toBeGreaterThanOrEqual(configuredMcpSessionTimeoutMs()); + expect( + disposal.span.tags["mcp.isolate.resident_runtimes"], + "the disposal records the isolate's resident-runtime gauge", + ).toBeDefined(); + + // ---- 3. the next call restores underneath the client ------------------ + const after = `after-${randomUUID().slice(0, 8)}`; + const second = yield* Effect.promise(() => + execute( + target.mcpUrl, + bearer, + sessionId, + "execute-after-idle", + `return ${JSON.stringify(after)};`, + ), + ); + expect( + second, + "the same session serves the next call with the right answer after restoring", + ).toContain(after); + + stream.abort(); + }), +); diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts index d7dfcb0e8..f0ea19469 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts @@ -34,6 +34,11 @@ import { pausedLeaseExtensionLog, runningLeaseExtensionLog, } from "./session-alarm-policy"; +import { + acquireResidentRuntime, + releaseResidentRuntime, + residencyAttributes, +} from "./session-runtime-residency"; export type IncomingTraceHeaders = IncomingPropagationHeaders; @@ -246,6 +251,10 @@ export abstract class McpAgentSessionDOBase< private dbHandle: TDbHandle | null = null; private sessionMeta: SessionMeta | null = null; private initialized = false; + /** Whether this instance is currently counted in the isolate's residency + * gauge. Tracked separately from `engine` because `closeRuntime` runs on + * paths where nothing was ever built, and it must not decrement then. */ + private countedAsResident = false; private onStartPromise: Promise | null = null; private lastActivityMs = 0; private approvalResponses = new Map(); @@ -442,6 +451,34 @@ export abstract class McpAgentSessionDOBase< ]); } + /** + * Keep this session's idle deadline armed across the SDK's own alarm + * bookkeeping. + * + * The agents SDK recomputes the Durable Object alarm from its schedule table + * and keep-alive refcount, and when it finds neither it does not leave the + * alarm alone — it DELETES it. It releases the last keep-alive ref at the end + * of every ordinary request, from a `waitUntil`, so the idle alarm that + * `markActivity` had just armed was being erased moments after the response + * went out. A session that had just served a tool call was therefore left + * with no alarm at all: the idle policy never ran again, and its runtime + * stayed resident until the platform evicted the whole object. + * + * The idle deadline belongs to this class, not to the SDK's scheduler, so it + * is re-asserted after the SDK has arranged whatever it needs. Only while a + * runtime is actually resident — once there is nothing left to reclaim the + * SDK's answer is right and re-arming would spin the alarm forever. + */ + protected async ensureIdleAlarmArmed(): Promise { + if (!this.hasResidentRuntime()) return; + const lastActivityMs = await this.loadLastActivity(); + if (lastActivityMs <= 0) return; + const idleDeadlineMs = lastActivityMs + this.sessionTimeoutMs(); + const armed = await this.ctx.storage.getAlarm(); + if (armed !== null && armed <= idleDeadlineMs) return; + await this.ctx.storage.setAlarm(Math.max(idleDeadlineMs, Date.now() + 1)); + } + private async loadLastActivity(): Promise { if (this.lastActivityMs > 0) return this.lastActivityMs; const stored = await this.ctx.storage.get(LAST_ACTIVITY_KEY); @@ -520,9 +557,20 @@ export abstract class McpAgentSessionDOBase< ); } + /** Whether anything is currently holding isolate memory for this session. */ + private hasResidentRuntime(): boolean { + return this.initialized || this.engine !== null || this.dbHandle !== null; + } + + /** + * Drop this session's execution runtime because it has gone idle, returning + * its memory to the isolate. Nothing durable is discarded, so the next + * request restores the session and the client sees only restore latency. + */ private async disposeIdleRuntime(input: { readonly idleMs: number; readonly pausedExecutionCount: number; + readonly activeStreamCount: number; }): Promise { console.info( JSON.stringify({ @@ -530,15 +578,34 @@ export abstract class McpAgentSessionDOBase< sessionId: this.sessionId, idleMs: input.idleMs, pausedExecutionCount: input.pausedExecutionCount, + activeStreamCount: input.activeStreamCount, }), ); - await Effect.runPromise(this.closeRuntime()); - await Effect.runPromise( - Effect.all([ - Effect.ignore(Effect.tryPromise(() => this.ctx.storage.deleteAlarm())), - Effect.ignore(Effect.tryPromise(() => this.ctx.storage.delete(LAST_ACTIVITY_KEY))), - ]), + const self = this; + const program = Effect.gen(function* () { + yield* self.closeRuntime(); + yield* Effect.all([ + Effect.ignore(Effect.tryPromise(() => self.ctx.storage.deleteAlarm())), + Effect.ignore(Effect.tryPromise(() => self.ctx.storage.delete(LAST_ACTIVITY_KEY))), + ]); + // Read the gauge AFTER the release, so the attribute reports what the + // isolate is actually holding now rather than what it held a moment ago. + yield* Effect.annotateCurrentSpan(residencyAttributes()); + }).pipe( + Effect.withSpan("mcp.session.idle_runtime_dispose", { + attributes: { + "mcp.session.id": self.sessionId, + "mcp.session.idle_ms": input.idleMs, + "mcp.session.paused_execution_count": input.pausedExecutionCount, + "mcp.session.active_stream_count": input.activeStreamCount, + }, + }), ); + // The alarm has no incoming trace context of its own, so this starts a new + // trace. It still has to be flushed explicitly — the alarm is not on any + // request's response path, and without the flush the span dies with the + // isolate and the mechanism stays unobservable in production. + await Effect.runPromise(this.withSpanFlush(this.withTelemetry(program))); } private resolveAndStoreSessionMeta(token: McpSessionInit) { @@ -749,6 +816,10 @@ export abstract class McpAgentSessionDOBase< yield* Effect.promise(() => Promise.resolve(dbHandle.end())).pipe(Effect.ignore); } self.initialized = false; + if (self.countedAsResident) { + self.countedAsResident = false; + releaseResidentRuntime(); + } }); } @@ -824,6 +895,14 @@ export abstract class McpAgentSessionDOBase< self.server = mcpServer; self.engine = engine; self.initialized = true; + if (!self.countedAsResident) { + self.countedAsResident = true; + acquireResidentRuntime(); + } + // The gauge on the way up. Paired with the same attributes on + // `mcp.session.idle_runtime_dispose`, this is what shows whether idle + // sessions are actually giving their runtimes back in production. + yield* Effect.annotateCurrentSpan(residencyAttributes()); // Last statement, and pure bookkeeping: the runtime above is already // installed and serving. Losing the timestamp/alarm write to a platform // reset must not undo any of it — the in-memory clock is already set and @@ -1117,7 +1196,7 @@ export abstract class McpAgentSessionDOBase< return; } - await this.disposeIdleRuntime({ idleMs, pausedExecutionCount }); + await this.disposeIdleRuntime({ idleMs, pausedExecutionCount, activeStreamCount }); } private validateApprovalIdentity( @@ -1494,3 +1573,36 @@ export abstract class McpAgentSessionDOBase< await Effect.runPromise(this.closeRuntime()); } } + +/** + * Install the idle-deadline repair described on + * {@link McpAgentSessionDOBase.ensureIdleAlarmArmed}. + * + * At runtime this is an ordinary method override — the agents SDK calls + * `this._scheduleNextAlarm()` and gets this one. It cannot be written in the + * class body because the SDK declares that member `private`, and TypeScript + * refuses to let a subclass redeclare a private base member at all. So it is + * installed on the prototype, which is the same thing to the runtime and the + * only form the compiler accepts. + */ +type BoundAsyncMethod = (this: object) => Promise; + +{ + // Read off the INHERITED prototype, so this captures the SDK's own + // implementation rather than the wrapper being installed just below. + const inherited: object = Object.getPrototypeOf(McpAgentSessionDOBase.prototype); + const scheduleNextAlarm = Reflect.get(inherited, "_scheduleNextAlarm") as + | BoundAsyncMethod + | undefined; + if (scheduleNextAlarm) { + Reflect.set( + McpAgentSessionDOBase.prototype, + "_scheduleNextAlarm", + async function (this: object): Promise { + await scheduleNextAlarm.call(this); + const ensureIdleAlarmArmed = Reflect.get(this, "ensureIdleAlarmArmed") as BoundAsyncMethod; + await ensureIdleAlarmArmed.call(this); + }, + ); + } +} diff --git a/packages/hosts/cloudflare/src/mcp/session-runtime-residency.ts b/packages/hosts/cloudflare/src/mcp/session-runtime-residency.ts new file mode 100644 index 000000000..e1b61d166 --- /dev/null +++ b/packages/hosts/cloudflare/src/mcp/session-runtime-residency.ts @@ -0,0 +1,96 @@ +/** + * How many MCP session execution runtimes are resident in THIS isolate. + * + * A single session Durable Object holds at most one runtime, so a per-instance + * flag would say nothing useful. The pressure that matters is per-ISOLATE: + * workerd colocates many Durable Objects onto one isolate with one heap, and a + * session runtime — the execution engine and its executor closure, the built + * tool catalog, and a live database handle — is the largest thing any of them + * holds. Several resident at once is the condition under which the isolate runs + * out of memory, and the allocation that fails is whichever comes next, which + * is why the symptom tends to name storage rather than the runtimes that + * actually consumed the heap. + * + * Note this counts SESSION runtimes only. The QuickJS WASM module is preloaded + * once per isolate and shared by every session on it, so it is deliberately not + * part of this gauge: disposal cannot release it and counting it would imply + * otherwise. + * + * Module scope is exactly isolate scope on Workers, so this counter measures + * the thing we want and resets naturally when the isolate is recycled. + */ +let residentRuntimeCount = 0; + +/** Peak residency seen in this isolate, so a gauge sampled per request still + * reveals a burst that had already receded by the time anything asked. */ +let peakResidentRuntimeCount = 0; + +export const acquireResidentRuntime = (): number => { + residentRuntimeCount += 1; + if (residentRuntimeCount > peakResidentRuntimeCount) { + peakResidentRuntimeCount = residentRuntimeCount; + } + return residentRuntimeCount; +}; + +export const releaseResidentRuntime = (): number => { + residentRuntimeCount = Math.max(0, residentRuntimeCount - 1); + return residentRuntimeCount; +}; + +export const currentResidentRuntimeCount = (): number => residentRuntimeCount; + +export const peakResidentRuntimeCountInIsolate = (): number => peakResidentRuntimeCount; + +/** Test-only: isolate-scoped module state outlives a single test case. */ +export const resetResidentRuntimeCountForTest = (): void => { + residentRuntimeCount = 0; + peakResidentRuntimeCount = 0; +}; + +type MemoryCapablePerformance = { + readonly memory?: { + readonly usedJSHeapSize?: unknown; + readonly totalJSHeapSize?: unknown; + readonly jsHeapSizeLimit?: unknown; + }; +}; + +const finiteNumber = (value: unknown): number | undefined => + typeof value === "number" && Number.isFinite(value) ? value : undefined; + +/** + * Isolate heap usage, when the runtime exposes it. + * + * workerd does not currently implement `performance.memory` (nor the async + * `measureUserAgentSpecificMemory`), so on production Workers this returns an + * empty object and the attributes are simply absent. It is feature-detected + * rather than omitted so that the day workerd does expose it, the gauge starts + * reporting with no further change — and so the local `workerd`/Node harnesses + * that DO expose a heap size record one today. + */ +export const isolateMemoryAttributes = (): Record => { + const perf = (globalThis as { readonly performance?: MemoryCapablePerformance }).performance; + const memory = perf?.memory; + if (!memory) return {}; + const used = finiteNumber(memory.usedJSHeapSize); + const total = finiteNumber(memory.totalJSHeapSize); + const limit = finiteNumber(memory.jsHeapSizeLimit); + return { + ...(used === undefined ? {} : { "mcp.isolate.heap_used_bytes": used }), + ...(total === undefined ? {} : { "mcp.isolate.heap_total_bytes": total }), + ...(limit === undefined ? {} : { "mcp.isolate.heap_limit_bytes": limit }), + }; +}; + +/** + * The residency gauge as span attributes. Attached to every runtime build and + * every idle disposal, so production can confirm the mechanism directly: + * residency should now fall back toward zero as sessions go idle instead of + * climbing with the number of connected-but-quiet clients. + */ +export const residencyAttributes = (): Record => ({ + "mcp.isolate.resident_runtimes": currentResidentRuntimeCount(), + "mcp.isolate.peak_resident_runtimes": peakResidentRuntimeCountInIsolate(), + ...isolateMemoryAttributes(), +});