diff --git a/.changeset/mcp-cross-session-resume.md b/.changeset/mcp-cross-session-resume.md new file mode 100644 index 0000000000..718b774f29 --- /dev/null +++ b/.changeset/mcp-cross-session-resume.md @@ -0,0 +1,5 @@ +--- +"@executor-js/host-mcp": patch +--- + +Allow model-mode MCP clients to resume paused executions after reconnecting while preserving identity, resource, approval and authorization boundaries. 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..89917ddb32 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.test.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "@effect/vitest"; import { Effect, type Cause } from "effect"; +import { EXTENSION_ID, RESOURCE_MIME_TYPE } from "@modelcontextprotocol/ext-apps/server"; import type { ExecutionEngine } from "@executor-js/execution"; import { FormElicitation, ToolAddress, createExecutor } from "@executor-js/sdk"; @@ -11,7 +12,7 @@ import { type McpBuildServer, type McpBuildServerOptions, } from "./in-memory-session-store"; -import { defaultMcpResource, type Principal } from "./seams"; +import { defaultMcpResource, type McpResource, type Principal } from "./seams"; import { createExecutorMcpServer } from "./tool-server"; const TEST_PRINCIPAL: Principal = { @@ -121,16 +122,27 @@ const makeLatchedTestEngine = (): { const IDLE_TTL_MS = 60_000; type TestSessionStore = ReturnType; +type OpenSessionOptions = { + readonly resource?: McpResource; + readonly elicitationMode?: "browser" | "model" | "native"; + readonly principal?: Principal; + readonly appTools?: boolean; +}; /** Open a session on `sessions` and return its minted id. */ const openSession = async ( sessions: TestSessionStore, - principal: Principal = TEST_PRINCIPAL, - requestUrl = "https://executor.test/mcp", + { + resource = defaultMcpResource, + elicitationMode = "model", + principal = TEST_PRINCIPAL, + appTools = false, + }: OpenSessionOptions = {}, ): Promise => { + const path = resource.kind === "default" ? "/mcp" : `/mcp/toolkits/${resource.slug}`; const response = (await Effect.runPromise( sessions.store.dispatch({ - request: new Request(requestUrl, { + request: new Request(`https://executor.test${path}?elicitation_mode=${elicitationMode}`, { method: "POST", headers: { "content-type": "application/json", @@ -142,13 +154,15 @@ const openSession = async ( method: "initialize", params: { protocolVersion: "2025-06-18", - capabilities: {}, - clientInfo: { name: "idle-test", version: "1.0.0" }, + capabilities: appTools + ? { extensions: { [EXTENSION_ID]: { mimeTypes: [RESOURCE_MIME_TYPE] } } } + : {}, + clientInfo: { name: "session-store-test", version: "1.0.0" }, }, }), }), principal, - resource: defaultMcpResource, + resource, sessionId: null, method: "POST", }), @@ -156,6 +170,21 @@ const openSession = async ( expect(response.status).toBe(200); const sessionId = response.headers.get("mcp-session-id") ?? ""; expect(sessionId).not.toBe(""); + if (appTools) { + await Effect.runPromise( + sessions.store.dispatch({ + request: new Request(`https://executor.test${path}`, { + method: "POST", + headers: { ...MCP_POST_HEADERS, "mcp-session-id": sessionId }, + body: JSON.stringify({ jsonrpc: "2.0", method: "notifications/initialized" }), + }), + principal, + resource, + sessionId, + method: "POST", + }), + ); + } return sessionId; }; @@ -192,7 +221,7 @@ it("keeps overlapping warm-session workspace writes bound to their request roles createExecutorMcpServer({ engine }).pipe(Effect.map((mcpServer) => ({ mcpServer, engine }))), ); const admin = { ...TEST_PRINCIPAL, orgRole: "admin" as const }; - const sessionId = await openSession(sessions, admin); + const sessionId = await openSession(sessions, { principal: admin }); const call = (id: number, role: "admin" | "member") => Effect.runPromise( sessions.store.dispatch({ @@ -273,7 +302,7 @@ it("binds a paused workspace write to the resuming principal after demotion", as createExecutorMcpServer({ engine }).pipe(Effect.map((mcpServer) => ({ mcpServer, engine }))), ); const admin = { ...TEST_PRINCIPAL, orgRole: "admin" as const }; - const sessionId = await openSession(sessions, admin); + const sessionId = await openSession(sessions, { principal: admin }); const call = (id: number, principal: Principal, name: "execute" | "resume", args: unknown) => Effect.runPromise( sessions.store.dispatch({ @@ -347,11 +376,10 @@ it("uses the browser approver's demoted role after an admin starts waiting", asy ); const admin = { ...TEST_PRINCIPAL, orgRole: "admin" as const }; const member = { ...admin, orgRole: "member" as const }; - const sessionId = await openSession( - sessions, - admin, - "https://executor.test/mcp?elicitation_mode=browser", - ); + const sessionId = await openSession(sessions, { + principal: admin, + elicitationMode: "browser", + }); const call = (id: number, name: "execute" | "resume", args: unknown) => Effect.runPromise( @@ -777,3 +805,316 @@ describe("pre-initialize dispatch through the in-memory session store", () => { expect(sessions.sessionCount()).toBe(0); }); }); + +describe("cross-session model resume boundaries and lifetime", () => { + type FixtureOptions = { + readonly latchResume?: boolean; + readonly appTools?: boolean; + readonly resumeEffect?: () => ReturnType["resume"]>; + }; + type ResumeOptions = { + readonly resource?: McpResource; + readonly requestId?: number; + readonly principal?: Principal; + readonly executionId?: string; + readonly toolName?: "resume" | "execute-action-resume"; + }; + + const executionId = "exec_cross_session"; + const pausedExecution = { + id: executionId, + elicitationContext: { + address: ToolAddress.make("executor.coreTools.policies.create"), + args: { owner: "org", pattern: "cross-session.*", action: "block" }, + request: FormElicitation.make({ message: "Approve?", requestedSchema: {} }), + }, + }; + const completed = { + status: "completed" as const, + result: { result: "owner-resumed" }, + }; + + const fixture = (options: FixtureOptions = {}) => { + let built = 0; + let paused = true; + let settled = false; + let resuming = false; + let resumeCalls = 0; + let ownerShutdowns = 0; + const started = Promise.withResolvers(); + const joined = Promise.withResolvers(); + const gate = Promise.withResolvers(); + const ownerEngine: ExecutionEngine = { + ...makeIdleTestEngine(), + getPausedExecution: (id) => + Effect.sync(() => (id === executionId && paused ? pausedExecution : null)), + isExecutionSettled: (id) => Effect.sync(() => id === executionId && settled), + resume: (id) => + Effect.gen(function* () { + if (id !== executionId) return null; + if (settled) return completed; + paused = false; + if (!resuming) { + resumeCalls += 1; + resuming = true; + } else joined.resolve(); + started.resolve(); + if (options.latchResume) yield* Effect.promise(() => gate.promise); + if (options.resumeEffect) return yield* options.resumeEffect(); + settled = true; + return completed; + }), + shutdown: Effect.sync(() => { + ownerShutdowns += 1; + }), + }; + const sessions = makeInMemoryMcpSessionStore( + (_principal, buildOptions) => { + const engine = ++built === 1 ? ownerEngine : makeIdleTestEngine(); + return createExecutorMcpServer({ + engine, + ...(options.appTools ? { loadAppShellHtml: async () => "" } : {}), + ...(buildOptions ?? {}), + }).pipe(Effect.map((mcpServer) => ({ mcpServer, engine }))); + }, + { sessionIdleTtlMs: IDLE_TTL_MS, sessionSweepIntervalMs: IDLE_TTL_MS }, + ); + return { + sessions, + started: started.promise, + joined: joined.promise, + release: gate.resolve, + resumeCalls: () => resumeCalls, + ownerShutdowns: () => ownerShutdowns, + }; + }; + + const resume = async ( + sessions: TestSessionStore, + sessionId: string, + { + resource = defaultMcpResource, + requestId = 2, + principal = TEST_PRINCIPAL, + executionId: requestedExecutionId = executionId, + toolName = "resume", + }: ResumeOptions = {}, + ) => { + const path = resource.kind === "default" ? "/mcp" : `/mcp/toolkits/${resource.slug}`; + const response = await Effect.runPromise( + sessions.store.dispatch({ + request: new Request(`https://executor.test${path}`, { + method: "POST", + headers: { ...MCP_POST_HEADERS, "mcp-session-id": sessionId }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: requestId, + method: "tools/call", + params: { + name: toolName, + arguments: { executionId: requestedExecutionId, action: "accept", content: "{}" }, + }, + }), + }), + principal, + resource, + sessionId, + method: "POST", + }), + ); + expect(response).toBeInstanceOf(Response); + const body = (await (response as Response).json()) as { + result?: { isError?: boolean; structuredContent?: Record }; + }; + return body.result ?? {}; + }; + + const withFixture = ( + run: (value: ReturnType) => Promise, + options: FixtureOptions = {}, + ) => { + const value = fixture(options); + return Effect.runPromise( + Effect.promise(() => run(value)).pipe( + Effect.ensuring( + Effect.promise(async () => { + value.release(); + await value.sessions.close(); + }), + ), + ), + ); + }; + + const boundaries: ReadonlyArray<{ + readonly name: string; + readonly owner?: OpenSessionOptions; + readonly requester?: OpenSessionOptions; + }> = [ + { + name: "MCP resource", + requester: { resource: { kind: "toolkit", slug: "restricted" } }, + }, + { name: "approval mode", owner: { elicitationMode: "browser" } }, + { + name: "account", + requester: { principal: { ...TEST_PRINCIPAL, accountId: "acct_other" } }, + }, + { + name: "organization", + requester: { principal: { ...TEST_PRINCIPAL, organizationId: "org_other" } }, + }, + ]; + + it.each(boundaries)("does not cross the $name boundary", ({ owner, requester = {} }) => + withFixture(async (f) => { + await openSession(f.sessions, owner); + const next = await openSession(f.sessions, requester); + const result = await resume(f.sessions, next, requester); + expect(result.isError).toBe(true); + expect(result.structuredContent).toMatchObject({ status: "execution_forbidden" }); + expect(f.resumeCalls()).toBe(0); + }), + ); + + it.each(["browser", "native"] as const)( + "does not let %s-mode app tools resume another session's model pause", + (elicitationMode) => + withFixture( + async (f) => { + await openSession(f.sessions); + const next = await openSession(f.sessions, { elicitationMode, appTools: true }); + const result = await resume(f.sessions, next, { toolName: "execute-action-resume" }); + expect(result.isError).toBe(true); + expect(result.structuredContent).toMatchObject({ status: "execution_not_found" }); + expect(f.resumeCalls()).toBe(0); + }, + { appTools: true }, + ), + ); + + it("keeps the owning session alive while another session resumes its execution", () => + withFixture( + async (f) => { + await openSession(f.sessions); + const next = await openSession(f.sessions); + const pending = resume(f.sessions, next); + await f.started; + expect(await f.sessions.sweepIdleSessions(Date.now() + IDLE_TTL_MS + 1000)).toBe(0); + expect(f.ownerShutdowns()).toBe(0); + f.release(); + await pending; + }, + { latchResume: true }, + )); + + it("replays a settled resume across another fresh session without repeating side effects", () => + withFixture(async (f) => { + await openSession(f.sessions); + const next = await openSession(f.sessions); + expect((await resume(f.sessions, next)).structuredContent).toMatchObject({ + status: "completed", + }); + const retry = await openSession(f.sessions); + const replay = await resume(f.sessions, retry, { requestId: 3 }); + expect(replay.isError).toBeFalsy(); + expect(replay.structuredContent).toMatchObject({ + status: "completed", + result: "owner-resumed", + }); + expect(f.resumeCalls()).toBe(1); + })); + + it("joins an in-flight resume from another fresh session", () => + withFixture( + async (f) => { + await openSession(f.sessions); + const first = resume(f.sessions, await openSession(f.sessions)); + await f.started; + const retry = resume(f.sessions, await openSession(f.sessions)); + // The pause has been consumed but its continuation is still running. + const joined = await Promise.race([f.joined.then(() => true), retry.then(() => false)]); + expect(joined).toBe(true); + f.release(); + const [initial, repeated] = await Promise.all([first, retry]); + expect(initial.structuredContent).toMatchObject({ status: "completed" }); + expect(repeated.structuredContent).toMatchObject({ + status: "completed", + result: "owner-resumed", + }); + expect(f.resumeCalls()).toBe(1); + }, + { latchResume: true }, + )); + + it("reports a missing id without resuming a different execution", () => + withFixture(async (f) => { + await openSession(f.sessions); + const next = await openSession(f.sessions); + const result = await resume(f.sessions, next, { executionId: "exec_unknown" }); + expect(result.structuredContent).toMatchObject({ status: "execution_not_found" }); + expect(f.resumeCalls()).toBe(0); + })); + + it("does not resurrect an execution after its owner session is disposed", () => + withFixture(async (f) => { + const owner = await openSession(f.sessions); + const next = await openSession(f.sessions); + await Effect.runPromise(f.sessions.store.dispose(owner)); + const result = await resume(f.sessions, next); + expect(result.structuredContent).toMatchObject({ status: "execution_not_found" }); + expect(f.resumeCalls()).toBe(0); + })); + + it("returns an opaque execution failure rather than recovery instructions for a missing pause", () => + withFixture( + async (f) => { + await openSession(f.sessions); + const result = await resume(f.sessions, await openSession(f.sessions)); + expect(result.isError).toBe(true); + expect(result.structuredContent).toMatchObject({ status: "error" }); + expect(result.structuredContent?.error).toMatch(/Internal tool error/); + expect(JSON.stringify(result)).not.toContain("sensitive continuation detail"); + expect(JSON.stringify(result)).not.toContain("run the execute tool again"); + }, + { resumeEffect: () => Effect.die("sensitive continuation detail") }, + )); + + it("uses the current resuming request's role after both sessions were initialized as admin", async () => { + const executor = await Effect.runPromise( + createExecutor({ ...makeTestConfig(), orgWrites: "request" }), + ); + await Effect.runPromise( + Effect.promise(() => + withFixture( + async (f) => { + const admin: Principal = { ...TEST_PRINCIPAL, orgRole: "admin" }; + await openSession(f.sessions, { principal: admin }); + const next = await openSession(f.sessions, { principal: admin }); + const result = await resume(f.sessions, next, { + principal: { ...admin, orgRole: "member" }, + }); + expect(result.isError).toBe(true); + expect(result.structuredContent).toMatchObject({ status: "error" }); + expect(await Effect.runPromise(executor.policies.list())).toEqual([]); + }, + { + resumeEffect: () => + executor.policies + .create({ + owner: "org", + pattern: "cross-session-demotion.*", + action: "block", + }) + .pipe( + Effect.map((policy) => ({ + status: "completed" as const, + result: { result: policy }, + })), + ), + }, + ), + ).pipe(Effect.ensuring(executor.close().pipe(Effect.orDie))), + ); + }); +}); diff --git a/packages/hosts/mcp/src/in-memory-session-store.ts b/packages/hosts/mcp/src/in-memory-session-store.ts index 952f9bd83f..47214e2ad8 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.ts @@ -2,7 +2,11 @@ import { Cause, Data, Effect, Layer } from "effect"; import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; -import { formatPausedExecution, type ExecutionEngine } from "@executor-js/execution"; +import { + formatPausedExecution, + type ExecutionEngine, + type ResumeResponse, +} from "@executor-js/execution"; import type { Executor, OrgWriteAccess } from "@executor-js/sdk"; import { @@ -13,6 +17,7 @@ import { readElicitationMode, readSearchToolsEnabled, readToolMode, + type McpElicitationMode, type McpToolMode, } from "./browser-approval"; import { @@ -33,7 +38,13 @@ import { type Principal, type McpResource, } from "./seams"; -import type { BrowserApprovalStore, McpPassthroughUnavailableError } from "./tool-server"; +import { + formatMcpExecutionOutcome, + toMcpFailureResult, + type BrowserApprovalStore, + type McpPassthroughUnavailableError, + type ResumeFallbackOutcome, +} from "./tool-server"; // --------------------------------------------------------------------------- // In-process McpSessionStore — the single-node serving store, shared by every @@ -121,6 +132,11 @@ export interface McpBuildServerOptions { readonly searchToolsEnabled?: boolean; /** The tool surface (`?mode=`): codemode (default) or passthrough. */ readonly mode?: McpToolMode; + /** Resume on another live session with the same identity, resource and approval mode. */ + readonly resumeFallback?: ( + executionId: string, + response: ResumeResponse, + ) => Effect.Effect; } /** Build the per-session `McpServer` + engine for a principal (the host's engine + tools). */ @@ -214,6 +230,7 @@ const RESUME_PATH = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)\/res interface SessionOwner { readonly principal: Principal; readonly resource: McpResource; + readonly elicitationMode: McpElicitationMode; } const sessionOwnerMatches = ( @@ -277,17 +294,11 @@ export const makeInMemoryMcpSessionStore = ( activeRequests.set(id, (activeRequests.get(id) ?? 0) + 1); }; - /** - * Release the claim and restamp: a call that ran for an hour leaves the - * session idle from the moment it FINISHED, not from the moment it started. - * `touch` is a no-op once the session is gone, so this can never resurrect a - * disposed id. - */ + /** Release one in-flight request claim. */ const endRequest = (id: string): void => { const remaining = (activeRequests.get(id) ?? 1) - 1; if (remaining > 0) activeRequests.set(id, remaining); else activeRequests.delete(id); - touch(id); }; /** @@ -394,7 +405,7 @@ export const makeInMemoryMcpSessionStore = ( const owner = owners.get(sessionId); if (!transport || !owner) return Effect.succeed("not-found"); if (!sessionOwnerMatches(owner, principal, resource)) return Effect.succeed("forbidden"); - owners.set(sessionId, { principal, resource }); + owners.set(sessionId, { ...owner, principal, resource }); touch(sessionId); // Claim before the await, release in the finalizer — `runHandleRequest` // already recovers every failure to a 500, but `ensuring` also covers an @@ -402,7 +413,13 @@ export const makeInMemoryMcpSessionStore = ( // make the session immortal, the opposite leak). beginRequest(sessionId); return runHandleRequest(transport, request, orgWriteAccessForPrincipal(principal)).pipe( - Effect.ensuring(Effect.sync(() => endRequest(sessionId))), + Effect.ensuring( + Effect.sync(() => { + endRequest(sessionId); + // A long request becomes idle when it finishes, not when it starts. + touch(sessionId); + }), + ), ); }; @@ -445,6 +462,57 @@ export const makeInMemoryMcpSessionStore = ( }; }; + const opaqueResumeFailure = (cause: Cause.Cause) => + Effect.succeed({ status: "result" as const, result: toMcpFailureResult(cause) }); + + /** + * Resume through a live model session with the same identity and resource. + * Calling its engine directly preserves pending joins and settled-result + * replay; probing mismatched sessions prevents their ids becoming lookup misses. + */ + const resumeAcrossSessions = + (requester: Principal, resource: McpResource) => + (executionId: string, response: ResumeResponse): Effect.Effect => + Effect.gen(function* () { + for (const [sid, engine] of engines) { + const owner = owners.get(sid); + if (!owner) continue; + if ( + !sessionOwnerMatches(owner, requester, resource) || + owner.elicitationMode !== "model" + ) { + const paused = yield* engine.getPausedExecution(executionId); + const settled = engine.isExecutionSettled + ? yield* engine.isExecutionSettled(executionId) + : false; + if (paused || settled) return { status: "execution_forbidden" as const }; + continue; + } + + beginRequest(sid); + const result = yield* Effect.gen(function* () { + // The caller's CurrentOrgWriteAccess reaches engine.resume, which + // rebinds the detached continuation before waking it. + const outcome = yield* engine.resume(executionId, response); + if (outcome) + return { status: "result" as const, result: formatMcpExecutionOutcome(outcome) }; + const settled = engine.isExecutionSettled + ? yield* engine.isExecutionSettled(executionId) + : false; + return settled ? { status: "execution_already_settled" as const } : null; + }).pipe( + // The generic fallback hook treats thrown failures as a lookup + // miss. Return the normal opaque MCP failure instead, so a failed + // continuation never instructs the client to execute it again. + Effect.catchCause(opaqueResumeFailure), + Effect.tap((outcome) => (outcome ? Effect.sync(() => touch(sid)) : Effect.void)), + Effect.ensuring(Effect.sync(() => endRequest(sid))), + ); + if (result) return result; + } + return null; + }).pipe(Effect.catchCause(opaqueResumeFailure)); + /** Open a new session: build the server, connect a transport, drive the request. */ const openSession = ( principal: Principal, @@ -455,6 +523,9 @@ export const makeInMemoryMcpSessionStore = ( return buildServer(principal, { ...buildOptionsFor(request, () => createdSessionId), resource, + ...(readElicitationMode(request) === "model" + ? { resumeFallback: resumeAcrossSessions(principal, resource) } + : {}), }).pipe( Effect.flatMap(({ mcpServer, engine, executor, close }) => Effect.gen(function* () { @@ -465,7 +536,11 @@ export const makeInMemoryMcpSessionStore = ( createdSessionId = sid; transports.set(sid, transport); servers.set(sid, mcpServer); - owners.set(sid, { principal, resource }); + owners.set(sid, { + principal, + resource, + elicitationMode: readElicitationMode(request), + }); engines.set(sid, engine); if (executor) executors.set(sid, executor); if (close) closers.set(sid, close); diff --git a/packages/hosts/mcp/src/tool-server.ts b/packages/hosts/mcp/src/tool-server.ts index f7c1a349f7..d3d41b44c2 100644 --- a/packages/hosts/mcp/src/tool-server.ts +++ b/packages/hosts/mcp/src/tool-server.ts @@ -827,7 +827,7 @@ const formatResumeApprovalRequired = (input: { }, }); -const toMcpFailureResult = (cause: Cause.Cause): McpToolResult => { +export const toMcpFailureResult = (cause: Cause.Cause): McpToolResult => { const correlationId = newCorrelationId(); const defect = Cause.findDefect(cause); const nativeElicitationFailed =