Reclaim idle MCP session runtimes

.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.

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<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-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<OpenStream> => {
+  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<string> => {
+  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();
+  }),
+);