History

fix(host-mcp): evict idle MCP sessions instead of leaking them (#1685)

* 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. * 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. * 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. * Add a self-host e2e scenario for idle MCP session eviction * 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. ---------

Davies Ayo committed 2 months ago

Commit 2b4e1061 parent 1908dd610 files · +656 −20Browse files

Unified · Split

10 files
  1. .changeset/mcp-session-idle-eviction.md +7 −0
  2. apps/host-selfhost/src/app.ts +1 −3
  3. apps/host-selfhost/src/config.ts +23 −0
  4. apps/host-selfhost/src/mcp/index.ts +14 −2
  5. apps/host-selfhost/src/mcp/session-store.ts +5 −1
  6. apps/host-selfhost/src/serve.ts +16 −2
  7. e2e/selfhost/mcp-session-idle-eviction.test.ts +172 −0
  8. e2e/setup/selfhost.boot.ts +7 −0
  9. packages/hosts/mcp/src/in-memory-session-store.test.ts +222 −0
  10. packages/hosts/mcp/src/in-memory-session-store.ts +189 −12
.changeset/mcp-session-idle-eviction.md+7−0
@@ -0,0 +1,7 @@
1---
2"@executor-js/host-mcp": patch
3---
4
5Evict 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.
6
7A 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.
apps/host-selfhost/src/app.ts+1−3
@@ -76,9 +76,7 @@ export const makeSelfHostApp = async (options: MakeSelfHostAppOptions = {}) => { · 75 unmodified lines
76 const { identityLayer, authHandler, betterAuth } = await resolveAuthProviders(dbHandle);76 const { identityLayer, authHandler, betterAuth } = await resolveAuthProviders(dbHandle);
77 77
78 // ---- the in-process MCP serving seams (+ shutdown hook) ----------------78 // ---- the in-process MCP serving seams (+ shutdown hook) ----------------
79 // Pass the pinned public origin so browser-approval URLs are reachable behind79 const mcp = makeSelfHostMcpSeams(dbHandle, betterAuth, config);
80 // a reverse proxy (not the internal 127.0.0.1 bind from the request URL).
81 const mcp = makeSelfHostMcpSeams(dbHandle, betterAuth, config.webBaseUrl);
82 80
83 // CLI device-login discovery (`executor login`). Points the CLI at Better81 // CLI device-login discovery (`executor login`). Points the CLI at Better
84 // Auth's device endpoints; `requestFormat: "json"` because those endpoints82 // Auth's device endpoints; `requestFormat: "json"` because those endpoints
apps/host-selfhost/src/config.ts+23−0
@@ -51,6 +51,11 @@ export interface SelfHostConfig { · 50 unmodified lines
51 * minutes (the same pattern as MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS on cloud).51 * minutes (the same pattern as MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS on cloud).
52 */52 */
53 readonly sandboxTimeoutMs: number | undefined;53 readonly sandboxTimeoutMs: number | undefined;
54 /**
55 * How long an MCP session may sit idle before the in-process store evicts it,
56 * or undefined for the store's own default (30 minutes). 0 disables eviction.
57 */
58 readonly mcpSessionIdleTtlMs: number | undefined;
54 /**59 /**
55 * How long a connection's persisted remote tool catalog stays fresh, in ms.60 * How long a connection's persisted remote tool catalog stays fresh, in ms.
56 * `undefined` takes the SDK default (15 minutes); `null` disables time-based61 * `undefined` takes the SDK default (15 minutes); `null` disables time-based
@@ -163,6 +168,7 @@ export const loadConfig = (): SelfHostConfig => { · 106 unmodified lines
163 organizationName: process.env.EXECUTOR_ORG_NAME ?? "Default",168 organizationName: process.env.EXECUTOR_ORG_NAME ?? "Default",
164 orgSlug: resolveOrgSlug(),169 orgSlug: resolveOrgSlug(),
165 sandboxTimeoutMs: resolveSandboxTimeoutMs(),170 sandboxTimeoutMs: resolveSandboxTimeoutMs(),
171 mcpSessionIdleTtlMs: resolveMcpSessionIdleTtlMs(),
166 toolsSyncTtlMs: resolveToolsSyncTtlMs(),172 toolsSyncTtlMs: resolveToolsSyncTtlMs(),
167 };173 };
168};174};
@@ -183,6 +189,23 @@ const resolveSandboxTimeoutMs = (): number | undefined => { · 14 unmodified lines
183 return Math.floor(parsed);189 return Math.floor(parsed);
184};190};
185 191
192// How long an MCP session may sit idle before the store evicts it. 0 disables
193// eviction, which restores the old behaviour of holding every session for the
194// lifetime of the process — only useful for diagnosing a client that cannot
195// tolerate re-initializing.
196const resolveMcpSessionIdleTtlMs = (): number | undefined => {
197 const raw = process.env.EXECUTOR_MCP_SESSION_IDLE_TTL_MS;
198 if (!raw) return undefined;
199 const parsed = Number(raw);
200 if (!Number.isFinite(parsed) || parsed < 0) {
201 // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: refuse to boot on a malformed operator knob
202 throw new Error(
203 `EXECUTOR_MCP_SESSION_IDLE_TTL_MS ${JSON.stringify(raw)} is not a non-negative number of milliseconds`,
204 );
205 }
206 return Math.floor(parsed);
207};
208
186// The org slug doubles as a URL segment (`/<slug>/policies`), so an209// The org slug doubles as a URL segment (`/<slug>/policies`), so an
187// operator-set value must fit the shared grammar and avoid reserved root210// operator-set value must fit the shared grammar and avoid reserved root
188// segments (api, mcp, login, …) — a colliding slug would shadow real routes.211// segments (api, mcp, login, …) — a colliding slug would shadow real routes.
apps/host-selfhost/src/mcp/index.ts+14−2
@@ -10,6 +10,7 @@ import type { · 9 unmodified lines
10 10
11import { BetterAuth, type BetterAuthHandle } from "../auth/better-auth";11import { BetterAuth, type BetterAuthHandle } from "../auth/better-auth";
12import type { SelfHostDbHandle } from "../db/self-host-db";12import type { SelfHostDbHandle } from "../db/self-host-db";
13import type { SelfHostConfig } from "../config";
13import { selfHostMcpAuth } from "./auth";14import { selfHostMcpAuth } from "./auth";
14import {15import {
15 makeSelfHostMcpSessionStore,16 makeSelfHostMcpSessionStore,
@@ -128,13 +129,24 @@ const makeApprovalHandler = · 112 unmodified lines
128 * instance provided; it still requires `IdentityProvider` from the resolved129 * instance provided; it still requires `IdentityProvider` from the resolved
129 * identity seam. Returns the three seam Layers plus the `close()` lifetime hook130 * identity seam. Returns the three seam Layers plus the `close()` lifetime hook
130 * the app wires into shutdown.131 * the app wires into shutdown.
132 *
133 * Takes the already-resolved `SelfHostConfig` rather than reading it here: the
134 * app loads it once at boot, and `loadConfig()` refuses to boot on a malformed
135 * operator knob, so calling it from a seam factory would both hide an env read
136 * behind construction and move that failure off the boot path.
131 */137 */
132export const makeSelfHostMcpSeams = (138export const makeSelfHostMcpSeams = (
133 dbHandle: SelfHostDbHandle,139 dbHandle: SelfHostDbHandle,
134 betterAuth: BetterAuthHandle,140 betterAuth: BetterAuthHandle,
135 webBaseUrl?: string,141 config: SelfHostConfig,
136): SelfHostMcpSeams => {142): SelfHostMcpSeams => {
137 const sessionStore = makeSelfHostMcpSessionStore(dbHandle, webBaseUrl);143 // The pinned public origin keeps browser-approval URLs reachable behind a
144 // reverse proxy (not the internal 127.0.0.1 bind from the request URL).
145 const sessionStore = makeSelfHostMcpSessionStore(
146 dbHandle,
147 config.webBaseUrl,
148 config.mcpSessionIdleTtlMs,
149 );
138 const auth: Layer.Layer<McpAuthProvider, never, IdentityProvider> = selfHostMcpAuth.pipe(150 const auth: Layer.Layer<McpAuthProvider, never, IdentityProvider> = selfHostMcpAuth.pipe(
139 Layer.provide(Layer.succeed(BetterAuth)(betterAuth)),151 Layer.provide(Layer.succeed(BetterAuth)(betterAuth)),
140 );152 );
apps/host-selfhost/src/mcp/session-store.ts+5−1
@@ -36,6 +36,7 @@ export { McpEngineBuildError } from "@executor-js/host-mcp/in-memory-session-sto · 35 unmodified lines
36export const makeSelfHostMcpSessionStore = (36export const makeSelfHostMcpSessionStore = (
37 db: SelfHostDbHandle,37 db: SelfHostDbHandle,
38 webBaseUrl?: string,38 webBaseUrl?: string,
39 sessionIdleTtlMs?: number,
39): InMemoryMcpSessionStore =>40): InMemoryMcpSessionStore =>
40 makeInMemoryMcpSessionStore(41 makeInMemoryMcpSessionStore(
41 makeMcpBuildServer(42 makeMcpBuildServer(
@@ -48,7 +49,10 @@ export const makeSelfHostMcpSessionStore = ( · 6 unmodified lines
48 selfHostAnalytics.record(`artifact_${action}`, { via: "agent" }),49 selfHostAnalytics.record(`artifact_${action}`, { via: "agent" }),
49 },50 },
50 ),51 ),
51 { webBaseUrl },52 {
53 ...(webBaseUrl === undefined ? {} : { webBaseUrl }),
54 ...(sessionIdleTtlMs === undefined ? {} : { sessionIdleTtlMs }),
55 },
52 );56 );
53 57
54/** The `McpSessionStore` envelope seam over a freshly built in-process store. */58/** The `McpSessionStore` envelope seam over a freshly built in-process store. */
apps/host-selfhost/src/serve.ts+16−2
@@ -109,7 +109,7 @@ const selfHostHttpMiddleware = (betterAuth: BetterAuthHandle) => · 108 unmodified lines
109 109
110export const startServer = async (): Promise<void> => {110export const startServer = async (): Promise<void> => {
111 const config = loadConfig();111 const config = loadConfig();
112 const { AppLayer, betterAuth } = await makeSelfHostApp();112 const { AppLayer, betterAuth, closeDb } = await makeSelfHostApp();
113 113
114 // Serve the built SPA, split by cacheability so a redeploy is picked up at114 // Serve the built SPA, split by cacheability so a redeploy is picked up at
115 // once instead of stranding browsers on a stale shell:115 // once instead of stranding browsers on a stale shell:
@@ -141,6 +141,16 @@ export const startServer = async (): Promise<void> => { · 25 unmodified lines
141 Effect.addFinalizer(() => Effect.promise(() => disposeAnalytics())),141 Effect.addFinalizer(() => Effect.promise(() => disposeAnalytics())),
142 );142 );
143 143
144 // Server-scope finalizer: release what the app opened at boot — every live
145 // MCP session (transport, server, and its execution engine, whose detached
146 // sandbox fibers keep querying the DB until the engine is shut down) and then
147 // the shared libSQL handle itself. `makeSelfHostApiHandler` runs this in its
148 // `dispose`, so tests have always released it; the long-lived server dropped
149 // the hook on the floor and leaked both across a graceful shutdown.
150 const AppResourcesLive = Layer.effectDiscard(
151 Effect.addFinalizer(() => Effect.promise(() => closeDb())),
152 );
153
144 // OTLP export, or `Layer.empty` when no collector is configured (see154 // OTLP export, or `Layer.empty` when no collector is configured (see
145 // ./telemetry). The `http.server` envelope span each request's `withSpan`155 // ./telemetry). The `http.server` envelope span each request's `withSpan`
146 // children parent under is NOT wired here: `HttpEffect.toHandled` already156 // children parent under is NOT wired here: `HttpEffect.toHandled` already
@@ -162,7 +172,11 @@ export const startServer = async (): Promise<void> => { · 15 unmodified lines
162 // in scope while the app's layers build, or spans created during construction172 // in scope while the app's layers build, or spans created during construction
163 // resolve the default no-op tracer and are silently dropped.173 // resolve the default no-op tracer and are silently dropped.
164 await BunRuntime.runMain(174 await BunRuntime.runMain(
165 Layer.launch(Layer.merge(ServerLive, AnalyticsFlushLive).pipe(Layer.provide(TelemetryLive))),175 Layer.launch(
176 Layer.mergeAll(ServerLive, AnalyticsFlushLive, AppResourcesLive).pipe(
177 Layer.provide(TelemetryLive),
178 ),
179 ),
166 );180 );
167};181};
168 182
e2e/selfhost/mcp-session-idle-eviction.test.ts+172−0
@@ -0,0 +1,172 @@
1// Selfhost-only: an MCP session that goes idle past the store's TTL is
2// reclaimed, and the next request on that session id gets the 404 / -32001 cue
3// that tells a client to re-initialize.
4//
5// Why this scenario boots its OWN instance instead of using the shared one:
6// the idle window is a BOOT-TIME operator knob
7// (EXECUTOR_MCP_SESSION_IDLE_TTL_MS), so there is no way to shrink it on a
8// running server. Waiting out the 30-minute default is not an option, and
9// shrinking it on the SHARED instance would silently evict sessions underneath
10// every other selfhost scenario. A dedicated instance on its own port and data
11// dir keeps the short window contained to this file.
12//
13// Why the wait is a single silent sleep rather than a poll loop: every request
14// the store forwards restamps that session's last-seen time. A poll loop is
15// itself the traffic that keeps the session alive, so it could never observe an
16// eviction. The scenario stays completely silent for one window, then makes
17// exactly one request.
18//
19// Auth is the Better Auth session cookie, not an OAuth bearer: self-host's MCP
20// auth provider accepts the cookie/api-key identity path, and the credential is
21// not what is under test here — session lifetime is.
22import { mkdtempSync, rmSync } from "node:fs";
23import { tmpdir } from "node:os";
24import { join } from "node:path";
25
26import { expect } from "@effect/vitest";
27import { Effect } from "effect";
28
29import { scenario } from "../src/scenario";
30import { RunDir, Target } from "../src/services";
31import { claimAndBoot } from "../src/ports";
32import { isBootReadinessTimeout } from "../setup/boot";
33import { bootSelfhost } from "../setup/selfhost.boot";
34import { SELFHOST_ADMIN, signInSession } from "../targets/selfhost";
35
36/** The idle window this instance runs with — small enough that the sweep, not
37 * the TTL, sets the pace. */
38const IDLE_TTL_MS = 2_000;
39
40/**
41 * How long to leave the session untouched. The store floors its sweep interval
42 * at 30s, so an idle session is disposed at the first tick that falls at least
43 * one TTL after its last request. Waiting two full sweep intervals plus the TTL
44 * means the assertion can never race a tick that has not fired yet.
45 */
46const QUIET_WINDOW_MS = 2 * 30_000 + IDLE_TTL_MS + 5_000;
47
48interface JsonRpcErrorBody {
49 readonly error?: { readonly code?: number; readonly message?: string };
50}
51
52scenario(
53 "MCP · an idle self-host session is evicted and answers 404 -32001 until the client re-initializes",
54 // Own vite dev boot (cold on a fresh checkout) plus a 67s quiet window, so
55 // this needs materially more than the project's 180s default.
56 { timeout: 420_000 },
57 Effect.gen(function* () {
58 // Selfhost-shaped scenario: yielded for the target name in failures, and so
59 // the file reads like its neighbours.
60 yield* Target;
61 const runDir = yield* RunDir;
62
63 const dataDir = mkdtempSync(join(tmpdir(), "executor-selfhost-idle-ttl-"));
64
65 // A distinct env var (not E2E_SELFHOST_PORT, which the shared instance has
66 // already published into this worker's env) so the claim actually probes
67 // and locks a free port instead of returning the shared one.
68 const booted = yield* Effect.promise(() =>
69 claimAndBoot(
70 [{ envVar: "E2E_SELFHOST_IDLE_TTL_PORT", offset: 6, label: "selfhost idle-ttl vite dev" }],
71 async (ports) => {
72 const port = ports.E2E_SELFHOST_IDLE_TTL_PORT!;
73 const baseUrl = `http://localhost:${port}`;
74 const procs = await bootSelfhost({
75 port,
76 webBaseUrl: baseUrl,
77 admin: SELFHOST_ADMIN,
78 dataDir,
79 logFile: join(runDir, "idle-ttl-boot.log"),
80 mcpSessionIdleTtlMs: IDLE_TTL_MS,
81 });
82 return { teardown: procs.teardown, value: baseUrl };
83 },
84 { label: "selfhost idle-ttl", retryWhen: isBootReadinessTimeout },
85 ),
86 );
87
88 yield* Effect.gen(function* () {
89 const baseUrl = booted.value;
90 const mcpUrl = new URL("/mcp", baseUrl).toString();
91 const { cookieHeader } = yield* Effect.promise(() => signInSession(baseUrl, SELFHOST_ADMIN));
92
93 const initialize = async (): Promise<Response> =>
94 fetch(mcpUrl, {
95 method: "POST",
96 headers: {
97 cookie: cookieHeader,
98 "content-type": "application/json",
99 accept: "application/json, text/event-stream",
100 },
101 body: JSON.stringify({
102 jsonrpc: "2.0",
103 id: 1,
104 method: "initialize",
105 params: {
106 protocolVersion: "2025-06-18",
107 capabilities: {},
108 clientInfo: { name: "idle-ttl-e2e", version: "1" },
109 },
110 }),
111 });
112
113 const listTools = async (sessionId: string, id: number): Promise<Response> =>
114 fetch(mcpUrl, {
115 method: "POST",
116 headers: {
117 cookie: cookieHeader,
118 "mcp-session-id": sessionId,
119 "mcp-protocol-version": "2025-06-18",
120 "content-type": "application/json",
121 accept: "application/json, text/event-stream",
122 },
123 body: JSON.stringify({ jsonrpc: "2.0", id, method: "tools/list" }),
124 });
125
126 // 1. A client initializes and the session serves.
127 const opened = yield* Effect.promise(initialize);
128 expect(opened.status, "initialize succeeds").toBe(200);
129 const sessionId = opened.headers.get("mcp-session-id");
130 expect(sessionId, "initialize returns a session id").toEqual(expect.any(String));
131
132 const working = yield* Effect.promise(() => listTools(sessionId!, 2));
133 expect(working.status, "the fresh session serves a request").toBe(200);
134
135 // 2. Idleness, driven by the clock and nothing else. Touching the session
136 // here — even to poll — would restamp it and defeat the measurement.
137 yield* Effect.sleep(`${QUIET_WINDOW_MS} millis`);
138
139 // 3. The evicted id is gone, and says so in the shape a client acts on:
140 // 404 tells it the id is dead, -32001 is the session-lifecycle code.
141 const afterIdle = yield* Effect.promise(() => listTools(sessionId!, 3));
142 const afterIdleBody = yield* Effect.promise(() => afterIdle.text());
143 // A session that was NOT evicted answers with the whole tool catalog, so
144 // the diagnostic is truncated — the status is the assertion, and a full
145 // tools/list dump in the failure output helps nobody.
146 expect(
147 afterIdle.status,
148 `an idle session is evicted, so its id 404s; body starts: ${afterIdleBody.slice(0, 200)}`,
149 ).toBe(404);
150 // 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
151 const parsed = JSON.parse(afterIdleBody) as JsonRpcErrorBody;
152 expect(parsed.error?.code, "the 404 carries the session-lifecycle code").toBe(-32001);
153
154 // 4. The cue is actionable: re-initializing gets a NEW, working session.
155 const reopened = yield* Effect.promise(initialize);
156 expect(reopened.status, "the client can re-initialize after eviction").toBe(200);
157 const newSessionId = reopened.headers.get("mcp-session-id");
158 expect(newSessionId, "re-initialize returns a session id").toEqual(expect.any(String));
159 expect(newSessionId, "re-initialize issues a different session").not.toBe(sessionId);
160
161 const afterReinit = yield* Effect.promise(() => listTools(newSessionId!, 4));
162 expect(afterReinit.status, "the re-initialized session serves a request").toBe(200);
163 }).pipe(
164 Effect.ensuring(
165 Effect.promise(async () => {
166 await booted.teardown();
167 rmSync(dataDir, { recursive: true, force: true });
168 }),
169 ),
170 );
171 }),
172);
e2e/setup/selfhost.boot.ts+7−0
@@ -24,6 +24,10 @@ export interface SelfhostBootOptions { · 23 unmodified lines
24 /** Shrink the sandbox execution budget (EXECUTOR_SANDBOX_TIMEOUT_MS) so24 /** Shrink the sandbox execution budget (EXECUTOR_SANDBOX_TIMEOUT_MS) so
25 * deadline scenarios prove their race in seconds. Omit for production. */25 * deadline scenarios prove their race in seconds. Omit for production. */
26 readonly sandboxTimeoutMs?: number;26 readonly sandboxTimeoutMs?: number;
27 /** Shrink the MCP session idle window (EXECUTOR_MCP_SESSION_IDLE_TTL_MS) so
28 * the eviction scenario proves in seconds what otherwise takes 30 minutes.
29 * Omit for production; the app then uses the store's own default. */
30 readonly mcpSessionIdleTtlMs?: number;
27}31}
28 32
29export const bootSelfhost = async (options: SelfhostBootOptions): Promise<BootedProcesses> => {33export const bootSelfhost = async (options: SelfhostBootOptions): Promise<BootedProcesses> => {
@@ -57,6 +61,9 @@ export const bootSelfhost = async (options: SelfhostBootOptions): Promise<Booted · 27 unmodified lines
57 ...(options.sandboxTimeoutMs !== undefined61 ...(options.sandboxTimeoutMs !== undefined
58 ? { EXECUTOR_SANDBOX_TIMEOUT_MS: String(options.sandboxTimeoutMs) }62 ? { EXECUTOR_SANDBOX_TIMEOUT_MS: String(options.sandboxTimeoutMs) }
59 : {}),63 : {}),
64 ...(options.mcpSessionIdleTtlMs !== undefined
65 ? { EXECUTOR_MCP_SESSION_IDLE_TTL_MS: String(options.mcpSessionIdleTtlMs) }
66 : {}),
60 },67 },
61 logFile: options.logFile,68 logFile: options.logFile,
62 },69 },
packages/hosts/mcp/src/in-memory-session-store.test.ts+222−0
@@ -57,6 +57,228 @@ it("preserves native elicitation mode when creating an in-memory MCP session", a · 56 unmodified lines
57 expect(buildOptions?.elicitationMode).toEqual({ mode: "native" });57 expect(buildOptions?.elicitationMode).toEqual({ mode: "native" });
58});58});
59 59
60/** A do-nothing engine: the eviction test drives session lifetime, not tools. */
61const makeIdleTestEngine = (): ExecutionEngine => ({
62 execute: () => Effect.succeed({ result: "unused" }),
63 executeWithPause: () => Effect.succeed({ status: "completed", result: { result: "unused" } }),
64 resume: () => Effect.succeed(null),
65 getPausedExecution: () => Effect.succeed(null),
66 pausedExecutionCount: () => Effect.succeed(0),
67 hasPausedExecutions: () => Effect.succeed(false),
68 getDescription: Effect.succeed("idle-eviction test executor"),
69 shutdown: Effect.void,
70});
71
72/**
73 * An engine whose `execute` parks until the test releases it, so a request can
74 * be held inside `transport.handleRequest` while the sweep runs. `shutdowns`
75 * counts `engine.shutdown` runs — the disposal step that ends the detached
76 * sandbox fibers, and which dropping the engine reference does not do.
77 */
78const makeLatchedTestEngine = (): {
79 readonly engine: ExecutionEngine;
80 readonly started: Promise<void>;
81 readonly release: () => void;
82 readonly shutdowns: () => number;
83} => {
84 let signalStarted: () => void = () => {};
85 const started = new Promise<void>((resolve) => {
86 signalStarted = resolve;
87 });
88 let openGate: () => void = () => {};
89 const gate = new Promise<void>((resolve) => {
90 openGate = resolve;
91 });
92 let shutdowns = 0;
93 const park = <A>(value: A): Effect.Effect<A> =>
94 Effect.promise(async () => {
95 signalStarted();
96 await gate;
97 return value;
98 });
99 const engine: ExecutionEngine = {
100 ...makeIdleTestEngine(),
101 execute: () => park({ result: "released" }),
102 executeWithPause: () => park({ status: "completed", result: { result: "released" } }),
103 shutdown: Effect.sync(() => {
104 shutdowns += 1;
105 }),
106 };
107 return { engine, started, release: () => openGate(), shutdowns: () => shutdowns };
108};
109
110// A long TTL keeps the sweep's own timer out of the way; the assertions drive
111// `sweepIdleSessions` directly with an explicit instant instead of sleeping
112// through a real window, so the test is deterministic rather than timing-raced.
113const IDLE_TTL_MS = 60_000;
114
115type TestSessionStore = ReturnType<typeof makeInMemoryMcpSessionStore>;
116
117/** Open a session on `sessions` and return its minted id. */
118const openSession = async (sessions: TestSessionStore): Promise<string> => {
119 const response = (await Effect.runPromise(
120 sessions.store.dispatch({
121 request: new Request("https://executor.test/mcp", {
122 method: "POST",
123 headers: {
124 "content-type": "application/json",
125 accept: "application/json, text/event-stream",
126 },
127 body: JSON.stringify({
128 jsonrpc: "2.0",
129 id: 1,
130 method: "initialize",
131 params: {
132 protocolVersion: "2025-06-18",
133 capabilities: {},
134 clientInfo: { name: "idle-test", version: "1.0.0" },
135 },
136 }),
137 }),
138 principal: TEST_PRINCIPAL,
139 resource: defaultMcpResource,
140 sessionId: null,
141 method: "POST",
142 }),
143 )) as Response;
144 expect(response.status).toBe(200);
145 const sessionId = response.headers.get("mcp-session-id") ?? "";
146 expect(sessionId).not.toBe("");
147 return sessionId;
148};
149
150it("evicts a session that goes idle past the TTL and keeps a busy one", async () => {
151 const engine = makeIdleTestEngine();
152 const sessions = makeInMemoryMcpSessionStore(
153 () =>
154 createExecutorMcpServer({ engine }).pipe(Effect.map((mcpServer) => ({ mcpServer, engine }))),
155 { sessionIdleTtlMs: IDLE_TTL_MS },
156 );
157
158 const open = (): Promise<string> => openSession(sessions);
159
160 const call = (sessionId: string, id: number) =>
161 Effect.runPromise(
162 sessions.store.dispatch({
163 request: new Request("https://executor.test/mcp", {
164 method: "POST",
165 headers: {
166 "content-type": "application/json",
167 accept: "application/json, text/event-stream",
168 "mcp-session-id": sessionId,
169 },
170 body: JSON.stringify({ jsonrpc: "2.0", id, method: "tools/list" }),
171 }),
172 principal: TEST_PRINCIPAL,
173 resource: defaultMcpResource,
174 sessionId,
175 method: "POST",
176 }),
177 );
178
179 // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: always close the store
180 try {
181 const idle = await open();
182 const busy = await open();
183 expect(sessions.sessionCount()).toBe(2);
184
185 // Neither is stale yet, so a sweep at the current instant takes nothing.
186 expect(await sessions.sweepIdleSessions()).toBe(0);
187 expect(sessions.sessionCount()).toBe(2);
188
189 // Let the wall clock advance so the two sessions' stamps are separable,
190 // then keep working on one of them: `forward` restamps that one and only
191 // that one.
192 await new Promise((resolve) => setTimeout(resolve, 25));
193 const restampedAt = Date.now();
194 await call(busy, 2);
195
196 // Sweep one TTL after the restamp, less a millisecond: `busy` was stamped
197 // at or after `restampedAt` so it cannot have aged a full TTL, while `idle`
198 // was stamped at least 25ms earlier and must have. Exactly one goes.
199 expect(await sessions.sweepIdleSessions(restampedAt + IDLE_TTL_MS - 1)).toBe(1);
200 expect(sessions.sessionCount()).toBe(1);
201
202 // The evicted id is gone; the store reports it the way the envelope 404s.
203 expect(await call(idle, 3)).toBe("not-found");
204 // The busy one still serves.
205 expect(await call(busy, 4)).toBeInstanceOf(Response);
206 } finally {
207 await sessions.close();
208 }
209});
210
211it("never evicts a session while one of its requests is still in flight", async () => {
212 const latched = makeLatchedTestEngine();
213 const sessions = makeInMemoryMcpSessionStore(
214 () =>
215 createExecutorMcpServer({ engine: latched.engine }).pipe(
216 Effect.map((mcpServer) => ({ mcpServer, engine: latched.engine })),
217 ),
218 { sessionIdleTtlMs: IDLE_TTL_MS },
219 );
220
221 const callExecute = (sessionId: string): Promise<unknown> =>
222 Effect.runPromise(
223 sessions.store.dispatch({
224 request: new Request("https://executor.test/mcp", {
225 method: "POST",
226 headers: {
227 "content-type": "application/json",
228 accept: "application/json, text/event-stream",
229 "mcp-session-id": sessionId,
230 },
231 body: JSON.stringify({
232 jsonrpc: "2.0",
233 id: 2,
234 method: "tools/call",
235 params: { name: "execute", arguments: { code: "return 1" } },
236 }),
237 }),
238 principal: TEST_PRINCIPAL,
239 resource: defaultMcpResource,
240 sessionId,
241 method: "POST",
242 }),
243 );
244
245 // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: always release the latch and close the store
246 try {
247 const sessionId = await openSession(sessions);
248
249 // Start a call and park it inside the engine. `forward` stamps last-seen
250 // BEFORE it awaits the transport, so from here on the stamp only ages — a
251 // request slower than the TTL is indistinguishable from an abandoned
252 // session unless the store also counts what is in flight.
253 const startedAt = Date.now();
254 const inFlight = callExecute(sessionId);
255 await latched.started;
256
257 // Sweep a full TTL past the moment the call began. Without the in-flight
258 // counter this evicts the session and closes the transport, the server, and
259 // the engine underneath the request that is still using them.
260 expect(await sessions.sweepIdleSessions(startedAt + IDLE_TTL_MS)).toBe(0);
261 expect(sessions.sessionCount()).toBe(1);
262 expect(latched.shutdowns()).toBe(0);
263
264 // The parked request still completes, on the transport it started on.
265 latched.release();
266 const response = await inFlight;
267 expect(response).toBeInstanceOf(Response);
268 expect((response as Response).status).toBe(200);
269
270 // And the reprieve is only for the duration of the call: the session is
271 // restamped as it ends, so the next idle window still reclaims it — engine
272 // shutdown included, which is what ends the detached sandbox fibers.
273 expect(await sessions.sweepIdleSessions(Date.now() + IDLE_TTL_MS)).toBe(1);
274 expect(sessions.sessionCount()).toBe(0);
275 expect(latched.shutdowns()).toBe(1);
276 } finally {
277 latched.release();
278 await sessions.close();
279 }
280});
281
60// ---------------------------------------------------------------------------282// ---------------------------------------------------------------------------
61// The pre-initialize guard, through the real store path.283// The pre-initialize guard, through the real store path.
62//284//
packages/hosts/mcp/src/in-memory-session-store.ts+189−12
@@ -53,6 +53,38 @@ import type { BrowserApprovalStore } from "./tool-server"; · 52 unmodified lines
53// - "forbidden" (session owned by another bearer) -> envelope renders 403 -3200353// - "forbidden" (session owned by another bearer) -> envelope renders 403 -32003
54// ---------------------------------------------------------------------------54// ---------------------------------------------------------------------------
55 55
56// A streamable-HTTP session only leaves these maps when the client sends
57// `DELETE /mcp`, and nothing sends it: `StreamableHTTPClientTransport.close()`
58// aborts locally and puts nothing on the wire (only `terminateSession()` sends
59// the DELETE, and `Client.close()` does not call it), and a client that crashes
60// or is killed cannot send it at all. Without a sweep, one abandoned session
61// pins its `McpServer`, its tool registry, and its `ExecutionEngine` for the
62// lifetime of the process.
63//
64// The standalone SSE stream is NOT a substitute teardown signal, and it is not
65// absent either. `enableJsonResponse` governs only how a POST carrying requests
66// answers; a POST carrying just the `notifications/initialized` notification
67// still gets a bare 202, which is exactly the cue the client SDK uses to open
68// the long-lived `GET /mcp` stream. So essentially every session holds an open
69// server-to-client stream for its whole life. That stream is silent by design
70// (it exists for server-initiated messages) and this transport does no max-age
71// rotation, so it produces no recurring request to stamp against — an open
72// stream tells us the socket is up, never that the peer is still working.
73//
74// So the store treats a session as abandoned once it has gone `idleTtlMs`
75// without a REQUEST and disposes it, open stream or not. That mirrors cloud's
76// `decideSessionAlarm`, where an active stream extends the lease only up to
77// `MAX_RUNNING_SESSION_IDLE_MS` and the session is then destroyed regardless;
78// the default here is that same order of ceiling. It is also what the
79// streamable-HTTP spec allows a server to do: a request carrying an evicted id
80// gets the store's existing "not-found" (404, -32001), the client's cue to
81// re-initialize. The cost is bounded and visible — a connected-but-quiet client
82// loses its stream at the ceiling and re-initializes on its next call.
83/** Idle window after which an untouched session is evicted. */
84const DEFAULT_SESSION_IDLE_TTL_MS = 30 * 60 * 1000;
85/** Floor on the sweep interval, so a small TTL cannot spin the timer. */
86const MIN_SWEEP_INTERVAL_MS = 30 * 1000;
87
56/** Engine construction failed for a principal. The store surfaces it as a 500. */88/** Engine construction failed for a principal. The store surfaces it as a 500. */
57export class McpEngineBuildError extends Data.TaggedError("McpEngineBuildError")<{89export class McpEngineBuildError extends Data.TaggedError("McpEngineBuildError")<{
58 readonly cause: unknown;90 readonly cause: unknown;
@@ -108,19 +140,55 @@ export interface InMemoryMcpSessionStore { · 49 unmodified lines
108 request: Request,140 request: Request,
109 principal?: Principal,141 principal?: Principal,
110 ) => Promise<Response | null>;142 ) => Promise<Response | null>;
143 /** Number of live initialized sessions currently owned by this store. */
144 readonly sessionCount: () => number;
145 /**
146 * Dispose every session idle past the store's TTL and return how many went.
147 * Runs on a timer; exposed so a host (or a test) can drive it directly.
148 */
149 readonly sweepIdleSessions: (now?: number) => Promise<number>;
111 /** Dispose every live session — wire into the host's shutdown (not a seam). */150 /** Dispose every live session — wire into the host's shutdown (not a seam). */
112 readonly close: () => Promise<void>;151 readonly close: () => Promise<void>;
113}152}
114 153
115const ignoreClose = (close: (() => Promise<void>) | undefined): Promise<void> =>
116 close
117 ? Effect.runPromise(Effect.ignore(Effect.tryPromise({ try: close, catch: () => undefined })))
118 : Promise.resolve();
119
120const formatBoundaryError = (error: unknown): unknown =>154const formatBoundaryError = (error: unknown): unknown =>
121 // oxlint-disable-next-line executor/no-instanceof-error, executor/no-unknown-error-message -- boundary: log unknown MCP SDK/runtime failures155 // oxlint-disable-next-line executor/no-instanceof-error, executor/no-unknown-error-message -- boundary: log unknown MCP SDK/runtime failures
122 error instanceof Error ? (error.stack ?? error.message) : error;156 error instanceof Error ? (error.stack ?? error.message) : error;
123 157
158/** One session handle refused to close. Reported, never propagated. */
159class McpHandleCloseError extends Data.TaggedError("McpHandleCloseError")<{
160 readonly cause: unknown;
161}> {}
162
163/**
164 * Release one session handle, best effort. Disposal must finish even when a
165 * handle refuses to close — the other handles still have to go, and a rejection
166 * here would surface as an unhandled rejection on a sweep tick nobody awaits.
167 * But a silently swallowed failure is a leaked transport, server, or engine that
168 * nothing can see, so name the handle and the session in a warning.
169 */
170const ignoreClose = (
171 sessionId: string | null,
172 handle: string,
173 close: (() => Promise<void>) | undefined,
174): Promise<void> => {
175 if (!close) return Promise.resolve();
176 const warn = (detail: unknown): Effect.Effect<void> =>
177 Effect.sync(() => {
178 console.warn(
179 `[mcp] failed to close ${handle} for session ${sessionId ?? "<uninitialized>"}:`,
180 formatBoundaryError(detail),
181 );
182 });
183 return Effect.runPromise(
184 Effect.tryPromise({ try: close, catch: (cause) => new McpHandleCloseError({ cause }) }).pipe(
185 Effect.catch((error) => warn(error.cause)),
186 // A defect cannot escape either: this runs detached from any request.
187 Effect.catchCause((cause) => warn(Cause.squash(cause))),
188 ),
189 );
190};
191
124// The store's error bodies are INNER responses (no CORS): the serving envelope192// The store's error bodies are INNER responses (no CORS): the serving envelope
125// re-wraps the store `Response` with CORS before it leaves the origin, so the193// re-wraps the store `Response` with CORS before it leaves the origin, so the
126// canonical renderer is called with `cors: false` (content-type only).194// canonical renderer is called with `cors: false` (content-type only).
@@ -159,23 +227,83 @@ export const makeInMemoryMcpSessionStore = ( · 32 unmodified lines
159 // proxy) it is preferred over the request URL — whose host would be the227 // proxy) it is preferred over the request URL — whose host would be the
160 // internal bind address (127.0.0.1:PORT), unreachable for the user. Omit it on228 // internal bind address (127.0.0.1:PORT), unreachable for the user. Omit it on
161 // loopback hosts (local/desktop), where the request URL is already correct.229 // loopback hosts (local/desktop), where the request URL is already correct.
162 options: { readonly webBaseUrl?: string } = {},230 options: {
231 readonly webBaseUrl?: string;
232 /** Idle window before a session is evicted. 0 disables eviction. */
233 readonly sessionIdleTtlMs?: number;
234 /** How often the sweep runs. Defaults to a quarter of the TTL. */
235 readonly sessionSweepIntervalMs?: number;
236 } = {},
163): InMemoryMcpSessionStore => {237): InMemoryMcpSessionStore => {
164 const transports = new Map<string, WebStandardStreamableHTTPServerTransport>();238 const transports = new Map<string, WebStandardStreamableHTTPServerTransport>();
165 const servers = new Map<string, McpServer>();239 const servers = new Map<string, McpServer>();
166 const owners = new Map<string, SessionOwner>();240 const owners = new Map<string, SessionOwner>();
167 const engines = new Map<string, ExecutionEngine<Cause.YieldableError>>();241 const engines = new Map<string, ExecutionEngine<Cause.YieldableError>>();
168 const approvals: InProcessBrowserApprovalStore = makeInProcessBrowserApprovalStore();242 const approvals: InProcessBrowserApprovalStore = makeInProcessBrowserApprovalStore();
243 // Monotonic-ish last-touch stamp per live session, the first input the idle
244 // sweep reads. Written on create and on every forwarded request.
245 const lastSeen = new Map<string, number>();
246 // Requests currently inside `transport.handleRequest` for a session, the
247 // sweep's second input. A stamp alone cannot describe a long call: it is
248 // written BEFORE the await, so a single `execute` that outruns the TTL (a
249 // browser approval waiting on a human, a slow upstream) would look exactly
250 // like an abandoned session and have its transport, server, and engine closed
251 // out from under the request that is still using them. Counting requests in
252 // flight makes "idle" mean what it says.
253 const activeRequests = new Map<string, number>();
254
255 const idleTtlMs = options.sessionIdleTtlMs ?? DEFAULT_SESSION_IDLE_TTL_MS;
256 const sweepIntervalMs =
257 options.sessionSweepIntervalMs ?? Math.max(MIN_SWEEP_INTERVAL_MS, Math.floor(idleTtlMs / 4));
258
259 const touch = (id: string): void => {
260 if (lastSeen.has(id)) lastSeen.set(id, Date.now());
261 };
262
263 /** Claim a session for one in-flight request, so the sweep cannot take it. */
264 const beginRequest = (id: string): void => {
265 activeRequests.set(id, (activeRequests.get(id) ?? 0) + 1);
266 };
267
268 /**
269 * Release the claim and restamp: a call that ran for an hour leaves the
270 * session idle from the moment it FINISHED, not from the moment it started.
271 * `touch` is a no-op once the session is gone, so this can never resurrect a
272 * disposed id.
273 */
274 const endRequest = (id: string): void => {
275 const remaining = (activeRequests.get(id) ?? 1) - 1;
276 if (remaining > 0) activeRequests.set(id, remaining);
277 else activeRequests.delete(id);
278 touch(id);
279 };
280
281 /**
282 * Shut down a session's engine. Dropping the reference is not enough: the
283 * engine's paused executions hold detached sandbox fibers that keep running —
284 * and keep querying the host's database handle — until `shutdown` interrupts
285 * them. Every disposal path goes through here.
286 */
287 const shutdownEngine = (
288 id: string | null,
289 engine: ExecutionEngine<Cause.YieldableError> | undefined,
290 ): Promise<void> =>
291 ignoreClose(id, "engine", engine ? () => Effect.runPromise(engine.shutdown) : undefined);
169 292
170 const dispose = async (id: string, opts: { transport?: boolean; server?: boolean } = {}) => {293 const dispose = async (id: string, opts: { transport?: boolean; server?: boolean } = {}) => {
171 const transport = transports.get(id);294 const transport = transports.get(id);
172 const server = servers.get(id);295 const server = servers.get(id);
296 const engine = engines.get(id);
173 transports.delete(id);297 transports.delete(id);
174 servers.delete(id);298 servers.delete(id);
175 owners.delete(id);299 owners.delete(id);
176 engines.delete(id);300 engines.delete(id);
177 if (opts.transport) await ignoreClose(transport ? () => transport.close() : undefined);301 lastSeen.delete(id);
178 if (opts.server) await ignoreClose(server ? () => server.close() : undefined);302 activeRequests.delete(id);
303 if (opts.transport)
304 await ignoreClose(id, "transport", transport ? () => transport.close() : undefined);
305 if (opts.server) await ignoreClose(id, "server", server ? () => server.close() : undefined);
306 await shutdownEngine(id, engine);
179 };307 };
180 308
181 /**309 /**
@@ -214,7 +342,15 @@ export const makeInMemoryMcpSessionStore = ( · 32 unmodified lines
214 const owner = owners.get(sessionId);342 const owner = owners.get(sessionId);
215 if (!transport || !owner) return Effect.succeed("not-found");343 if (!transport || !owner) return Effect.succeed("not-found");
216 if (!sessionOwnerMatches(owner, principal, resource)) return Effect.succeed("forbidden");344 if (!sessionOwnerMatches(owner, principal, resource)) return Effect.succeed("forbidden");
217 return runHandleRequest(transport, request);345 touch(sessionId);
346 // Claim before the await, release in the finalizer — `runHandleRequest`
347 // already recovers every failure to a 500, but `ensuring` also covers an
348 // interrupt, so the counter cannot be left permanently raised (which would
349 // make the session immortal, the opposite leak).
350 beginRequest(sessionId);
351 return runHandleRequest(transport, request).pipe(
352 Effect.ensuring(Effect.sync(() => endRequest(sessionId))),
353 );
218 };354 };
219 355
220 /**356 /**
@@ -273,6 +409,7 @@ export const makeInMemoryMcpSessionStore = ( · 52 unmodified lines
273 servers.set(sid, mcpServer);409 servers.set(sid, mcpServer);
274 owners.set(sid, { principal, resource });410 owners.set(sid, { principal, resource });
275 engines.set(sid, engine);411 engines.set(sid, engine);
412 lastSeen.set(sid, Date.now());
276 },413 },
277 onsessionclosed: (sid) => void dispose(sid, { server: true }),414 onsessionclosed: (sid) => void dispose(sid, { server: true }),
278 });415 });
@@ -284,8 +421,12 @@ export const makeInMemoryMcpSessionStore = ( · 5 unmodified lines
284 // The session id is minted on the first (initialize) request, so we421 // The session id is minted on the first (initialize) request, so we
285 // drive `handleRequest` here; if no id results we close eagerly.422 // drive `handleRequest` here; if no id results we close eagerly.
286 return yield* runHandleRequest(transport, request, () => {423 return yield* runHandleRequest(transport, request, () => {
287 void ignoreClose(() => transport.close());424 // Nothing was ever registered under a session id, so `dispose` has
288 void ignoreClose(() => mcpServer.close());425 // no entry to work from — release the three handles by hand, engine
426 // included.
427 void ignoreClose(null, "transport", () => transport.close());
428 void ignoreClose(null, "server", () => mcpServer.close());
429 void shutdownEngine(null, engine);
289 });430 });
290 }),431 }),
291 ),432 ),
@@ -399,12 +540,48 @@ export const makeInMemoryMcpSessionStore = ( · 107 unmodified lines
399 });540 });
400 };541 };
401 542
543 /**
544 * Dispose every session whose last request is older than the idle window AND
545 * which has nothing in flight. A session serving a request is busy, however
546 * long ago that request started; it gets a fresh stamp the moment it ends, so
547 * a later sweep still reclaims it if the client then goes quiet.
548 */
549 const sweepIdleSessions = async (now: number = Date.now()): Promise<number> => {
550 if (idleTtlMs <= 0) return 0;
551 const stale = [...lastSeen.entries()]
552 .filter(([id, seen]) => now - seen >= idleTtlMs && (activeRequests.get(id) ?? 0) === 0)
553 .map(([id]) => id);
554 // Both flags: an evicted session's transport has no other owner, and leaving
555 // it open would keep the very handles the eviction exists to release.
556 await Promise.all(stale.map((id) => dispose(id, { transport: true, server: true })));
557 return stale.length;
558 };
559
560 // `unref` so the sweep never keeps a host process alive on its own. Node and
561 // Bun both return a Timeout with it; the DOM typing does not, hence the guard.
562 const sweepTimer: ReturnType<typeof setInterval> | undefined =
563 idleTtlMs > 0
564 ? setInterval(() => {
565 // Same shape as `ignoreClose`: a sweep failure is not the host's
566 // problem and must never surface as an unhandled rejection.
567 void Effect.runPromise(
568 Effect.ignore(
569 Effect.tryPromise({ try: () => sweepIdleSessions(), catch: () => undefined }),
570 ),
571 );
572 }, sweepIntervalMs)
573 : undefined;
574 (sweepTimer as { unref?: () => void } | undefined)?.unref?.();
575
402 return {576 return {
403 store,577 store,
404 handlePausedRequest,578 handlePausedRequest,
405 handleApprovalRequest,579 handleApprovalRequest,
580 sessionCount: () => transports.size,
581 sweepIdleSessions,
406 close: async () => {582 close: async () => {
407 const ids = new Set([...transports.keys(), ...servers.keys()]);583 if (sweepTimer !== undefined) clearInterval(sweepTimer);
584 const ids = new Set([...transports.keys(), ...servers.keys(), ...engines.keys()]);
408 await Promise.all([...ids].map((id) => dispose(id, { transport: true, server: true })));585 await Promise.all([...ids].map((id) => dispose(id, { transport: true, server: true })));
409 },586 },
410 };587 };