diff --git a/.changeset/mcp-session-idle-eviction.md b/.changeset/mcp-session-idle-eviction.md new file mode 100644 index 0000000000..246ee0c06f --- /dev/null +++ b/.changeset/mcp-session-idle-eviction.md @@ -0,0 +1,7 @@ +--- +"@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). 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/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 acb77204f3..c218582e89 100644 --- a/apps/host-selfhost/src/config.ts +++ b/apps/host-selfhost/src/config.ts @@ -51,6 +51,11 @@ 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; /** * How long a connection's persisted remote tool catalog stays fresh, in ms. * `undefined` takes the SDK default (15 minutes); `null` disables time-based @@ -163,6 +168,7 @@ export const loadConfig = (): SelfHostConfig => { organizationName: process.env.EXECUTOR_ORG_NAME ?? "Default", orgSlug: resolveOrgSlug(), sandboxTimeoutMs: resolveSandboxTimeoutMs(), + mcpSessionIdleTtlMs: resolveMcpSessionIdleTtlMs(), toolsSyncTtlMs: resolveToolsSyncTtlMs(), }; }; @@ -183,6 +189,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 52287518cd..5992a6193c 100644 --- a/apps/host-selfhost/src/mcp/index.ts +++ b/apps/host-selfhost/src/mcp/index.ts @@ -10,6 +10,7 @@ import type { import { BetterAuth, type BetterAuthHandle } from "../auth/better-auth"; import type { SelfHostDbHandle } from "../db/self-host-db"; +import type { SelfHostConfig } from "../config"; import { selfHostMcpAuth } from "./auth"; import { makeSelfHostMcpSessionStore, @@ -128,13 +129,24 @@ 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 => { - const sessionStore = makeSelfHostMcpSessionStore(dbHandle, webBaseUrl); + // 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, + config.webBaseUrl, + config.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 c17a8fa4db..f9d29b1f05 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 }), + }, ); /** The `McpSessionStore` envelope seam over a freshly built in-process store. */ 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/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 ({ + execute: () => Effect.succeed({ result: "unused" }), + executeWithPause: () => Effect.succeed({ status: "completed", result: { result: "unused" } }), + resume: () => Effect.succeed(null), + getPausedExecution: () => Effect.succeed(null), + pausedExecutionCount: () => Effect.succeed(0), + hasPausedExecutions: () => Effect.succeed(false), + getDescription: Effect.succeed("idle-eviction test executor"), + 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( + () => + createExecutorMcpServer({ engine }).pipe(Effect.map((mcpServer) => ({ mcpServer, engine }))), + { sessionIdleTtlMs: IDLE_TTL_MS }, + ); + + const open = (): Promise => openSession(sessions); + + const call = (sessionId: string, id: number) => + 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, method: "tools/list" }), + }), + principal: TEST_PRINCIPAL, + resource: defaultMcpResource, + sessionId, + method: "POST", + }), + ); + + // 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 at the current instant takes nothing. + expect(await sessions.sweepIdleSessions()).toBe(0); + expect(sessions.sessionCount()).toBe(2); + + // Let the wall clock advance so the two sessions' stamps are separable, + // then keep working on one of them: `forward` restamps that one and only + // that one. + await new Promise((resolve) => setTimeout(resolve, 25)); + const restampedAt = Date.now(); + await call(busy, 2); + + // Sweep one TTL after the restamp, less a millisecond: `busy` was stamped + // at or after `restampedAt` so it cannot have aged a full TTL, while `idle` + // was stamped at least 25ms earlier and must have. Exactly one goes. + expect(await sessions.sweepIdleSessions(restampedAt + IDLE_TTL_MS - 1)).toBe(1); + expect(sessions.sessionCount()).toBe(1); + + // The evicted id is gone; the store reports it the way the envelope 404s. + expect(await call(idle, 3)).toBe("not-found"); + // The busy one still serves. + expect(await call(busy, 4)).toBeInstanceOf(Response); + } finally { + 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(); + } +}); + // --------------------------------------------------------------------------- // The pre-initialize guard, through the real store path. // diff --git a/packages/hosts/mcp/src/in-memory-session-store.ts b/packages/hosts/mcp/src/in-memory-session-store.ts index 81cc4aa2de..d66d30d84d 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.ts @@ -53,6 +53,38 @@ import type { BrowserApprovalStore } from "./tool-server"; // - "forbidden" (session owned by another bearer) -> envelope renders 403 -32003 // --------------------------------------------------------------------------- +// A streamable-HTTP session only leaves these maps when the client sends +// `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, 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. */ +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; @@ -108,19 +140,55 @@ export interface InMemoryMcpSessionStore { request: Request, principal?: Principal, ) => 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; } -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). @@ -159,23 +227,83 @@ 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(); const owners = new Map(); const engines = new Map>(); const approvals: InProcessBrowserApprovalStore = makeInProcessBrowserApprovalStore(); + // 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 = + 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()); + }; + + /** 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); - if (opts.transport) await ignoreClose(transport ? () => transport.close() : undefined); - if (opts.server) await ignoreClose(server ? () => server.close() : undefined); + lastSeen.delete(id); + 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); }; /** @@ -214,7 +342,15 @@ 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"); - return runHandleRequest(transport, request); + touch(sessionId); + // 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))), + ); }; /** @@ -273,6 +409,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 }), }); @@ -284,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); }); }), ), @@ -399,12 +540,48 @@ export const makeInMemoryMcpSessionStore = ( }); }; + /** + * 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(([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. + 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 () => { - const ids = new Set([...transports.keys(), ...servers.keys()]); + if (sweepTimer !== undefined) clearInterval(sweepTimer); + const ids = new Set([...transports.keys(), ...servers.keys(), ...engines.keys()]); await Promise.all([...ids].map((id) => dispose(id, { transport: true, server: true }))); }, };