diff --git a/apps/host-selfhost/src/app.ts b/apps/host-selfhost/src/app.ts index 6f39d3d09..f6930dd11 100644 --- a/apps/host-selfhost/src/app.ts +++ b/apps/host-selfhost/src/app.ts @@ -85,6 +85,10 @@ export const makeSelfHostApp = async (options: MakeSelfHostAppOptions = {}) => { routes: [ // Better Auth owns /api/auth/* — the full path reaches it unmodified. HttpRouter.add("*", "/api/auth/*", HttpEffect.fromWebHandler(authHandler)), + // Browser approval of paused MCP executions: the console resume page + // reads paused detail (GET) and records the decision (POST .../resume), + // session-cookie-gated, delegating to the in-process MCP store. + HttpRouter.add("*", "/api/mcp-sessions/*", HttpEffect.fromWebHandler(mcp.approvalHandler)), // App-local admin (invite-code) API, served under /api/admin/*. makeSelfHostAdminApiLayer({ betterAuth, db: dbHandle, mountPrefix: "/api" }), // Public system API: /api/health + /api/setup-status (unauthenticated). diff --git a/apps/host-selfhost/src/mcp/index.ts b/apps/host-selfhost/src/mcp/index.ts index f6b212c87..b45eba388 100644 --- a/apps/host-selfhost/src/mcp/index.ts +++ b/apps/host-selfhost/src/mcp/index.ts @@ -1,4 +1,4 @@ -import { Layer } from "effect"; +import { Effect, Layer } from "effect"; import { IdentityProvider } from "@executor-js/api/server"; import type { McpAuthProvider, McpErrorReporter, McpSessionStore } from "@executor-js/host-mcp"; @@ -50,10 +50,49 @@ export interface SelfHostMcpSeams { readonly sessions: Layer.Layer; /** Route 500 defects through the host's console `ErrorCapture`. */ readonly reporter: Layer.Layer; + /** + * The browser-approval HTTP handler, mounted by the app at + * `/api/mcp-sessions/*`: a session-cookie-gated web handler that serves the + * paused-execution detail (GET) and records the human's decision (POST + * `/resume`) for the console approval page. Browser elicitation mode only. + */ + readonly approvalHandler: (request: Request) => Promise; /** Dispose all live in-process MCP sessions at shutdown (not a seam). */ readonly close: () => Promise; } +const jsonResponse = (value: unknown, status: number): Response => + new Response(JSON.stringify(value), { status, headers: { "content-type": "application/json" } }); + +/** + * Gate the browser-approval endpoints behind a valid Better Auth session (the + * console page calls them with the user's cookie), then delegate to the + * in-process store's paused/resume handlers. Single-tenant: any authenticated + * user of the one org may act on a session it still holds — the store confirms + * the execution belongs to the addressed session before recording. + */ +const makeApprovalHandler = + ( + store: ReturnType, + betterAuth: BetterAuthHandle, + ): ((request: Request) => Promise) => + async (request) => { + // A malformed cookie must read as unauthenticated, not 500. + const session = await Effect.runPromise( + Effect.tryPromise({ + try: () => betterAuth.auth.api.getSession({ headers: request.headers }), + catch: () => "session lookup failed", + }).pipe(Effect.orElseSucceed(() => null)), + ); + if (!session) return jsonResponse({ error: "Unauthorized" }, 401); + + return ( + (await store.handlePausedRequest(request)) ?? + (await store.handleApprovalRequest(request)) ?? + jsonResponse({ error: "Not found" }, 404) + ); + }; + /** * Build the self-host MCP serving seams over the long-lived DB handle. The auth * seam is `selfHostMcpAuth` (Better Auth mcp() OAuth), with the Better Auth @@ -73,6 +112,7 @@ export const makeSelfHostMcpSeams = ( auth, sessions: selfHostMcpSessions(sessionStore), reporter: selfHostMcpReporter, + approvalHandler: makeApprovalHandler(sessionStore, betterAuth), close: sessionStore.close, }; }; diff --git a/apps/local/src/mcp.ts b/apps/local/src/mcp.ts index 4b9b80977..b26114a5f 100644 --- a/apps/local/src/mcp.ts +++ b/apps/local/src/mcp.ts @@ -1,4 +1,4 @@ -import { Deferred, Effect } from "effect"; +import { Effect } from "effect"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; @@ -14,7 +14,8 @@ import { formatResumeAcknowledgement, readElicitationMode, } from "@executor-js/host-mcp/browser-approval"; -import type { ResumeResponse } from "@executor-js/execution"; +import { makeInProcessBrowserApprovalStore } from "@executor-js/host-mcp/browser-approval-store"; +import { formatPausedExecution, type ResumeResponse } from "@executor-js/execution"; import { startIntegrationsRefresh } from "./integrations"; @@ -24,6 +25,9 @@ import { startIntegrationsRefresh } from "./integrations"; export type McpRequestHandler = { readonly handleRequest: (request: Request) => Promise; + /** GET `/api/mcp-sessions/:id/executions/:id` — paused detail for the console. */ + readonly handlePausedRequest: (request: Request) => Promise; + /** POST `/api/mcp-sessions/:id/executions/:id/resume` — record the decision. */ readonly handleApprovalRequest: (request: Request) => Promise; readonly close: () => Promise; }; @@ -52,6 +56,7 @@ const ignoreClose = (close: (() => Promise) | undefined): Promise => ) : Promise.resolve(); +const pausedRequestPattern = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)$/; const approvalRequestPattern = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)\/resume$/; const json = (value: unknown, status = 200): Response => @@ -77,16 +82,29 @@ const resumeApprovalResult = (executionId: string, response: ResumeResponse) => export const createMcpRequestHandler = (config: ExecutorMcpServerConfig): McpRequestHandler => { const transports = new Map(); const servers = new Map(); - const approvalResponses = new Map>(); - const approvalWaiters = new Map>>(); + const approvals = makeInProcessBrowserApprovalStore(); + // Local runs one shared engine across every MCP session (main.ts builds it and + // passes it in), so the paused-execution lookup for browser approval reads it + // directly — there is no per-session engine to track. + const engine = "engine" in config ? config.engine : null; + + const pausedDetail = ( + executionId: string, + ): Promise | null> => + engine + ? Effect.runPromise( + engine.getPausedExecution(executionId).pipe( + Effect.map((paused) => (paused ? formatPausedExecution(paused) : null)), + Effect.orElseSucceed(() => null), + ), + ) + : Promise.resolve(null); const dispose = async (id: string, opts: { transport?: boolean; server?: boolean } = {}) => { const t = transports.get(id); const s = servers.get(id); transports.delete(id); servers.delete(id); - approvalResponses.delete(id); - approvalWaiters.delete(id); if (opts.transport) await ignoreClose(t ? () => t.close() : undefined); if (opts.server) await ignoreClose(s ? () => s.close() : undefined); }; @@ -125,48 +143,7 @@ export const createMcpRequestHandler = (config: ExecutorMcpServerConfig): McpReq created = await Effect.runPromise( createExecutorMcpServer({ ...config, - browserApprovalStore: { - takeResponse: (executionId) => - Effect.sync(() => { - if (!createdSessionId) return null; - const sessionApprovals = approvalResponses.get(createdSessionId); - const response = sessionApprovals?.get(executionId) ?? null; - sessionApprovals?.delete(executionId); - return response; - }), - waitForResponse: (executionId) => - Effect.gen(function* () { - if (!createdSessionId) return null; - const sessionApprovals = approvalResponses.get(createdSessionId); - const response = sessionApprovals?.get(executionId) ?? null; - if (response) { - sessionApprovals?.delete(executionId); - return response; - } - - const sessionWaiters = - approvalWaiters.get(createdSessionId) ?? - new Map>(); - const waiter = - sessionWaiters.get(executionId) ?? (yield* Deferred.make()); - sessionWaiters.set(executionId, waiter); - approvalWaiters.set(createdSessionId, sessionWaiters); - - yield* Deferred.await(waiter).pipe( - Effect.ensuring( - Effect.sync(() => { - if (sessionWaiters.get(executionId) === waiter) { - sessionWaiters.delete(executionId); - } - }), - ), - ); - const approvals = approvalResponses.get(createdSessionId); - const approved = approvals?.get(executionId) ?? null; - approvals?.delete(executionId); - return approved; - }), - }, + browserApprovalStore: approvals.store, elicitationMode: elicitationMode === "browser" ? { @@ -197,26 +174,29 @@ export const createMcpRequestHandler = (config: ExecutorMcpServerConfig): McpReq } }, + handlePausedRequest: async (request) => { + const match = pausedRequestPattern.exec(new URL(request.url).pathname); + if (!match) return json({ error: "Not found" }, 404); + if (request.method !== "GET") return json({ error: "Method not allowed" }, 405); + + const paused = await pausedDetail(decodeURIComponent(match[2])); + if (!paused) return json({ error: "Paused execution not found" }, 404); + return json({ text: paused.text, structured: paused.structured }); + }, + handleApprovalRequest: async (request) => { - const url = new URL(request.url); - const match = approvalRequestPattern.exec(url.pathname); + const match = approvalRequestPattern.exec(new URL(request.url).pathname); if (!match) return json({ error: "Not found" }, 404); if (request.method !== "POST") return json({ error: "Method not allowed" }, 405); - const sessionId = decodeURIComponent(match[1]); const executionId = decodeURIComponent(match[2]); - if (!servers.has(sessionId)) return json({ error: "MCP session not found" }, 404); + // The shared engine must still hold the paused execution — guards stale ids. + if (!(await pausedDetail(executionId))) return json({ error: "MCP session not found" }, 404); const response = await readResumeResponse(request); if (!response) return json({ error: "Invalid approval response" }, 400); - const sessionApprovals = - approvalResponses.get(sessionId) ?? new Map(); - sessionApprovals.set(executionId, response); - approvalResponses.set(sessionId, sessionApprovals); - const waiter = approvalWaiters.get(sessionId)?.get(executionId); - if (waiter) await Effect.runPromise(Deferred.succeed(waiter, response)); - + await Effect.runPromise(approvals.recordResponse(executionId, response)); return json(resumeApprovalResult(executionId, response)); }, diff --git a/apps/local/src/serve.test.ts b/apps/local/src/serve.test.ts index 9b46bc400..4786abe33 100644 --- a/apps/local/src/serve.test.ts +++ b/apps/local/src/serve.test.ts @@ -21,6 +21,7 @@ const startTestServer = async (): Promise => { mcp: { handleRequest: async () => new Response("ok"), handleApprovalRequest: async () => new Response("ok"), + handlePausedRequest: async () => new Response("ok"), close: async () => {}, }, }, @@ -80,6 +81,7 @@ describe("startServer network bind auth", () => { mcp: { handleRequest: async () => new Response("ok"), handleApprovalRequest: async () => new Response("ok"), + handlePausedRequest: async () => new Response("ok"), close: async () => {}, }, }, @@ -101,6 +103,7 @@ describe("startServer network bind auth", () => { mcp: { handleRequest: async () => new Response("ok"), handleApprovalRequest: async () => new Response("ok"), + handlePausedRequest: async () => new Response("ok"), close: async () => {}, }, }, @@ -131,6 +134,7 @@ describe("startServer network bind auth", () => { mcp: { handleRequest: async () => new Response("ok"), handleApprovalRequest: async () => new Response("ok"), + handlePausedRequest: async () => new Response("ok"), close: async () => {}, }, }, @@ -166,6 +170,7 @@ describe("startServer network bind auth", () => { mcp: { handleRequest: async () => new Response("ok"), handleApprovalRequest: async () => new Response("ok"), + handlePausedRequest: async () => new Response("ok"), close: async () => {}, }, }, diff --git a/apps/local/src/serve.ts b/apps/local/src/serve.ts index cc6ef3b9c..2fd49ff0c 100644 --- a/apps/local/src/serve.ts +++ b/apps/local/src/serve.ts @@ -326,7 +326,11 @@ export async function startServer(opts: StartServerOptions = {}): Promise - (principal: Principal) => + (principal: Principal, options?: McpBuildServerOptions) => makeExecutionStack( principal.accountId, principal.organizationId, @@ -46,7 +47,11 @@ export const makeMcpBuildServer = Effect.map(({ engine }) => engine), Effect.provide(executionStack), Effect.mapError((cause) => new McpEngineBuildError({ cause })), - Effect.flatMap((engine) => createExecutorMcpServer({ engine })), + Effect.flatMap((engine) => + createExecutorMcpServer({ engine, ...(options ?? {}) }).pipe( + Effect.map((mcpServer) => ({ mcpServer, engine })), + ), + ), ); /** diff --git a/packages/hosts/mcp/package.json b/packages/hosts/mcp/package.json index ac5655b31..31d591afa 100644 --- a/packages/hosts/mcp/package.json +++ b/packages/hosts/mcp/package.json @@ -19,6 +19,10 @@ "./browser-approval": { "types": "./src/browser-approval.ts", "default": "./src/browser-approval.ts" + }, + "./browser-approval-store": { + "types": "./src/browser-approval-store.ts", + "default": "./src/browser-approval-store.ts" } }, "scripts": { diff --git a/packages/hosts/mcp/src/browser-approval-store.ts b/packages/hosts/mcp/src/browser-approval-store.ts new file mode 100644 index 000000000..812b7c49e --- /dev/null +++ b/packages/hosts/mcp/src/browser-approval-store.ts @@ -0,0 +1,71 @@ +// --------------------------------------------------------------------------- +// In-process browser-approval store — the single-process equivalent of the +// Durable Object's persisted approval responses (apps/cloud, host-cloudflare). +// +// It is the bridge between the two halves of a browser approval: +// - the MCP `resume` tool long-polls `store.waitForResponse(executionId)`, +// - the HTTP approval endpoint records the human's decision via +// `recordResponse(executionId, response)`, which wakes that waiter. +// +// Keyed by executionId alone — execution ids are unique per execution, so one +// store serves every session in the process. The in-memory MCP session store +// and the local app both build on it. +// --------------------------------------------------------------------------- + +import { Deferred, Effect } from "effect"; + +import type { ResumeResponse } from "@executor-js/execution"; + +import type { BrowserApprovalStore } from "./tool-server"; + +export interface InProcessBrowserApprovalStore { + /** The store the MCP server awaits a decision on (browser elicitation mode). */ + readonly store: BrowserApprovalStore; + /** Record a human's decision, waking any in-flight `waitForResponse`. */ + readonly recordResponse: (executionId: string, response: ResumeResponse) => Effect.Effect; + /** Drop a pending decision/waiter (e.g. when its session is torn down). */ + readonly forget: (executionId: string) => void; +} + +export const makeInProcessBrowserApprovalStore = (): InProcessBrowserApprovalStore => { + const responses = new Map(); + const waiters = new Map>(); + + const take = (executionId: string): Effect.Effect => + Effect.sync(() => { + const response = responses.get(executionId) ?? null; + if (response) responses.delete(executionId); + return response; + }); + + const waitFor = (executionId: string): Effect.Effect => + Effect.gen(function* () { + const existing = yield* take(executionId); + if (existing) return existing; + + const waiter = waiters.get(executionId) ?? (yield* Deferred.make()); + waiters.set(executionId, waiter); + yield* Deferred.await(waiter).pipe( + Effect.ensuring( + Effect.sync(() => { + if (waiters.get(executionId) === waiter) waiters.delete(executionId); + }), + ), + ); + return yield* take(executionId); + }); + + return { + store: { takeResponse: take, waitForResponse: waitFor }, + recordResponse: (executionId, response) => + Effect.gen(function* () { + responses.set(executionId, response); + const waiter = waiters.get(executionId); + if (waiter) yield* Deferred.succeed(waiter, response); + }), + forget: (executionId) => { + responses.delete(executionId); + waiters.delete(executionId); + }, + }; +}; diff --git a/packages/hosts/mcp/src/in-memory-session-store.ts b/packages/hosts/mcp/src/in-memory-session-store.ts index d22f8a16a..393c829b0 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.ts @@ -1,7 +1,19 @@ -import { Data, Effect, Layer } from "effect"; +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 { + approvalUrlForRequest, + decodeResumeResponse, + formatResumeAcknowledgement, + readElicitationMode, +} from "./browser-approval"; +import { + makeInProcessBrowserApprovalStore, + type InProcessBrowserApprovalStore, +} from "./browser-approval-store"; import { jsonRpcErrorBody } from "./envelope"; import { McpSessionStore, @@ -10,23 +22,26 @@ import { type McpDispatchResult, type Principal, } from "./seams"; +import type { BrowserApprovalStore } from "./tool-server"; // --------------------------------------------------------------------------- // In-process McpSessionStore — the single-node serving store, shared by every -// host that has no cross-isolate session backend (self-host, the Cloudflare -// QuickJS host). Cloud's Durable Object store is the cross-isolate variant of -// the same `McpSessionStore` seam. +// host that has no cross-isolate session backend (self-host, the local app). +// Cloud's Durable Object store is the cross-isolate variant of the same +// `McpSessionStore` seam. // // In the two-seam envelope the store owns the ENTIRE session lifecycle via // `dispatch`: create (no session id + POST initialize), forward (session id -// present), and ownership (cross-bearer). Three Maps keyed by mcp-session-id — -// transports, servers, owners — hold the live in-process sessions. Closing a -// session is just closing its transport + server. +// present), and ownership (cross-bearer). Maps keyed by mcp-session-id hold the +// live in-process sessions: transports, servers, owners, and — for the browser +// approval flow — the per-session engines. // -// The engine is a store implementation detail, not an envelope seam: the store -// builds each per-session `McpServer` through the host-supplied `buildServer` -// (the host's execution stack over its own DB + code substrate). The two-seam -// envelope has no engine seam — the store owns engine construction. +// Browser approval: when the create request carries `?elicitation_mode=browser`, +// the store builds the session's server in browser mode (an `approvalUrl` + the +// shared in-process approval store) and keeps the session's engine so the HTTP +// approval endpoints (`handlePausedRequest` / `handleApprovalRequest`) can read +// the paused execution and record the human's decision. The Durable Object +// hosts do the equivalent with `ctx.storage`. // // `dispatch` returns the transport `Response` to pass through, or: // - "not-found" (unknown session id) -> envelope renders 404 -32001 @@ -38,14 +53,42 @@ export class McpEngineBuildError extends Data.TaggedError("McpEngineBuildError") readonly cause: unknown; }> {} -/** Build the per-session `McpServer` for a principal (the host's engine + tools). */ +/** The connected MCP server plus the engine the approval endpoints drive. */ +export interface BuiltMcpServer { + readonly mcpServer: McpServer; + readonly engine: ExecutionEngine; +} + +/** The browser-mode wiring the store hands a build call when a session opts in. */ +export interface McpBuildServerOptions { + readonly elicitationMode?: + | { readonly mode: "browser"; readonly approvalUrl: (executionId: string) => string } + | { readonly mode: "model" } + | { readonly mode: "native" }; + readonly browserApprovalStore?: BrowserApprovalStore; +} + +/** Build the per-session `McpServer` + engine for a principal (the host's engine + tools). */ export type McpBuildServer = ( principal: Principal, -) => Effect.Effect; + options?: McpBuildServerOptions, +) => Effect.Effect; export interface InMemoryMcpSessionStore { /** The `McpSessionStore` seam value to hand to `inMemoryMcpSessionsLayer`. */ readonly store: McpSessionStore["Service"]; + /** + * Serve `GET /api/mcp-sessions/:sessionId/executions/:executionId` — the + * paused-execution detail the console approval page renders. Returns the + * paused `{ text, structured }` or a 404. Null if the path does not match. + */ + readonly handlePausedRequest: (request: Request) => Promise; + /** + * Serve `POST /api/mcp-sessions/:sessionId/executions/:executionId/resume` — + * record the human's decision and wake the long-polling `resume` tool call. + * Null if the path does not match. + */ + readonly handleApprovalRequest: (request: Request) => Promise; /** Dispose every live session — wire into the host's shutdown (not a seam). */ readonly close: () => Promise; } @@ -65,6 +108,12 @@ const formatBoundaryError = (error: unknown): unknown => const jsonRpcError = (status: number, code: number, message: string): Response => jsonRpcErrorBody(status, code, message, { cors: false }); +const json = (value: unknown, status = 200): Response => + new Response(JSON.stringify(value), { status, headers: { "content-type": "application/json" } }); + +const PAUSED_PATH = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)$/; +const RESUME_PATH = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)\/resume$/; + /** * Build the in-process session store plus an explicit `close()` that disposes * all live sessions. `close()` is not part of the seam — it is the host lifetime @@ -77,6 +126,8 @@ export const makeInMemoryMcpSessionStore = ( const transports = new Map(); const servers = new Map(); const owners = new Map(); + const engines = new Map>(); + const approvals: InProcessBrowserApprovalStore = makeInProcessBrowserApprovalStore(); const dispose = async (id: string, opts: { transport?: boolean; server?: boolean } = {}) => { const transport = transports.get(id); @@ -84,6 +135,7 @@ export const makeInMemoryMcpSessionStore = ( transports.delete(id); servers.delete(id); owners.delete(id); + engines.delete(id); if (opts.transport) await ignoreClose(transport ? () => transport.close() : undefined); if (opts.server) await ignoreClose(server ? () => server.close() : undefined); }; @@ -126,18 +178,44 @@ export const makeInMemoryMcpSessionStore = ( return runHandleRequest(transport, request); }; + /** + * The browser-mode wiring for a create request: when the client asks for + * `elicitation_mode=browser`, build the server with an `approvalUrl` (anchored + * at the request origin + the session id, minted on initialize) and the shared + * approval store. Otherwise pass the bare model/native mode through. + */ + const buildOptionsFor = ( + request: Request, + sessionId: () => string | null, + ): McpBuildServerOptions => { + if (readElicitationMode(request) !== "browser") return { elicitationMode: { mode: "model" } }; + return { + elicitationMode: { + mode: "browser", + approvalUrl: (executionId) => approvalUrlForRequest(request, executionId, sessionId()), + }, + browserApprovalStore: approvals.store, + }; + }; + /** Open a new session: build the server, connect a transport, drive the request. */ - const create = (principal: Principal, request: Request): Effect.Effect => - buildServer(principal).pipe( - Effect.flatMap((server) => + const create = (principal: Principal, request: Request): Effect.Effect => { + let createdSessionId: string | null = null; + return buildServer( + principal, + buildOptionsFor(request, () => createdSessionId), + ).pipe( + Effect.flatMap(({ mcpServer, engine }) => Effect.gen(function* () { const transport = new WebStandardStreamableHTTPServerTransport({ sessionIdGenerator: () => crypto.randomUUID(), enableJsonResponse: true, onsessioninitialized: (sid) => { + createdSessionId = sid; transports.set(sid, transport); - servers.set(sid, server); + servers.set(sid, mcpServer); owners.set(sid, principal); + engines.set(sid, engine); }, onsessionclosed: (sid) => void dispose(sid, { server: true }), }); @@ -145,12 +223,12 @@ export const makeInMemoryMcpSessionStore = ( const sid = transport.sessionId; if (sid) void dispose(sid, { server: true }); }; - yield* Effect.promise(() => server.connect(transport)); + yield* Effect.promise(() => mcpServer.connect(transport)); // The session id is minted on the first (initialize) request, so we // drive `handleRequest` here; if no id results we close eagerly. return yield* runHandleRequest(transport, request, () => { void ignoreClose(() => transport.close()); - void ignoreClose(() => server.close()); + void ignoreClose(() => mcpServer.close()); }); }), ), @@ -159,6 +237,7 @@ export const makeInMemoryMcpSessionStore = ( Effect.succeed(jsonRpcError(500, -32603, "Internal server error")), ), ); + }; const store: McpSessionStore["Service"] = { dispatch: ({ request, principal, sessionId }: McpDispatchInput) => @@ -167,8 +246,65 @@ export const makeInMemoryMcpSessionStore = ( Effect.promise(() => dispose(sessionId, { transport: true, server: true })), }; + /** Resolve a paused execution from the session that owns it, for HTTP approval. */ + const pausedFromSession = ( + sessionId: string, + executionId: string, + ): Promise | null> => { + const engine = engines.get(sessionId); + if (!engine) return Promise.resolve(null); + return Effect.runPromise( + engine.getPausedExecution(executionId).pipe( + Effect.map((paused) => (paused ? formatPausedExecution(paused) : null)), + Effect.orElseSucceed(() => null), + ), + ); + }; + + const handlePausedRequest = async (request: Request): Promise => { + const match = PAUSED_PATH.exec(new URL(request.url).pathname); + if (!match) return null; + if (request.method !== "GET") return json({ error: "Method not allowed" }, 405); + const paused = await pausedFromSession( + decodeURIComponent(match[1]!), + decodeURIComponent(match[2]!), + ); + if (!paused) return json({ error: "Paused execution not found" }, 404); + return json({ text: paused.text, structured: paused.structured }); + }; + + const handleApprovalRequest = async (request: Request): Promise => { + const match = RESUME_PATH.exec(new URL(request.url).pathname); + if (!match) return null; + if (request.method !== "POST") return json({ error: "Method not allowed" }, 405); + + const sessionId = decodeURIComponent(match[1]!); + const executionId = decodeURIComponent(match[2]!); + // The session must still hold the paused execution — guards stale ids and + // confirms the execution belongs to this session before recording. + const paused = await pausedFromSession(sessionId, executionId); + if (!paused) return json({ error: "Paused execution not found" }, 404); + + const raw = await Effect.runPromise( + Effect.tryPromise({ try: () => request.json(), catch: () => null }).pipe( + Effect.orElseSucceed(() => null), + ), + ); + const response = raw === null ? null : decodeResumeResponse(raw); + if (!response) return json({ error: "Invalid approval response" }, 400); + + await Effect.runPromise(approvals.recordResponse(executionId, response)); + return json({ + status: "completed", + ...formatResumeAcknowledgement(executionId, response), + isError: false, + }); + }; + return { store, + handlePausedRequest, + handleApprovalRequest, close: async () => { const ids = new Set([...transports.keys(), ...servers.keys()]); await Promise.all([...ids].map((id) => dispose(id, { transport: true, server: true }))); diff --git a/packages/react/src/routes/resume.$executionId.tsx b/packages/react/src/routes/resume.$executionId.tsx index 5077f6569..8ce0878e4 100644 --- a/packages/react/src/routes/resume.$executionId.tsx +++ b/packages/react/src/routes/resume.$executionId.tsx @@ -5,7 +5,6 @@ import * as Atom from "effect/unstable/reactivity/Atom"; import { createFileRoute } from "@tanstack/react-router"; import { ResumeApprovalPage, ResumeApprovalPageView } from "../pages/resume-approval"; -import { pausedExecutionAtom } from "../api/atoms"; import type { ElicitationAction } from "../components/elicitation-approval"; const SearchParams = Schema.toStandardSchemaV1( @@ -31,6 +30,48 @@ class LocalMcpResumeError extends Data.TaggedError("LocalMcpResumeError")<{ readonly message: string; }> {} +const McpPausedExecutionInfo = Schema.Struct({ + text: Schema.String, + structured: Schema.Unknown, +}); +const decodeMcpPausedExecutionInfo = Schema.decodeUnknownOption(McpPausedExecutionInfo); + +// Paused-execution detail for the in-process / Cloudflare hosts, fetched +// session-scoped: the MCP paused execution lives in its session's engine, so it +// is resolved through `/api/mcp-sessions/:id/...`, not the session-less +// `/api/executions/:id` (which hits a different engine and can't see it). Cloud +// serves the equivalent through its own route + Durable Object RPC. +const mcpPausedExecutionAtom = Atom.family( + (key: { readonly mcpSessionId: string; readonly executionId: string }) => + Atom.make( + Effect.gen(function* () { + const response = yield* Effect.tryPromise({ + try: () => + fetch( + `/api/mcp-sessions/${encodeURIComponent(key.mcpSessionId)}/executions/${encodeURIComponent(key.executionId)}`, + ), + catch: () => new LocalMcpResumeError({ message: "Failed to load the paused execution." }), + }); + if (!response.ok) { + return yield* new LocalMcpResumeError({ + message: `Paused execution unavailable (${response.status}).`, + }); + } + const body = yield* Effect.tryPromise({ + try: () => response.json(), + catch: () => new LocalMcpResumeError({ message: "Paused response was not valid JSON." }), + }); + const decoded = decodeMcpPausedExecutionInfo(body); + if (Option.isNone(decoded)) { + return yield* new LocalMcpResumeError({ + message: "Paused response had an unexpected shape.", + }); + } + return decoded.value; + }), + ), +); + type LocalMcpResumeInput = { readonly mcpSessionId: string; readonly executionId: string; @@ -96,7 +137,9 @@ function RouteComponent() { } function LocalMcpResumeApproval(props: { executionId: string; mcpSessionId: string }) { - const paused = useAtomValue(pausedExecutionAtom(props.executionId)); + const paused = useAtomValue( + mcpPausedExecutionAtom({ mcpSessionId: props.mcpSessionId, executionId: props.executionId }), + ); const doResume = useAtomSet(resumeLocalMcpExecution, { mode: "promiseExit" }); const resume = useCallback( (executionId: string, action: ElicitationAction, content?: Record) =>