diff --git a/apps/cloud/src/api/protected.test.ts b/apps/cloud/src/api/protected.test.ts index 5a1a02283e..858f9710f4 100644 --- a/apps/cloud/src/api/protected.test.ts +++ b/apps/cloud/src/api/protected.test.ts @@ -17,6 +17,8 @@ const makeBaseEngine = (): ExecutionEngine => result: { result: "ok", logs: [] }, }), getPausedExecution: () => Effect.succeed(null), + pausedExecutionCount: () => Effect.succeed(0), + hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed("desc"), }) as ExecutionEngine; diff --git a/apps/cloud/src/engine/execution-usage.ts b/apps/cloud/src/engine/execution-usage.ts index c2dafc33d6..701eeb160b 100644 --- a/apps/cloud/src/engine/execution-usage.ts +++ b/apps/cloud/src/engine/execution-usage.ts @@ -19,5 +19,7 @@ export const withExecutionUsageTracking = ( // resume doesn't count as usage resume: (executionId, response) => engine.resume(executionId, response), getPausedExecution: (executionId) => engine.getPausedExecution(executionId), + pausedExecutionCount: () => engine.pausedExecutionCount(), + hasPausedExecutions: () => engine.hasPausedExecutions(), getDescription: engine.getDescription, }); diff --git a/apps/host-cloudflare/src/mcp/session-durable-object.ts b/apps/host-cloudflare/src/mcp/session-durable-object.ts index 386f09028b..b09449d285 100644 --- a/apps/host-cloudflare/src/mcp/session-durable-object.ts +++ b/apps/host-cloudflare/src/mcp/session-durable-object.ts @@ -93,6 +93,7 @@ export class McpSessionDO extends McpAgentSessionDOBase=1.3.0 <1.10.0" } }, "sha512-lID/vxSuKWXM55XhAKNoYXu9Cutoq5hFdkbTdI/zDKQktXzcWBVhNsOkiZFTMU9UtEWuGRNe0HUgmsFldIdxVA=="], "@opentelemetry/exporter-logs-otlp-proto/@opentelemetry/sdk-trace-base": ["@opentelemetry/sdk-trace-base@2.6.1", "", { "dependencies": { "@opentelemetry/core": "2.6.1", "@opentelemetry/resources": "2.6.1", "@opentelemetry/semantic-conventions": "^1.29.0" }, "peerDependencies": { "@opentelemetry/api": ">=1.3.0 <1.10.0" } }, "sha512-r86ut4T1e8vNwB35CqCcKd45yzqH6/6Wzvpk2/cZB8PsPLlZFTvrh8yfOS3CYZYcUmAx4hHTZJ8AO8Dj8nrdhw=="], @@ -6312,8 +6308,6 @@ "drizzle-kit/esbuild": ["esbuild@0.25.12", "", { "optionalDependencies": { "@esbuild/aix-ppc64": "0.25.12", "@esbuild/android-arm": "0.25.12", "@esbuild/android-arm64": "0.25.12", "@esbuild/android-x64": "0.25.12", "@esbuild/darwin-arm64": "0.25.12", "@esbuild/darwin-x64": "0.25.12", "@esbuild/freebsd-arm64": "0.25.12", "@esbuild/freebsd-x64": "0.25.12", "@esbuild/linux-arm": "0.25.12", "@esbuild/linux-arm64": "0.25.12", "@esbuild/linux-ia32": "0.25.12", "@esbuild/linux-loong64": "0.25.12", "@esbuild/linux-mips64el": "0.25.12", "@esbuild/linux-ppc64": "0.25.12", "@esbuild/linux-riscv64": "0.25.12", "@esbuild/linux-s390x": "0.25.12", "@esbuild/linux-x64": "0.25.12", "@esbuild/netbsd-arm64": "0.25.12", "@esbuild/netbsd-x64": "0.25.12", "@esbuild/openbsd-arm64": "0.25.12", "@esbuild/openbsd-x64": "0.25.12", "@esbuild/openharmony-arm64": "0.25.12", "@esbuild/sunos-x64": "0.25.12", "@esbuild/win32-arm64": "0.25.12", "@esbuild/win32-ia32": "0.25.12", "@esbuild/win32-x64": "0.25.12" }, "bin": { "esbuild": "bin/esbuild" } }, "sha512-bbPBYYrtZbkt6Os6FiTLCTFxvq4tt3JKall1vRwshA3fdVztsLAatFaZobhkBC8/BrPetoa0oksYoKXoG4ryJg=="], - "effect/yaml": ["yaml@2.8.3", "", { "bin": { "yaml": "bin.mjs" } }, "sha512-AvbaCLOO2Otw/lW5bmh9d/WEdcDFdQp2Z2ZUH3pX9U2ihyUY0nvLv7J6TrWowklRGPYbB/IuIMfYgxaCPg5Bpg=="], - "electron-builder/chalk": ["chalk@4.1.2", "", { "dependencies": { "ansi-styles": "^4.1.0", "supports-color": "^7.1.0" } }, "sha512-oKnbhFyRIXpUuez8iBMmyEa4nbj4IOQyuhc/wy9kY7/WVPcwIO9VA668Pu8RkO7+0G76SLROeyw9CpQ061i4mA=="], "electron-builder/fs-extra": ["fs-extra@10.1.0", "", { "dependencies": { "graceful-fs": "^4.2.0", "jsonfile": "^6.0.1", "universalify": "^2.0.0" } }, "sha512-oRXApq54ETRj4eMiFzGnHWGy+zo5raudjuxN0b8H7s/RU2oW0Wvsx9O0ACRN/kRq9E8Vu/ReskGB5o3ji+FzHQ=="], @@ -6382,12 +6376,12 @@ "just-bash/quickjs-emscripten": ["quickjs-emscripten@0.32.0", "", { "dependencies": { "@jitl/quickjs-wasmfile-debug-asyncify": "0.32.0", "@jitl/quickjs-wasmfile-debug-sync": "0.32.0", "@jitl/quickjs-wasmfile-release-asyncify": "0.32.0", "@jitl/quickjs-wasmfile-release-sync": "0.32.0", "quickjs-emscripten-core": "0.32.0" } }, "sha512-So0Sqw869y/S2oE3Nuc0uT3Dhqgvsj8FSrwBdsuTosVsG8ME5/OcudU1GxsrIFdFABgy17GHnTVO9TYV/bLQcA=="], + "just-bash/yaml": ["yaml@2.9.0", "", { "bin": { "yaml": "bin.mjs" } }, "sha512-2AvhNX3mb8zd6Zy7INTtSpl1F15HW6Wnqj0srWlkKLcpYl/gMIMJiyuGq2KeI2YFxUPjdlB+3Lc10seMLtL4cA=="], + "katex/commander": ["commander@8.3.0", "", {}, "sha512-OkTL9umf+He2DZkUq8f8J9of7yL6RJKI24dVITBmNfZBmri9zYZQrKkuXiKhyfPSu8tUhnVBB1iKXevvnlR4Ww=="], "knip/jiti": ["jiti@2.6.1", "", { "bin": { "jiti": "lib/jiti-cli.mjs" } }, "sha512-ekilCSN1jwRvIbgeg/57YFh8qQDNbwDb9xT/qu2DAHbFFZUicIl4ygVaAvzveMhMVr3LnpSKTNnwt8PoOfmKhQ=="], - "knip/yaml": ["yaml@2.8.3", "", { "bin": { "yaml": "bin.mjs" } }, "sha512-AvbaCLOO2Otw/lW5bmh9d/WEdcDFdQp2Z2ZUH3pX9U2ihyUY0nvLv7J6TrWowklRGPYbB/IuIMfYgxaCPg5Bpg=="], - "libsql/detect-libc": ["detect-libc@2.0.2", "", {}, "sha512-UX6sGumvvqSaXgdKGUsgZWqcUyIXZ/vZTrlRT/iobiKhGL0zL4d3osHj3uqllWJK+i+sixDS/3COVEOFbupFyw=="], "magicast/@babel/parser": ["@babel/parser@7.29.2", "", { "dependencies": { "@babel/types": "^7.29.0" }, "bin": "./bin/babel-parser.js" }, "sha512-4GgRzy/+fsBa72/RZVJmGKPmZu9Byn8o4MoLpmNe1m8ZfYnz5emHLQz3U4gLud6Zwl0RZIcgiLD7Uq7ySFuDLA=="], @@ -6706,6 +6700,8 @@ "@executor-js/cloudflare/agents/partysocket": ["partysocket@1.2.0", "", { "dependencies": { "event-target-polyfill": "^0.0.4" }, "peerDependencies": { "react": ">=17" }, "optionalPeers": ["react"] }, "sha512-xgXql4N0b3umN263lHkrZMvvtYC906h4YjY8l63LNcF0x1bfJmQtZLQGqNQWzFHAxLns/6h6/rjdLW4EdYqQwA=="], + "@executor-js/cloudflare/agents/yaml": ["yaml@2.9.0", "", { "bin": { "yaml": "bin.mjs" } }, "sha512-2AvhNX3mb8zd6Zy7INTtSpl1F15HW6Wnqj0srWlkKLcpYl/gMIMJiyuGq2KeI2YFxUPjdlB+3Lc10seMLtL4cA=="], + "@executor-js/cloudflare/agents/yargs": ["yargs@18.0.0", "", { "dependencies": { "cliui": "^9.0.1", "escalade": "^3.1.1", "get-caller-file": "^2.0.5", "string-width": "^7.2.0", "y18n": "^5.0.5", "yargs-parser": "^22.0.0" } }, "sha512-4UEqdc2RYGHZc7Doyqkrqiln3p9X2DZVxaGbwhn2pi7MrRagKaOcIKe8L3OxYcbhXLgLFUS3zAYuQjKBQgmuNg=="], "@executor-js/e2e/@types/node/undici-types": ["undici-types@7.24.6", "", {}, "sha512-WRNW+sJgj5OBN4/0JpHFqtqzhpbnV0GuB+OozA9gCL7a993SmU+1JBZCzLNxYsbMfIeDL+lTsphD5jN5N+n0zg=="], diff --git a/packages/core/execution/src/engine.test.ts b/packages/core/execution/src/engine.test.ts index 89f13e367b..593cf6b2b9 100644 --- a/packages/core/execution/src/engine.test.ts +++ b/packages/core/execution/src/engine.test.ts @@ -85,3 +85,18 @@ describe("executeWithPause failure propagation", () => { }), ); }); + +describe("pausedExecutionCount", () => { + it.effect("starts at zero", () => + Effect.gen(function* () { + const executor = yield* makeExecutor(); + const engine = createExecutionEngine({ + executor, + codeExecutor: succeedingExecutor, + }); + + expect(yield* engine.pausedExecutionCount()).toBe(0); + expect(yield* engine.hasPausedExecutions()).toBe(false); + }), + ); +}); diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts index 8325c5c997..31e1fbc953 100644 --- a/packages/core/execution/src/engine.ts +++ b/packages/core/execution/src/engine.ts @@ -421,6 +421,12 @@ export type ExecutionEngine */ readonly getPausedExecution: (executionId: string) => Effect.Effect; + /** Count of executions currently paused awaiting resume. */ + readonly pausedExecutionCount: () => Effect.Effect; + + /** Whether any executions are paused awaiting resume. */ + readonly hasPausedExecutions: () => Effect.Effect; + /** * Get the dynamic tool description (workflow + namespaces). */ @@ -852,6 +858,8 @@ export const createExecutionEngine = Effect.sync(() => pausedExecutions.get(executionId) ?? null), + pausedExecutionCount: () => Effect.sync(() => pausedExecutions.size), + hasPausedExecutions: () => Effect.sync(() => pausedExecutions.size > 0), getDescription: buildExecuteDescription(executor), }; }; diff --git a/packages/core/execution/src/promise.ts b/packages/core/execution/src/promise.ts index cf57304a59..87878b95af 100644 --- a/packages/core/execution/src/promise.ts +++ b/packages/core/execution/src/promise.ts @@ -47,6 +47,8 @@ export type ExecutionEngine = { response: ResumeResponse, ) => Promise; readonly getPausedExecution: (executionId: string) => Promise; + readonly pausedExecutionCount: () => Promise; + readonly hasPausedExecutions: () => Promise; readonly getDescription: () => Promise; }; @@ -152,6 +154,8 @@ export const toPromiseExecutionEngine = ( executeWithPause: (code) => Effect.runPromise(engine.executeWithPause(code)), resume: (executionId, response) => Effect.runPromise(engine.resume(executionId, response)), getPausedExecution: (executionId) => Effect.runPromise(engine.getPausedExecution(executionId)), + pausedExecutionCount: () => Effect.runPromise(engine.pausedExecutionCount()), + hasPausedExecutions: () => Effect.runPromise(engine.hasPausedExecutions()), getDescription: () => Effect.runPromise(engine.getDescription), }); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index f02e40312b..3ab3757dc7 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -3885,8 +3885,9 @@ export const createExecutor = void; + timeout: ReturnType | null; + expiring: boolean; +}; + export type McpSessionApprovalResult = | { readonly status: "ok"; @@ -81,9 +97,22 @@ export interface BrowserApprovalStore { const SESSION_META_KEY = "session-meta"; const LAST_ACTIVITY_KEY = "last-activity-ms"; -const SESSION_TIMEOUT_MS = 5 * 60 * 1000; +const PARTYSERVER_NAME_KEY = "__ps_name"; +const MCP_HTTP_METHOD_HEADER = "cf-mcp-method"; +const MCP_MESSAGE_HEADER = "cf-mcp-message"; +const ACTIVE_POST_RESPONSE_WAIT_POLL_MS = 25; +const ACTIVE_POST_RESPONSE_WAIT_MAX_MS = PAUSED_APPROVAL_TIMEOUT_MS + 30_000; const approvalResponseKey = (executionId: string) => `approval-response:${executionId}`; +type JsonRpcRequestId = string | number; +const JsonRpcRequestWithId = Schema.Struct({ + id: Schema.Union([Schema.String, Schema.Number]), + method: Schema.String, +}); +const JsonRpcPostPayload = Schema.fromJsonString(Schema.Unknown); +const decodeJsonRpcPostPayload = Schema.decodeUnknownOption(JsonRpcPostPayload); +const decodeJsonRpcRequestWithId = Schema.decodeUnknownOption(JsonRpcRequestWithId); + const resumeApprovalResult = ( executionId: string, response: ResumeResponse, @@ -115,6 +144,46 @@ const isSessionProps = (props: unknown): props is McpSessionProps => typeof (props as { readonly session?: unknown }).session === "object" && (props as { readonly session?: unknown }).session !== null; +const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); + +const readActivePostRequestIds = (request: Request): readonly JsonRpcRequestId[] => { + if (request.headers.get(MCP_HTTP_METHOD_HEADER) !== "POST") return []; + const encoded = request.headers.get(MCP_MESSAGE_HEADER); + if (!encoded) return []; + const decoded = Effect.runSyncExit( + Effect.try({ + try: () => atob(encoded), + catch: () => "invalid_base64" as const, + }), + ); + if (Exit.isFailure(decoded)) { + console.warn( + JSON.stringify({ + event: "mcp_active_post_response_wait_parse_failed", + reason: "invalid_base64", + }), + ); + return []; + } + const parsed = decodeJsonRpcPostPayload(decoded.value); + if (Option.isNone(parsed)) { + console.warn( + JSON.stringify({ + event: "mcp_active_post_response_wait_parse_failed", + reason: "invalid_json", + }), + ); + return []; + } + const messages = Array.isArray(parsed.value) ? parsed.value : [parsed.value]; + const requestIds: JsonRpcRequestId[] = []; + for (const message of messages) { + const decoded = decodeJsonRpcRequestWithId(message); + if (Option.isSome(decoded)) requestIds.push(decoded.value.id); + } + return requestIds; +}; + export abstract class McpAgentSessionDOBase< Env extends Cloudflare.Env = Cloudflare.Env, TDbHandle extends SessionDbHandle = SessionDbHandle, @@ -127,6 +196,7 @@ export abstract class McpAgentSessionDOBase< private lastActivityMs = 0; private approvalResponses = new Map(); private approvalWaiters = new Map>(); + private pendingApprovalLeases = new Map(); protected abstract openSessionDb(): TDbHandle | Promise; @@ -163,10 +233,53 @@ export abstract class McpAgentSessionDOBase< waitForResponse: (executionId) => this.waitForApprovalResponse(executionId), }; + protected readonly pausedExecutionHooks: PausedExecutionHooks = { + onExecutionPaused: (executionId) => + Effect.sync(() => { + this.queuePendingApprovalLeaseStart(executionId); + }), + onResumeStarted: (executionId) => this.beginPendingApprovalResume(executionId), + onResumeSettled: (executionId) => this.finishPendingApprovalResume(executionId), + }; + + override async onConnect(conn: Connection, context: ConnectionContext): Promise { + const requestIds = readActivePostRequestIds(context.request); + if (requestIds.length === 0) { + await super.onConnect(conn, context); + return; + } + + await this.keepAliveWhile(async () => { + await super.onConnect(conn, context); + await this.waitForActivePostResponseDrain(conn.id); + }); + } + private openSessionDbHandle(): Effect.Effect { return Effect.promise(() => Promise.resolve(this.openSessionDb())); } + private async waitForActivePostResponseDrain(streamId: string): Promise { + const startedAt = Date.now(); + for (;;) { + const requestIds = await this.getStreamRequestIds(streamId); + if (!requestIds || requestIds.length === 0) return; + if (Date.now() - startedAt >= ACTIVE_POST_RESPONSE_WAIT_MAX_MS) { + console.warn( + JSON.stringify({ + event: "mcp_active_post_response_wait_timeout", + sessionId: this.sessionId, + streamId, + requestIds, + elapsedMs: Date.now() - startedAt, + }), + ); + return; + } + await sleep(ACTIVE_POST_RESPONSE_WAIT_POLL_MS); + } + } + private loadSessionMeta(): Effect.Effect { return Effect.promise(async () => { if (this.sessionMeta) return this.sessionMeta; @@ -196,6 +309,22 @@ export abstract class McpAgentSessionDOBase< return this.lastActivityMs; } + private async hasPartyServerName(): Promise { + if (this.ctx.id.name) return true; + const stored = await this.ctx.storage.get(PARTYSERVER_NAME_KEY); + return !!stored; + } + + private async cleanupUnaddressableSessionAlarm(): Promise { + await Effect.runPromise(this.closeRuntime()); + await Effect.runPromise( + Effect.all([ + Effect.ignore(Effect.tryPromise(() => this.ctx.storage.deleteAlarm())), + Effect.ignore(Effect.tryPromise(() => this.ctx.storage.delete(LAST_ACTIVITY_KEY))), + ]), + ); + } + private resolveAndStoreSessionMeta(token: McpSessionInit) { const self = this; return Effect.gen(function* () { @@ -238,6 +367,7 @@ export abstract class McpAgentSessionDOBase< private closeRuntime(): Effect.Effect { const self = this; return Effect.gen(function* () { + yield* self.releaseAllPendingApprovalLeases(); if (self.server) { const server = self.server; yield* Effect.promise(() => server.close()).pipe(Effect.ignore); @@ -399,12 +529,7 @@ export abstract class McpAgentSessionDOBase< const paused = yield* self.engine.getPausedExecution(executionId); if (!paused) return { status: "not_found" } as const; - self.approvalResponses.set(executionId, response); - yield* Effect.promise(() => - self.ctx.storage.put(approvalResponseKey(executionId), response), - ); - const waiter = self.approvalWaiters.get(executionId); - if (waiter) yield* Deferred.succeed(waiter, response); + yield* self.recordApprovalResponse(executionId, response); return resumeApprovalResult(executionId, response); }).pipe( Effect.withSpan("McpSessionDO.resumeExecutionForApproval", { @@ -423,13 +548,42 @@ export abstract class McpAgentSessionDOBase< await super.destroy(); } + private async pausedExecutionCount(): Promise { + if (!this.engine) return 0; + return Effect.runPromise(this.engine.pausedExecutionCount()); + } + override async alarm(): Promise { + if (!(await this.hasPartyServerName())) { + await this.cleanupUnaddressableSessionAlarm(); + return; + } const lastActivityMs = await this.loadLastActivity(); - if (lastActivityMs > 0 && Date.now() - lastActivityMs >= SESSION_TIMEOUT_MS) { - await this.destroy(); + const idleMs = lastActivityMs > 0 ? Date.now() - lastActivityMs : 0; + const pausedExecutionCount = await this.pausedExecutionCount(); + const decision = decideSessionAlarm({ idleMs, pausedExecutionCount }); + + if (decision.kind === "idle_within_timeout") { + await super.alarm(); + return; + } + + if (decision.kind === "extend_paused_lease") { + console.info( + JSON.stringify( + pausedLeaseExtensionLog({ + sessionId: this.sessionId, + pausedExecutionCount, + idleMs, + leaseMs: decision.leaseMs, + }), + ), + ); + await this.ctx.storage.setAlarm(Date.now() + decision.leaseMs); return; } - await super.alarm(); + + await this.destroy(); } private validateApprovalIdentity( @@ -446,6 +600,141 @@ export abstract class McpAgentSessionDOBase< }).pipe(Effect.withSpan("mcp.session.validate_approval_identity")); } + private startPendingApprovalLease(executionId: string): Effect.Effect { + const self = this; + return Effect.gen(function* () { + if (self.pendingApprovalLeases.has(executionId)) return; + + yield* Effect.promise(() => self.markActivity()).pipe( + Effect.withSpan("McpSessionDO.markActivity"), + ); + const disposeKeepAlive = yield* Effect.promise(() => self.keepAlive()); + const timeout = setTimeout(() => { + self.queuePendingApprovalLeaseExpiration(executionId); + }, PAUSED_APPROVAL_TIMEOUT_MS); + self.pendingApprovalLeases.set(executionId, { disposeKeepAlive, timeout, expiring: false }); + }).pipe( + Effect.withSpan("McpSessionDO.pending_approval_lease.start", { + attributes: { "mcp.execution.id": executionId }, + }), + Effect.tapCause((cause) => + Effect.sync(() => { + console.error("[mcp-session] pending approval lease start failed:", Cause.pretty(cause)); + self.captureCause(cause); + }), + ), + Effect.ignore, + ); + } + + private queuePendingApprovalLeaseStart(executionId: string): void { + this.ctx.waitUntil(Effect.runPromise(this.startPendingApprovalLease(executionId))); + } + + private queuePendingApprovalLeaseExpiration(executionId: string): void { + this.ctx.waitUntil( + Effect.runPromise( + this.expirePendingApproval(executionId).pipe( + Effect.tapCause((cause) => + Effect.sync(() => { + console.error( + "[mcp-session] pending approval lease expiration failed:", + Cause.pretty(cause), + ); + this.captureCause(cause); + }), + ), + Effect.ignore, + ), + ), + ); + } + + private beginPendingApprovalResume(executionId: string): Effect.Effect { + return Effect.sync(() => { + const lease = this.pendingApprovalLeases.get(executionId); + if (!lease || lease.expiring) return; + if (lease.timeout) clearTimeout(lease.timeout); + lease.timeout = null; + }).pipe( + Effect.withSpan("McpSessionDO.pending_approval_lease.begin_resume", { + attributes: { "mcp.execution.id": executionId }, + }), + ); + } + + private finishPendingApprovalResume(executionId: string): Effect.Effect { + return this.releasePendingApprovalLease(executionId).pipe( + Effect.withSpan("McpSessionDO.pending_approval_lease.finish", { + attributes: { "mcp.execution.id": executionId }, + }), + ); + } + + private releasePendingApprovalLease(executionId: string): Effect.Effect { + return Effect.sync(() => { + const lease = this.pendingApprovalLeases.get(executionId); + if (!lease) return; + if (lease.timeout) clearTimeout(lease.timeout); + this.pendingApprovalLeases.delete(executionId); + lease.disposeKeepAlive(); + }); + } + + private releaseAllPendingApprovalLeases(): Effect.Effect { + return Effect.sync(() => { + for (const executionId of this.pendingApprovalLeases.keys()) { + const lease = this.pendingApprovalLeases.get(executionId); + if (!lease) continue; + if (lease.timeout) clearTimeout(lease.timeout); + lease.disposeKeepAlive(); + } + this.pendingApprovalLeases.clear(); + }); + } + + private recordApprovalResponse( + executionId: string, + response: ResumeResponse, + ): Effect.Effect { + const self = this; + return Effect.gen(function* () { + self.approvalResponses.set(executionId, response); + yield* Effect.promise(() => self.ctx.storage.put(approvalResponseKey(executionId), response)); + const waiter = self.approvalWaiters.get(executionId); + if (waiter) yield* Deferred.succeed(waiter, response); + }); + } + + private expirePendingApproval(executionId: string): Effect.Effect { + const self = this; + return Effect.gen(function* () { + const lease = self.pendingApprovalLeases.get(executionId); + if (!lease || lease.expiring) return; + lease.expiring = true; + if (lease.timeout) clearTimeout(lease.timeout); + lease.timeout = null; + if (self.approvalResponses.has(executionId)) return; + + const response = { + action: "decline", + content: { reason: "approval_timeout" }, + } satisfies ResumeResponse; + yield* Effect.sync(() => { + console.info(JSON.stringify({ event: "mcp_pending_approval_lease_expire", executionId })); + }); + yield* self.recordApprovalResponse(executionId, response); + if (self.engine && !self.approvalWaiters.has(executionId)) { + yield* self.engine.resume(executionId, response).pipe(Effect.ignore); + } + }).pipe( + Effect.ensuring(self.releasePendingApprovalLease(executionId)), + Effect.withSpan("McpSessionDO.pending_approval_lease.expire", { + attributes: { "mcp.execution.id": executionId }, + }), + ); + } + private takeApprovalResponse(executionId: string): Effect.Effect { const self = this; return Effect.promise(async () => { diff --git a/packages/hosts/cloudflare/src/mcp/session-alarm-policy.test.ts b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.test.ts new file mode 100644 index 0000000000..48db8c918a --- /dev/null +++ b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.test.ts @@ -0,0 +1,58 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { + MAX_PAUSED_SESSION_IDLE_MS, + PAUSED_EXECUTION_LEASE_MS, + SESSION_TIMEOUT_MS, + decideSessionAlarm, +} from "./session-alarm-policy"; + +describe("decideSessionAlarm", () => { + it("keeps the session within the idle timeout", () => { + expect( + decideSessionAlarm({ + idleMs: SESSION_TIMEOUT_MS - 1, + pausedExecutionCount: 0, + }), + ).toEqual({ kind: "idle_within_timeout" }); + }); + + it("stays within the idle timeout regardless of paused work", () => { + expect( + decideSessionAlarm({ + idleMs: SESSION_TIMEOUT_MS - 1, + pausedExecutionCount: 3, + }), + ).toEqual({ kind: "idle_within_timeout" }); + }); + + it("destroys an idle session with no paused work", () => { + expect( + decideSessionAlarm({ + idleMs: SESSION_TIMEOUT_MS, + pausedExecutionCount: 0, + }), + ).toEqual({ kind: "destroy_idle_session" }); + }); + + it("extends the lease when paused continuations exist", () => { + expect( + decideSessionAlarm({ + idleMs: SESSION_TIMEOUT_MS, + pausedExecutionCount: 1, + }), + ).toEqual({ + kind: "extend_paused_lease", + leaseMs: PAUSED_EXECUTION_LEASE_MS, + }); + }); + + it("destroys a paused session once the lease cap elapses", () => { + expect( + decideSessionAlarm({ + idleMs: MAX_PAUSED_SESSION_IDLE_MS, + pausedExecutionCount: 1, + }), + ).toEqual({ kind: "destroy_idle_session" }); + }); +}); diff --git a/packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts new file mode 100644 index 0000000000..f4c79caa0d --- /dev/null +++ b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts @@ -0,0 +1,47 @@ +import { PAUSED_APPROVAL_TIMEOUT_MS } from "@executor-js/host-mcp/tool-server"; + +/** Idle timeout for MCP sessions with no paused continuations. */ +export const SESSION_TIMEOUT_MS = 5 * 60 * 1000; + +/** Lease extension while paused executions block hibernation (matches browser approval wait). */ +export const PAUSED_EXECUTION_LEASE_MS = PAUSED_APPROVAL_TIMEOUT_MS; + +/** + * Hard upper bound on idle time before a paused session is torn down regardless + * of outstanding paused work. The lease grants a single approval window of grace + * past the idle timeout — once it elapses the browser approval wait has already + * timed out (see tool-server `waitForBrowserApprovalResponse`), so the paused + * execution is no longer resumable and the DO must not keep extending forever. + */ +export const MAX_PAUSED_SESSION_IDLE_MS = SESSION_TIMEOUT_MS + PAUSED_EXECUTION_LEASE_MS; + +export type SessionAlarmDecision = + | { readonly kind: "idle_within_timeout" } + | { readonly kind: "destroy_idle_session" } + | { readonly kind: "extend_paused_lease"; readonly leaseMs: number }; + +export const decideSessionAlarm = (input: { + readonly idleMs: number; + readonly pausedExecutionCount: number; +}): SessionAlarmDecision => { + if (input.idleMs < SESSION_TIMEOUT_MS) { + return { kind: "idle_within_timeout" }; + } + if (input.pausedExecutionCount > 0 && input.idleMs < MAX_PAUSED_SESSION_IDLE_MS) { + return { kind: "extend_paused_lease", leaseMs: PAUSED_EXECUTION_LEASE_MS }; + } + return { kind: "destroy_idle_session" }; +}; + +export const pausedLeaseExtensionLog = (input: { + readonly sessionId: string; + readonly pausedExecutionCount: number; + readonly idleMs: number; + readonly leaseMs: number; +}): Record => ({ + event: "mcp_session_paused_lease_extension", + sessionId: input.sessionId, + pausedExecutionCount: input.pausedExecutionCount, + idleMs: input.idleMs, + leaseMs: input.leaseMs, +}); diff --git a/packages/hosts/cloudflare/src/mcp/session-durable-object.ts b/packages/hosts/cloudflare/src/mcp/session-durable-object.ts index de83d370e2..9f7e8bf539 100644 --- a/packages/hosts/cloudflare/src/mcp/session-durable-object.ts +++ b/packages/hosts/cloudflare/src/mcp/session-durable-object.ts @@ -27,6 +27,7 @@ import { } from "@executor-js/execution"; import { makeMcpWorkerTransport, type McpWorkerTransport } from "./worker-transport"; +import { decideSessionAlarm, pausedLeaseExtensionLog } from "./session-alarm-policy"; import { INTERNAL_ACCOUNT_ID_HEADER, INTERNAL_ORGANIZATION_ID_HEADER, @@ -82,7 +83,6 @@ const resumeApprovalResult = ( }); const HEARTBEAT_MS = 30 * 1000; -const SESSION_TIMEOUT_MS = 5 * 60 * 1000; const TRANSPORT_STATE_KEY = "transport"; const SESSION_META_KEY = "session-meta"; const LAST_ACTIVITY_KEY = "last-activity-ms"; @@ -968,14 +968,38 @@ export abstract class McpSessionDOBase< ); } + private async pausedExecutionCount(): Promise { + if (!this.engine) return 0; + return Effect.runPromise(this.engine.pausedExecutionCount()); + } + private async runAlarm(): Promise { const lastActivityMs = await this.loadLastActivity(); const idleMs = Date.now() - lastActivityMs; - if (idleMs >= SESSION_TIMEOUT_MS) { + const pausedExecutionCount = await this.pausedExecutionCount(); + const decision = decideSessionAlarm({ idleMs, pausedExecutionCount }); + + if (decision.kind === "extend_paused_lease") { + console.info( + JSON.stringify( + pausedLeaseExtensionLog({ + sessionId: this.sessionId, + pausedExecutionCount, + idleMs, + leaseMs: decision.leaseMs, + }), + ), + ); + await this.ctx.storage.setAlarm(Date.now() + decision.leaseMs); + return; + } + + if (decision.kind === "destroy_idle_session") { await Effect.runPromise(this.closeRuntime()); await this.ctx.storage.deleteAlarm(); return; } + await this.ctx.storage.setAlarm(Date.now() + HEARTBEAT_MS); } diff --git a/packages/hosts/mcp/src/tool-server.test.ts b/packages/hosts/mcp/src/tool-server.test.ts index 0f16b9b76f..e06a1c2af6 100644 --- a/packages/hosts/mcp/src/tool-server.test.ts +++ b/packages/hosts/mcp/src/tool-server.test.ts @@ -38,6 +38,8 @@ const makeStubEngine = (overrides: { (() => Effect.succeed({ status: "completed", result: { result: "default" } })), resume: overrides.resume ?? (() => Effect.succeed(null)), getPausedExecution: () => Effect.succeed(null), + pausedExecutionCount: () => Effect.succeed(0), + hasPausedExecutions: () => Effect.succeed(false), getDescription: Effect.succeed(overrides.description ?? "test executor"), }); @@ -46,7 +48,10 @@ const withClient = async ( engine: ExecutionEngine, capabilities: ClientCapabilities, fn: (client: Client) => Promise, - config?: Pick, "debug" | "elicitationMode" | "browserApprovalStore">, + config?: Pick< + ExecutorMcpServerConfig, + "debug" | "elicitationMode" | "browserApprovalStore" | "pausedExecutionHooks" + >, ) => { const mcpServer = await Effect.runPromise(createExecutorMcpServer({ engine, ...config })); const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); @@ -1042,6 +1047,49 @@ describe("MCP host server — client without elicitation (pause/resume)", () => }); }); + it("reports pause and resume lifecycle boundaries", async () => { + const events: string[] = []; + const approval = FormElicitation.make({ message: "Approve", requestedSchema: {} }); + const engine = makeStubEngine({ + executeWithPause: () => Effect.succeed(makePausedResult("exec_1", approval)), + resume: () => Effect.succeed(makePausedResult("exec_2", approval)), + }); + + await withClient( + engine, + NO_CAPS, + async (client) => { + await client.callTool({ + name: "execute", + arguments: { code: "pause" }, + }); + expect(events).toEqual(["pause:exec_1"]); + + await client.callTool({ + name: "resume", + arguments: { executionId: "exec_1", action: "accept", content: "{}" }, + }); + expect(events).toEqual([ + "pause:exec_1", + "resume-start:exec_1", + "resume-settle:exec_1", + "pause:exec_2", + ]); + }, + { + elicitationMode: { mode: "model" }, + pausedExecutionHooks: { + onExecutionPaused: (executionId) => + Effect.sync(() => events.push(`pause:${executionId}`)), + onResumeStarted: (executionId) => + Effect.sync(() => events.push(`resume-start:${executionId}`)), + onResumeSettled: (executionId) => + Effect.sync(() => events.push(`resume-settle:${executionId}`)), + }, + }, + ); + }); + it("default model resume mode explains empty form schemas as model-side confirmation", async () => { const engine = makeStubEngine({ executeWithPause: () => diff --git a/packages/hosts/mcp/src/tool-server.ts b/packages/hosts/mcp/src/tool-server.ts index 39de36db1d..bd482b88c4 100644 --- a/packages/hosts/mcp/src/tool-server.ts +++ b/packages/hosts/mcp/src/tool-server.ts @@ -1,4 +1,4 @@ -import { Effect, Match, Option, Schema } from "effect"; +import { Duration, Effect, Match, Option, Schema } from "effect"; import * as Cause from "effect/Cause"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { ContentBlockSchema, type ContentBlock } from "@modelcontextprotocol/sdk/types.js"; @@ -99,6 +99,12 @@ type SharedMcpServerConfig = { * it, runs carry no trigger (the historical behaviour). */ readonly trigger?: ExecutionTrigger; + /** + * Host-owned lifecycle for paused executions. The MCP server reports pause + * boundaries; the host decides whether that means a keepAlive lease, browser + * wait, durable record, or no-op. + */ + readonly pausedExecutionHooks?: PausedExecutionHooks; }; export type ExecutorMcpServerConfig = @@ -112,6 +118,15 @@ export type BrowserApprovalStore = { readonly waitForResponse?: (executionId: string) => Effect.Effect; }; +export const PAUSED_APPROVAL_TIMEOUT_MS = 4 * 60 * 1000; +const BROWSER_APPROVAL_WAIT_TIMEOUT_MS = PAUSED_APPROVAL_TIMEOUT_MS + 1000; + +export type PausedExecutionHooks = { + readonly onExecutionPaused?: (executionId: string) => Effect.Effect; + readonly onResumeStarted?: (executionId: string) => Effect.Effect; + readonly onResumeSettled?: (executionId: string) => Effect.Effect; +}; + // --------------------------------------------------------------------------- // Elicitation bridge // --------------------------------------------------------------------------- @@ -553,6 +568,17 @@ export const createExecutorMcpServer = ( ({ mode: "model", } as const); + const onExecutionPaused = (executionId: string): Effect.Effect => + config.pausedExecutionHooks?.onExecutionPaused?.(executionId) ?? Effect.void; + const onResumeStarted = (executionId: string): Effect.Effect => + config.pausedExecutionHooks?.onResumeStarted?.(executionId) ?? Effect.void; + const onResumeSettled = (executionId: string): Effect.Effect => + config.pausedExecutionHooks?.onResumeSettled?.(executionId) ?? Effect.void; + const resumeWithLifecycle = (executionId: string, response: ResumeResponse) => + Effect.gen(function* () { + yield* onResumeStarted(executionId); + return yield* engine.resume(executionId, response); + }).pipe(Effect.ensuring(onResumeSettled(executionId))); const resolveParentSpan = (): Tracer.AnySpan | undefined => { const ps = config.parentSpan; @@ -604,6 +630,14 @@ export const createExecutorMcpServer = ( ? pausedInteractionKind(outcome.execution.elicitationContext.request) : undefined, }); + if (outcome.status === "paused") { + yield* Effect.annotateCurrentSpan({ + "mcp.execute.paused": true, + "mcp.execute.paused_execution_id": outcome.execution.id, + "mcp.execute.pause_source": "execute", + }); + yield* onExecutionPaused(outcome.execution.id); + } return outcome.status === "completed" ? toMcpResult(outcome.result) : elicitationMode.mode === "browser" @@ -630,7 +664,7 @@ export const createExecutorMcpServer = ( hasContent: content !== undefined, clientCapabilities: server.server.getClientCapabilities() ?? null, }); - const outcome = yield* engine.resume(executionId, { action, content }); + const outcome = yield* resumeWithLifecycle(executionId, { action, content }); if (!outcome) { debugLog("resume.missing_execution", { executionId }); return missingExecutionResult(executionId); @@ -644,6 +678,14 @@ export const createExecutorMcpServer = ( ? pausedInteractionKind(outcome.execution.elicitationContext.request) : undefined, }); + if (outcome.status === "paused") { + yield* Effect.annotateCurrentSpan({ + "mcp.execute.paused": true, + "mcp.execute.paused_execution_id": outcome.execution.id, + "mcp.execute.pause_source": "resume", + }); + yield* onExecutionPaused(outcome.execution.id); + } return outcome.status === "completed" ? toMcpResult(outcome.result) : toMcpPausedResult(formatPausedExecution(outcome.execution)); @@ -692,7 +734,7 @@ export const createExecutorMcpServer = ( return waitForResponse(executionId).pipe( Effect.timeoutOrElse({ - duration: "10 minutes", + duration: Duration.millis(BROWSER_APPROVAL_WAIT_TIMEOUT_MS), orElse: () => Effect.succeed(null), }), ); @@ -703,10 +745,18 @@ export const createExecutorMcpServer = ( const response = yield* waitForBrowserApprovalResponse(executionId); if (!response) return yield* requireUserResumeApproval(executionId); - const outcome = yield* engine.resume(executionId, response); + const outcome = yield* resumeWithLifecycle(executionId, response); if (!outcome) { return missingExecutionResult(executionId); } + if (outcome.status === "paused") { + yield* Effect.annotateCurrentSpan({ + "mcp.execute.paused": true, + "mcp.execute.paused_execution_id": outcome.execution.id, + "mcp.execute.pause_source": "browser_resume", + }); + yield* onExecutionPaused(outcome.execution.id); + } return outcome.status === "completed" ? toMcpResult(outcome.result) : yield* requireUserResumeApproval(outcome.execution.id); diff --git a/packages/plugins/execution-history/src/sdk/store.ts b/packages/plugins/execution-history/src/sdk/store.ts index a00513f686..802a4fda38 100644 --- a/packages/plugins/execution-history/src/sdk/store.ts +++ b/packages/plugins/execution-history/src/sdk/store.ts @@ -62,6 +62,63 @@ const ownerOf = (binding: OwnerBinding): Owner => (binding.subject != null ? "us /** Hoisted: `Schema.is` compiles a guard, so it must not be rebuilt per row. */ const isRunStatus = Schema.is(RunStatus); +const MAX_INTERACTION_ARG_KEYS = 25; +const MAX_INTERACTION_STRING_CHARS = 300; + +const isRecord = (value: unknown): value is Readonly> => + typeof value === "object" && value !== null && !Array.isArray(value); + +const previewScalar = (value: unknown): unknown => { + if (value === null || typeof value === "boolean" || typeof value === "number") return value; + if (typeof value === "string") { + return value.length > MAX_INTERACTION_STRING_CHARS + ? `${value.slice(0, MAX_INTERACTION_STRING_CHARS)}...` + : value; + } + if (Array.isArray(value)) return `[Array(${value.length})]`; + if (isRecord(value)) return `[Object(${Object.keys(value).length})]`; + return `[${typeof value}]`; +}; + +const argsPreviewOf = (args: unknown): unknown => { + if (!isRecord(args)) return previewScalar(args); + const entries = Object.entries(args).slice(0, MAX_INTERACTION_ARG_KEYS); + const preview: Record = {}; + for (const [key, value] of entries) { + preview[key] = previewScalar(value); + } + const omitted = Object.keys(args).length - entries.length; + return omitted > 0 ? { ...preview, __omittedKeys: omitted } : preview; +}; + +const interactionPayloadOf = ( + context: Extract["context"], +) => { + const request = context.request; + if (Predicate.isTagged(request, "UrlElicitation")) { + return { + address: String(context.address), + argsPreview: argsPreviewOf(context.args), + request: { + kind: "UrlElicitation", + message: request.message, + url: request.url, + elicitationId: request.elicitationId, + }, + }; + } + + return { + address: String(context.address), + argsPreview: argsPreviewOf(context.args), + request: { + kind: "FormElicitation", + message: request.message, + requestedSchema: request.requestedSchema, + }, + }; +}; + /** First dot-delimited segment of a tool path (its namespace), or null. */ const namespaceOf = (path: string): string | null => { const index = path.indexOf("."); @@ -456,7 +513,7 @@ export const makeExecutionHistoryStore = (deps: StorageDeps): ExecutionHistorySt status: "pending", kind, purpose: request.message, - payloadJson: toJson(event.context), + payloadJson: toJson(interactionPayloadOf(event.context)), responseJson: null, errorText: null, startedAt: event.startedAt.getTime(), diff --git a/packages/plugins/openapi/src/sdk/backing.ts b/packages/plugins/openapi/src/sdk/backing.ts index 2415eeb6cb..0018671e9e 100644 --- a/packages/plugins/openapi/src/sdk/backing.ts +++ b/packages/plugins/openapi/src/sdk/backing.ts @@ -651,15 +651,14 @@ export const resolveOpenApiBackedAnnotations = (input: { readonly toolRows: readonly { readonly name: string }[]; }) => Effect.gen(function* () { - const ops = yield* input.ctx.storage.listOperations(String(input.integration)); - const byName = new Map(); - for (const op of ops) byName.set(op.toolName, op.binding); const out: Record> = {}; for (const row of input.toolRows) { - const binding = byName.get(row.name); - if (binding) { - out[row.name] = annotationsForOperation(binding.method, binding.pathTemplate); - } + const operation = yield* input.ctx.storage.getOperation(input.integration, row.name); + if (!operation) continue; + out[row.name] = annotationsForOperation( + operation.binding.method, + operation.binding.pathTemplate, + ); } return out; });