Make MCP pause/resume idempotent with collision-proof ids and honest expiry errors (#976)

RhysSullivan · web-flow · commit 8244fee567cb · 2026-06-14T22:29:26.000-07:00
* Add failing e2e: paused approval must survive the session runtime idle teardown

Reproduces the reported bug: with model elicitation over streamable HTTP,
execute returns a paused executionId, but the session DO's 5-minute idle
alarm calls closeRuntime() and rebuilds the engine from storage on the next
request — the in-memory pause map (fiber + Deferred) is gone, so resume
answers 'No paused execution: exec_1'.

A 20-second gap passes (the runtime is still warm); the 6-minute gap fails
deterministically, so the loss is the app's own idle teardown, not a
workerd eviction race.

* Make pause/resume honest and idempotent; ids unique across engine rebuilds

Production telemetry showed two failure classes behind 'No paused execution':
duplicate resumes (client retries a lost response seconds after a successful
resume) and pauses outliving the session runtime (idle teardown, isolate
eviction). Durable pause state is intentionally out of scope; instead:

- resume() is idempotent per executionId: settled outcomes are cached (exits,
  so failures re-fail typed) and replayed on retry; a concurrent duplicate
  joins the in-flight resume via a Deferred instead of missing the consumed
  pause.
- execution ids are exec_&lt;uuid&gt; instead of a per-engine exec_N counter, so a
  rebuilt engine (cold restore, redeploy) can never reissue an id a stale
  client still holds and bind its resume to a different execution's pause.
- a sandbox fiber that settles without a resume (timeout, failure) drops its
  abandoned pause entries and records the terminal outcome, so
  getPausedExecution stops reporting dead pauses and the map no longer leaks.
- the resume miss now explains itself: paused executions expire, recovery is
  re-running execute. Structured status execution_not_found, recovery
  re_execute.

The idle-teardown e2e scenario now specs the accepted contract: an approval
paused past the idle window expires with the re-run guidance and the session
keeps working (fresh execute pauses under a new id and resumes to completion);
a new scenario locks duplicate-resume replay end-to-end.
diff --git a/.changeset/pause-resume-idempotent.md b/.changeset/pause-resume-idempotent.md
@@ -0,0 +1,12 @@
+---
+"@executor-js/execution": patch
+"@executor-js/host-mcp": patch
+---
+
+Make paused-execution resume reliable: `resume` is now idempotent (a retried
+resume replays the recorded outcome instead of failing with "No paused
+execution"), execution ids are globally unique so a rebuilt engine can never
+re-mint an id a stale client still holds, pauses abandoned by a dead sandbox
+are dropped and their terminal outcome kept for late resumes, and an expired
+or lost pause now returns recovery guidance (re-run execute) instead of a bare
+miss.
diff --git a/e2e/cloud/mcp-client-sessions.test.ts b/e2e/cloud/mcp-client-sessions.test.ts
@@ -213,6 +213,154 @@ const result = await tools.executor.coreTools.policies.list({});
 return JSON.stringify(result);
 `;
 
+const executionIdOf = (text: string): string | undefined =>
+  /\bexecutionId:\s*(\S+)/.exec(text)?.[1];
+
+// The session DO tears its runtime down after 5 minutes without a request
+// (SESSION_TIMEOUT_MS in McpSessionDOBase) and rebuilds it from storage on
+// the next one — the same engine-state wipe a workerd eviction or a deploy
+// causes. Paused approvals deliberately do NOT survive this (durable pause
+// state is out of scope); the contract is that an expired pause fails with
+// recovery guidance and the session keeps working.
+const IDLE_TEARDOWN_GAP = "6 minutes";
+
+scenario(
+  "MCP sessions · an approval paused past the idle window expires with re-run guidance, not a dead end",
+  { timeout: 480_000 },
+  Effect.gen(function* () {
+    const target = yield* Target;
+    const { client } = yield* Api;
+    const mcp = yield* Mcp;
+    const identity = yield* target.newIdentity();
+    const api = yield* client(coreApi, identity);
+    const bearer = yield* mcp.mintBearer(emailOf(identity));
+
+    const policy = yield* api.policies.create({
+      payload: { owner: "org", pattern: APPROVAL_TARGET_TOOL, action: "require_approval" },
+    });
+
+    yield* Effect.gen(function* () {
+      const first = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer));
+      const sessionId = first.transport.sessionId;
+      const paused = yield* Effect.promise(() =>
+        first.client.callTool({ name: "execute", arguments: { code: GATED_CODE } }),
+      ).pipe(Effect.ensuring(closeQuietly(first)));
+      const pausedText = textOf(paused);
+      expect(pausedText, "the gated call pauses instead of completing").toContain(
+        "Execution paused",
+      );
+      const executionId = executionIdOf(pausedText);
+      expect(executionId, "the paused result carries the executionId").toEqual(expect.any(String));
+
+      // The user thinks the approval over for longer than the session keeps
+      // its runtime warm; the pause is gone when they come back.
+      yield* Effect.sleep(IDLE_TEARDOWN_GAP);
+
+      const second = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer, sessionId));
+      yield* Effect.gen(function* () {
+        const resumed = yield* Effect.promise(() =>
+          second.client.callTool({
+            name: "resume",
+            arguments: { executionId: executionId ?? "", action: "accept", content: "{}" },
+          }),
+        );
+        expect(resumed.isError, "the expired resume is an error, not a silent success").toBe(true);
+        expect(
+          textOf(resumed),
+          "the error tells the model how to recover, not just that the pause is gone",
+        ).toContain("run the execute tool again");
+
+        // The advertised recovery actually works on the same session: a fresh
+        // execute pauses with a NEW id (stale ids are never reused), and
+        // resuming that completes the gated call.
+        const reExecuted = yield* Effect.promise(() =>
+          second.client.callTool({ name: "execute", arguments: { code: GATED_CODE } }),
+        );
+        const reExecutedText = textOf(reExecuted);
+        expect(reExecutedText, "the re-run pauses for approval again").toContain(
+          "Execution paused",
+        );
+        const freshExecutionId = executionIdOf(reExecutedText);
+        expect(freshExecutionId, "the re-run mints a different executionId").not.toBe(executionId);
+
+        const resumedFresh = yield* Effect.promise(() =>
+          second.client.callTool({
+            name: "resume",
+            arguments: { executionId: freshExecutionId ?? "", action: "accept", content: "{}" },
+          }),
+        );
+        expect(resumedFresh.isError, "the fresh approval resumes to completion").not.toBe(true);
+        expect(textOf(resumedFresh), "the gated tool's result comes back after approval").toContain(
+          APPROVAL_TARGET_TOOL,
+        );
+      }).pipe(Effect.ensuring(closeQuietly(second)));
+    }).pipe(
+      Effect.ensuring(
+        api.policies
+          .remove({ params: { policyId: policy.id }, payload: { owner: "org" } })
+          .pipe(Effect.ignore),
+      ),
+    );
+  }),
+);
+
+scenario(
+  "MCP sessions · a duplicate resume replays the outcome instead of losing the approval",
+  {},
+  Effect.gen(function* () {
+    const target = yield* Target;
+    const { client } = yield* Api;
+    const mcp = yield* Mcp;
+    const identity = yield* target.newIdentity();
+    const api = yield* client(coreApi, identity);
+    const bearer = yield* mcp.mintBearer(emailOf(identity));
+
+    const policy = yield* api.policies.create({
+      payload: { owner: "org", pattern: APPROVAL_TARGET_TOOL, action: "require_approval" },
+    });
+
+    yield* Effect.gen(function* () {
+      const session = yield* Effect.promise(() => connectClient(target.mcpUrl, bearer));
+      yield* Effect.gen(function* () {
+        const paused = yield* Effect.promise(() =>
+          session.client.callTool({ name: "execute", arguments: { code: GATED_CODE } }),
+        );
+        const executionId = executionIdOf(textOf(paused));
+        expect(executionId, "the paused result carries the executionId").toEqual(
+          expect.any(String),
+        );
+
+        const resume = () =>
+          Effect.promise(() =>
+            session.client.callTool({
+              name: "resume",
+              arguments: { executionId: executionId ?? "", action: "accept", content: "{}" },
+            }),
+          );
+
+        const first = yield* resume();
+        expect(first.isError, "the first resume completes the execution").not.toBe(true);
+
+        // MCP clients retry resume when a response is lost in transit. The
+        // retry must replay the same completed outcome — the production
+        // failure mode was "No paused execution" seconds after a successful
+        // resume.
+        const retry = yield* resume();
+        expect(retry.isError, "the duplicate resume is not an error").not.toBe(true);
+        expect(textOf(retry), "the duplicate resume returns the same completed result").toContain(
+          APPROVAL_TARGET_TOOL,
+        );
+      }).pipe(Effect.ensuring(closeQuietly(session)));
+    }).pipe(
+      Effect.ensuring(
+        api.policies
+          .remove({ params: { policyId: policy.id }, payload: { owner: "org" } })
+          .pipe(Effect.ignore),
+      ),
+    );
+  }),
+);
+
 scenario(
   "MCP sessions · a paused approval survives the client reconnecting and resumes",
   {},
diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts
@@ -1,5 +1,6 @@
 import { Deferred, Effect, Fiber, Predicate, Queue } from "effect";
 import type * as Cause from "effect/Cause";
+import * as Exit from "effect/Exit";
 
 import type {
   Executor,
@@ -377,7 +378,29 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
 ): ExecutionEngine<E> => {
   const { executor, codeExecutor, toolDiscoveryProvider = defaultToolDiscoveryProvider } = config;
   const pausedExecutions = new Map<string, InternalPausedExecution<E>>();
-  let nextId = 0;
+  // 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
+  // already-delivered resume answers "no paused execution" (observed in
+  // production seconds after a successful resume). Bounded FIFO — pause
+  // volume is tiny (human approvals), so a small window is plenty.
+  const settledOutcomes = new Map<string, Exit.Exit<ExecutionResult, E>>();
+  const SETTLED_OUTCOME_LIMIT = 64;
+  // Resumes whose outcome is still being computed, so a concurrent duplicate
+  // awaits the same result instead of missing the (already-consumed) pause.
+  const pendingResumes = new Map<string, Deferred.Deferred<ExecutionResult, E>>();
+
+  // Exits (not just successes) so a replayed failure re-fails through the
+  // typed channel — hosts render engine failures opaquely, and a replay must
+  // not bypass that by flattening the cause into result text.
+  const recordSettledOutcome = (executionId: string, exit: Exit.Exit<ExecutionResult, E>): void => {
+    settledOutcomes.set(executionId, exit);
+    while (settledOutcomes.size > SETTLED_OUTCOME_LIMIT) {
+      const oldest = settledOutcomes.keys().next().value;
+      if (oldest === undefined) break;
+      settledOutcomes.delete(oldest);
+    }
+  };
 
   /**
    * Race a running fiber against the pause queue. Returns when either
@@ -427,7 +450,11 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
     const elicitationHandler: ElicitationHandler = (ctx) =>
       Effect.gen(function* () {
         const responseDeferred = yield* Deferred.make<typeof ElicitationResponse.Type>();
-        const id = `exec_${++nextId}`;
+        // Globally unique — engine instances are rebuilt on host restarts
+        // (Durable Object cold restores, redeploys), so a counter would
+        // re-mint the same ids and let a stale client resume bind to a
+        // different execution's pause.
+        const id = `exec_${crypto.randomUUID()}`;
 
         const paused: InternalPausedExecution<E> = {
           id,
@@ -453,12 +480,41 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
       codeExecutor.execute(code, invoker).pipe(Effect.withSpan("executor.code.exec")),
     );
 
+    // When the fiber settles on its own (sandbox timeout, failure) while
+    // pauses are still outstanding, drop them: getPausedExecution must not
+    // report a pause whose fiber can no longer consume a response, and the
+    // map must not grow forever. A resume retry still finds the terminal
+    // outcome via the settled-outcome cache.
+    const sandboxFiber = fiber;
+    yield* Effect.forkDetach(
+      Fiber.await(sandboxFiber).pipe(
+        Effect.flatMap((exit) =>
+          Effect.sync(() => {
+            const outcome = Exit.map(
+              exit,
+              (result): ExecutionResult => ({ status: "completed", result }),
+            );
+            for (const [id, paused] of pausedExecutions) {
+              if (paused.fiber !== sandboxFiber) continue;
+              pausedExecutions.delete(id);
+              recordSettledOutcome(id, outcome);
+            }
+          }),
+        ),
+      ),
+    );
+
     return (yield* awaitCompletionOrPause(fiber, pauseQueue)) as ExecutionResult;
   });
 
   /**
    * Resume a paused execution. Completes the response Deferred to unblock the
    * fiber, then races completion against the next queued or future pause.
+   *
+   * Idempotent per executionId: MCP clients retry `resume` when a response is
+   * lost in transit, so a duplicate of an already-delivered resume replays the
+   * recorded outcome, and a duplicate that arrives while the first is still
+   * in flight awaits the same outcome instead of reporting a missing pause.
    */
   const resumeExecution = Effect.fn("mcp.execute.resume")(function* (
     executionId: string,
@@ -468,16 +524,39 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
       "mcp.execute.resume.action": response.action,
     });
 
+    const settled = settledOutcomes.get(executionId);
+    if (settled) {
+      yield* Effect.annotateCurrentSpan({ "mcp.execute.resume.replayed": true });
+      return (yield* settled) as ExecutionResult;
+    }
+
+    const pending = pendingResumes.get(executionId);
+    if (pending) {
+      yield* Effect.annotateCurrentSpan({ "mcp.execute.resume.joined_inflight": true });
+      return (yield* Deferred.await(pending)) as ExecutionResult;
+    }
+
     const paused = pausedExecutions.get(executionId);
     if (!paused) return null;
     pausedExecutions.delete(executionId);
 
+    const inflight = yield* Deferred.make<ExecutionResult, E>();
+    pendingResumes.set(executionId, inflight);
+
     yield* Deferred.succeed(paused.response, {
       action: response.action as typeof ElicitationResponse.Type.action,
       content: response.content,
     });
 
-    return (yield* awaitCompletionOrPause(paused.fiber, paused.pauseQueue)) as ExecutionResult;
+    return (yield* awaitCompletionOrPause(paused.fiber, paused.pauseQueue).pipe(
+      Effect.onExit((exit) =>
+        Effect.gen(function* () {
+          recordSettledOutcome(executionId, exit);
+          pendingResumes.delete(executionId);
+          yield* Deferred.done(inflight, exit);
+        }),
+      ),
+    )) as ExecutionResult;
   });
 
   /**
diff --git a/packages/core/execution/src/tool-invoker.test.ts b/packages/core/execution/src/tool-invoker.test.ts
@@ -1290,6 +1290,94 @@ describe("pause/resume with multiple elicitations", () => {
     { timeout: 10000 },
   );
 
+  it.effect(
+    "execution ids are unique across engine instances (no counter reuse)",
+    () =>
+      Effect.gen(function* () {
+        const executor = yield* makeElicitingExecutor();
+        const engineA = createExecutionEngine({ executor, codeExecutor });
+        const engineB = createExecutionEngine({ executor, codeExecutor });
+        const code = "return await tools.api.org.main.singleApproval({});";
+
+        const pausedA = yield* engineA.executeWithPause(code);
+        const pausedB = yield* engineB.executeWithPause(code);
+        expect(pausedA.status).toBe("paused");
+        expect(pausedB.status).toBe("paused");
+        if (pausedA.status !== "paused" || pausedB.status !== "paused") return;
+
+        // A rebuilt engine (host restart) must never re-mint an id a client
+        // may still hold — a stale resume would bind to the wrong pause.
+        expect(pausedA.execution.id).not.toBe(pausedB.execution.id);
+
+        yield* engineA.resume(pausedA.execution.id, { action: "accept" });
+        yield* engineB.resume(pausedB.execution.id, { action: "accept" });
+      }),
+    { timeout: 10000 },
+  );
+
+  it.effect(
+    "a duplicate resume replays the delivered outcome instead of reporting a missing pause",
+    () =>
+      Effect.gen(function* () {
+        const executor = yield* makeElicitingExecutor();
+        const engine = createExecutionEngine({ executor, codeExecutor });
+        const code = "return await tools.api.org.main.singleApproval({});";
+
+        const outcome1 = yield* engine.executeWithPause(code);
+        expect(outcome1.status).toBe("paused");
+        if (outcome1.status !== "paused") return;
+        const executionId = outcome1.execution.id;
+
+        const first = yield* engine.resume(executionId, { action: "accept" });
+        expect(first?.status).toBe("completed");
+
+        // The MCP client retries resume when the response is lost in
+        // transit; the retry must return the same completed outcome.
+        const retry = yield* engine.resume(executionId, { action: "accept" });
+        expect(retry?.status).toBe("completed");
+        const completed = retry as Extract<NonNullable<typeof retry>, { status: "completed" }>;
+        expect(completed.result.error).toBeUndefined();
+        expect(completed.result.result).toMatchObject({ ok: true });
+      }),
+    { timeout: 10000 },
+  );
+
+  // live clock: the sandbox timeout is a real timer, so the test must
+  // actually wait for it rather than suspend on the virtual TestClock.
+  it.live(
+    "a pause abandoned by a failing sandbox is dropped and its resume replays the failure outcome",
+    () =>
+      Effect.gen(function* () {
+        const executor = yield* makeElicitingExecutor();
+        // Sandbox times out while suspended on the elicitation, so the fiber
+        // settles without a resume ever arriving.
+        const engine = createExecutionEngine({
+          executor,
+          codeExecutor: makeQuickJsExecutor({ timeoutMs: 250 }),
+        });
+        const code = "return await tools.api.org.main.singleApproval({});";
+
+        const outcome1 = yield* engine.executeWithPause(code);
+        expect(outcome1.status).toBe("paused");
+        if (outcome1.status !== "paused") return;
+        const executionId = outcome1.execution.id;
+
+        // Wait for the sandbox timeout to settle the detached fiber.
+        yield* Effect.sleep("600 millis");
+
+        // The dead pause must no longer be reported as live...
+        const stillPaused = yield* engine.getPausedExecution(executionId);
+        expect(stillPaused).toBeNull();
+
+        // ...and a late resume surfaces the terminal outcome, not a miss.
+        const late = yield* engine.resume(executionId, { action: "accept" });
+        expect(late?.status).toBe("completed");
+        const completed = late as Extract<NonNullable<typeof late>, { status: "completed" }>;
+        expect(completed.result.error).toContain("timed out");
+      }),
+    { timeout: 10000 },
+  );
+
   // Regression: use separate top-level runPromise calls to match HTTP/CLI
   // pause/resume, and a single-elicit tool so no later pause can mask a dead
   // sandbox fiber.
diff --git a/packages/hosts/mcp/src/tool-server.ts b/packages/hosts/mcp/src/tool-server.ts
