From 016420c360792175798e461926df8dd6cc60e4cb Mon Sep 17 00:00:00 2001 From: Davies Ayo Date: Wed, 19 Aug 2026 23:03:31 +1000 Subject: [PATCH 1/5] fix(host-mcp): evict idle MCP sessions instead of leaking them The in-process session store keyed transports, servers, owners, and engines by session id and only ever deleted an entry on `onsessionclosed`, which the SDK fires on `DELETE /mcp`. Nothing sends that DELETE: the client SDK's `transport.close()` aborts locally and puts nothing on the wire, a crashed client cannot send it, and `enableJsonResponse` leaves no stream whose teardown could stand in for it. Every initialize therefore pinned an McpServer, its tool registry, and an ExecutionEngine until the process exited. Measured against ghcr.io/usefulsoftwareco/executor-selfhost:1.5.42, 500 sessions opened without a DELETE grow RSS by 346 MiB (709 KiB each, linear, no plateau); the same 500 with a DELETE grow it by 13 MiB. Stamp each session on create and on every forwarded request, then sweep on a timer and dispose anything idle past the TTL. Eviction is what the streamable HTTP spec allows a server to do, and the store already renders an unknown id as the existing "not-found" (404 -32001), which is a client's cue to re-initialize. --- .changeset/mcp-session-idle-eviction.md | 5 + .../mcp/src/in-memory-session-store.test.ts | 100 ++++++++++++++++++ .../hosts/mcp/src/in-memory-session-store.ts | 74 ++++++++++++- 3 files changed, 178 insertions(+), 1 deletion(-) create mode 100644 .changeset/mcp-session-idle-eviction.md diff --git a/.changeset/mcp-session-idle-eviction.md b/.changeset/mcp-session-idle-eviction.md new file mode 100644 index 0000000000..e7c075d65b --- /dev/null +++ b/.changeset/mcp-session-idle-eviction.md @@ -0,0 +1,5 @@ +--- +"@executor-js/host-mcp": patch +--- + +Evict idle MCP sessions instead of holding them for the lifetime of the process. The in-process session store only released a session when the client sent `DELETE /mcp`, which the MCP client SDK's `transport.close()` never sends and a crashed client cannot send, so every `initialize` permanently retained an `McpServer`, its tool registry, and an `ExecutionEngine`. Sessions are now stamped on create and on each request, and a timer disposes anything idle past `sessionIdleTtlMs` (30 minutes by default). diff --git a/packages/hosts/mcp/src/in-memory-session-store.test.ts b/packages/hosts/mcp/src/in-memory-session-store.test.ts index 6c61e9bcab..3118de164c 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.test.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.test.ts @@ -112,6 +112,106 @@ describe("in-memory MCP session store", () => { expect(buildOptions?.requestStateSigningKey).toBeInstanceOf(Uint8Array); }); + it("evicts a session that goes idle past the TTL and keeps a busy one", async () => { + const { engine } = makeElicitingEngine(); + const sessions = makeInMemoryMcpSessionStore( + (_principal, options) => + buildMcpServer({ + engine, + ...options, + loadAppShellHtml: async () => "", + }).pipe(Effect.map((mcpServer) => ({ mcpServer, engine }))), + { sessionIdleTtlMs: 300 }, + ); + + const open = async (): Promise => { + const response = (await Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "idle-test", version: "1.0.0" }, + }, + }), + }), + principal: TEST_PRINCIPAL, + resource: defaultMcpResource, + sessionId: null, + method: "POST", + }), + )) as Response; + expect(response.status).toBe(200); + const sessionId = response.headers.get("mcp-session-id") ?? ""; + expect(sessionId).not.toBe(""); + return sessionId; + }; + + // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: always close the store + try { + const idle = await open(); + const busy = await open(); + expect(sessions.sessionCount()).toBe(2); + + // Neither is stale yet, so a sweep now must not touch them. + expect(await sessions.sweepIdleSessions()).toBe(0); + expect(sessions.sessionCount()).toBe(2); + + // Let both age past the idle window, then keep working on one of them: + // `forward` restamps that session and only that session. + await new Promise((resolve) => setTimeout(resolve, 500)); + await Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json, text/event-stream", + "mcp-session-id": busy, + }, + body: JSON.stringify({ jsonrpc: "2.0", id: 2, method: "tools/list" }), + }), + principal: TEST_PRINCIPAL, + resource: defaultMcpResource, + sessionId: busy, + method: "POST", + }), + ); + + // The idle session is now well past the window and the busy one was just + // restamped, so the sweep takes exactly one. + expect(await sessions.sweepIdleSessions()).toBe(1); + expect(sessions.sessionCount()).toBe(1); + + // The evicted id is gone; the store reports it the way the envelope 404s. + const afterEviction = await Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { "content-type": "application/json", "mcp-session-id": idle }, + body: JSON.stringify({ jsonrpc: "2.0", id: 3, method: "tools/list" }), + }), + principal: TEST_PRINCIPAL, + resource: defaultMcpResource, + sessionId: idle, + method: "POST", + }), + ); + expect(afterEviction).toBe("not-found"); + } finally { + await sessions.close(); + } + }); + it("serves a legacy client with live Apps capabilities, elicitation, and reuse", async () => { const { engine, resumedWith } = makeElicitingEngine(); const sessions = makeInMemoryMcpSessionStore((_principal, options) => diff --git a/packages/hosts/mcp/src/in-memory-session-store.ts b/packages/hosts/mcp/src/in-memory-session-store.ts index 64ca5f9d92..856185682b 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.ts @@ -54,6 +54,23 @@ import { mcpRequestStatePrincipal, type BrowserApprovalStore } from "./tool-serv // - "forbidden" (session owned by another bearer) -> envelope renders 403 -32003 // --------------------------------------------------------------------------- +// A streamable-HTTP session only leaves these maps when the client sends +// `DELETE /mcp`. Nothing else can free it: with `enableJsonResponse` there is no +// stream whose teardown signals the client is gone, and a client that crashes, +// is killed, or simply calls the MCP SDK's `transport.close()` (which aborts +// locally and sends nothing) never issues that DELETE. Without a sweep, one +// abandoned session pins its `McpServer`, its tool registry, and its +// `ExecutionEngine` for the lifetime of the process. +// +// So the store treats a session as abandoned once it has gone `idleTtlMs` +// without a request and disposes it. That is what the streamable-HTTP spec +// allows a server to do: a request carrying an evicted id gets the store's +// existing "not-found" (404, -32001), which is the client's cue to re-initialize. +/** Idle window after which an untouched session is evicted. */ +const DEFAULT_SESSION_IDLE_TTL_MS = 30 * 60 * 1000; +/** Floor on the sweep interval, so a small TTL cannot spin the timer. */ +const MIN_SWEEP_INTERVAL_MS = 30 * 1000; + /** Engine construction failed for a principal. The store surfaces it as a 500. */ export class McpEngineBuildError extends Data.TaggedError("McpEngineBuildError")<{ readonly cause: unknown; @@ -116,6 +133,11 @@ export interface InMemoryMcpSessionStore { ) => Promise; /** Number of live initialized sessions currently owned by this store. */ readonly sessionCount: () => number; + /** + * Dispose every session idle past the store's TTL and return how many went. + * Runs on a timer; exposed so a host (or a test) can drive it directly. + */ + readonly sweepIdleSessions: (now?: number) => Promise; /** Dispose every live session — wire into the host's shutdown (not a seam). */ readonly close: () => Promise; } @@ -172,7 +194,13 @@ export const makeInMemoryMcpSessionStore = ( // proxy) it is preferred over the request URL — whose host would be the // internal bind address (127.0.0.1:PORT), unreachable for the user. Omit it on // loopback hosts (local/desktop), where the request URL is already correct. - options: { readonly webBaseUrl?: string } = {}, + options: { + readonly webBaseUrl?: string; + /** Idle window before a session is evicted. 0 disables eviction. */ + readonly sessionIdleTtlMs?: number; + /** How often the sweep runs. Defaults to a quarter of the TTL. */ + readonly sessionSweepIntervalMs?: number; + } = {}, ): InMemoryMcpSessionStore => { const transports = new Map(); const servers = new Map(); @@ -180,6 +208,17 @@ export const makeInMemoryMcpSessionStore = ( const engines = new Map>(); const approvals: InProcessBrowserApprovalStore = makeInProcessBrowserApprovalStore(); const requestStateSigningKey = crypto.getRandomValues(new Uint8Array(32)); + // Monotonic-ish last-touch stamp per live session, the only input the idle + // sweep reads. Written on create and on every forwarded request. + const lastSeen = new Map(); + + const idleTtlMs = options.sessionIdleTtlMs ?? DEFAULT_SESSION_IDLE_TTL_MS; + const sweepIntervalMs = + options.sessionSweepIntervalMs ?? Math.max(MIN_SWEEP_INTERVAL_MS, Math.floor(idleTtlMs / 4)); + + const touch = (id: string): void => { + if (lastSeen.has(id)) lastSeen.set(id, Date.now()); + }; const dispose = async (id: string, opts: { transport?: boolean; server?: boolean } = {}) => { const transport = transports.get(id); @@ -188,6 +227,7 @@ export const makeInMemoryMcpSessionStore = ( servers.delete(id); owners.delete(id); engines.delete(id); + lastSeen.delete(id); if (opts.transport) await ignoreClose(transport ? () => transport.close() : undefined); if (opts.server) await ignoreClose(server ? () => server.close() : undefined); }; @@ -228,6 +268,7 @@ export const makeInMemoryMcpSessionStore = ( const owner = owners.get(sessionId); if (!transport || !owner) return Effect.succeed("not-found"); if (!sessionOwnerMatches(owner, principal, resource)) return Effect.succeed("forbidden"); + touch(sessionId); return runHandleRequest(transport, request); }; @@ -290,6 +331,7 @@ export const makeInMemoryMcpSessionStore = ( servers.set(sid, mcpServer); owners.set(sid, { principal, resource }); engines.set(sid, engine); + lastSeen.set(sid, Date.now()); }, onsessionclosed: (sid) => void dispose(sid, { server: true }), }); @@ -397,12 +439,42 @@ export const makeInMemoryMcpSessionStore = ( }); }; + /** Dispose every session whose last request is older than the idle window. */ + const sweepIdleSessions = async (now: number = Date.now()): Promise => { + if (idleTtlMs <= 0) return 0; + const stale = [...lastSeen.entries()] + .filter(([, seen]) => now - seen >= idleTtlMs) + .map(([id]) => id); + // Both flags: an evicted session's transport has no other owner, and leaving + // it open would keep the very handles the eviction exists to release. + await Promise.all(stale.map((id) => dispose(id, { transport: true, server: true }))); + return stale.length; + }; + + // `unref` so the sweep never keeps a host process alive on its own. Node and + // Bun both return a Timeout with it; the DOM typing does not, hence the guard. + const sweepTimer: ReturnType | undefined = + idleTtlMs > 0 + ? setInterval(() => { + // Same shape as `ignoreClose`: a sweep failure is not the host's + // problem and must never surface as an unhandled rejection. + void Effect.runPromise( + Effect.ignore( + Effect.tryPromise({ try: () => sweepIdleSessions(), catch: () => undefined }), + ), + ); + }, sweepIntervalMs) + : undefined; + (sweepTimer as { unref?: () => void } | undefined)?.unref?.(); + return { store, handlePausedRequest, handleApprovalRequest, sessionCount: () => transports.size, + sweepIdleSessions, close: async () => { + if (sweepTimer !== undefined) clearInterval(sweepTimer); const ids = new Set([...transports.keys(), ...servers.keys()]); await Promise.all([...ids].map((id) => dispose(id, { transport: true, server: true }))); }, From 00863debad8215ec6d4a990d11b7d5e9c14ebc45 Mon Sep 17 00:00:00 2001 From: Davies Ayo Date: Wed, 19 Aug 2026 23:04:58 +1000 Subject: [PATCH 2/5] feat(host-selfhost): expose EXECUTOR_MCP_SESSION_IDLE_TTL_MS The store's idle window is only useful if an operator can tune it: a client that cannot tolerate re-initializing needs a longer TTL, and diagnosing one needs eviction off entirely (0). Parse it the same way as EXECUTOR_SANDBOX_TIMEOUT_MS, refusing to boot on a malformed value. --- apps/host-selfhost/src/config.ts | 19 +++++++++++++++++++ apps/host-selfhost/src/mcp/index.ts | 7 ++++++- apps/host-selfhost/src/mcp/session-store.ts | 6 +++++- 3 files changed, 30 insertions(+), 2 deletions(-) diff --git a/apps/host-selfhost/src/config.ts b/apps/host-selfhost/src/config.ts index 20728fabff..63f5adc655 100644 --- a/apps/host-selfhost/src/config.ts +++ b/apps/host-selfhost/src/config.ts @@ -53,6 +53,7 @@ export interface SelfHostConfig { * minutes (the same pattern as MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS on cloud). */ readonly sandboxTimeoutMs: number | undefined; + readonly mcpSessionIdleTtlMs: number | undefined; } export const resolveDataDir = (): string => @@ -160,6 +161,7 @@ export const loadConfig = (): SelfHostConfig => { organizationName: process.env.EXECUTOR_ORG_NAME ?? "Default", orgSlug: resolveOrgSlug(), sandboxTimeoutMs: resolveSandboxTimeoutMs(), + mcpSessionIdleTtlMs: resolveMcpSessionIdleTtlMs(), }; }; @@ -179,6 +181,23 @@ const resolveSandboxTimeoutMs = (): number | undefined => { return Math.floor(parsed); }; +// How long an MCP session may sit idle before the store evicts it. 0 disables +// eviction, which restores the old behaviour of holding every session for the +// lifetime of the process — only useful for diagnosing a client that cannot +// tolerate re-initializing. +const resolveMcpSessionIdleTtlMs = (): number | undefined => { + const raw = process.env.EXECUTOR_MCP_SESSION_IDLE_TTL_MS; + if (!raw) return undefined; + const parsed = Number(raw); + if (!Number.isFinite(parsed) || parsed < 0) { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: refuse to boot on a malformed operator knob + throw new Error( + `EXECUTOR_MCP_SESSION_IDLE_TTL_MS ${JSON.stringify(raw)} is not a non-negative number of milliseconds`, + ); + } + return Math.floor(parsed); +}; + // The org slug doubles as a URL segment (`//policies`), so an // operator-set value must fit the shared grammar and avoid reserved root // segments (api, mcp, login, …) — a colliding slug would shadow real routes. diff --git a/apps/host-selfhost/src/mcp/index.ts b/apps/host-selfhost/src/mcp/index.ts index a11d5cd8c9..8a5fd27024 100644 --- a/apps/host-selfhost/src/mcp/index.ts +++ b/apps/host-selfhost/src/mcp/index.ts @@ -11,6 +11,7 @@ import type { import { BetterAuth, type BetterAuthHandle } from "../auth/better-auth"; import type { SelfHostDbHandle } from "../db/self-host-db"; +import { loadConfig } from "../config"; import { selfHostMcpAuth } from "./auth"; import { makeSelfHostMcpSessionStore, @@ -142,7 +143,11 @@ export const makeSelfHostMcpSeams = ( webBaseUrl?: string, modernEnabled = true, ): SelfHostMcpSeams => { - const sessionStore = makeSelfHostMcpSessionStore(dbHandle, webBaseUrl); + const sessionStore = makeSelfHostMcpSessionStore( + dbHandle, + webBaseUrl, + loadConfig().mcpSessionIdleTtlMs, + ); const auth: Layer.Layer = selfHostMcpAuth.pipe( Layer.provide(Layer.succeed(BetterAuth)(betterAuth)), ); diff --git a/apps/host-selfhost/src/mcp/session-store.ts b/apps/host-selfhost/src/mcp/session-store.ts index 436671f3a2..a31d987ab2 100644 --- a/apps/host-selfhost/src/mcp/session-store.ts +++ b/apps/host-selfhost/src/mcp/session-store.ts @@ -36,6 +36,7 @@ export { McpEngineBuildError } from "@executor-js/host-mcp/in-memory-session-sto export const makeSelfHostMcpSessionStore = ( db: SelfHostDbHandle, webBaseUrl?: string, + sessionIdleTtlMs?: number, ): InMemoryMcpSessionStore => makeInMemoryMcpSessionStore( makeMcpBuildServer( @@ -48,7 +49,10 @@ export const makeSelfHostMcpSessionStore = ( selfHostAnalytics.record(`artifact_${action}`, { via: "agent" }), }, ), - { webBaseUrl }, + { + ...(webBaseUrl === undefined ? {} : { webBaseUrl }), + ...(sessionIdleTtlMs === undefined ? {} : { sessionIdleTtlMs }), + }, ); /** Build the stateless MCP server seam over the same self-host stack/config. */ From c3724edc751ade37202a60a49cb23bf3189566e4 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 27 Aug 2026 21:40:16 -0700 Subject: [PATCH 3/5] Thread self-host config into the MCP seams and correct the eviction rationale makeSelfHostMcpSeams called loadConfig() inside the factory, hiding an env read behind construction and moving a boot-time throw (loadConfig refuses a malformed EXECUTOR_MCP_SESSION_IDLE_TTL_MS) off the boot path. The app already has the resolved config, so pass it in. The store comment claimed enableJsonResponse means no long-lived stream exists. It only governs how a POST carrying requests answers; the bare 202 for notifications/initialized still cues the client to open the GET stream, so nearly every session holds one. Eviction ignores it on purpose, which is what cloud already does past its running-lease ceiling. --- .changeset/mcp-session-idle-eviction.md | 2 +- apps/host-selfhost/src/app.ts | 4 +-- apps/host-selfhost/src/config.ts | 4 +++ apps/host-selfhost/src/mcp/index.ts | 15 ++++++--- .../hosts/mcp/src/in-memory-session-store.ts | 33 ++++++++++++++----- 5 files changed, 41 insertions(+), 17 deletions(-) diff --git a/.changeset/mcp-session-idle-eviction.md b/.changeset/mcp-session-idle-eviction.md index e7c075d65b..43ed1c604f 100644 --- a/.changeset/mcp-session-idle-eviction.md +++ b/.changeset/mcp-session-idle-eviction.md @@ -2,4 +2,4 @@ "@executor-js/host-mcp": patch --- -Evict idle MCP sessions instead of holding them for the lifetime of the process. The in-process session store only released a session when the client sent `DELETE /mcp`, which the MCP client SDK's `transport.close()` never sends and a crashed client cannot send, so every `initialize` permanently retained an `McpServer`, its tool registry, and an `ExecutionEngine`. Sessions are now stamped on create and on each request, and a timer disposes anything idle past `sessionIdleTtlMs` (30 minutes by default). +Evict idle MCP sessions instead of holding them for the lifetime of the process. The in-process session store only released a session when the client sent `DELETE /mcp`, which the MCP client SDK's `transport.close()` never sends and a crashed client cannot send, so every `initialize` permanently retained an `McpServer`, its tool registry, and an `ExecutionEngine`. Sessions are now stamped on create and on each request, and a timer disposes anything idle past `sessionIdleTtlMs` (30 minutes by default). An open server-to-client stream does not defer eviction, matching how cloud's session alarm destroys a session once it passes its running-lease ceiling; an evicted id answers 404 `-32001`, which is the client's cue to re-initialize. diff --git a/apps/host-selfhost/src/app.ts b/apps/host-selfhost/src/app.ts index 4eaf631f54..a2341702a8 100644 --- a/apps/host-selfhost/src/app.ts +++ b/apps/host-selfhost/src/app.ts @@ -76,9 +76,7 @@ export const makeSelfHostApp = async (options: MakeSelfHostAppOptions = {}) => { const { identityLayer, authHandler, betterAuth } = await resolveAuthProviders(dbHandle); // ---- the in-process MCP serving seams (+ shutdown hook) ---------------- - // Pass the pinned public origin so browser-approval URLs are reachable behind - // a reverse proxy (not the internal 127.0.0.1 bind from the request URL). - const mcp = makeSelfHostMcpSeams(dbHandle, betterAuth, config.webBaseUrl); + const mcp = makeSelfHostMcpSeams(dbHandle, betterAuth, config); // CLI device-login discovery (`executor login`). Points the CLI at Better // Auth's device endpoints; `requestFormat: "json"` because those endpoints diff --git a/apps/host-selfhost/src/config.ts b/apps/host-selfhost/src/config.ts index e79cc56e94..daad96fcbf 100644 --- a/apps/host-selfhost/src/config.ts +++ b/apps/host-selfhost/src/config.ts @@ -51,6 +51,10 @@ export interface SelfHostConfig { * minutes (the same pattern as MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS on cloud). */ readonly sandboxTimeoutMs: number | undefined; + /** + * How long an MCP session may sit idle before the in-process store evicts it, + * or undefined for the store's own default (30 minutes). 0 disables eviction. + */ readonly mcpSessionIdleTtlMs: number | undefined; } diff --git a/apps/host-selfhost/src/mcp/index.ts b/apps/host-selfhost/src/mcp/index.ts index af1af47395..5992a6193c 100644 --- a/apps/host-selfhost/src/mcp/index.ts +++ b/apps/host-selfhost/src/mcp/index.ts @@ -10,7 +10,7 @@ import type { import { BetterAuth, type BetterAuthHandle } from "../auth/better-auth"; import type { SelfHostDbHandle } from "../db/self-host-db"; -import { loadConfig } from "../config"; +import type { SelfHostConfig } from "../config"; import { selfHostMcpAuth } from "./auth"; import { makeSelfHostMcpSessionStore, @@ -129,16 +129,23 @@ const makeApprovalHandler = * instance provided; it still requires `IdentityProvider` from the resolved * identity seam. Returns the three seam Layers plus the `close()` lifetime hook * the app wires into shutdown. + * + * Takes the already-resolved `SelfHostConfig` rather than reading it here: the + * app loads it once at boot, and `loadConfig()` refuses to boot on a malformed + * operator knob, so calling it from a seam factory would both hide an env read + * behind construction and move that failure off the boot path. */ export const makeSelfHostMcpSeams = ( dbHandle: SelfHostDbHandle, betterAuth: BetterAuthHandle, - webBaseUrl?: string, + config: SelfHostConfig, ): SelfHostMcpSeams => { + // The pinned public origin keeps browser-approval URLs reachable behind a + // reverse proxy (not the internal 127.0.0.1 bind from the request URL). const sessionStore = makeSelfHostMcpSessionStore( dbHandle, - webBaseUrl, - loadConfig().mcpSessionIdleTtlMs, + config.webBaseUrl, + config.mcpSessionIdleTtlMs, ); const auth: Layer.Layer = selfHostMcpAuth.pipe( Layer.provide(Layer.succeed(BetterAuth)(betterAuth)), diff --git a/packages/hosts/mcp/src/in-memory-session-store.ts b/packages/hosts/mcp/src/in-memory-session-store.ts index bf8eb5b708..472e6170c9 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.ts @@ -54,17 +54,32 @@ import type { BrowserApprovalStore } from "./tool-server"; // --------------------------------------------------------------------------- // A streamable-HTTP session only leaves these maps when the client sends -// `DELETE /mcp`. Nothing else can free it: with `enableJsonResponse` there is no -// stream whose teardown signals the client is gone, and a client that crashes, -// is killed, or simply calls the MCP SDK's `transport.close()` (which aborts -// locally and sends nothing) never issues that DELETE. Without a sweep, one -// abandoned session pins its `McpServer`, its tool registry, and its -// `ExecutionEngine` for the lifetime of the process. +// `DELETE /mcp`, and nothing sends it: `StreamableHTTPClientTransport.close()` +// aborts locally and puts nothing on the wire (only `terminateSession()` sends +// the DELETE, and `Client.close()` does not call it), and a client that crashes +// or is killed cannot send it at all. Without a sweep, one abandoned session +// pins its `McpServer`, its tool registry, and its `ExecutionEngine` for the +// lifetime of the process. +// +// The standalone SSE stream is NOT a substitute teardown signal, and it is not +// absent either. `enableJsonResponse` governs only how a POST carrying requests +// answers; a POST carrying just the `notifications/initialized` notification +// still gets a bare 202, which is exactly the cue the client SDK uses to open +// the long-lived `GET /mcp` stream. So essentially every session holds an open +// server-to-client stream for its whole life. That stream is silent by design +// (it exists for server-initiated messages) and this transport does no max-age +// rotation, so it produces no recurring request to stamp against — an open +// stream tells us the socket is up, never that the peer is still working. // // So the store treats a session as abandoned once it has gone `idleTtlMs` -// without a request and disposes it. That is what the streamable-HTTP spec -// allows a server to do: a request carrying an evicted id gets the store's -// existing "not-found" (404, -32001), which is the client's cue to re-initialize. +// without a REQUEST and disposes it, open stream or not. That mirrors cloud's +// `decideSessionAlarm`, where an active stream extends the lease only up to +// `MAX_RUNNING_SESSION_IDLE_MS` and the session is then destroyed regardless; +// the default here is that same order of ceiling. It is also what the +// streamable-HTTP spec allows a server to do: a request carrying an evicted id +// gets the store's existing "not-found" (404, -32001), the client's cue to +// re-initialize. The cost is bounded and visible — a connected-but-quiet client +// loses its stream at the ceiling and re-initializes on its next call. /** Idle window after which an untouched session is evicted. */ const DEFAULT_SESSION_IDLE_TTL_MS = 30 * 60 * 1000; /** Floor on the sweep interval, so a small TTL cannot spin the timer. */ From f0e3f2d7c23c74997093e181c2372faf02c1e68d Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 27 Aug 2026 22:54:57 -0700 Subject: [PATCH 4/5] Add a self-host e2e scenario for idle MCP session eviction --- .../mcp-session-idle-eviction.test.ts | 172 ++++++++++++++++++ e2e/setup/selfhost.boot.ts | 7 + 2 files changed, 179 insertions(+) create mode 100644 e2e/selfhost/mcp-session-idle-eviction.test.ts diff --git a/e2e/selfhost/mcp-session-idle-eviction.test.ts b/e2e/selfhost/mcp-session-idle-eviction.test.ts new file mode 100644 index 0000000000..1ea5236891 --- /dev/null +++ b/e2e/selfhost/mcp-session-idle-eviction.test.ts @@ -0,0 +1,172 @@ +// Selfhost-only: an MCP session that goes idle past the store's TTL is +// reclaimed, and the next request on that session id gets the 404 / -32001 cue +// that tells a client to re-initialize. +// +// Why this scenario boots its OWN instance instead of using the shared one: +// the idle window is a BOOT-TIME operator knob +// (EXECUTOR_MCP_SESSION_IDLE_TTL_MS), so there is no way to shrink it on a +// running server. Waiting out the 30-minute default is not an option, and +// shrinking it on the SHARED instance would silently evict sessions underneath +// every other selfhost scenario. A dedicated instance on its own port and data +// dir keeps the short window contained to this file. +// +// Why the wait is a single silent sleep rather than a poll loop: every request +// the store forwards restamps that session's last-seen time. A poll loop is +// itself the traffic that keeps the session alive, so it could never observe an +// eviction. The scenario stays completely silent for one window, then makes +// exactly one request. +// +// Auth is the Better Auth session cookie, not an OAuth bearer: self-host's MCP +// auth provider accepts the cookie/api-key identity path, and the credential is +// not what is under test here — session lifetime is. +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; + +import { scenario } from "../src/scenario"; +import { RunDir, Target } from "../src/services"; +import { claimAndBoot } from "../src/ports"; +import { isBootReadinessTimeout } from "../setup/boot"; +import { bootSelfhost } from "../setup/selfhost.boot"; +import { SELFHOST_ADMIN, signInSession } from "../targets/selfhost"; + +/** The idle window this instance runs with — small enough that the sweep, not + * the TTL, sets the pace. */ +const IDLE_TTL_MS = 2_000; + +/** + * How long to leave the session untouched. The store floors its sweep interval + * at 30s, so an idle session is disposed at the first tick that falls at least + * one TTL after its last request. Waiting two full sweep intervals plus the TTL + * means the assertion can never race a tick that has not fired yet. + */ +const QUIET_WINDOW_MS = 2 * 30_000 + IDLE_TTL_MS + 5_000; + +interface JsonRpcErrorBody { + readonly error?: { readonly code?: number; readonly message?: string }; +} + +scenario( + "MCP · an idle self-host session is evicted and answers 404 -32001 until the client re-initializes", + // Own vite dev boot (cold on a fresh checkout) plus a 67s quiet window, so + // this needs materially more than the project's 180s default. + { timeout: 420_000 }, + Effect.gen(function* () { + // Selfhost-shaped scenario: yielded for the target name in failures, and so + // the file reads like its neighbours. + yield* Target; + const runDir = yield* RunDir; + + const dataDir = mkdtempSync(join(tmpdir(), "executor-selfhost-idle-ttl-")); + + // A distinct env var (not E2E_SELFHOST_PORT, which the shared instance has + // already published into this worker's env) so the claim actually probes + // and locks a free port instead of returning the shared one. + const booted = yield* Effect.promise(() => + claimAndBoot( + [{ envVar: "E2E_SELFHOST_IDLE_TTL_PORT", offset: 6, label: "selfhost idle-ttl vite dev" }], + async (ports) => { + const port = ports.E2E_SELFHOST_IDLE_TTL_PORT!; + const baseUrl = `http://localhost:${port}`; + const procs = await bootSelfhost({ + port, + webBaseUrl: baseUrl, + admin: SELFHOST_ADMIN, + dataDir, + logFile: join(runDir, "idle-ttl-boot.log"), + mcpSessionIdleTtlMs: IDLE_TTL_MS, + }); + return { teardown: procs.teardown, value: baseUrl }; + }, + { label: "selfhost idle-ttl", retryWhen: isBootReadinessTimeout }, + ), + ); + + yield* Effect.gen(function* () { + const baseUrl = booted.value; + const mcpUrl = new URL("/mcp", baseUrl).toString(); + const { cookieHeader } = yield* Effect.promise(() => signInSession(baseUrl, SELFHOST_ADMIN)); + + const initialize = async (): Promise => + fetch(mcpUrl, { + method: "POST", + headers: { + cookie: cookieHeader, + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "idle-ttl-e2e", version: "1" }, + }, + }), + }); + + const listTools = async (sessionId: string, id: number): Promise => + fetch(mcpUrl, { + method: "POST", + headers: { + cookie: cookieHeader, + "mcp-session-id": sessionId, + "mcp-protocol-version": "2025-06-18", + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify({ jsonrpc: "2.0", id, method: "tools/list" }), + }); + + // 1. A client initializes and the session serves. + const opened = yield* Effect.promise(initialize); + expect(opened.status, "initialize succeeds").toBe(200); + const sessionId = opened.headers.get("mcp-session-id"); + expect(sessionId, "initialize returns a session id").toEqual(expect.any(String)); + + const working = yield* Effect.promise(() => listTools(sessionId!, 2)); + expect(working.status, "the fresh session serves a request").toBe(200); + + // 2. Idleness, driven by the clock and nothing else. Touching the session + // here — even to poll — would restamp it and defeat the measurement. + yield* Effect.sleep(`${QUIET_WINDOW_MS} millis`); + + // 3. The evicted id is gone, and says so in the shape a client acts on: + // 404 tells it the id is dead, -32001 is the session-lifecycle code. + const afterIdle = yield* Effect.promise(() => listTools(sessionId!, 3)); + const afterIdleBody = yield* Effect.promise(() => afterIdle.text()); + // A session that was NOT evicted answers with the whole tool catalog, so + // the diagnostic is truncated — the status is the assertion, and a full + // tools/list dump in the failure output helps nobody. + expect( + afterIdle.status, + `an idle session is evicted, so its id 404s; body starts: ${afterIdleBody.slice(0, 200)}`, + ).toBe(404); + // oxlint-disable-next-line executor/no-json-parse -- boundary: the raw JSON-RPC error frame this scenario asserts on, never decoded into a domain type + const parsed = JSON.parse(afterIdleBody) as JsonRpcErrorBody; + expect(parsed.error?.code, "the 404 carries the session-lifecycle code").toBe(-32001); + + // 4. The cue is actionable: re-initializing gets a NEW, working session. + const reopened = yield* Effect.promise(initialize); + expect(reopened.status, "the client can re-initialize after eviction").toBe(200); + const newSessionId = reopened.headers.get("mcp-session-id"); + expect(newSessionId, "re-initialize returns a session id").toEqual(expect.any(String)); + expect(newSessionId, "re-initialize issues a different session").not.toBe(sessionId); + + const afterReinit = yield* Effect.promise(() => listTools(newSessionId!, 4)); + expect(afterReinit.status, "the re-initialized session serves a request").toBe(200); + }).pipe( + Effect.ensuring( + Effect.promise(async () => { + await booted.teardown(); + rmSync(dataDir, { recursive: true, force: true }); + }), + ), + ); + }), +); diff --git a/e2e/setup/selfhost.boot.ts b/e2e/setup/selfhost.boot.ts index 90b29bc87d..41b3d38ea2 100644 --- a/e2e/setup/selfhost.boot.ts +++ b/e2e/setup/selfhost.boot.ts @@ -24,6 +24,10 @@ export interface SelfhostBootOptions { /** Shrink the sandbox execution budget (EXECUTOR_SANDBOX_TIMEOUT_MS) so * deadline scenarios prove their race in seconds. Omit for production. */ readonly sandboxTimeoutMs?: number; + /** Shrink the MCP session idle window (EXECUTOR_MCP_SESSION_IDLE_TTL_MS) so + * the eviction scenario proves in seconds what otherwise takes 30 minutes. + * Omit for production; the app then uses the store's own default. */ + readonly mcpSessionIdleTtlMs?: number; } export const bootSelfhost = async (options: SelfhostBootOptions): Promise => { @@ -57,6 +61,9 @@ export const bootSelfhost = async (options: SelfhostBootOptions): Promise Date: Thu, 27 Aug 2026 23:35:58 -0700 Subject: [PATCH 5/5] Hold an MCP session while a request is in flight, and shut its engine down The idle sweep read only a last-seen stamp, written before the store awaits transport.handleRequest. A call slower than the idle window was therefore indistinguishable from an abandoned session, and the sweep closed the transport, the server, and the engine underneath the request still using them. Count the requests inside handleRequest per session, skip a session with any, and restamp when a call ends so idleness measures from completion. Disposal deleted the engine reference without running engine.shutdown, so the detached sandbox fibers a paused execution holds kept running - and kept querying the host database handle - after the session was gone. Every disposal path now shuts the engine down: sweep eviction, the dispose seam, store close, onsessionclosed, and the eager close of a transport that never minted an id. A close failure was swallowed whole, which made a leaked handle invisible. Keep it best-effort, but log it at warning with the session id and the handle. Wire the store close hook into the self-host server. startServer discarded closeDb, so a graceful shutdown left every live session and the libSQL handle open; only the test web-handler path ever released them. --- .changeset/mcp-session-idle-eviction.md | 2 + apps/host-selfhost/src/serve.ts | 18 +- .../mcp/src/in-memory-session-store.test.ts | 176 +++++++++++++++--- .../hosts/mcp/src/in-memory-session-store.ts | 115 ++++++++++-- 4 files changed, 264 insertions(+), 47 deletions(-) diff --git a/.changeset/mcp-session-idle-eviction.md b/.changeset/mcp-session-idle-eviction.md index 43ed1c604f..246ee0c06f 100644 --- a/.changeset/mcp-session-idle-eviction.md +++ b/.changeset/mcp-session-idle-eviction.md @@ -3,3 +3,5 @@ --- Evict idle MCP sessions instead of holding them for the lifetime of the process. The in-process session store only released a session when the client sent `DELETE /mcp`, which the MCP client SDK's `transport.close()` never sends and a crashed client cannot send, so every `initialize` permanently retained an `McpServer`, its tool registry, and an `ExecutionEngine`. Sessions are now stamped on create and on each request, and a timer disposes anything idle past `sessionIdleTtlMs` (30 minutes by default). An open server-to-client stream does not defer eviction, matching how cloud's session alarm destroys a session once it passes its running-lease ceiling; an evicted id answers 404 `-32001`, which is the client's cue to re-initialize. + +A request in flight holds its session: idleness counts from when a call ends, not from when it started, so a tool call slower than the idle window is never cut off mid-flight. Disposal also shuts the session's execution engine down rather than only dropping the reference, which is what ends its detached sandbox fibers, and a handle that fails to close is now logged with its session id instead of being discarded silently. diff --git a/apps/host-selfhost/src/serve.ts b/apps/host-selfhost/src/serve.ts index f5700c53fe..37939da929 100644 --- a/apps/host-selfhost/src/serve.ts +++ b/apps/host-selfhost/src/serve.ts @@ -109,7 +109,7 @@ const selfHostHttpMiddleware = (betterAuth: BetterAuthHandle) => export const startServer = async (): Promise => { const config = loadConfig(); - const { AppLayer, betterAuth } = await makeSelfHostApp(); + const { AppLayer, betterAuth, closeDb } = await makeSelfHostApp(); // Serve the built SPA, split by cacheability so a redeploy is picked up at // once instead of stranding browsers on a stale shell: @@ -141,6 +141,16 @@ export const startServer = async (): Promise => { Effect.addFinalizer(() => Effect.promise(() => disposeAnalytics())), ); + // Server-scope finalizer: release what the app opened at boot — every live + // MCP session (transport, server, and its execution engine, whose detached + // sandbox fibers keep querying the DB until the engine is shut down) and then + // the shared libSQL handle itself. `makeSelfHostApiHandler` runs this in its + // `dispose`, so tests have always released it; the long-lived server dropped + // the hook on the floor and leaked both across a graceful shutdown. + const AppResourcesLive = Layer.effectDiscard( + Effect.addFinalizer(() => Effect.promise(() => closeDb())), + ); + // OTLP export, or `Layer.empty` when no collector is configured (see // ./telemetry). The `http.server` envelope span each request's `withSpan` // children parent under is NOT wired here: `HttpEffect.toHandled` already @@ -162,7 +172,11 @@ export const startServer = async (): Promise => { // in scope while the app's layers build, or spans created during construction // resolve the default no-op tracer and are silently dropped. await BunRuntime.runMain( - Layer.launch(Layer.merge(ServerLive, AnalyticsFlushLive).pipe(Layer.provide(TelemetryLive))), + Layer.launch( + Layer.mergeAll(ServerLive, AnalyticsFlushLive, AppResourcesLive).pipe( + Layer.provide(TelemetryLive), + ), + ), ); }; diff --git a/packages/hosts/mcp/src/in-memory-session-store.test.ts b/packages/hosts/mcp/src/in-memory-session-store.test.ts index 88c40ddb52..6e9e72c710 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.test.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.test.ts @@ -68,11 +68,84 @@ const makeIdleTestEngine = (): ExecutionEngine => ({ shutdown: Effect.void, }); +/** + * An engine whose `execute` parks until the test releases it, so a request can + * be held inside `transport.handleRequest` while the sweep runs. `shutdowns` + * counts `engine.shutdown` runs — the disposal step that ends the detached + * sandbox fibers, and which dropping the engine reference does not do. + */ +const makeLatchedTestEngine = (): { + readonly engine: ExecutionEngine; + readonly started: Promise; + readonly release: () => void; + readonly shutdowns: () => number; +} => { + let signalStarted: () => void = () => {}; + const started = new Promise((resolve) => { + signalStarted = resolve; + }); + let openGate: () => void = () => {}; + const gate = new Promise((resolve) => { + openGate = resolve; + }); + let shutdowns = 0; + const park = (value: A): Effect.Effect => + Effect.promise(async () => { + signalStarted(); + await gate; + return value; + }); + const engine: ExecutionEngine = { + ...makeIdleTestEngine(), + execute: () => park({ result: "released" }), + executeWithPause: () => park({ status: "completed", result: { result: "released" } }), + shutdown: Effect.sync(() => { + shutdowns += 1; + }), + }; + return { engine, started, release: () => openGate(), shutdowns: () => shutdowns }; +}; + // A long TTL keeps the sweep's own timer out of the way; the assertions drive // `sweepIdleSessions` directly with an explicit instant instead of sleeping // through a real window, so the test is deterministic rather than timing-raced. const IDLE_TTL_MS = 60_000; +type TestSessionStore = ReturnType; + +/** Open a session on `sessions` and return its minted id. */ +const openSession = async (sessions: TestSessionStore): Promise => { + const response = (await Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "idle-test", version: "1.0.0" }, + }, + }), + }), + principal: TEST_PRINCIPAL, + resource: defaultMcpResource, + sessionId: null, + method: "POST", + }), + )) as Response; + expect(response.status).toBe(200); + const sessionId = response.headers.get("mcp-session-id") ?? ""; + expect(sessionId).not.toBe(""); + return sessionId; +}; + it("evicts a session that goes idle past the TTL and keeps a busy one", async () => { const engine = makeIdleTestEngine(); const sessions = makeInMemoryMcpSessionStore( @@ -81,37 +154,7 @@ it("evicts a session that goes idle past the TTL and keeps a busy one", async () { sessionIdleTtlMs: IDLE_TTL_MS }, ); - const open = async (): Promise => { - const response = (await Effect.runPromise( - sessions.store.dispatch({ - request: new Request("https://executor.test/mcp", { - method: "POST", - headers: { - "content-type": "application/json", - accept: "application/json, text/event-stream", - }, - body: JSON.stringify({ - jsonrpc: "2.0", - id: 1, - method: "initialize", - params: { - protocolVersion: "2025-06-18", - capabilities: {}, - clientInfo: { name: "idle-test", version: "1.0.0" }, - }, - }), - }), - principal: TEST_PRINCIPAL, - resource: defaultMcpResource, - sessionId: null, - method: "POST", - }), - )) as Response; - expect(response.status).toBe(200); - const sessionId = response.headers.get("mcp-session-id") ?? ""; - expect(sessionId).not.toBe(""); - return sessionId; - }; + const open = (): Promise => openSession(sessions); const call = (sessionId: string, id: number) => Effect.runPromise( @@ -163,3 +206,74 @@ it("evicts a session that goes idle past the TTL and keeps a busy one", async () await sessions.close(); } }); + +it("never evicts a session while one of its requests is still in flight", async () => { + const latched = makeLatchedTestEngine(); + const sessions = makeInMemoryMcpSessionStore( + () => + createExecutorMcpServer({ engine: latched.engine }).pipe( + Effect.map((mcpServer) => ({ mcpServer, engine: latched.engine })), + ), + { sessionIdleTtlMs: IDLE_TTL_MS }, + ); + + const callExecute = (sessionId: string): Promise => + Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { + "content-type": "application/json", + accept: "application/json, text/event-stream", + "mcp-session-id": sessionId, + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 2, + method: "tools/call", + params: { name: "execute", arguments: { code: "return 1" } }, + }), + }), + principal: TEST_PRINCIPAL, + resource: defaultMcpResource, + sessionId, + method: "POST", + }), + ); + + // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: always release the latch and close the store + try { + const sessionId = await openSession(sessions); + + // Start a call and park it inside the engine. `forward` stamps last-seen + // BEFORE it awaits the transport, so from here on the stamp only ages — a + // request slower than the TTL is indistinguishable from an abandoned + // session unless the store also counts what is in flight. + const startedAt = Date.now(); + const inFlight = callExecute(sessionId); + await latched.started; + + // Sweep a full TTL past the moment the call began. Without the in-flight + // counter this evicts the session and closes the transport, the server, and + // the engine underneath the request that is still using them. + expect(await sessions.sweepIdleSessions(startedAt + IDLE_TTL_MS)).toBe(0); + expect(sessions.sessionCount()).toBe(1); + expect(latched.shutdowns()).toBe(0); + + // The parked request still completes, on the transport it started on. + latched.release(); + const response = await inFlight; + expect(response).toBeInstanceOf(Response); + expect((response as Response).status).toBe(200); + + // And the reprieve is only for the duration of the call: the session is + // restamped as it ends, so the next idle window still reclaims it — engine + // shutdown included, which is what ends the detached sandbox fibers. + expect(await sessions.sweepIdleSessions(Date.now() + IDLE_TTL_MS)).toBe(1); + expect(sessions.sessionCount()).toBe(0); + expect(latched.shutdowns()).toBe(1); + } finally { + latched.release(); + await sessions.close(); + } +}); diff --git a/packages/hosts/mcp/src/in-memory-session-store.ts b/packages/hosts/mcp/src/in-memory-session-store.ts index 472e6170c9..74b157b854 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.ts @@ -151,15 +151,44 @@ export interface InMemoryMcpSessionStore { readonly close: () => Promise; } -const ignoreClose = (close: (() => Promise) | undefined): Promise => - close - ? Effect.runPromise(Effect.ignore(Effect.tryPromise({ try: close, catch: () => undefined }))) - : Promise.resolve(); - const formatBoundaryError = (error: unknown): unknown => // oxlint-disable-next-line executor/no-instanceof-error, executor/no-unknown-error-message -- boundary: log unknown MCP SDK/runtime failures error instanceof Error ? (error.stack ?? error.message) : error; +/** One session handle refused to close. Reported, never propagated. */ +class McpHandleCloseError extends Data.TaggedError("McpHandleCloseError")<{ + readonly cause: unknown; +}> {} + +/** + * Release one session handle, best effort. Disposal must finish even when a + * handle refuses to close — the other handles still have to go, and a rejection + * here would surface as an unhandled rejection on a sweep tick nobody awaits. + * But a silently swallowed failure is a leaked transport, server, or engine that + * nothing can see, so name the handle and the session in a warning. + */ +const ignoreClose = ( + sessionId: string | null, + handle: string, + close: (() => Promise) | undefined, +): Promise => { + if (!close) return Promise.resolve(); + const warn = (detail: unknown): Effect.Effect => + Effect.sync(() => { + console.warn( + `[mcp] failed to close ${handle} for session ${sessionId ?? ""}:`, + formatBoundaryError(detail), + ); + }); + return Effect.runPromise( + Effect.tryPromise({ try: close, catch: (cause) => new McpHandleCloseError({ cause }) }).pipe( + Effect.catch((error) => warn(error.cause)), + // A defect cannot escape either: this runs detached from any request. + Effect.catchCause((cause) => warn(Cause.squash(cause))), + ), + ); +}; + // The store's error bodies are INNER responses (no CORS): the serving envelope // re-wraps the store `Response` with CORS before it leaves the origin, so the // canonical renderer is called with `cors: false` (content-type only). @@ -211,9 +240,17 @@ export const makeInMemoryMcpSessionStore = ( const owners = new Map(); const engines = new Map>(); const approvals: InProcessBrowserApprovalStore = makeInProcessBrowserApprovalStore(); - // Monotonic-ish last-touch stamp per live session, the only input the idle + // Monotonic-ish last-touch stamp per live session, the first input the idle // sweep reads. Written on create and on every forwarded request. const lastSeen = new Map(); + // Requests currently inside `transport.handleRequest` for a session, the + // sweep's second input. A stamp alone cannot describe a long call: it is + // written BEFORE the await, so a single `execute` that outruns the TTL (a + // browser approval waiting on a human, a slow upstream) would look exactly + // like an abandoned session and have its transport, server, and engine closed + // out from under the request that is still using them. Counting requests in + // flight makes "idle" mean what it says. + const activeRequests = new Map(); const idleTtlMs = options.sessionIdleTtlMs ?? DEFAULT_SESSION_IDLE_TTL_MS; const sweepIntervalMs = @@ -223,16 +260,50 @@ export const makeInMemoryMcpSessionStore = ( if (lastSeen.has(id)) lastSeen.set(id, Date.now()); }; + /** Claim a session for one in-flight request, so the sweep cannot take it. */ + const beginRequest = (id: string): void => { + activeRequests.set(id, (activeRequests.get(id) ?? 0) + 1); + }; + + /** + * Release the claim and restamp: a call that ran for an hour leaves the + * session idle from the moment it FINISHED, not from the moment it started. + * `touch` is a no-op once the session is gone, so this can never resurrect a + * disposed id. + */ + const endRequest = (id: string): void => { + const remaining = (activeRequests.get(id) ?? 1) - 1; + if (remaining > 0) activeRequests.set(id, remaining); + else activeRequests.delete(id); + touch(id); + }; + + /** + * Shut down a session's engine. Dropping the reference is not enough: the + * engine's paused executions hold detached sandbox fibers that keep running — + * and keep querying the host's database handle — until `shutdown` interrupts + * them. Every disposal path goes through here. + */ + const shutdownEngine = ( + id: string | null, + engine: ExecutionEngine | undefined, + ): Promise => + ignoreClose(id, "engine", engine ? () => Effect.runPromise(engine.shutdown) : undefined); + const dispose = async (id: string, opts: { transport?: boolean; server?: boolean } = {}) => { const transport = transports.get(id); const server = servers.get(id); + const engine = engines.get(id); transports.delete(id); servers.delete(id); owners.delete(id); engines.delete(id); lastSeen.delete(id); - if (opts.transport) await ignoreClose(transport ? () => transport.close() : undefined); - if (opts.server) await ignoreClose(server ? () => server.close() : undefined); + activeRequests.delete(id); + if (opts.transport) + await ignoreClose(id, "transport", transport ? () => transport.close() : undefined); + if (opts.server) await ignoreClose(id, "server", server ? () => server.close() : undefined); + await shutdownEngine(id, engine); }; /** @@ -272,7 +343,14 @@ export const makeInMemoryMcpSessionStore = ( if (!transport || !owner) return Effect.succeed("not-found"); if (!sessionOwnerMatches(owner, principal, resource)) return Effect.succeed("forbidden"); touch(sessionId); - return runHandleRequest(transport, request); + // Claim before the await, release in the finalizer — `runHandleRequest` + // already recovers every failure to a 500, but `ensuring` also covers an + // interrupt, so the counter cannot be left permanently raised (which would + // make the session immortal, the opposite leak). + beginRequest(sessionId); + return runHandleRequest(transport, request).pipe( + Effect.ensuring(Effect.sync(() => endRequest(sessionId))), + ); }; /** @@ -343,8 +421,12 @@ export const makeInMemoryMcpSessionStore = ( // The session id is minted on the first (initialize) request, so we // drive `handleRequest` here; if no id results we close eagerly. return yield* runHandleRequest(transport, request, () => { - void ignoreClose(() => transport.close()); - void ignoreClose(() => mcpServer.close()); + // Nothing was ever registered under a session id, so `dispose` has + // no entry to work from — release the three handles by hand, engine + // included. + void ignoreClose(null, "transport", () => transport.close()); + void ignoreClose(null, "server", () => mcpServer.close()); + void shutdownEngine(null, engine); }); }), ), @@ -439,11 +521,16 @@ export const makeInMemoryMcpSessionStore = ( }); }; - /** Dispose every session whose last request is older than the idle window. */ + /** + * Dispose every session whose last request is older than the idle window AND + * which has nothing in flight. A session serving a request is busy, however + * long ago that request started; it gets a fresh stamp the moment it ends, so + * a later sweep still reclaims it if the client then goes quiet. + */ const sweepIdleSessions = async (now: number = Date.now()): Promise => { if (idleTtlMs <= 0) return 0; const stale = [...lastSeen.entries()] - .filter(([, seen]) => now - seen >= idleTtlMs) + .filter(([id, seen]) => now - seen >= idleTtlMs && (activeRequests.get(id) ?? 0) === 0) .map(([id]) => id); // Both flags: an evicted session's transport has no other owner, and leaving // it open would keep the very handles the eviction exists to release. @@ -475,7 +562,7 @@ export const makeInMemoryMcpSessionStore = ( sweepIdleSessions, close: async () => { if (sweepTimer !== undefined) clearInterval(sweepTimer); - const ids = new Set([...transports.keys(), ...servers.keys()]); + const ids = new Set([...transports.keys(), ...servers.keys(), ...engines.keys()]); await Promise.all([...ids].map((id) => dispose(id, { transport: true, server: true }))); }, };