From 0a47f3a2415a73961b71d10894906bae6dd14db5 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 27 Aug 2026 20:49:00 -0700 Subject: [PATCH] End a request's sandbox fibers before its database connection closes executeWithPause forks the sandbox as a daemon so a pause can outlive the caller that observed it. That fiber closes over the executor, and the FumaDB client captures its handle at construction rather than resolving it per operation, so the fiber keeps a live reference to the connection the host opened for the scope that built the engine. On the HTTP executor plane that scope is the request, and its finalizer closes the pool as soon as the response is written. A sandbox still parked at that moment issues its next query against a closed pool. Add ExecutionEngine.shutdown, which interrupts the engine's in-flight sandbox fibers and waits for them to unwind, and run it from the shared execution-stack middleware so it happens before the request scope tears the connection down. Nothing is lost by ending it there: on a per-request engine the pause is already unreachable once the response is written, since a resume lands on a different engine and replays the call. The MCP session Durable Object does not use this middleware and keeps its session-lifetime pauses. shutdown is a required member so a decorator that rebuilds the engine object cannot drop it silently. --- apps/cloud/src/api/protected.test.ts | 2 + apps/cloud/src/engine/execution-gate.ts | 3 + .../engine/execution-rate-limit.node.test.ts | 2 + apps/cloud/src/engine/execution-usage.ts | 3 + packages/core/analytics/src/engine.test.ts | 2 + .../src/server/execution-stack-middleware.ts | 22 ++++ packages/core/execution/src/engine.ts | 53 +++++++++ .../core/execution/src/tool-invoker.test.ts | 101 ++++++++++++++++++ .../mcp/agent-session-durable-object.test.ts | 2 + .../mcp/agent-session-model-resume.test.ts | 2 + .../src/shell-resource.smoke.test.ts | 2 + .../src/shell/mcp-app.browser.test.ts | 2 + .../hosts/mcp/src/artifacts-tools.test.ts | 2 + .../mcp/src/namespace-search-tools.test.ts | 2 + packages/hosts/mcp/src/tool-server.test.ts | 2 + 15 files changed, 202 insertions(+) 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<