diff --git a/apps/cloud/src/api/protected.test.ts b/apps/cloud/src/api/protected.test.ts index 858f9710f..f5a81f0ae 100644 --- a/apps/cloud/src/api/protected.test.ts +++ b/apps/cloud/src/api/protected.test.ts @@ -20,6 +20,8 @@ const makeBaseEngine = (): ExecutionEngine => pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed("desc"), + // The fake forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, }) as ExecutionEngine; describe("withExecutionUsageTracking", () => { diff --git a/apps/cloud/src/engine/execution-gate.ts b/apps/cloud/src/engine/execution-gate.ts index 82d5f4c2a..60c102663 100644 --- a/apps/cloud/src/engine/execution-gate.ts +++ b/apps/cloud/src/engine/execution-gate.ts @@ -108,6 +108,9 @@ export const withPreExecutionGate = ( pausedExecutionCount: () => engine.pausedExecutionCount(), hasPausedExecutions: () => engine.hasPausedExecutions(), getDescription: engine.getDescription, + // Forwarded, not re-implemented: the wrapped engine owns the sandbox fibers, + // so the host's request-scope teardown has to reach through this decorator. + shutdown: engine.shutdown, }); // --------------------------------------------------------------------------- diff --git a/apps/cloud/src/engine/execution-rate-limit.node.test.ts b/apps/cloud/src/engine/execution-rate-limit.node.test.ts index 678697edc..4cefbdbf8 100644 --- a/apps/cloud/src/engine/execution-rate-limit.node.test.ts +++ b/apps/cloud/src/engine/execution-rate-limit.node.test.ts @@ -31,6 +31,8 @@ const engineStub: ExecutionEngine = { pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed("stub"), + // The stub forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, }; /** Counter that hands out a caller-controlled sequence of counts. */ diff --git a/apps/cloud/src/engine/execution-usage.ts b/apps/cloud/src/engine/execution-usage.ts index 94b85cb39..d5b2bc8e9 100644 --- a/apps/cloud/src/engine/execution-usage.ts +++ b/apps/cloud/src/engine/execution-usage.ts @@ -25,4 +25,7 @@ export const withExecutionUsageTracking = ( pausedExecutionCount: () => engine.pausedExecutionCount(), hasPausedExecutions: () => engine.hasPausedExecutions(), getDescription: engine.getDescription, + // Forwarded, not re-implemented: the wrapped engine owns the sandbox fibers, + // so the host's request-scope teardown has to reach through this decorator. + shutdown: engine.shutdown, }); diff --git a/packages/core/analytics/src/engine.test.ts b/packages/core/analytics/src/engine.test.ts index ebca8c527..d3104fd7b 100644 --- a/packages/core/analytics/src/engine.test.ts +++ b/packages/core/analytics/src/engine.test.ts @@ -60,6 +60,8 @@ const makeFakeEngine = ( pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed("fake"), + // The fake forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, }); describe("withExecutionAnalytics", () => { diff --git a/packages/core/api/src/server/execution-stack-middleware.ts b/packages/core/api/src/server/execution-stack-middleware.ts index b3e6749b8..501f85737 100644 --- a/packages/core/api/src/server/execution-stack-middleware.ts +++ b/packages/core/api/src/server/execution-stack-middleware.ts @@ -259,6 +259,25 @@ export const makeExecutionStackMiddleware = < Effect.provideService(ExecutorService, executor), Effect.provideService(ExecutionEngineService, engine), provideExecutorExtensions(executor), + // This engine belongs to THIS request: the stack above was built + // from the host's request-scoped DB handle, and `executeWithPause` + // forks its sandbox as a daemon that would otherwise outlive the + // handler. The host closes the connection when the request scope + // closes — which happens AFTER this effect returns — so ending the + // engine here is what keeps a sandbox fiber from waking up on a + // closed pool. + // + // Nothing is lost by ending it: on a per-request engine the paused + // fiber is already unreachable once the response is written (a + // resume lands on a different engine and replays the call instead), + // so the fiber could only ever have failed. The MCP session Durable + // Object does NOT come through here — it builds its own stack over a + // session-lifetime handle, so its pauses still survive between + // requests, which is the whole point of that plane. + // + // `ensuring`, not `tap`: interruption and failure have to end the + // fiber too, and it must not run before the response is produced. + Effect.ensuring(engine.shutdown), ); // Provide the boot-captured context; uncaptured deps (cloud's // request-scoped `DbService`) remain residual and flow through here. @@ -315,6 +334,9 @@ const readOnlyExecutionEngine: ExecutionEngine = { getPausedExecution: () => Effect.succeed(null), pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), + // Nothing is ever forked here — the platform branch cannot execute — so there + // is no sandbox fiber to end. + shutdown: Effect.void, // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: only the MCP tool server reads this, and the MCP plane never serves a platform credential getDescription: Effect.die(new PlatformEngineUnavailable({ member: "getDescription" })), }; diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts index 4e08e3188..36a5c8cbf 100644 --- a/packages/core/execution/src/engine.ts +++ b/packages/core/execution/src/engine.ts @@ -484,6 +484,21 @@ export type ExecutionEngine * Get the dynamic tool description (workflow + namespaces). */ readonly getDescription: Effect.Effect; + + /** + * End this engine's in-flight sandbox fibers and wait for them to unwind. + * + * `executeWithPause` forks the sandbox as a daemon so a pause can outlive the + * caller that observed it. That fiber holds the executor, and so the DB handle + * belonging to whichever scope built this engine. A host that builds an engine + * per request MUST call this before that scope's connection is closed, or the + * fiber outlives the pool it queries. + * + * Required, not optional: a decorator that wrapped the engine and quietly + * dropped this member would silently reopen that race, so the type makes + * forwarding it a compile error. + */ + readonly shutdown: Effect.Effect; }; export const createExecutionEngine = ( @@ -491,6 +506,17 @@ export const createExecutionEngine = => { const { executor, codeExecutor, toolDiscoveryProvider = defaultToolDiscoveryProvider } = config; const pausedExecutions = new Map>(); + // Every sandbox fiber `startPausableExecution` still has in flight. + // + // Those fibers are daemons (`Effect.forkDetach`) so a pause can outlive the + // caller that observed it. But they close over `executor`, and the executor + // closes over the FumaDB handle the host opened for whatever scope built THIS + // engine — `makeFumaClient` captures `db` at construction, not per operation. + // A host that builds one engine per HTTP request therefore needs a way to end + // that fiber's life with the request; otherwise it wakes up after the + // request's postgres pool has been closed and every query it makes lands on a + // dead pool. `shutdown` below is that seam. + const liveSandboxFibers = new Set>(); // Outcomes of executions that already settled (resumed to completion, hit a // new pause, or died while paused). MCP clients retry `resume` when a // response gets lost in transit; without this cache the retry of an @@ -613,6 +639,7 @@ export const createExecutionEngine = Effect.sync(() => { + // Settled on its own — it can no longer touch the host's DB handle, + // so it is not `shutdown`'s problem any more. + liveSandboxFibers.delete(sandboxFiber); const outcome = Exit.map( exit, (result): ExecutionResult => ({ status: "completed", result }), @@ -727,10 +757,33 @@ export const createExecutionEngine = = Effect.suspend(() => { + const fibers = Array.from(liveSandboxFibers); + liveSandboxFibers.clear(); + for (const [id, paused] of pausedExecutions) { + if (fibers.includes(paused.fiber)) pausedExecutions.delete(id); + } + return Fiber.interruptAll(fibers); + }); + return { execute: runInlineExecution, executeWithPause: startPausableExecution, resume: resumeExecution, + shutdown, isExecutionSettled: (executionId) => Effect.sync(() => settledExecutionIds.has(executionId)), getPausedExecution: (executionId) => Effect.sync(() => pausedExecutions.get(executionId) ?? null), diff --git a/packages/core/execution/src/tool-invoker.test.ts b/packages/core/execution/src/tool-invoker.test.ts index 536fa31f6..747bd2310 100644 --- a/packages/core/execution/src/tool-invoker.test.ts +++ b/packages/core/execution/src/tool-invoker.test.ts @@ -27,6 +27,7 @@ import { typeCheckOutputTypeScript, } from "@executor-js/sdk/testing"; import { makeQuickJsExecutor } from "@executor-js/runtime-quickjs"; +import type { CodeExecutor, ExecuteResult } from "@executor-js/codemode-core"; import { createExecutionEngine } from "./engine"; import { ExecutionToolError } from "./errors"; import { @@ -1875,4 +1876,104 @@ describe("pause/resume with multiple elicitations", () => { expect(resumed.result.error).toBeUndefined(); expect(resumed.result.result).toMatchObject({ ok: true }); }, 10000); + + // ------------------------------------------------------------------------- + // Regression: a paused execution's sandbox fiber must not outlive the + // database handle it is holding. + // + // `executeWithPause` forks the sandbox with `Effect.forkDetach` so a pause can + // outlive the caller that observed it. What that fiber closes over is the + // problem: its invoker is built from the engine's `executor`, and the FumaDB + // client captures its handle at CONSTRUCTION rather than resolving it per + // operation. So the parked fiber keeps a live reference to whichever + // connection the host opened for the scope that built this engine. + // + // On the cloud `/api/*` plane that scope is the HTTP request, and its + // finalizer closes the postgres pool as soon as the response is written. A + // sandbox fiber still parked at that moment issues its next query against a + // pool that has already had `end()` called on it — which postgres.js answers + // with CONNECTION_ENDED, and which workerd answers with "Cannot perform I/O on + // behalf of a different request" once the socket is reached from a later + // request. Those are the storage faults seen in production on reads like + // `tool.findMany` and `plugin_storage.findMany`. + // + // `engine.shutdown` is the seam the host uses to end the fiber BEFORE closing + // the connection. Nothing is lost by ending it here: on a per-request engine + // the pause is already unreachable once the response is written (a resume + // lands on a different engine and replays the call instead). + // ------------------------------------------------------------------------- + it.effect( + "shutdown ends a paused execution's sandbox fiber so it cannot outlive the request", + () => + Effect.gen(function* () { + const executor = yield* makeElicitingExecutor(); + + // Instrumented in place of the QuickJS runtime so the test can observe + // the one thing that actually matters: whether the forked sandbox fiber + // is still alive. It pauses through the real approval-gated tool, so + // `executeWithPause` returns on the caller's own fiber exactly as it + // does in the HTTP handler. + let sandboxEnded = false; + const probeCodeExecutor: CodeExecutor = { + execute: (_code, toolInvoker) => + Effect.gen(function* () { + yield* Effect.orDie( + toolInvoker.invoke({ path: "api.org.main.singleApproval", args: {} }), + ); + return { result: "unreachable", logs: [] } satisfies ExecuteResult; + }).pipe( + Effect.onInterrupt(() => + Effect.sync(() => { + sandboxEnded = true; + }), + ), + ), + }; + + const engine = createExecutionEngine({ executor, codeExecutor: probeCodeExecutor }); + + const outcome = yield* engine.executeWithPause("ignored by the probe executor"); + expect(outcome.status).toBe("paused"); + const paused = outcome as Extract; + expect(yield* engine.pausedExecutionCount()).toBe(1); + + // This is the leak: the response is ready, but the sandbox is still + // parked on the approval, still holding the executor and its handle. + expect(sandboxEnded, "the sandbox is still alive while the request is served").toBe(false); + + // Called from the same fiber that forked the sandbox — the shape the + // HTTP middleware runs `Effect.ensuring(engine.shutdown)` in. Bounded, + // because a shutdown that blocked would hold the request open for as + // long as the sandbox sat parked, which is worse than the leak. + const ended = yield* Effect.race( + engine.shutdown.pipe(Effect.as("ended" as const)), + Effect.sleep("5 seconds").pipe(Effect.as("hung" as const)), + ); + expect(ended, "shutdown must not block the request it runs in").toBe("ended"); + + // The assertion the fix exists for: the fiber is actually gone by the + // time shutdown returns, so it can no longer reach the executor — and + // therefore can no longer query the connection the host closes next. + expect(sandboxEnded, "shutdown interrupted the parked sandbox fiber").toBe(true); + + // The pause goes with it: one whose fiber is dead can never consume a + // response, so leaving it behind would only mislead a later resume. + expect(yield* engine.pausedExecutionCount()).toBe(0); + expect(yield* engine.getPausedExecution(paused.execution.id)).toBeNull(); + }), + { timeout: 15000 }, + ); + + it.effect("shutdown is a no-op when nothing is in flight", () => + Effect.gen(function* () { + const executor = yield* makeElicitingExecutor(); + const engine = createExecutionEngine({ executor, codeExecutor }); + + // Runs on every request, including the overwhelming majority that never + // pause, so it has to be inert rather than an error path. + yield* engine.shutdown; + yield* engine.shutdown; + expect(yield* engine.pausedExecutionCount()).toBe(0); + }), + ); }); diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts index 05c257cb2..276a5803d 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts @@ -173,6 +173,8 @@ const makeEngine = ( pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed("test engine"), + // The fake forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, }, }; }; diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts b/packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts index 3fa87cdc4..6877b2486 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-model-resume.test.ts @@ -263,6 +263,8 @@ const makeEngine = ( pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed("test engine"), + // The fake forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, }; return { calls, engine, resume }; }; diff --git a/packages/hosts/mcp-apps-shell/src/shell-resource.smoke.test.ts b/packages/hosts/mcp-apps-shell/src/shell-resource.smoke.test.ts index 046138968..83cac8494 100644 --- a/packages/hosts/mcp-apps-shell/src/shell-resource.smoke.test.ts +++ b/packages/hosts/mcp-apps-shell/src/shell-resource.smoke.test.ts @@ -28,6 +28,8 @@ const stubEngine: ExecutionEngine = { pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed("smoke"), + // The fake forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, }; // oxlint-disable-next-line executor/no-double-cast -- boundary: MCP SDK ClientCapabilities predates the ext-apps `extensions` field diff --git a/packages/hosts/mcp-apps-shell/src/shell/mcp-app.browser.test.ts b/packages/hosts/mcp-apps-shell/src/shell/mcp-app.browser.test.ts index 057684955..acf79110e 100644 --- a/packages/hosts/mcp-apps-shell/src/shell/mcp-app.browser.test.ts +++ b/packages/hosts/mcp-apps-shell/src/shell/mcp-app.browser.test.ts @@ -1311,6 +1311,8 @@ const startMcpHarness = async (openApi: OpenApiServer): Promise => { const startSchemaElicitationMcpHarness = (): Promise => startMcpHarnessForEngine({ getDescription: Effect.succeed("schema elicitation test executor"), + // The fake forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, execute: () => Effect.succeed({ result: null }), executeWithPause: () => Effect.succeed( diff --git a/packages/hosts/mcp/src/artifacts-tools.test.ts b/packages/hosts/mcp/src/artifacts-tools.test.ts index a8a3b794b..bc0a958b2 100644 --- a/packages/hosts/mcp/src/artifacts-tools.test.ts +++ b/packages/hosts/mcp/src/artifacts-tools.test.ts @@ -54,6 +54,8 @@ const makeStubEngine = (overrides: { pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed("test executor"), + // The fake forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, }); /** diff --git a/packages/hosts/mcp/src/namespace-search-tools.test.ts b/packages/hosts/mcp/src/namespace-search-tools.test.ts index 6e798f0f8..428fb091f 100644 --- a/packages/hosts/mcp/src/namespace-search-tools.test.ts +++ b/packages/hosts/mcp/src/namespace-search-tools.test.ts @@ -35,6 +35,8 @@ const makeRecordingEngine = (): { pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed("test executor"), + // The fake forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, }, }; }; diff --git a/packages/hosts/mcp/src/tool-server.test.ts b/packages/hosts/mcp/src/tool-server.test.ts index 813faeb48..311d6ee26 100644 --- a/packages/hosts/mcp/src/tool-server.test.ts +++ b/packages/hosts/mcp/src/tool-server.test.ts @@ -56,6 +56,8 @@ const makeStubEngine = (overrides: { pausedExecutionCount: () => Effect.succeed(0), hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed(overrides.description ?? "test executor"), + // The fake forks nothing, so there is no sandbox fiber to end. + shutdown: Effect.void, }); type TestServerConfig = Pick<