From 6a94828a88f28d4a43a5a80fc9a90407c668f3d2 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 23 Jun 2026 16:19:22 +0530 Subject: [PATCH 1/7] feat(execution): add paused execution introspection API Expose pausedExecutionCount and hasPausedExecutions on ExecutionEngine so Cloudflare MCP session DOs can distinguish idle cleanup from paused continuations. --- packages/core/execution/src/engine.test.ts | 15 +++++++++++++++ packages/core/execution/src/engine.ts | 8 ++++++++ packages/core/execution/src/promise.ts | 4 ++++ packages/hosts/mcp/src/tool-server.test.ts | 2 ++ 4 files changed, 29 insertions(+) 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/hosts/mcp/src/tool-server.test.ts b/packages/hosts/mcp/src/tool-server.test.ts index 0f16b9b76f..52fcd409f5 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"), }); From c03fb17d719efb7a34fdf7913aa1858f186c9726 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 23 Jun 2026 16:19:28 +0530 Subject: [PATCH 2/7] feat(cloudflare): extend MCP session lease while paused continuations exist Add shared alarm policy so idle sessions still destroy after 5 minutes, but sessions with in-memory paused executions reschedule a 10-minute lease instead of tearing down the runtime. Preserves the legacy unaddressable-alarm guard. --- .../src/mcp/agent-session-durable-object.ts | 58 +++++++++++++++++-- .../src/mcp/session-alarm-policy.test.ts | 39 +++++++++++++ .../src/mcp/session-alarm-policy.ts | 36 ++++++++++++ .../src/mcp/session-durable-object.ts | 28 ++++++++- 4 files changed, 155 insertions(+), 6 deletions(-) create mode 100644 packages/hosts/cloudflare/src/mcp/session-alarm-policy.test.ts create mode 100644 packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts index 2dbf1a58ad..1deb2c9b8a 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts @@ -12,6 +12,11 @@ import { } from "@executor-js/execution"; import type { IncomingPropagationHeaders, McpElicitationMode } from "./do-headers"; +import { + SESSION_TIMEOUT_MS, + decideSessionAlarm, + pausedLeaseExtensionLog, +} from "./session-alarm-policy"; export type IncomingTraceHeaders = IncomingPropagationHeaders; @@ -81,7 +86,7 @@ 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 approvalResponseKey = (executionId: string) => `approval-response:${executionId}`; const resumeApprovalResult = ( @@ -196,6 +201,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* () { @@ -423,13 +444,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; } - await super.alarm(); + + 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 this.destroy(); } private validateApprovalIdentity( 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..7c99d3832f --- /dev/null +++ b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { + 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("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, + }); + }); +}); 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..9942e133bf --- /dev/null +++ b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts @@ -0,0 +1,36 @@ +/** 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 = 10 * 60 * 1000; + +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) { + 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); } From 88050b4511a3c3501ae0bd1d2fd273f1e78e5e02 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 23 Jun 2026 16:28:30 +0530 Subject: [PATCH 3/7] fix(cloud): forward paused execution introspection --- apps/cloud/src/api/protected.test.ts | 2 ++ apps/cloud/src/engine/execution-usage.ts | 2 ++ 2 files changed, 4 insertions(+) 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, }); From 16006fb5a24ec0c3a3adabca929fffd6c232b522 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 23 Jun 2026 17:11:09 +0530 Subject: [PATCH 4/7] fix(cloudflare): cap paused-session lease at one approval window (greptile) The extend_paused_lease branch rescheduled the alarm unconditionally, so a session whose paused execution was never resumed (browser tab closed, client disconnect) would extend the lease forever and keep the Durable Object alive and billable indefinitely. Bound the lease to a single approval window: once idle exceeds SESSION_TIMEOUT_MS + PAUSED_EXECUTION_LEASE_MS (15 min), the session is torn down regardless of outstanding paused work. By that point the browser approval wait (tool-server, 10 min) has already timed out, so the paused execution is no longer resumable. idleMs grows across extensions because activity is not marked during the wait, so decideSessionAlarm stays pure. Also add the missing coverage: idle_within_timeout wins regardless of paused work, and the new cap forces destroy_idle_session. --- .../src/mcp/session-alarm-policy.test.ts | 19 +++++++++++++++++++ .../src/mcp/session-alarm-policy.ts | 11 ++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/packages/hosts/cloudflare/src/mcp/session-alarm-policy.test.ts b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.test.ts index 7c99d3832f..48db8c918a 100644 --- a/packages/hosts/cloudflare/src/mcp/session-alarm-policy.test.ts +++ b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "@effect/vitest"; import { + MAX_PAUSED_SESSION_IDLE_MS, PAUSED_EXECUTION_LEASE_MS, SESSION_TIMEOUT_MS, decideSessionAlarm, @@ -16,6 +17,15 @@ describe("decideSessionAlarm", () => { ).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({ @@ -36,4 +46,13 @@ describe("decideSessionAlarm", () => { 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 index 9942e133bf..6ec5800cb1 100644 --- a/packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts +++ b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts @@ -4,6 +4,15 @@ 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 = 10 * 60 * 1000; +/** + * 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" } @@ -16,7 +25,7 @@ export const decideSessionAlarm = (input: { if (input.idleMs < SESSION_TIMEOUT_MS) { return { kind: "idle_within_timeout" }; } - if (input.pausedExecutionCount > 0) { + 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" }; From e05b68f0130320d3a02879561e1ea7298066115d Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Wed, 24 Jun 2026 05:12:43 +0530 Subject: [PATCH 5/7] fix(mcp): settle approval pauses under hibernation Add host-owned paused execution hooks so Cloudflare MCP sessions can hold a bounded pending-approval lease, keep active POST requests alive until their response drains, and release hibernation blockers on resume or timeout. Also bound the approval wait window, compact execution-history interaction payloads, and resolve OpenAPI-backed annotations with point lookups so large specs do not reset the Durable Object before returning a pause envelope. --- .../src/mcp/session-durable-object.ts | 1 + bun.lock | 14 +- packages/core/sdk/src/executor.ts | 3 +- .../src/mcp/agent-session-durable-object.ts | 258 +++++++++++++++++- .../src/mcp/session-alarm-policy.ts | 4 +- packages/hosts/mcp/src/tool-server.test.ts | 48 +++- packages/hosts/mcp/src/tool-server.ts | 119 +++++++- .../execution-history/src/sdk/store.ts | 59 +++- packages/plugins/openapi/src/sdk/backing.ts | 13 +- 9 files changed, 484 insertions(+), 35 deletions(-) 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/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"; @@ -87,8 +98,21 @@ export interface BrowserApprovalStore { const SESSION_META_KEY = "session-meta"; const LAST_ACTIVITY_KEY = "last-activity-ms"; 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, @@ -120,6 +144,30 @@ 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 parsed = decodeJsonRpcPostPayload(atob(encoded)); + if (Option.isNone(parsed)) { + console.warn( + JSON.stringify({ + event: "mcp_active_post_response_wait_parse_failed", + }), + ); + 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, @@ -132,6 +180,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; @@ -168,10 +217,68 @@ 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 () => { + console.info( + JSON.stringify({ + event: "mcp_active_post_response_wait_start", + sessionId: this.sessionId, + streamId: conn.id, + requestIds, + }), + ); + await super.onConnect(conn, context); + await this.waitForActivePostResponseDrain(conn.id); + console.info( + JSON.stringify({ + event: "mcp_active_post_response_wait_finish", + sessionId: this.sessionId, + streamId: 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; @@ -259,6 +366,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); @@ -420,12 +528,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", { @@ -496,6 +599,147 @@ 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.sync(() => { + console.info(JSON.stringify({ event: "mcp_pending_approval_lease_start", executionId })); + }); + 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 }); + yield* Effect.sync(() => { + console.info(JSON.stringify({ event: "mcp_pending_approval_lease_started", executionId })); + }); + }).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 { + console.info(JSON.stringify({ event: "mcp_pending_approval_lease_queued", executionId })); + 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; + + 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) { + 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.ts b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts index 6ec5800cb1..f4c79caa0d 100644 --- a/packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts +++ b/packages/hosts/cloudflare/src/mcp/session-alarm-policy.ts @@ -1,8 +1,10 @@ +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 = 10 * 60 * 1000; +export const PAUSED_EXECUTION_LEASE_MS = PAUSED_APPROVAL_TIMEOUT_MS; /** * Hard upper bound on idle time before a paused session is torn down regardless diff --git a/packages/hosts/mcp/src/tool-server.test.ts b/packages/hosts/mcp/src/tool-server.test.ts index 52fcd409f5..e06a1c2af6 100644 --- a/packages/hosts/mcp/src/tool-server.test.ts +++ b/packages/hosts/mcp/src/tool-server.test.ts @@ -48,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(); @@ -1044,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..e0f50f4238 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 // --------------------------------------------------------------------------- @@ -488,6 +503,14 @@ const toMcpFailureResult = (cause: Cause.Cause): McpToolResult => { }; }; +const logMcpPauseEvent = (event: string, data: Record): void => { + console.info(JSON.stringify({ event, ...data })); +}; + +const logMcpToolEvent = (event: string, data: Record): void => { + console.info(JSON.stringify({ event, ...data })); +}; + // A paused execution lives in the session runtime's memory: it expires when // the user takes too long to answer, and dies early when the runtime is // rebuilt (host restart, redeploy). Either way the recovery is the same and @@ -553,6 +576,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; @@ -595,7 +629,16 @@ export const createExecutorMcpServer = ( }); return toMcpResult(result); } + logMcpToolEvent("mcp_tool_execute_with_pause_start", { + codeLength: code.length, + elicitationMode: elicitationMode.mode, + }); const outcome = yield* engine.executeWithPause(code, { trigger }); + logMcpToolEvent("mcp_tool_execute_with_pause_result", { + status: outcome.status, + executionId: outcome.status === "paused" ? outcome.execution.id : undefined, + elicitationMode: elicitationMode.mode, + }); debugLog("execute.paused_flow_result", { status: outcome.status, executionId: outcome.status === "paused" ? outcome.execution.id : undefined, @@ -604,6 +647,19 @@ export const createExecutorMcpServer = ( ? pausedInteractionKind(outcome.execution.elicitationContext.request) : undefined, }); + if (outcome.status === "paused") { + logMcpPauseEvent("mcp_pause_created", { + executionId: outcome.execution.id, + mode: elicitationMode.mode, + source: "execute", + }); + yield* onExecutionPaused(outcome.execution.id); + logMcpPauseEvent("mcp_pause_returning", { + executionId: outcome.execution.id, + mode: elicitationMode.mode, + source: "execute", + }); + } return outcome.status === "completed" ? toMcpResult(outcome.result) : elicitationMode.mode === "browser" @@ -630,7 +686,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 +700,19 @@ export const createExecutorMcpServer = ( ? pausedInteractionKind(outcome.execution.elicitationContext.request) : undefined, }); + if (outcome.status === "paused") { + logMcpPauseEvent("mcp_pause_created", { + executionId: outcome.execution.id, + mode: elicitationMode.mode, + source: "resume", + }); + yield* onExecutionPaused(outcome.execution.id); + logMcpPauseEvent("mcp_pause_returning", { + executionId: outcome.execution.id, + mode: elicitationMode.mode, + source: "resume", + }); + } return outcome.status === "completed" ? toMcpResult(outcome.result) : toMcpPausedResult(formatPausedExecution(outcome.execution)); @@ -692,7 +761,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 +772,23 @@ 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") { + logMcpPauseEvent("mcp_pause_created", { + executionId: outcome.execution.id, + mode: elicitationMode.mode, + source: "browser_resume", + }); + yield* onExecutionPaused(outcome.execution.id); + logMcpPauseEvent("mcp_pause_returning", { + executionId: outcome.execution.id, + mode: elicitationMode.mode, + source: "browser_resume", + }); + } return outcome.status === "completed" ? toMcpResult(outcome.result) : yield* requireUserResumeApproval(outcome.execution.id); @@ -728,7 +810,13 @@ export const createExecutorMcpServer = ( description, inputSchema: { code: z.string().trim().min(1) }, }, - ({ code }) => runToolEffect(executeCode(code)), + ({ code }) => { + logMcpToolEvent("mcp_tool_execute_entry", { + codeLength: code.length, + elicitationMode: elicitationMode.mode, + }); + return runToolEffect(executeCode(code)); + }, ), ).pipe( Effect.withSpan("mcp.host.register_tool", { @@ -760,8 +848,16 @@ export const createExecutorMcpServer = ( .default("{}"), }, }, - ({ executionId, action, content: rawContent }) => - runToolEffect(resumeExecution(executionId, action, parseJsonContent(rawContent))), + ({ executionId, action, content: rawContent }) => { + logMcpToolEvent("mcp_tool_resume_entry", { + executionId, + action, + elicitationMode: elicitationMode.mode, + }); + return runToolEffect( + resumeExecution(executionId, action, parseJsonContent(rawContent)), + ); + }, ); } @@ -777,7 +873,14 @@ export const createExecutorMcpServer = ( executionId: z.string().describe("The execution ID from the paused result"), }, }, - ({ executionId }) => runToolEffect(resumeAfterBrowserApproval(executionId)), + ({ executionId }) => { + logMcpToolEvent("mcp_tool_resume_entry", { + executionId, + action: "browser_wait", + elicitationMode: elicitationMode.mode, + }); + return runToolEffect(resumeAfterBrowserApproval(executionId)); + }, ); }).pipe( Effect.withSpan("mcp.host.register_tool", { 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; }); From 692d69567bc7d9cf5a1879a243f7bf6cca1260b7 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Wed, 24 Jun 2026 05:33:32 +0530 Subject: [PATCH 6/7] refactor(mcp): align pause lifecycle telemetry Remove MCP host console logging wrappers and rely on existing debug logging plus span annotations for pause lifecycle state. Keep Cloudflare Durable Object logs focused on exceptional boundary events by dropping happy-path active POST and pending approval lease messages. --- .../src/mcp/agent-session-durable-object.ts | 22 ----- packages/hosts/mcp/src/tool-server.ts | 85 ++++--------------- 2 files changed, 16 insertions(+), 91 deletions(-) diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts index 761ef2d343..a4df24ce77 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts @@ -234,23 +234,8 @@ export abstract class McpAgentSessionDOBase< } await this.keepAliveWhile(async () => { - console.info( - JSON.stringify({ - event: "mcp_active_post_response_wait_start", - sessionId: this.sessionId, - streamId: conn.id, - requestIds, - }), - ); await super.onConnect(conn, context); await this.waitForActivePostResponseDrain(conn.id); - console.info( - JSON.stringify({ - event: "mcp_active_post_response_wait_finish", - sessionId: this.sessionId, - streamId: conn.id, - }), - ); }); } @@ -604,9 +589,6 @@ export abstract class McpAgentSessionDOBase< return Effect.gen(function* () { if (self.pendingApprovalLeases.has(executionId)) return; - yield* Effect.sync(() => { - console.info(JSON.stringify({ event: "mcp_pending_approval_lease_start", executionId })); - }); yield* Effect.promise(() => self.markActivity()).pipe( Effect.withSpan("McpSessionDO.markActivity"), ); @@ -615,9 +597,6 @@ export abstract class McpAgentSessionDOBase< self.queuePendingApprovalLeaseExpiration(executionId); }, PAUSED_APPROVAL_TIMEOUT_MS); self.pendingApprovalLeases.set(executionId, { disposeKeepAlive, timeout, expiring: false }); - yield* Effect.sync(() => { - console.info(JSON.stringify({ event: "mcp_pending_approval_lease_started", executionId })); - }); }).pipe( Effect.withSpan("McpSessionDO.pending_approval_lease.start", { attributes: { "mcp.execution.id": executionId }, @@ -633,7 +612,6 @@ export abstract class McpAgentSessionDOBase< } private queuePendingApprovalLeaseStart(executionId: string): void { - console.info(JSON.stringify({ event: "mcp_pending_approval_lease_queued", executionId })); this.ctx.waitUntil(Effect.runPromise(this.startPendingApprovalLease(executionId))); } diff --git a/packages/hosts/mcp/src/tool-server.ts b/packages/hosts/mcp/src/tool-server.ts index e0f50f4238..bd482b88c4 100644 --- a/packages/hosts/mcp/src/tool-server.ts +++ b/packages/hosts/mcp/src/tool-server.ts @@ -503,14 +503,6 @@ const toMcpFailureResult = (cause: Cause.Cause): McpToolResult => { }; }; -const logMcpPauseEvent = (event: string, data: Record): void => { - console.info(JSON.stringify({ event, ...data })); -}; - -const logMcpToolEvent = (event: string, data: Record): void => { - console.info(JSON.stringify({ event, ...data })); -}; - // A paused execution lives in the session runtime's memory: it expires when // the user takes too long to answer, and dies early when the runtime is // rebuilt (host restart, redeploy). Either way the recovery is the same and @@ -629,16 +621,7 @@ export const createExecutorMcpServer = ( }); return toMcpResult(result); } - logMcpToolEvent("mcp_tool_execute_with_pause_start", { - codeLength: code.length, - elicitationMode: elicitationMode.mode, - }); const outcome = yield* engine.executeWithPause(code, { trigger }); - logMcpToolEvent("mcp_tool_execute_with_pause_result", { - status: outcome.status, - executionId: outcome.status === "paused" ? outcome.execution.id : undefined, - elicitationMode: elicitationMode.mode, - }); debugLog("execute.paused_flow_result", { status: outcome.status, executionId: outcome.status === "paused" ? outcome.execution.id : undefined, @@ -648,17 +631,12 @@ export const createExecutorMcpServer = ( : undefined, }); if (outcome.status === "paused") { - logMcpPauseEvent("mcp_pause_created", { - executionId: outcome.execution.id, - mode: elicitationMode.mode, - source: "execute", + 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); - logMcpPauseEvent("mcp_pause_returning", { - executionId: outcome.execution.id, - mode: elicitationMode.mode, - source: "execute", - }); } return outcome.status === "completed" ? toMcpResult(outcome.result) @@ -701,17 +679,12 @@ export const createExecutorMcpServer = ( : undefined, }); if (outcome.status === "paused") { - logMcpPauseEvent("mcp_pause_created", { - executionId: outcome.execution.id, - mode: elicitationMode.mode, - source: "resume", + 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); - logMcpPauseEvent("mcp_pause_returning", { - executionId: outcome.execution.id, - mode: elicitationMode.mode, - source: "resume", - }); } return outcome.status === "completed" ? toMcpResult(outcome.result) @@ -777,17 +750,12 @@ export const createExecutorMcpServer = ( return missingExecutionResult(executionId); } if (outcome.status === "paused") { - logMcpPauseEvent("mcp_pause_created", { - executionId: outcome.execution.id, - mode: elicitationMode.mode, - source: "browser_resume", + 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); - logMcpPauseEvent("mcp_pause_returning", { - executionId: outcome.execution.id, - mode: elicitationMode.mode, - source: "browser_resume", - }); } return outcome.status === "completed" ? toMcpResult(outcome.result) @@ -810,13 +778,7 @@ export const createExecutorMcpServer = ( description, inputSchema: { code: z.string().trim().min(1) }, }, - ({ code }) => { - logMcpToolEvent("mcp_tool_execute_entry", { - codeLength: code.length, - elicitationMode: elicitationMode.mode, - }); - return runToolEffect(executeCode(code)); - }, + ({ code }) => runToolEffect(executeCode(code)), ), ).pipe( Effect.withSpan("mcp.host.register_tool", { @@ -848,16 +810,8 @@ export const createExecutorMcpServer = ( .default("{}"), }, }, - ({ executionId, action, content: rawContent }) => { - logMcpToolEvent("mcp_tool_resume_entry", { - executionId, - action, - elicitationMode: elicitationMode.mode, - }); - return runToolEffect( - resumeExecution(executionId, action, parseJsonContent(rawContent)), - ); - }, + ({ executionId, action, content: rawContent }) => + runToolEffect(resumeExecution(executionId, action, parseJsonContent(rawContent))), ); } @@ -873,14 +827,7 @@ export const createExecutorMcpServer = ( executionId: z.string().describe("The execution ID from the paused result"), }, }, - ({ executionId }) => { - logMcpToolEvent("mcp_tool_resume_entry", { - executionId, - action: "browser_wait", - elicitationMode: elicitationMode.mode, - }); - return runToolEffect(resumeAfterBrowserApproval(executionId)); - }, + ({ executionId }) => runToolEffect(resumeAfterBrowserApproval(executionId)), ); }).pipe( Effect.withSpan("mcp.host.register_tool", { From df61efca9b2e118eff80d12d52a7584970a48c33 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Wed, 24 Jun 2026 11:03:15 +0530 Subject: [PATCH 7/7] fix(mcp): guard approval timeout races Handle malformed active POST metadata without failing the Agent connection. Avoid overwriting a stored browser approval on timeout, and only resume timed-out executions directly when no MCP waiter is already registered. --- .../src/mcp/agent-session-durable-object.ts | 23 ++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts index a4df24ce77..a79de190fe 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts @@ -1,4 +1,4 @@ -import { Cause, Deferred, Effect, Option, Schema } from "effect"; +import { Cause, Deferred, Effect, Exit, Option, Schema } from "effect"; import type * as Tracer from "effect/Tracer"; import type { Connection, ConnectionContext } from "agents"; import { McpAgent } from "agents/mcp"; @@ -150,11 +150,27 @@ 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 parsed = decodeJsonRpcPostPayload(atob(encoded)); + 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 []; @@ -698,6 +714,7 @@ export abstract class McpAgentSessionDOBase< lease.expiring = true; if (lease.timeout) clearTimeout(lease.timeout); lease.timeout = null; + if (self.approvalResponses.has(executionId)) return; const response = { action: "decline", @@ -707,7 +724,7 @@ export abstract class McpAgentSessionDOBase< console.info(JSON.stringify({ event: "mcp_pending_approval_lease_expire", executionId })); }); yield* self.recordApprovalResponse(executionId, response); - if (self.engine) { + if (self.engine && !self.approvalWaiters.has(executionId)) { yield* self.engine.resume(executionId, response).pipe(Effect.ignore); } }).pipe(