From 2550e8be814c6c329a4f8d827c33403b1bfdb092 Mon Sep 17 00:00:00 2001 From: James Brocklehurst <232689185+jamieb72-afk@users.noreply.github.com> Date: Mon, 17 Aug 2026 22:19:34 +1000 Subject: [PATCH] fix(local): reuse database for toolkit MCP sessions --- .changeset/toolkit-mcp-shared-database.md | 5 ++ apps/local/src/executor.ts | 86 +++++++++++++++-------- apps/local/src/main.ts | 4 +- apps/local/src/mcp-modern.test.ts | 51 ++++++++++++++ apps/local/src/mcp.ts | 49 ++++++++----- apps/local/src/serve.test.ts | 25 +++++++ 6 files changed, 169 insertions(+), 51 deletions(-) create mode 100644 .changeset/toolkit-mcp-shared-database.md diff --git a/.changeset/toolkit-mcp-shared-database.md b/.changeset/toolkit-mcp-shared-database.md new file mode 100644 index 0000000000..292c066f71 --- /dev/null +++ b/.changeset/toolkit-mcp-shared-database.md @@ -0,0 +1,5 @@ +--- +"executor": patch +--- + +Fix toolkit MCP routes in the local daemon by reusing its owned SQLite database and sharing one scoped executor per toolkit instead of reacquiring storage and connector processes for every client. diff --git a/apps/local/src/executor.ts b/apps/local/src/executor.ts index dd91838f35..cbb8747c9c 100644 --- a/apps/local/src/executor.ts +++ b/apps/local/src/executor.ts @@ -11,6 +11,7 @@ import { runSqliteDataMigrations, type AnyPlugin, type Executor, + type FumaDb, } from "@executor-js/sdk"; import { collectTables } from "@executor-js/api/server"; import { loadPluginsFromJsonc } from "@executor-js/config"; @@ -92,6 +93,7 @@ const loadLocalPlugins = (options: LocalExecutorOptions = {}) => interface LocalExecutorBundle { readonly executor: Executor; readonly plugins: LocalPlugins; + readonly db: FumaDb; /** Where this daemon's web UI is reachable, resolved once at boot. Surfaced * so callers building user-facing links (MCP artifact deep links) use the * same origin the executor itself was configured with. */ @@ -142,6 +144,42 @@ const handleOrNull = (promise: ReturnType) => ), ); +const createExecutorBundleForDb = (db: FumaDb, cwd: string, plugins: LocalPlugins) => + Effect.gen(function* () { + const tenantId = makeTenantId(cwd); + // webBaseUrl is where the executor's web UI listens - same port as the + // daemon API since the daemon serves both. Mirrors serve.ts's port + // resolution so a custom $PORT flows through. EXECUTOR_WEB_BASE_URL + // overrides entirely for deployments where the UI is on a different host. + const webBaseUrl = + process.env.EXECUTOR_WEB_BASE_URL ?? `http://localhost:${process.env.PORT ?? "4788"}`; + + const executor = yield* createExecutor({ + tenant: Tenant.make(tenantId), + subject: Subject.make(LOCAL_SUBJECT), + db, + plugins, + onIntegrationChange: (event) => + localAnalytics.record( + event.kind === "added" ? "integration_added" : "integration_removed", + { plugin_key: event.pluginKey }, + ), + onElicitation: "accept-all", + oauthEndpointUrlPolicy: { allowHttp: true }, + // EXPLICIT OAuth callback - the daemon serves the v2 `/api/oauth/callback` + // route on the same origin as the web UI. Derived from `webBaseUrl` + // (loopback localhost is correct + intended for the local CLI, but it + // is wired explicitly here rather than relying on a hidden default). + redirectUri: new URL("/api/oauth/callback", webBaseUrl).toString(), + // Built-in agent-facing tools (integrations / connections / policies). + coreTools: { + webBaseUrl, + }, + }); + + return { executor, plugins, db, webBaseUrl }; + }); + const createLocalExecutorLayer = (options: LocalExecutorOptions = {}) => { const storage = resolveStorage(); @@ -184,35 +222,8 @@ const createLocalExecutorLayer = (options: LocalExecutorOptions = {}) => { ), ); - // webBaseUrl is where the executor's web UI listens — same port as the - // daemon API since the daemon serves both. Mirrors serve.ts's port - // resolution so a custom $PORT flows through. EXECUTOR_WEB_BASE_URL - // overrides entirely for deployments where the UI is on a different host. - const webBaseUrl = - process.env.EXECUTOR_WEB_BASE_URL ?? `http://localhost:${process.env.PORT ?? "4788"}`; - - const executor = yield* createExecutor({ - tenant: Tenant.make(tenantId), - subject: Subject.make(LOCAL_SUBJECT), - db: sqlite.db, - plugins, - onIntegrationChange: (event) => - localAnalytics.record( - event.kind === "added" ? "integration_added" : "integration_removed", - { plugin_key: event.pluginKey }, - ), - onElicitation: "accept-all", - oauthEndpointUrlPolicy: { allowHttp: true }, - // EXPLICIT OAuth callback — the daemon serves the v2 `/api/oauth/callback` - // route on the same origin as the web UI. Derived from `webBaseUrl` - // (loopback localhost is correct + intended for the local CLI, but it - // is wired explicitly here rather than relying on a hidden default). - redirectUri: new URL("/api/oauth/callback", webBaseUrl).toString(), - // Built-in agent-facing tools (integrations / connections / policies). - coreTools: { - webBaseUrl, - }, - }); + const bundle = yield* createExecutorBundleForDb(sqlite.db, cwd, plugins); + const executor = bundle.executor; if (migration.migrated) { console.warn( @@ -243,7 +254,7 @@ const createLocalExecutorLayer = (options: LocalExecutorOptions = {}) => { ); } - return { executor, plugins, webBaseUrl }; + return bundle; }), ); }; @@ -257,6 +268,7 @@ export const createExecutorHandle = async (options: LocalExecutorOptions = {}) = executor: bundle.executor, plugins: bundle.plugins, webBaseUrl: bundle.webBaseUrl, + db: bundle.db, dispose: async () => { await Effect.runPromise(Effect.ignore(bundle.executor.close())); await ignorePromiseFailure("disposeRuntime", () => runtime.dispose()); @@ -310,6 +322,20 @@ const loadSharedHandle = (): Promise => { export const getExecutor = () => loadSharedHandle().then((handle) => handle.executor); export const getExecutorBundle = () => loadSharedHandle(); +export const createScopedExecutorHandle = async (options: LocalExecutorOptions = {}) => { + const shared = await getExecutorBundle(); + const { cwd, plugins } = await Effect.runPromise(loadLocalPlugins(options)); + const bundle = await Effect.runPromise(createExecutorBundleForDb(shared.db, cwd, plugins)); + + return { + executor: bundle.executor, + plugins: bundle.plugins, + dispose: async () => { + await Effect.runPromise(Effect.ignore(bundle.executor.close())); + }, + }; +}; + export const disposeExecutor = async (): Promise => { const currentHandlePromise = sharedHandlePromise; sharedHandlePromise = null; diff --git a/apps/local/src/main.ts b/apps/local/src/main.ts index 14e272f37a..5d92cc8122 100644 --- a/apps/local/src/main.ts +++ b/apps/local/src/main.ts @@ -8,7 +8,7 @@ import { smokeRenderArtifact } from "@executor-js/mcp-apps-shell/smoke-render"; import { makeQuickJsExecutor } from "@executor-js/runtime-quickjs"; import { localAnalytics } from "./analytics"; import { makeLocalApiHandler } from "./app"; -import { createExecutorHandle, disposeExecutor, getExecutorBundle } from "./executor"; +import { createScopedExecutorHandle, disposeExecutor, getExecutorBundle } from "./executor"; import { createMcpRequestHandler, type McpRequestHandler } from "./mcp"; // --------------------------------------------------------------------------- @@ -137,7 +137,7 @@ export const createServerHandlers = async (token: string): Promise { + it("shares a toolkit resource config across clients until handler shutdown", async () => { + let createCalls = 0; + let closeCalls = 0; + const mcp = createMcpRequestHandler({ + defaultConfig: { engine }, + createConfigForResource: () => { + createCalls += 1; + return { + config: { engine }, + close: async () => { + closeCalls += 1; + }, + }; + }, + }); + const connectClient = async () => { + const transport = new StreamableHTTPClientTransport( + new URL("http://local.test/mcp/toolkits/shared-toolkit"), + { + fetch: (input, init) => + mcp.handleRequest( + input instanceof Request + ? new Request(input, init) + : new Request(input.toString(), init), + ), + }, + ); + const client = new Client( + { name: "local-modern-toolkit-test", version: "1.0.0" }, + { capabilities: {}, versionNegotiation: { mode: { pin: "2026-07-28" } } }, + ); + await client.connect(transport); + return client; + }; + + const first = await connectClient(); + const second = await connectClient(); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- test boundary: always close both clients and the shared local handler + try { + await first.close(); + await second.close(); + expect(createCalls).toBe(1); + expect(closeCalls).toBe(0); + } finally { + await first.close(); + await second.close(); + await mcp.close(); + } + expect(closeCalls).toBe(1); + }); + it("discovers, lists tools, and executes without creating a legacy session", async () => { const mcp = createMcpRequestHandler({ engine }); const sessionHeaders: Array = []; diff --git a/apps/local/src/mcp.ts b/apps/local/src/mcp.ts index c7efd3c179..2a9f696fea 100644 --- a/apps/local/src/mcp.ts +++ b/apps/local/src/mcp.ts @@ -140,8 +140,8 @@ export const createMcpRequestHandler = ( const servers = new Map(); const resources = new Map(); const sessionEngines = new Map(); - const sessionClosers = new Map Promise>(); const modernHandlers = new Map(); + const resourceConfigs = new Map>(); const modernRequestBodies = new WeakMap(); const approvals = makeInProcessBrowserApprovalStore(); const defaultEngine = engineFromConfig(handlerConfig.defaultConfig); @@ -165,21 +165,45 @@ export const createMcpRequestHandler = ( const configForResource = async (resource: McpResource): Promise => { if (!handlerConfig.createConfigForResource) return { config: handlerConfig.defaultConfig }; - return handlerConfig.createConfigForResource(resource); + const key = mcpResourceKey(resource); + const cached = resourceConfigs.get(key); + if (cached) return cached; + + const pending = Promise.resolve(handlerConfig.createConfigForResource(resource)); + resourceConfigs.set(key, pending); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: preserve the Promise factory's original rejection while evicting its cache entry + try { + return await pending; + } catch (error) { + if (resourceConfigs.get(key) === pending) resourceConfigs.delete(key); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: callers already map this factory rejection to the MCP protocol error response + throw error; + } + }; + + const closeResourceConfigs = async (): Promise => { + const pending = [...resourceConfigs.values()]; + resourceConfigs.clear(); + const configs = await Promise.all( + pending.map((config) => + config.then( + (value) => value, + () => null, + ), + ), + ); + await Promise.all(configs.map((config) => ignoreClose(config?.close))); }; const dispose = async (id: string, opts: { transport?: boolean; server?: boolean } = {}) => { const t = transports.get(id); const s = servers.get(id); - const close = sessionClosers.get(id); transports.delete(id); servers.delete(id); resources.delete(id); sessionEngines.delete(id); - sessionClosers.delete(id); if (opts.transport) await ignoreClose(t ? () => t.close() : undefined); if (opts.server) await ignoreClose(s ? () => s.close() : undefined); - await ignoreClose(close); }; const modernHandlerFor = (resource: McpResource): McpHttpHandler => { @@ -214,17 +238,6 @@ export const createMcpRequestHandler = ( requestStatePrincipal: "local", ...(requestStateBinding === null ? {} : { requestStateBinding }), }); - if (resourceConfig.close) { - const closeServer = server.close.bind(server); - const closeConfig = resourceConfig.close; - let closed = false; - server.close = async () => { - if (closed) return; - closed = true; - await ignoreClose(closeServer); - await ignoreClose(closeConfig); - }; - } return server; }), ); @@ -280,7 +293,6 @@ export const createMcpRequestHandler = ( resources.set(sid, resource); const engine = resourceConfig ? engineFromConfig(resourceConfig.config) : null; if (engine) sessionEngines.set(sid, engine); - if (resourceConfig?.close) sessionClosers.set(sid, resourceConfig.close); }, onsessionclosed: (sid) => void dispose(sid, { server: true }), }); @@ -320,7 +332,6 @@ export const createMcpRequestHandler = ( await ignoreClose(() => transport.close()); const server = created; await ignoreClose(server ? () => server.close() : undefined); - await ignoreClose(resourceConfig?.close); } return response; } catch (error) { @@ -329,7 +340,6 @@ export const createMcpRequestHandler = ( await ignoreClose(() => transport.close()); const server = created; await ignoreClose(server ? () => server.close() : undefined); - await ignoreClose(resourceConfig?.close); } return jsonError(500, -32603, "Internal server error"); } @@ -370,6 +380,7 @@ export const createMcpRequestHandler = ( ...[...ids].map((id) => dispose(id, { transport: true, server: true })), ...[...modernHandlers.values()].map((handler) => handler.close()), ]); + await closeResourceConfigs(); }, }; }; diff --git a/apps/local/src/serve.test.ts b/apps/local/src/serve.test.ts index b7c7e1fc9b..01589a3464 100644 --- a/apps/local/src/serve.test.ts +++ b/apps/local/src/serve.test.ts @@ -85,6 +85,31 @@ describe("startServer static/SPA routing (unauthenticated)", () => { }); describe("startServer startup cleanup", () => { + it("opens a toolkit MCP session without reacquiring database ownership", async () => { + server = await startServer({ port: 0, clientDir, authToken: TOKEN }); + const response = await fetch(`http://127.0.0.1:${server.port}/mcp/toolkits/missing-toolkit`, { + method: "POST", + headers: { + authorization: `Bearer ${TOKEN}`, + accept: "application/json, text/event-stream", + "content-type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-03-26", + capabilities: {}, + clientInfo: { name: "local-toolkit-test", version: "1.0.0" }, + }, + }), + }); + + expect(response.status).toBe(200); + await response.body?.cancel(); + }); + it("releases the owned DB when a default-handler server stops", async () => { server = await startServer({ port: 0, clientDir, authToken: TOKEN }); await server.stop();