From 645da791d1fbf4c2fa0059b9c4ba91b5e2be2d54 Mon Sep 17 00:00:00 2001 From: Drake Thomsen Date: Fri, 2 Oct 2026 22:00:14 +0200 Subject: [PATCH 1/2] Fix model approval resume across self-hosted MCP sessions --- bun.lock | 1 + e2e/scenarios/policy-tool-approval.test.ts | 8 +- packages/hosts/mcp/package.json | 1 + .../mcp/src/in-memory-session-store.test.ts | 150 +++++++++++++++++- .../hosts/mcp/src/in-memory-session-store.ts | 61 ++++++- packages/hosts/mcp/src/tool-server.ts | 8 +- 6 files changed, 217 insertions(+), 12 deletions(-) diff --git a/bun.lock b/bun.lock index c9436aa422..6ca0db5087 100644 --- a/bun.lock +++ b/bun.lock @@ -713,6 +713,7 @@ }, "devDependencies": { "@effect/vitest": "catalog:", + "@executor-js/runtime-quickjs": "workspace:*", "@types/node": "catalog:", "bun-types": "catalog:", "vitest": "catalog:", diff --git a/e2e/scenarios/policy-tool-approval.test.ts b/e2e/scenarios/policy-tool-approval.test.ts index c6841f5421..8b04f50ecb 100644 --- a/e2e/scenarios/policy-tool-approval.test.ts +++ b/e2e/scenarios/policy-tool-approval.test.ts @@ -38,7 +38,7 @@ return JSON.stringify(result); `; scenario( - "Policy tools · policies.create pauses for approval from its own annotation, then runs once approved", + "Policy tools · policies.create pauses for approval from its own annotation, then resumes from a new MCP session", {}, Effect.gen(function* () { const target = yield* Target; @@ -85,7 +85,11 @@ scenario( "policy is not written while the approval is still pending", ).toBe(false); - const resumed = yield* session.approvePaused(paused.text); + const resumeSession = mcp.session(identity); + yield* resumeSession.listTools(); + const resumed = yield* resumeSession.approvePaused(paused.text); + const replayed = yield* resumeSession.approvePaused(paused.text); + expect(replayed.text, "retry replays the completed result").toBe(resumed.text); expect(resumed.ok, "resumed execution completed without error").toBe(true); const afterApproval = yield* client.policies.list(); diff --git a/packages/hosts/mcp/package.json b/packages/hosts/mcp/package.json index 1e8dd7c4fc..e55c79c68b 100644 --- a/packages/hosts/mcp/package.json +++ b/packages/hosts/mcp/package.json @@ -54,6 +54,7 @@ }, "devDependencies": { "@effect/vitest": "catalog:", + "@executor-js/runtime-quickjs": "workspace:*", "@types/node": "catalog:", "bun-types": "catalog:", "vitest": "catalog:" diff --git a/packages/hosts/mcp/src/in-memory-session-store.test.ts b/packages/hosts/mcp/src/in-memory-session-store.test.ts index bdc51db85d..558317fadb 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.test.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.test.ts @@ -1,8 +1,10 @@ -import { describe, expect, it } from "@effect/vitest"; +import { afterEach, describe, expect, it } from "@effect/vitest"; import { Effect, type Cause } from "effect"; -import type { ExecutionEngine } from "@executor-js/execution"; +import { createExecutionEngine, type ExecutionEngine } from "@executor-js/execution"; import { FormElicitation, ToolAddress, createExecutor } from "@executor-js/sdk"; +import { makeQuickJsExecutor } from "@executor-js/runtime-quickjs"; +import type { Executor } from "@executor-js/sdk"; import { makeTestConfig } from "@executor-js/sdk/testing"; import { @@ -11,7 +13,7 @@ import { type McpBuildServer, type McpBuildServerOptions, } from "./in-memory-session-store"; -import { defaultMcpResource, type Principal } from "./seams"; +import { defaultMcpResource, type Principal, type McpResource } from "./seams"; import { createExecutorMcpServer } from "./tool-server"; const TEST_PRINCIPAL: Principal = { @@ -127,6 +129,7 @@ const openSession = async ( sessions: TestSessionStore, principal: Principal = TEST_PRINCIPAL, requestUrl = "https://executor.test/mcp", + resource: McpResource = defaultMcpResource, ): Promise => { const response = (await Effect.runPromise( sessions.store.dispatch({ @@ -148,7 +151,7 @@ const openSession = async ( }), }), principal, - resource: defaultMcpResource, + resource, sessionId: null, method: "POST", }), @@ -159,6 +162,145 @@ const openSession = async ( return sessionId; }; +describe("model approvals across in-memory MCP sessions", () => { + const admin = { ...TEST_PRINCIPAL, orgRole: "admin" as const }; + const policyCode = `return await tools.executor.coreTools.policies.create({ + owner: "org", pattern: "cross-session-test.*", action: "block" + });`; + + const stores: TestSessionStore[] = []; + afterEach(async () => { + await Promise.all(stores.splice(0).map((store) => store.close())); + }); + + const setup = () => { + const executors: Executor[] = []; + const sessions = makeInMemoryMcpSessionStore((_principal, options) => + Effect.gen(function* () { + const executor = yield* createExecutor( + makeTestConfig({ coreTools: {}, orgWrites: "request" }), + ); + executors.push(executor); + const engine = createExecutionEngine({ executor, codeExecutor: makeQuickJsExecutor() }); + const mcpServer = yield* createExecutorMcpServer({ engine, ...options }); + return { engine, executor, mcpServer }; + }).pipe(Effect.mapError((cause) => new McpEngineBuildError({ cause }))), + ); + stores.push(sessions); + let rpcId = 1; + const call = async ( + sessionId: string, + name: string, + args: unknown, + principal: Principal = admin, + resource: McpResource = defaultMcpResource, + ) => { + const response = await Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { ...MCP_POST_HEADERS, "mcp-session-id": sessionId }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: ++rpcId, + method: "tools/call", + params: { name, arguments: args }, + }), + }), + principal, + resource, + sessionId, + method: "POST", + }), + ); + expect(response).toBeInstanceOf(Response); + return ( + (await (response as Response).json()) as { + result: { structuredContent: Record; isError?: boolean }; + } + ).result; + }; + return { sessions, call, executors }; + }; + + it("resumes a real paused engine from a new session and replays without repeating the write", async () => { + const { sessions, call, executors } = setup(); + { + const a = await openSession(sessions, admin); + const b = await openSession(sessions, admin); + expect((await call(a, "execute", { code: "return 42;" })).structuredContent.result).toBe(42); + const paused = await call(a, "execute", { code: policyCode }); + expect(paused.structuredContent.status).toBe("waiting_for_interaction"); + expect(await Effect.runPromise(executors[0]!.policies.list())).toHaveLength(0); + const args = { + executionId: paused.structuredContent.executionId, + action: "accept", + content: "{}", + }; + const resumed = await call(b, "resume", args); + expect(resumed.structuredContent.status).toBe("completed"); + expect(await Effect.runPromise(executors[0]!.policies.list())).toHaveLength(1); + expect((await call(b, "resume", args)).structuredContent).toEqual(resumed.structuredContent); + expect(await Effect.runPromise(executors[0]!.policies.list())).toHaveLength(1); + } + }); + + it("uses the resuming request's permissions after the owner is demoted", async () => { + const { sessions, call, executors } = setup(); + { + const a = await openSession(sessions, admin); + const b = await openSession(sessions, admin); + const paused = await call(a, "execute", { code: policyCode }); + const resumed = await call( + b, + "resume", + { executionId: paused.structuredContent.executionId, action: "accept", content: "{}" }, + { ...admin, orgRole: "member" }, + ); + expect(resumed.structuredContent.status).toBe("completed"); + expect(resumed.structuredContent.result).toMatchObject({ + ok: false, + error: { code: "org_write_denied" }, + }); + expect(await Effect.runPromise(executors[0]!.policies.list())).toHaveLength(0); + } + }); + + it.each(["account", "organization", "resource", "browser", "disposed"])( + "does not cross the %s boundary", + async (boundary) => { + const { sessions, call } = setup(); + { + const a = await openSession( + sessions, + admin, + boundary === "browser" ? "https://executor.test/mcp?elicitation_mode=browser" : undefined, + ); + const principal = { + ...admin, + ...(boundary === "account" ? { accountId: "other" } : {}), + ...(boundary === "organization" ? { organizationId: "other" } : {}), + }; + const resource: McpResource = + boundary === "resource" ? { kind: "toolkit", slug: "other" } : defaultMcpResource; + const b = await openSession(sessions, principal, undefined, resource); + const paused = await call(a, "execute", { code: policyCode }); + const executionId = paused.structuredContent.executionId; + expect(typeof executionId).toBe("string"); + if (boundary === "disposed") await Effect.runPromise(sessions.store.dispose(a)); + const resumed = await call( + b, + "resume", + { executionId, action: "accept", content: "{}" }, + principal, + resource, + ); + expect(resumed.structuredContent.status).toBe("execution_not_found"); + } + }, + ); +}); + it("keeps overlapping warm-session workspace writes bound to their request roles", async () => { const executor = await Effect.runPromise( createExecutor({ ...makeTestConfig(), orgWrites: "request" }), diff --git a/packages/hosts/mcp/src/in-memory-session-store.ts b/packages/hosts/mcp/src/in-memory-session-store.ts index 952f9bd83f..4ac190a8a5 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.ts @@ -33,7 +33,14 @@ import { type Principal, type McpResource, } from "./seams"; -import type { BrowserApprovalStore, McpPassthroughUnavailableError } from "./tool-server"; +import { + formatMcpExecutionOutcome, + formatMcpExecutionFailure, + type BrowserApprovalStore, + type McpPassthroughUnavailableError, + type ResumeFallbackOutcome, +} from "./tool-server"; +import type { ResumeResponse } from "@executor-js/execution"; // --------------------------------------------------------------------------- // In-process McpSessionStore — the single-node serving store, shared by every @@ -106,6 +113,10 @@ export interface BuiltMcpServer { /** The browser-mode wiring the store hands a build call when a session opts in. */ export interface McpBuildServerOptions { + readonly resumeFallback?: ( + executionId: string, + response: ResumeResponse, + ) => Effect.Effect; readonly resource?: McpResource; readonly elicitationMode?: | { readonly mode: "browser"; readonly approvalUrl: (executionId: string) => string } @@ -249,6 +260,7 @@ export const makeInMemoryMcpSessionStore = ( const servers = new Map(); const owners = new Map(); const engines = new Map>(); + const modelSessions = new Set(); const executors = new Map(); const closers = new Map Promise>(); const approvals: InProcessBrowserApprovalStore = makeInProcessBrowserApprovalStore(); @@ -283,11 +295,11 @@ export const makeInMemoryMcpSessionStore = ( * `touch` is a no-op once the session is gone, so this can never resurrect a * disposed id. */ - const endRequest = (id: string): void => { + const endRequest = (id: string, restamp = true): void => { const remaining = (activeRequests.get(id) ?? 1) - 1; if (remaining > 0) activeRequests.set(id, remaining); else activeRequests.delete(id); - touch(id); + if (restamp) touch(id); }; /** @@ -319,6 +331,7 @@ export const makeInMemoryMcpSessionStore = ( servers.delete(id); owners.delete(id); engines.delete(id); + modelSessions.delete(id); executors.delete(id); closers.delete(id); lastSeen.delete(id); @@ -455,6 +468,47 @@ export const makeInMemoryMcpSessionStore = ( return buildServer(principal, { ...buildOptionsFor(request, () => createdSessionId), resource, + // A client may initialize again between execute and resume. Keep engines + // session-owned, but route model approvals within the same authenticated + // account, organization, resource, and approval mode. Calling resume (not + // probing only paused state) also joins in-flight calls and replays the + // engine's bounded settled-result cache without executing a tool twice. + resumeFallback: (executionId, response) => + Effect.gen(function* () { + if (!createdSessionId || !modelSessions.has(createdSessionId)) return null; + const caller = owners.get(createdSessionId); + if (!caller) return null; + for (const [id, engine] of engines) { + const owner = owners.get(id); + if ( + id === createdSessionId || + !modelSessions.has(id) || + !owner || + !sessionOwnerMatches(owner, caller.principal, caller.resource) + ) + continue; + // Keep the owning session alive for the forwarded request. The + // Effect inherits the resumer's request-local workspace permissions. + beginRequest(id); + let matched = false; + const result = yield* engine.resume(executionId, response).pipe( + Effect.map((outcome) => (outcome ? formatMcpExecutionOutcome(outcome) : null)), + Effect.catchCause((cause) => Effect.succeed(formatMcpExecutionFailure(cause))), + Effect.tap((result) => + Effect.sync(() => { + matched = result !== null; + }), + ), + // A miss must not keep every unrelated session alive forever. + Effect.ensuring(Effect.sync(() => endRequest(id, matched))), + ); + if (result) return { status: "result" as const, result }; + if (yield* engine.isExecutionSettled?.(executionId) ?? Effect.succeed(false)) { + return { status: "execution_already_settled" as const }; + } + } + return null; + }), }).pipe( Effect.flatMap(({ mcpServer, engine, executor, close }) => Effect.gen(function* () { @@ -467,6 +521,7 @@ export const makeInMemoryMcpSessionStore = ( servers.set(sid, mcpServer); owners.set(sid, { principal, resource }); engines.set(sid, engine); + if (readElicitationMode(request) === "model") modelSessions.add(sid); if (executor) executors.set(sid, executor); if (close) closers.set(sid, close); lastSeen.set(sid, Date.now()); diff --git a/packages/hosts/mcp/src/tool-server.ts b/packages/hosts/mcp/src/tool-server.ts index 91562be746..df1eb08e72 100644 --- a/packages/hosts/mcp/src/tool-server.ts +++ b/packages/hosts/mcp/src/tool-server.ts @@ -822,7 +822,7 @@ const formatResumeApprovalRequired = (input: { }, }); -const toMcpFailureResult = (cause: Cause.Cause): McpToolResult => { +export const formatMcpExecutionFailure = (cause: Cause.Cause): McpToolResult => { const correlationId = newCorrelationId(); const defect = Cause.findDefect(cause); const nativeElicitationFailed = @@ -1286,7 +1286,7 @@ const registerPassthroughTools = ( CurrentOrgWriteAccess, makeOrgWriteAccessState(requestOrgWriteAccess(extra)), ), - Effect.catchCause((cause) => Effect.succeed(toMcpFailureResult(cause))), + Effect.catchCause((cause) => Effect.succeed(formatMcpExecutionFailure(cause))), ), ); yield* Effect.sync(() => { @@ -1638,7 +1638,7 @@ export const createExecutorMcpServer = ( CurrentOrgWriteAccess, makeOrgWriteAccessState(requestOrgWriteAccess(extra)), ), - Effect.catchCause((cause) => Effect.succeed(toMcpFailureResult(cause))), + Effect.catchCause((cause) => Effect.succeed(formatMcpExecutionFailure(cause))), ), ); @@ -1692,6 +1692,7 @@ export const createExecutorMcpServer = ( } const outcome = yield* engine.executeWithPause(code); debugLog("execute.paused_flow_result", { + ...joinKeyAttributes(extra), status: outcome.status, executionId: outcome.status === "paused" ? outcome.execution.id : undefined, interactionKind: @@ -1849,6 +1850,7 @@ export const createExecutorMcpServer = ( "mcp.execute.execution_id": executionId, }); debugLog("resume.call", { + ...joinKeyAttributes(extra), executionId, action: response.action, hasContent: response.content !== undefined, From 633ef3a49dabe1f5439a94603599948030a830c5 Mon Sep 17 00:00:00 2001 From: Drake Thomsen Date: Fri, 2 Oct 2026 22:13:07 +0200 Subject: [PATCH 2/2] Scope cross-session replay scenario to the self-host store --- e2e/scenarios/policy-tool-approval.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/e2e/scenarios/policy-tool-approval.test.ts b/e2e/scenarios/policy-tool-approval.test.ts index 8b04f50ecb..946f1f3574 100644 --- a/e2e/scenarios/policy-tool-approval.test.ts +++ b/e2e/scenarios/policy-tool-approval.test.ts @@ -38,7 +38,7 @@ return JSON.stringify(result); `; scenario( - "Policy tools · policies.create pauses for approval from its own annotation, then resumes from a new MCP session", + "Policy tools · policies.create pauses for approval from its own annotation, then resumes (across MCP sessions on self-host)", {}, Effect.gen(function* () { const target = yield* Target; @@ -85,7 +85,9 @@ scenario( "policy is not written while the approval is still pending", ).toBe(false); - const resumeSession = mcp.session(identity); + // The in-memory store supports replay across live sessions. Cloud uses + // a separate owner directory whose cross-session replay is out of scope. + const resumeSession = target.name.startsWith("selfhost") ? mcp.session(identity) : session; yield* resumeSession.listTools(); const resumed = yield* resumeSession.approvePaused(paused.text); const replayed = yield* resumeSession.approvePaused(paused.text);