// ---------------------------------------------------------------------------// Cloud MCP Session Durable Object — the cloud binding of the shared// `McpAgentSessionDOBase` (@executor-js/cloudflare). Hibernatable transport// serving, cold restore, the inactivity alarm, owner validation, browser// approval storage, and the per-request span bridge live in the base. Cloud// supplies ONLY its injected dependencies:// - openSessionDb → a long-lived postgres.js handle// - resolveSessionMeta → WorkOS/UserStore organization resolution// - buildMcpServer → the cloud execution stack + MCP tool server// - withTelemetry → the WebSdk tracer + W3C parent-span stitching// - captureCause → Sentry error capture// host-cloudflare binds the same base to D1 instead; the two stay byte-identical// except for these seams.// --------------------------------------------------------------------------- import { env } from "cloudflare:workers";import { Data, Effect, Layer } from "effect";import type { Cause } from "effect";import * as OtelTracer from "@effect/opentelemetry/Tracer";import { drizzle } from "drizzle-orm/postgres-js";import postgres, { type Sql } from "postgres"; import { PAUSED_APPROVAL_TIMEOUT_MS, createExecutorMcpServer,} from "@executor-js/host-mcp/tool-server";import { buildResumeApprovalUrl } from "@executor-js/host-mcp/browser-approval";import { artifactUrlFor } from "@executor-js/host-mcp/create-artifact";import { makeAssetsShellHtmlLoader } from "@executor-js/mcp-apps-shell/worker";import { smokeRenderArtifact } from "@executor-js/mcp-apps-shell/smoke-render";import { McpAgentSessionDOBase, type BuiltMcpServer, type IncomingTraceHeaders, type McpApprovalOwner, type McpSessionModelResumeResult, type McpSessionInit, type SessionMeta,} from "@executor-js/cloudflare/mcp/agent-durable-object";import { mcpExecutionOwnerDirectoryFromNamespace, type McpExecutionOwnerDirectory, type McpExecutionOwnerRoute,} from "@executor-js/cloudflare/mcp/execution-owner-directory";import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub";import { buildExecuteDescription, parseIntegrationInventory, type ResumeResponse,} from "@executor-js/execution";import { acquireBuildSlot, type BuildSlotHandle } from "./session-build-semaphore"; // The DO meters executions just like the HTTP `/api/*` plane: it builds its// engine with `CloudMeteredExecutionStackLayer`, so every MCP execution is// tracked to Autumn (the MCP server is the primary execution surface, so leaving// it unmetered silently dropped the bulk of real usage). The billing service// (`AutumnService.Default`) is provided LOCALLY to the metered stack below, so// the DO still imports the focused `CoreSharedServices` root (beside// `WorkOSClient`), NOT `../api/layers`, and its bundle stays free of the whole// HTTP API assembly. (This used to require a dedicated `core-shared-services.ts`// leaf to keep `auth/handlers.ts` -> `@tanstack/react-start` out of the DO// bundle; that coupling is gone now that `handlers.ts` queues cookies through// `SessionAuthLive` instead.)import { CoreSharedServices } from "../auth/workos";import { UserStoreService } from "../auth/context";import { resolveSessionMetaForToken } from "./session-meta";import { DbService, combinedSchema, resolveConnectionString, type DrizzleDb, type DbServiceShape,} from "../db/db";import { AutumnService } from "../extensions/billing/service";import { DoTelemetryLive, flushTracerProvider } from "../observability/telemetry";import { captureCause as reportCause, captureCauseEffect as reportCauseEffect, claimCauseHandledByDurableObject, tagCurrentSentryScopeWithCurrentOtelSpan,} from "../observability";import { parseTraceparent } from "./traceparent"; // Re-export the shared types so existing cloud importers// (`auth/handlers.ts`, etc.) keep their `../mcp/session-durable-object` path.export type { McpApprovalOwner, McpSessionApprovalResult, McpSessionModelResumeResult, McpSessionResumeApprovalResult, McpSessionInit, IncomingTraceHeaders,} from "@executor-js/cloudflare/mcp/agent-durable-object"; // ---------------------------------------------------------------------------// Cloud DB handle — one postgres.js client per session runtime// --------------------------------------------------------------------------- const LONG_LIVED_DB_IDLE_TIMEOUT_SECONDS = 5;const LONG_LIVED_DB_MAX_LIFETIME_SECONDS = 120;const TELEMETRY_FLUSH_TIMEOUT_MS = 1_000; const positiveMilliseconds = (raw: string | undefined): number | undefined => { if (!raw) return undefined; const parsed = Number(raw); if (!Number.isFinite(parsed) || parsed <= 0) return undefined; return Math.floor(parsed);}; /** Same shape as `positiveMilliseconds`, for a plain count rather than a * duration — used only by the resident-runtime soft-cap override below. */const positiveInteger = positiveMilliseconds; type CloudSessionDbHandle = DbServiceShape & { readonly sql: Sql; readonly end: () => Promise<void>;}; class McpModelResumeForwardError extends Data.TaggedError("McpModelResumeForwardError")<{ readonly cause: unknown;}> {} /** * The DO keeps one postgres.js client for the MCP session runtime. postgres.js * closes idle sockets quickly, while the runtime object stays alive so the MCP * server can preserve session-local protocol state across requests. */const makeDbHandle = (options: { readonly idleTimeout: number; readonly maxLifetime: number;}): CloudSessionDbHandle => { const sql = postgres(resolveConnectionString(), { max: 1, idle_timeout: options.idleTimeout, max_lifetime: options.maxLifetime, connect_timeout: 10, fetch_types: false, prepare: true, onnotice: () => undefined, }); return { sql, db: drizzle(sql, { schema: combinedSchema }) as DrizzleDb, // oxlint-disable-next-line executor/no-promise-catch -- boundary: postgres.js close is best-effort during DO/runtime cleanup end: () => sql.end({ timeout: 0 }).catch(() => undefined), };}; const makeEphemeralDb = (): CloudSessionDbHandle => makeDbHandle({ idleTimeout: 0, maxLifetime: 60 }); // The org-resolution + session-runtime services. They DON'T re-provide// `DoTelemetryLive` — that would install a second WebSdk tracer in the nested// Effect scope, disconnecting every child span from the outer DO-method trace.// Tracer comes from the outermost `withTelemetry` at the DO method boundary.const makeSessionServices = (dbHandle: CloudSessionDbHandle) => { const DbLive = Layer.succeed(DbService)({ sql: dbHandle.sql, db: dbHandle.db }); const UserStoreLive = UserStoreService.Live.pipe(Layer.provide(DbLive)); return Layer.mergeAll(DbLive, UserStoreLive, CoreSharedServices);}; // The `ui://executor/shell.html` resource, over the ASSETS binding: the// deployed Worker has no filesystem, so the document is the stable-named// asset the client build emitted (`mcpAppsShellAsset`), fetched at first// artifact resource read. Module scope so the fetch-and-verify happens once// per isolate, not once per session. The dev thunk carries the built shell// inline under `vite dev`, where no assets exist yet for the binding to find.const loadAppShellHtml = makeAssetsShellHtmlLoader({ assets: env.ASSETS, devShellHtml: () => import("virtual:executor-mcp-apps-shell-dev-html").then((mod) => mod.devShellHtml),}); // QuickJS-WASM must be loaded before the smoke render asks for a sandbox: the// default variant cannot fetch its own `.wasm` on Workers. `../quickjs` is// imported dynamically here, not at module scope, so a session that never// calls create_artifact/edit_artifact never pays for it — see the comment on// the dynamic import block in `buildMcpServer` for why that matters on a cold// isolate. `preloadQuickJs()` itself is memoized per isolate (and resets on// failure), so concurrent artifact calls, and repeat calls after the first,// are all free past the first successful load.const smokeRenderArtifactAfterQuickJsPreload: typeof smokeRenderArtifact = async (code) => { const { preloadQuickJs } = await import("../quickjs"); await preloadQuickJs(); return smokeRenderArtifact(code);}; // ---------------------------------------------------------------------------// Durable Object// --------------------------------------------------------------------------- export class McpSessionDOSqlite extends McpAgentSessionDOBase<Env, CloudSessionDbHandle> { // Set once per `buildMcpServer` call, read back by `sessionFootprintAttributes` // so the counts land on the base's `McpSessionDO.init` span alongside // `mcp.isolate.*` residency — see the comment at the `parseIntegrationInventory` // call site below for why this is free to compute. private lastSessionFootprint: Record<string, number> = {}; protected override sessionFootprintAttributes(): Record<string, number> { return this.lastSessionFootprint; } protected override sessionTimeoutMs(): number { return positiveMilliseconds(env.MCP_SESSION_TIMEOUT_MS) ?? super.sessionTimeoutMs(); } protected override maxPausedSessionIdleMs(): number { return ( positiveMilliseconds(env.MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS) ?? super.maxPausedSessionIdleMs() ); } protected override executionOwnerDirectory(): McpExecutionOwnerDirectory | null { return mcpExecutionOwnerDirectoryFromNamespace(env.MCP_EXECUTION_OWNER); } // Test-only override so e2e can exercise a REAL cross-DO eviction request in // workerd without registering 32 sessions. Unset in production, where this // falls through to the base class's `RESIDENT_RUNTIME_SOFT_CAP`. protected override residentRuntimeSoftCap(): number { return positiveInteger(env.MCP_RESIDENT_RUNTIME_SOFT_CAP) ?? super.residentRuntimeSoftCap(); } protected override supportsCapEviction(): boolean { return true; } protected override requestSelfEviction(): Promise<void> { // Routed through this session's OWN stub (never a direct in-process call) // so `requestCapEviction`'s teardown runs under an IoContext scoped to // this request, not whatever request happened to trigger the eviction // check — see the base class's `requestSelfEviction` doc comment. return mcpSessionStub(env.MCP_SESSION, this.sessionId).requestCapEviction(); } protected override forwardModelResumeToOwner( owner: McpExecutionOwnerRoute, identity: McpApprovalOwner, executionId: string, response: ResumeResponse, ): Effect.Effect<McpSessionModelResumeResult, unknown> { return Effect.tryPromise({ try: () => mcpSessionStub(env.MCP_SESSION, owner.sessionId).resumeExecutionForModel( executionId, identity, response, ), catch: (cause) => new McpModelResumeForwardError({ cause }), }); } protected override openSessionDb(): CloudSessionDbHandle { return makeDbHandle({ idleTimeout: LONG_LIVED_DB_IDLE_TIMEOUT_SECONDS, maxLifetime: LONG_LIVED_DB_MAX_LIFETIME_SECONDS, }); } protected override resolveSessionMeta( token: McpSessionInit, storedMeta: SessionMeta | null, ): Effect.Effect<SessionMeta> { // The database handle is opened LAZILY: on the props and stored paths — the // overwhelming majority of inits — nothing here touches Postgres at all, // which is the whole point. postgres.js only dials on first query, so // building the handle costs nothing; `ensuring` still closes it. const dbHandle = makeEphemeralDb(); return resolveSessionMetaForToken(token, storedMeta).pipe( Effect.withSpan("McpSessionDOSqlite.resolveSessionMeta"), Effect.provide(makeSessionServices(dbHandle)), Effect.ensuring(Effect.promise(() => dbHandle.end())), // The base's `resolveSessionMeta` seam has no error channel, and a // Durable Object's `init` can only reject its Promise — so a failure has // to leave as a defect. What changed is WHAT leaves: an unreachable // organization directory is now a bounded, classified // `McpSessionMetaUnavailableError` whose message the worker recognises // and renders as a retryable 503 (see `agent-handler.ts`), instead of an // unclassified Postgres cause that produced a 500 and a client hang. // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: the DO init seam is Promise-only; the failure is classified before it dies Effect.orDie, ); } protected override buildMcpServer( sessionMeta: SessionMeta, dbHandle: CloudSessionDbHandle, ): Effect.Effect<BuiltMcpServer> { const self = this; // Set synchronously, inside the `Effect.promise` executor below, before // any `await` — so there is no window where interruption can land after // the handle exists but before this is assigned. `cancel` is idempotent // (see `session-build-semaphore.ts`), so it is safe to invoke from both // the abort listener below AND the `Effect.ensuring` finalizer without // risking a double release. let buildSlot: BuildSlotHandle | undefined; const build = Effect.gen(function* () { // A burst of cold sessions landing on one isolate at once used to pay // full concurrent build cost; this bounds it to `MAX_CONCURRENT_BUILDS` // at a time and queues the rest FIFO, degrading to old concurrent // behavior (see `timedOut` below) rather than stalling if the queue // itself gets stuck. See `session-build-semaphore.ts`. // // `Effect.promise` hands its executor an `AbortSignal` tied to this // fiber's own interruption — firing it (rather than merely walking // away) is what lets a waiter still sitting in the semaphore's queue // be dequeued immediately on a client disconnect, instead of getting // granted a slot later with no one left to release it. const slot = yield* Effect.promise((signal) => { const handle = acquireBuildSlot(); buildSlot = handle; signal.addEventListener("abort", () => handle.cancel(), { once: true }); return handle.promise; }); if (slot.timedOut) { yield* Effect.annotateCurrentSpan({ "mcp.init.build_queue_timeout": true }); } else if (slot.waitMs > 0) { yield* Effect.annotateCurrentSpan({ "mcp.init.build_queue_ms": slot.waitMs }); } // Imported here rather than at module scope. Cloudflare requires a // Durable Object class to be exported from the Worker entry, so every // static import this module makes is evaluated by *every* cold isolate — // including the ones that only render a page or forward a passthrough // proxy and never open an MCP session. These two roots pull the whole // code-execution stack (sucrase, ajv): measured at 1.9 MB of the // Worker's startup closure, for code only a real session runs. // `apps/cloud/scripts/start-closure.mjs` reports that number and will // show it moving back if these become static again. QuickJS-WASM is a // separate dynamic import off the artifact smoke-render path only (see // `smokeRenderArtifactAfterQuickJsPreload` above) — it is never needed // during init, so it no longer lives in this Promise.all at all. const [{ makeExecutionStack }, { CloudMeteredExecutionStackLayer }] = yield* Effect.promise( () => Promise.all([ import("../engine/execution-stack"), import("../engine/execution-stack-metered"), ]) as Promise< [ typeof import("../engine/execution-stack"), typeof import("../engine/execution-stack-metered"), ] >, ); const { executor, engine } = yield* makeExecutionStack( sessionMeta.userId, sessionMeta.organizationId, sessionMeta.organizationName, { mcpResource: sessionMeta.resource, orgWrites: "request", }, ).pipe( // The metered stack tracks each execution to Autumn. It requires // `AutumnService | DbService`; `AutumnService.Default` is provided here // (it only reads `env`, no further deps), and `DbService` flows from the // outer `makeSessionServices`. When `AUTUMN_SECRET_KEY` is unset the // billing service degrades to a no-op tracker, so this stays inert in // cloud dev/preview environments that run without a billing backend. Effect.provide(CloudMeteredExecutionStackLayer.pipe(Layer.provide(AutumnService.Default))), Effect.withSpan("McpSessionDOSqlite.makeExecutionStack"), ); // Build the description here so `executor.connections.list()` stays under // the DO startup span and the MCP SDK receives a concrete string instead // of invoking `engine.getDescription` across its async boundary. const description = yield* buildExecuteDescription(executor).pipe( Effect.withSpan("mcp.execute.description.build"), ); // Cheap size proxy for the session's footprint: `parseIntegrationInventory` // is a regex walk over the description string already built above, the // same trick `createExecutorMcpServer` uses to derive its per-integration // search tools without a second `connections.list()` — no new query, no // catalog serialization. This is a *count* of distinct connected // integrations, not raw connections (several connections can share one // integration) and is capped at the description's 50-item inventory // limit, same as what the model itself sees. self.lastSessionFootprint = { "mcp.session.integration_count": parseIntegrationInventory(description).length, }; const sessionElicitationMode = sessionMeta.elicitationMode ?? "model"; const mcpServer = yield* createExecutorMcpServer({ engine, description, artifacts: executor.artifacts, connections: executor.connections, tools: executor.tools, integrations: executor.integrations, // Artifacts are on by default, opt-out per connection. A session // persisted without a value restores to the default, same as a fresh // connection whose URL says nothing about `?artifacts=`. artifactsEnabled: sessionMeta.artifactsEnabled ?? true, // Per-integration search tools are off by default, opt-in per // connection (`?search_tools=true`). Same restore rule as artifacts. searchToolsEnabled: sessionMeta.searchToolsEnabled ?? false, // The tool surface must survive a cold restore unchanged: the client // cached the names it saw at `initialize`. mode: sessionMeta.toolMode ?? "codemode", // Cold restores rebuild this server with no `initialize` to replay, so // the negotiated apps support comes back from storage instead. restoredAppsEnabled: sessionMeta.appsEnabled ?? false, onAppsEnabledChange: (appsEnabled) => self.persistAppsEnabled(appsEnabled), loadAppShellHtml, smokeRenderArtifact: smokeRenderArtifactAfterQuickJsPreload, artifactUrl: artifactUrlFor( env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", sessionMeta.organizationSlug, ), parentSpan: () => self.currentParentSpan(), debug: env.EXECUTOR_MCP_DEBUG === "true", browserApprovalStore: self.browserApprovalStore, pausedExecutionHooks: self.pausedExecutionHooks, pausedExecutionLeaseMs: PAUSED_APPROVAL_TIMEOUT_MS, resumeFallback: self.modelResumeFallback, elicitationMode: sessionElicitationMode === "browser" ? { mode: "browser" as const, approvalUrl: (executionId) => buildResumeApprovalUrl({ origin: env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", executionId, sessionId: self.sessionId, organizationSlug: sessionMeta.organizationSlug, }), } : { mode: sessionElicitationMode }, }).pipe(Effect.withSpan("McpSessionDOSqlite.createExecutorMcpServer")); return { mcpServer, engine } satisfies BuiltMcpServer; }); return build.pipe( Effect.withSpan("McpSessionDOSqlite.buildMcpServer"), // Always paired with the `acquireBuildSlot` above: releases on success, // failure, and interruption alike, so a build that throws — or a fiber // that is interrupted mid-build — never wedges the FIFO queue behind // it. `handle.cancel()` is idempotent and self-describing about // whether a slot was actually granted, so no separate "did we acquire" // flag is needed here; this and the abort listener above may both fire // for the same handle without double-releasing. Effect.ensuring( Effect.sync(() => { buildSlot?.cancel(); }), ), Effect.provide(makeSessionServices(dbHandle)), // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: runtime-build failures surface as the base's tapCause/cleanup defect Effect.orDie, ); } protected override withTelemetry<A, E>( effect: Effect.Effect<A, E>, incoming?: IncomingTraceHeaders, ): Effect.Effect<A, E> { const parsed = parseTraceparent(incoming?.traceparent, incoming?.tracestate); const traced = parsed ? OtelTracer.withSpanContext(effect, parsed) : effect; return traced.pipe(Effect.provide(DoTelemetryLive)); } protected override captureCause(cause: Cause.Cause<unknown>): void { reportCause(cause); } protected override captureCauseEffect( cause: Cause.Cause<unknown>, ): Effect.Effect<string | undefined> { return reportCauseEffect(cause); } protected override prepareErrorCaptureScope(): Effect.Effect<void> { return Effect.asVoid(tagCurrentSentryScopeWithCurrentOtelSpan); } // The DO owns this cause; `instrumentDurableObjectWithSentry` (server.ts) must // not report it a second time when the rejection escapes the method. protected override claimCauseHandled(_cause: Cause.Cause<unknown>): Effect.Effect<void> { return claimCauseHandledByDurableObject; } // Best-effort export the DO isolate's buffered spans after the RPC settles, // so a dying init/handleRequest can ship its own spans (and the exception + // stack recorded on them) — not just the worker-side `mcp.do.*` span. Keep it // off the response path and bounded: telemetry export must not hold a // successful MCP response open. protected override flushTelemetry(): Promise<void> { this.ctx.waitUntil( Effect.runPromise( Effect.tryPromise({ try: () => flushTracerProvider(), catch: () => undefined, }).pipe( Effect.ignore, Effect.timeoutOrElse({ duration: `${TELEMETRY_FLUSH_TIMEOUT_MS} millis`, orElse: () => Effect.void, }), ), ), ); return Promise.resolve();Lines 1–500 of 502. Load more
// --------------------------------------------------------------------------- // Cloud MCP Session Durable Object — the cloud binding of the shared // `McpAgentSessionDOBase` (@executor-js/cloudflare). Hibernatable transport // serving, cold restore, the inactivity alarm, owner validation, browser // approval storage, and the per-request span bridge live in the base. Cloud // supplies ONLY its injected dependencies: // - openSessionDb → a long-lived postgres.js handle // - resolveSessionMeta → WorkOS/UserStore organization resolution // - buildMcpServer → the cloud execution stack + MCP tool server // - withTelemetry → the WebSdk tracer + W3C parent-span stitching // - captureCause → Sentry error capture // host-cloudflare binds the same base to D1 instead; the two stay byte-identical // except for these seams. // --------------------------------------------------------------------------- import { env } from "cloudflare:workers"; import { Data, Effect, Layer } from "effect"; import type { Cause } from "effect"; import * as OtelTracer from "@effect/opentelemetry/Tracer"; import { drizzle } from "drizzle-orm/postgres-js"; import postgres, { type Sql } from "postgres"; import { PAUSED_APPROVAL_TIMEOUT_MS, createExecutorMcpServer, } from "@executor-js/host-mcp/tool-server"; import { buildResumeApprovalUrl } from "@executor-js/host-mcp/browser-approval"; import { artifactUrlFor } from "@executor-js/host-mcp/create-artifact"; import { makeAssetsShellHtmlLoader } from "@executor-js/mcp-apps-shell/worker"; import { smokeRenderArtifact } from "@executor-js/mcp-apps-shell/smoke-render"; import { McpAgentSessionDOBase, type BuiltMcpServer, type IncomingTraceHeaders, type McpApprovalOwner, type McpSessionModelResumeResult, type McpSessionInit, type SessionMeta, } from "@executor-js/cloudflare/mcp/agent-durable-object"; import { mcpExecutionOwnerDirectoryFromNamespace, type McpExecutionOwnerDirectory, type McpExecutionOwnerRoute, } from "@executor-js/cloudflare/mcp/execution-owner-directory"; import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; import { buildExecuteDescription, parseIntegrationInventory, type ResumeResponse, } from "@executor-js/execution"; import { acquireBuildSlot, type BuildSlotHandle } from "./session-build-semaphore"; // The DO meters executions just like the HTTP `/api/*` plane: it builds its // engine with `CloudMeteredExecutionStackLayer`, so every MCP execution is // tracked to Autumn (the MCP server is the primary execution surface, so leaving // it unmetered silently dropped the bulk of real usage). The billing service // (`AutumnService.Default`) is provided LOCALLY to the metered stack below, so // the DO still imports the focused `CoreSharedServices` root (beside // `WorkOSClient`), NOT `../api/layers`, and its bundle stays free of the whole // HTTP API assembly. (This used to require a dedicated `core-shared-services.ts` // leaf to keep `auth/handlers.ts` -> `@tanstack/react-start` out of the DO // bundle; that coupling is gone now that `handlers.ts` queues cookies through // `SessionAuthLive` instead.) import { CoreSharedServices } from "../auth/workos"; import { UserStoreService } from "../auth/context"; import { resolveSessionMetaForToken } from "./session-meta"; import { DbService, combinedSchema, resolveConnectionString, type DrizzleDb, type DbServiceShape, } from "../db/db"; import { AutumnService } from "../extensions/billing/service"; import { DoTelemetryLive, flushTracerProvider } from "../observability/telemetry"; import { captureCause as reportCause, captureCauseEffect as reportCauseEffect, claimCauseHandledByDurableObject, tagCurrentSentryScopeWithCurrentOtelSpan, } from "../observability"; import { parseTraceparent } from "./traceparent"; // Re-export the shared types so existing cloud importers // (`auth/handlers.ts`, etc.) keep their `../mcp/session-durable-object` path. export type { McpApprovalOwner, McpSessionApprovalResult, McpSessionModelResumeResult, McpSessionResumeApprovalResult, McpSessionInit, IncomingTraceHeaders, } from "@executor-js/cloudflare/mcp/agent-durable-object"; // --------------------------------------------------------------------------- // Cloud DB handle — one postgres.js client per session runtime // --------------------------------------------------------------------------- const LONG_LIVED_DB_IDLE_TIMEOUT_SECONDS = 5; const LONG_LIVED_DB_MAX_LIFETIME_SECONDS = 120; const TELEMETRY_FLUSH_TIMEOUT_MS = 1_000; const positiveMilliseconds = (raw: string | undefined): number | undefined => { if (!raw) return undefined; const parsed = Number(raw); if (!Number.isFinite(parsed) || parsed <= 0) return undefined; return Math.floor(parsed); }; /** Same shape as `positiveMilliseconds`, for a plain count rather than a * duration — used only by the resident-runtime soft-cap override below. */ const positiveInteger = positiveMilliseconds; type CloudSessionDbHandle = DbServiceShape & { readonly sql: Sql; readonly end: () => Promise<void>; }; class McpModelResumeForwardError extends Data.TaggedError("McpModelResumeForwardError")<{ readonly cause: unknown; }> {} /** * The DO keeps one postgres.js client for the MCP session runtime. postgres.js * closes idle sockets quickly, while the runtime object stays alive so the MCP * server can preserve session-local protocol state across requests. */ const makeDbHandle = (options: { readonly idleTimeout: number; readonly maxLifetime: number; }): CloudSessionDbHandle => { const sql = postgres(resolveConnectionString(), { max: 1, idle_timeout: options.idleTimeout, max_lifetime: options.maxLifetime, connect_timeout: 10, fetch_types: false, prepare: true, onnotice: () => undefined, }); return { sql, db: drizzle(sql, { schema: combinedSchema }) as DrizzleDb, // oxlint-disable-next-line executor/no-promise-catch -- boundary: postgres.js close is best-effort during DO/runtime cleanup end: () => sql.end({ timeout: 0 }).catch(() => undefined), }; }; const makeEphemeralDb = (): CloudSessionDbHandle => makeDbHandle({ idleTimeout: 0, maxLifetime: 60 }); // The org-resolution + session-runtime services. They DON'T re-provide // `DoTelemetryLive` — that would install a second WebSdk tracer in the nested // Effect scope, disconnecting every child span from the outer DO-method trace. // Tracer comes from the outermost `withTelemetry` at the DO method boundary. const makeSessionServices = (dbHandle: CloudSessionDbHandle) => { const DbLive = Layer.succeed(DbService)({ sql: dbHandle.sql, db: dbHandle.db }); const UserStoreLive = UserStoreService.Live.pipe(Layer.provide(DbLive)); return Layer.mergeAll(DbLive, UserStoreLive, CoreSharedServices); }; // The `ui://executor/shell.html` resource, over the ASSETS binding: the // deployed Worker has no filesystem, so the document is the stable-named // asset the client build emitted (`mcpAppsShellAsset`), fetched at first // artifact resource read. Module scope so the fetch-and-verify happens once // per isolate, not once per session. The dev thunk carries the built shell // inline under `vite dev`, where no assets exist yet for the binding to find. const loadAppShellHtml = makeAssetsShellHtmlLoader({ assets: env.ASSETS, devShellHtml: () => import("virtual:executor-mcp-apps-shell-dev-html").then((mod) => mod.devShellHtml), }); // QuickJS-WASM must be loaded before the smoke render asks for a sandbox: the // default variant cannot fetch its own `.wasm` on Workers. `../quickjs` is // imported dynamically here, not at module scope, so a session that never // calls create_artifact/edit_artifact never pays for it — see the comment on // the dynamic import block in `buildMcpServer` for why that matters on a cold // isolate. `preloadQuickJs()` itself is memoized per isolate (and resets on // failure), so concurrent artifact calls, and repeat calls after the first, // are all free past the first successful load. const smokeRenderArtifactAfterQuickJsPreload: typeof smokeRenderArtifact = async (code) => { const { preloadQuickJs } = await import("../quickjs"); await preloadQuickJs(); return smokeRenderArtifact(code); }; // --------------------------------------------------------------------------- // Durable Object // --------------------------------------------------------------------------- export class McpSessionDOSqlite extends McpAgentSessionDOBase<Env, CloudSessionDbHandle> { // Set once per `buildMcpServer` call, read back by `sessionFootprintAttributes` // so the counts land on the base's `McpSessionDO.init` span alongside // `mcp.isolate.*` residency — see the comment at the `parseIntegrationInventory` // call site below for why this is free to compute. private lastSessionFootprint: Record<string, number> = {}; protected override sessionFootprintAttributes(): Record<string, number> { return this.lastSessionFootprint; } protected override sessionTimeoutMs(): number { return positiveMilliseconds(env.MCP_SESSION_TIMEOUT_MS) ?? super.sessionTimeoutMs(); } protected override maxPausedSessionIdleMs(): number { return ( positiveMilliseconds(env.MCP_PAUSED_SESSION_IDLE_TIMEOUT_MS) ?? super.maxPausedSessionIdleMs() ); } protected override executionOwnerDirectory(): McpExecutionOwnerDirectory | null { return mcpExecutionOwnerDirectoryFromNamespace(env.MCP_EXECUTION_OWNER); } // Test-only override so e2e can exercise a REAL cross-DO eviction request in // workerd without registering 32 sessions. Unset in production, where this // falls through to the base class's `RESIDENT_RUNTIME_SOFT_CAP`. protected override residentRuntimeSoftCap(): number { return positiveInteger(env.MCP_RESIDENT_RUNTIME_SOFT_CAP) ?? super.residentRuntimeSoftCap(); } protected override supportsCapEviction(): boolean { return true; } protected override requestSelfEviction(): Promise<void> { // Routed through this session's OWN stub (never a direct in-process call) // so `requestCapEviction`'s teardown runs under an IoContext scoped to // this request, not whatever request happened to trigger the eviction // check — see the base class's `requestSelfEviction` doc comment. return mcpSessionStub(env.MCP_SESSION, this.sessionId).requestCapEviction(); } protected override forwardModelResumeToOwner( owner: McpExecutionOwnerRoute, identity: McpApprovalOwner, executionId: string, response: ResumeResponse, ): Effect.Effect<McpSessionModelResumeResult, unknown> { return Effect.tryPromise({ try: () => mcpSessionStub(env.MCP_SESSION, owner.sessionId).resumeExecutionForModel( executionId, identity, response, ), catch: (cause) => new McpModelResumeForwardError({ cause }), }); } protected override openSessionDb(): CloudSessionDbHandle { return makeDbHandle({ idleTimeout: LONG_LIVED_DB_IDLE_TIMEOUT_SECONDS, maxLifetime: LONG_LIVED_DB_MAX_LIFETIME_SECONDS, }); } protected override resolveSessionMeta( token: McpSessionInit, storedMeta: SessionMeta | null, ): Effect.Effect<SessionMeta> { // The database handle is opened LAZILY: on the props and stored paths — the // overwhelming majority of inits — nothing here touches Postgres at all, // which is the whole point. postgres.js only dials on first query, so // building the handle costs nothing; `ensuring` still closes it. const dbHandle = makeEphemeralDb(); return resolveSessionMetaForToken(token, storedMeta).pipe( Effect.withSpan("McpSessionDOSqlite.resolveSessionMeta"), Effect.provide(makeSessionServices(dbHandle)), Effect.ensuring(Effect.promise(() => dbHandle.end())), // The base's `resolveSessionMeta` seam has no error channel, and a // Durable Object's `init` can only reject its Promise — so a failure has // to leave as a defect. What changed is WHAT leaves: an unreachable // organization directory is now a bounded, classified // `McpSessionMetaUnavailableError` whose message the worker recognises // and renders as a retryable 503 (see `agent-handler.ts`), instead of an // unclassified Postgres cause that produced a 500 and a client hang. // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: the DO init seam is Promise-only; the failure is classified before it dies Effect.orDie, ); } protected override buildMcpServer( sessionMeta: SessionMeta, dbHandle: CloudSessionDbHandle, ): Effect.Effect<BuiltMcpServer> { const self = this; // Set synchronously, inside the `Effect.promise` executor below, before // any `await` — so there is no window where interruption can land after // the handle exists but before this is assigned. `cancel` is idempotent // (see `session-build-semaphore.ts`), so it is safe to invoke from both // the abort listener below AND the `Effect.ensuring` finalizer without // risking a double release. let buildSlot: BuildSlotHandle | undefined; const build = Effect.gen(function* () { // A burst of cold sessions landing on one isolate at once used to pay // full concurrent build cost; this bounds it to `MAX_CONCURRENT_BUILDS` // at a time and queues the rest FIFO, degrading to old concurrent // behavior (see `timedOut` below) rather than stalling if the queue // itself gets stuck. See `session-build-semaphore.ts`. // // `Effect.promise` hands its executor an `AbortSignal` tied to this // fiber's own interruption — firing it (rather than merely walking // away) is what lets a waiter still sitting in the semaphore's queue // be dequeued immediately on a client disconnect, instead of getting // granted a slot later with no one left to release it. const slot = yield* Effect.promise((signal) => { const handle = acquireBuildSlot(); buildSlot = handle; signal.addEventListener("abort", () => handle.cancel(), { once: true }); return handle.promise; }); if (slot.timedOut) { yield* Effect.annotateCurrentSpan({ "mcp.init.build_queue_timeout": true }); } else if (slot.waitMs > 0) { yield* Effect.annotateCurrentSpan({ "mcp.init.build_queue_ms": slot.waitMs }); } // Imported here rather than at module scope. Cloudflare requires a // Durable Object class to be exported from the Worker entry, so every // static import this module makes is evaluated by *every* cold isolate — // including the ones that only render a page or forward a passthrough // proxy and never open an MCP session. These two roots pull the whole // code-execution stack (sucrase, ajv): measured at 1.9 MB of the // Worker's startup closure, for code only a real session runs. // `apps/cloud/scripts/start-closure.mjs` reports that number and will // show it moving back if these become static again. QuickJS-WASM is a // separate dynamic import off the artifact smoke-render path only (see // `smokeRenderArtifactAfterQuickJsPreload` above) — it is never needed // during init, so it no longer lives in this Promise.all at all. const [{ makeExecutionStack }, { CloudMeteredExecutionStackLayer }] = yield* Effect.promise( () => Promise.all([ import("../engine/execution-stack"), import("../engine/execution-stack-metered"), ]) as Promise< [ typeof import("../engine/execution-stack"), typeof import("../engine/execution-stack-metered"), ] >, ); const { executor, engine } = yield* makeExecutionStack( sessionMeta.userId, sessionMeta.organizationId, sessionMeta.organizationName, { mcpResource: sessionMeta.resource, orgWrites: "request", }, ).pipe( // The metered stack tracks each execution to Autumn. It requires // `AutumnService | DbService`; `AutumnService.Default` is provided here // (it only reads `env`, no further deps), and `DbService` flows from the // outer `makeSessionServices`. When `AUTUMN_SECRET_KEY` is unset the // billing service degrades to a no-op tracker, so this stays inert in // cloud dev/preview environments that run without a billing backend. Effect.provide(CloudMeteredExecutionStackLayer.pipe(Layer.provide(AutumnService.Default))), Effect.withSpan("McpSessionDOSqlite.makeExecutionStack"), ); // Build the description here so `executor.connections.list()` stays under // the DO startup span and the MCP SDK receives a concrete string instead // of invoking `engine.getDescription` across its async boundary. const description = yield* buildExecuteDescription(executor).pipe( Effect.withSpan("mcp.execute.description.build"), ); // Cheap size proxy for the session's footprint: `parseIntegrationInventory` // is a regex walk over the description string already built above, the // same trick `createExecutorMcpServer` uses to derive its per-integration // search tools without a second `connections.list()` — no new query, no // catalog serialization. This is a *count* of distinct connected // integrations, not raw connections (several connections can share one // integration) and is capped at the description's 50-item inventory // limit, same as what the model itself sees. self.lastSessionFootprint = { "mcp.session.integration_count": parseIntegrationInventory(description).length, }; const sessionElicitationMode = sessionMeta.elicitationMode ?? "model"; const mcpServer = yield* createExecutorMcpServer({ engine, description, artifacts: executor.artifacts, connections: executor.connections, tools: executor.tools, integrations: executor.integrations, // Artifacts are on by default, opt-out per connection. A session // persisted without a value restores to the default, same as a fresh // connection whose URL says nothing about `?artifacts=`. artifactsEnabled: sessionMeta.artifactsEnabled ?? true, // Per-integration search tools are off by default, opt-in per // connection (`?search_tools=true`). Same restore rule as artifacts. searchToolsEnabled: sessionMeta.searchToolsEnabled ?? false, // The tool surface must survive a cold restore unchanged: the client // cached the names it saw at `initialize`. mode: sessionMeta.toolMode ?? "codemode", // Cold restores rebuild this server with no `initialize` to replay, so // the negotiated apps support comes back from storage instead. restoredAppsEnabled: sessionMeta.appsEnabled ?? false, onAppsEnabledChange: (appsEnabled) => self.persistAppsEnabled(appsEnabled), loadAppShellHtml, smokeRenderArtifact: smokeRenderArtifactAfterQuickJsPreload, artifactUrl: artifactUrlFor( env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", sessionMeta.organizationSlug, ), parentSpan: () => self.currentParentSpan(), debug: env.EXECUTOR_MCP_DEBUG === "true", browserApprovalStore: self.browserApprovalStore, pausedExecutionHooks: self.pausedExecutionHooks, pausedExecutionLeaseMs: PAUSED_APPROVAL_TIMEOUT_MS, resumeFallback: self.modelResumeFallback, elicitationMode: sessionElicitationMode === "browser" ? { mode: "browser" as const, approvalUrl: (executionId) => buildResumeApprovalUrl({ origin: env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", executionId, sessionId: self.sessionId, organizationSlug: sessionMeta.organizationSlug, }), } : { mode: sessionElicitationMode }, }).pipe(Effect.withSpan("McpSessionDOSqlite.createExecutorMcpServer")); return { mcpServer, engine } satisfies BuiltMcpServer; }); return build.pipe( Effect.withSpan("McpSessionDOSqlite.buildMcpServer"), // Always paired with the `acquireBuildSlot` above: releases on success, // failure, and interruption alike, so a build that throws — or a fiber // that is interrupted mid-build — never wedges the FIFO queue behind // it. `handle.cancel()` is idempotent and self-describing about // whether a slot was actually granted, so no separate "did we acquire" // flag is needed here; this and the abort listener above may both fire // for the same handle without double-releasing. Effect.ensuring( Effect.sync(() => { buildSlot?.cancel(); }), ), Effect.provide(makeSessionServices(dbHandle)), // oxlint-disable-next-line executor/no-effect-escape-hatch -- boundary: runtime-build failures surface as the base's tapCause/cleanup defect Effect.orDie, ); } protected override withTelemetry<A, E>( effect: Effect.Effect<A, E>, incoming?: IncomingTraceHeaders, ): Effect.Effect<A, E> { const parsed = parseTraceparent(incoming?.traceparent, incoming?.tracestate); const traced = parsed ? OtelTracer.withSpanContext(effect, parsed) : effect; return traced.pipe(Effect.provide(DoTelemetryLive)); } protected override captureCause(cause: Cause.Cause<unknown>): void { reportCause(cause); } protected override captureCauseEffect( cause: Cause.Cause<unknown>, ): Effect.Effect<string | undefined> { return reportCauseEffect(cause); } protected override prepareErrorCaptureScope(): Effect.Effect<void> { return Effect.asVoid(tagCurrentSentryScopeWithCurrentOtelSpan); } // The DO owns this cause; `instrumentDurableObjectWithSentry` (server.ts) must // not report it a second time when the rejection escapes the method. protected override claimCauseHandled(_cause: Cause.Cause<unknown>): Effect.Effect<void> { return claimCauseHandledByDurableObject; } // Best-effort export the DO isolate's buffered spans after the RPC settles, // so a dying init/handleRequest can ship its own spans (and the exception + // stack recorded on them) — not just the worker-side `mcp.do.*` span. Keep it // off the response path and bounded: telemetry export must not hold a // successful MCP response open. protected override flushTelemetry(): Promise<void> { this.ctx.waitUntil( Effect.runPromise( Effect.tryPromise({ try: () => flushTracerProvider(), catch: () => undefined, }).pipe( Effect.ignore, Effect.timeoutOrElse({ duration: `${TELEMETRY_FLUSH_TIMEOUT_MS} millis`, orElse: () => Effect.void, }), ), ), ); return Promise.resolve();