From cadf6baf2e372b1aa1d7bda946a990fbe4056511 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Thu, 27 Aug 2026 21:24:31 -0700 Subject: [PATCH] Resolve the MCP session name from a single source in the alarm path The alarm guard read the durable name record while every later read went through PartyServer, which consults only ctx.id.name and an in-memory field the alarm path never populates. An alarm could pass the guard and then throw on a session id read for a log line. Route all reads through one accessor and make observational reads total. --- .../mcp/agent-session-durable-object.test.ts | 176 +++++++++++++++++- .../src/mcp/agent-session-durable-object.ts | 131 +++++++++++-- 2 files changed, 289 insertions(+), 18 deletions(-) diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts index 05c257cb28..577c203a6a 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts @@ -1,6 +1,6 @@ // oxlint-disable executor/no-error-constructor, executor/no-try-catch-or-throw -- boundary: the storage fake reproduces the plain Errors the Cloudflare runtime throws, and rejecting is the only way a DurableObjectStorage reports them -import { describe, expect, it } from "@effect/vitest"; -import { Cause, Effect, Exit } from "effect"; +import { afterEach, describe, expect, it } from "@effect/vitest"; +import { Cause, Effect, Exit, Schema } from "effect"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; import type { JSONRPCMessage, MessageExtraInfo } from "@modelcontextprotocol/sdk/types.js"; @@ -19,6 +19,8 @@ class MemoryStorage { private readonly data = new Map(); alarm: number | undefined; + private idName: string | undefined = "streamable-http:session-reconnect"; + readonly sql = { exec: () => [], }; @@ -69,8 +71,18 @@ class MemoryStorage { return callback(); } - get id(): { readonly name: string } { - return { name: "streamable-http:session-reconnect" }; + get id(): { readonly name: string | undefined } { + return { name: this.idName }; + } + + /** + * Model a Durable Object invocation the runtime does not give a + * `ctx.id.name` — the shape an alarm fires in when it is running against an + * alarm record that never carried one. + */ + withoutIdName(): this { + this.idName = undefined; + return this; } get storage(): MemoryStorage { @@ -136,6 +148,9 @@ class RestoredTransport implements Transport { const makeServer = () => new McpServer({ name: "executor-test", version: "1.0.0" }); +/** The DO's structured logs are JSON lines; assertions read them back. */ +const decodeLogLine = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); + const makeDeferred = (): { readonly promise: Promise; readonly resolve: () => void } => { let resolve: () => void = () => undefined; const promise = new Promise((settle) => { @@ -750,3 +765,156 @@ describe("McpAgentSessionDOBase init survives a platform reset of its bookkeepin expect(captured, "a deploy reset is expected platform behaviour, not a defect").toEqual([]); }); }); + +// PartyServer answers "what is this DO's name" from three sources and does not +// consult them in one place: the `name` getter reads `ctx.id.name` and an +// in-memory field hydrated during initialization, while the durable `__ps_name` +// record is read only BY that initialization. The alarm entry point makes most +// of its decisions itself and delegates to `super.alarm()` on one branch only, +// so it never runs that hydration — an alarm could read the durable record, +// conclude the session was addressable, and then die reading the session id for +// a LOG LINE about the decision it had just made. +// +// This cannot be provoked through the dev stack, which addresses every Durable +// Object by name, so the shape is pinned here instead: an unnamed `ctx.id`, a +// faithful stand-in for PartyServer's throwing getter, and the alarm driven end +// to end. +describe("McpAgentSessionDOBase alarm name resolution", () => { + type NameSession = HarnessSession & { + /** Installed by {@link installPartyServerName}, as PartyServer installs it. */ + readonly name: string; + sessionIdForTelemetry: () => string; + }; + + const storedName = "streamable-http:session-stale-alarm"; + + const restoreConsole: Array<() => void> = []; + afterEach(() => { + while (restoreConsole.length > 0) restoreConsole.pop()?.(); + }); + + // Stand-in for PartyServer's `name` getter and the agents SDK's + // `getSessionId`: `ctx.id.name`, else the in-memory field, else the throw. + // `inMemoryName` stays absent by default because the alarm path is exactly + // the one that never runs the initialization which would populate it. + const installPartyServerName = ( + session: NameSession, + options: { readonly inMemoryName?: string } = {}, + ): void => { + Object.defineProperty(session, "name", { + configurable: true, + get: () => { + const ctxName = session.ctx.id.name; + if (ctxName !== undefined) return ctxName; + if (options.inMemoryName !== undefined) return options.inMemoryName; + throw new Error( + "Attempting to read .name on McpSessionDOSqlite, but this.ctx.id.name is not set and no __ps_name fallback record is available.", + ); + }, + }); + session.getSessionId = () => { + const [, sessionId] = session.name.split(":"); + if (!sessionId) throw new Error("Invalid session id."); + return sessionId; + }; + }; + + const makeUnnamedSession = async ( + options: { readonly storeName?: boolean } = {}, + ): Promise<{ session: NameSession; storage: MemoryStorage; logs: string[] }> => { + const storage = new MemoryStorage().withoutIdName(); + if (options.storeName ?? true) await storage.put("__ps_name", storedName); + + const session = (await makeHarnessSession()) as NameSession; + session.ctx = storage; + session.lastActivityMs = Date.now() - 10; + session.sessionTimeoutMs = () => 1; + installPartyServerName(session); + + const logs: string[] = []; + const info = console.info; + const warn = console.warn; + console.info = (line: unknown) => logs.push(String(line)); + console.warn = (line: unknown) => logs.push(String(line)); + restoreConsole.push(() => { + console.info = info; + console.warn = warn; + }); + + return { session, storage, logs }; + }; + + const parsed = (logs: readonly string[]): readonly unknown[] => + logs.map((line) => decodeLogLine(line)); + + it("disposes the right session when only the durable record carries the name", async () => { + const { session, storage, logs } = await makeUnnamedSession(); + + await expect( + session.alarm(), + "an alarm whose guard found a name must not die on the next read of that name", + ).resolves.toBeUndefined(); + + expect( + parsed(logs), + "the session the stored record names is the one reported as disposed", + ).toContainEqual( + expect.objectContaining({ + event: "mcp_session_idle_runtime_dispose", + sessionId: "session-stale-alarm", + }), + ); + expect(session.initialized, "the idle runtime is torn down").toBe(false); + expect(session.engine, "the execution engine is released").toBeNull(); + expect(storage.alarm, "the idle alarm is cleared").toBeUndefined(); + expect(await storage.get("last-activity-ms"), "the idle clock is cleared").toBeUndefined(); + }); + + // The lease branches log BEFORE they act, so a throwing read there loses the + // extension itself and not merely the line: the alarm dies and the session is + // left pinned without making progress. + it("extends a paused lease from the durable record instead of dying on its log", async () => { + const { session, storage, logs } = await makeUnnamedSession(); + session.maxPausedSessionIdleMs = () => 1_000_000; + session.engine = { + ...(session.engine as ExecutionEngine), + pausedExecutionCount: () => Effect.succeed(1), + }; + + await expect(session.alarm()).resolves.toBeUndefined(); + + expect(parsed(logs)).toContainEqual( + expect.objectContaining({ + event: "mcp_session_paused_lease_extension", + sessionId: "session-stale-alarm", + }), + ); + expect(storage.alarm, "the lease is actually extended").toBeGreaterThan(0); + expect(session.initialized, "a leased session keeps its runtime").toBe(true); + }); + + it("completes and cleans up when no source has a name at all", async () => { + const { session, storage, logs } = await makeUnnamedSession({ storeName: false }); + await storage.setAlarm(Date.now()); + + await expect( + session.alarm(), + "an unaddressable session exits rather than throwing into an endless alarm retry", + ).resolves.toBeUndefined(); + + expect(parsed(logs)).toContainEqual( + expect.objectContaining({ event: "mcp_session_unaddressable_alarm_cleanup" }), + ); + expect(storage.alarm, "the alarm does not retry forever").toBeUndefined(); + expect(session.initialized, "the runtime it cannot address is released").toBe(false); + }); + + it("never throws on an observational read of the session id", async () => { + const { session } = await makeUnnamedSession({ storeName: false }); + + expect(() => session.sessionIdForTelemetry()).not.toThrow(); + expect(session.sessionIdForTelemetry(), "a placeholder keeps the log shape stable").toBe( + "unresolved", + ); + }); +}); 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 d7dfcb0e8c..4c32af4d0a 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts @@ -151,6 +151,12 @@ export interface BrowserApprovalStore { const SESSION_META_KEY = "session-meta"; const LAST_ACTIVITY_KEY = "last-activity-ms"; const PARTYSERVER_NAME_KEY = "__ps_name"; +/** + * Stand-in session id for a log line written when the DO's name could not be + * resolved. A named placeholder keeps the field present and the log shape + * stable, and is countable when it happens. + */ +const UNRESOLVED_SESSION_ID = "unresolved"; /** The agents SDK's durable "condemned" marker (`_cf_scheduleDestroy`). */ const AGENTS_DESTROY_PENDING_KEY = "cf_agents_destroy_pending"; const MCP_HTTP_METHOD_HEADER = "cf-mcp-method"; @@ -248,6 +254,7 @@ export abstract class McpAgentSessionDOBase< private initialized = false; private onStartPromise: Promise | null = null; private lastActivityMs = 0; + private resolvedSessionName: string | undefined = undefined; private approvalResponses = new Map(); private approvalWaiters = new Map>(); private pendingApprovalLeases = new Map(); @@ -315,10 +322,98 @@ export abstract class McpAgentSessionDOBase< return Promise.resolve(); } + /** + * The session id as it appears in logs and span attributes. + * + * Purely observational, and therefore total: a log line or a span attribute + * must never be able to abort the work it is describing. The alarm path is + * the one that used to prove this the hard way — it logged its decision + * before doing it, and on an invocation where the name could not be resolved + * the LOG threw and took the whole alarm with it. + */ + protected sessionIdForTelemetry(): string { + return this.sessionIdOrUndefined() ?? UNRESOLVED_SESSION_ID; + } + + /** + * The session id, or `undefined` when this DO's name cannot be resolved from + * any source. Derived from {@link sessionNameOrUndefined} so it agrees with + * the guard that decided the DO was addressable in the first place. + */ + protected sessionIdOrUndefined(): string | undefined { + const name = this.sessionNameOrUndefined(); + if (name === undefined) return undefined; + const [, sessionId] = name.split(":"); + return sessionId ? sessionId : undefined; + } + + /** + * The session id for callers that cannot proceed without one — routing an + * execution back to its owner, addressing an approval URL. Those callers want + * the throw, because a wrong id is worse than a failure. + */ protected get sessionId(): string { + const resolved = this.sessionIdOrUndefined(); + if (resolved !== undefined) return resolved; return this.getSessionId(); } + /** + * The single place that answers "what is this Durable Object's name". + * + * PartyServer has three sources and does not consult them all in the same + * place: `this.name` reads `ctx.id.name` and an in-memory field that is only + * hydrated during initialization, while the durable `__ps_name` record is + * read *only* by that initialization. An entry point that skips + * initialization — the alarm override below, which handles most of its + * decisions itself and only delegates to `super.alarm()` on one branch — can + * therefore find the durable record present, conclude the session is + * addressable, and still have `this.name` throw on the very next read. + * + * Everything in this class goes through this accessor instead, so the guard + * and the reads that follow it can never disagree. + */ + private sessionNameOrUndefined(): string | undefined { + return this.ctx.id.name ?? this.resolvedSessionName ?? this.partyServerNameOrUndefined(); + } + + /** + * Ask every source, including durable storage, and remember the answer. + * + * The remembered value is what makes the synchronous accessor above agree + * with this one for the rest of the invocation: a name recovered from the + * `__ps_name` record stays available to callers that cannot await. + */ + private async resolveSessionName(): Promise { + const inMemory = this.sessionNameOrUndefined(); + if (inMemory !== undefined) { + this.resolvedSessionName = inMemory; + return inMemory; + } + const stored = await this.ctx.storage.get(PARTYSERVER_NAME_KEY); + if (!stored) return undefined; + this.resolvedSessionName = stored; + return stored; + } + + /** + * Read PartyServer's own name resolution without inheriting its throw. + * + * PartyServer exposes no non-throwing probe for "do you have a name yet", so + * "it throws" IS the signal for "not resolvable from memory" — the same shape + * as {@link connectionsOrNone}. In a unit harness the getter also + * throws on its uninitialized private field, which reads identically here and + * is equally correct. + */ + private partyServerNameOrUndefined(): string | undefined { + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: see doc comment; partyserver's `name` getter throws instead of reporting absence. + try { + return (this as { readonly name?: string }).name; + } catch { + return undefined; + } + } + protected currentParentSpan(): Tracer.AnySpan | undefined { return undefined; } @@ -449,12 +544,6 @@ 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 activeStreamCount(): number { return this.connectionsOrNone().length; } @@ -511,6 +600,16 @@ export abstract class McpAgentSessionDOBase< } private async cleanupUnaddressableSessionAlarm(): Promise { + // No name from any source means no session to act on: there is nothing to + // route, nothing to identify, and no id to put in this line. Say so, tear + // the runtime down, drop the alarm, and return normally — a session that + // cannot be addressed must not leave an alarm retrying forever. + console.warn( + JSON.stringify({ + event: "mcp_session_unaddressable_alarm_cleanup", + sessionId: this.sessionIdForTelemetry(), + }), + ); await Effect.runPromise(this.closeRuntime()); await Effect.runPromise( Effect.all([ @@ -527,7 +626,7 @@ export abstract class McpAgentSessionDOBase< console.info( JSON.stringify({ event: "mcp_session_idle_runtime_dispose", - sessionId: this.sessionId, + sessionId: this.sessionIdForTelemetry(), idleMs: input.idleMs, pausedExecutionCount: input.pausedExecutionCount, }), @@ -588,7 +687,7 @@ export abstract class McpAgentSessionDOBase< JSON.stringify({ event: "mcp_session_durable_object_reset", operation: input.operation, - sessionId: self.sessionId, + sessionId: self.sessionIdForTelemetry(), resetKind: input.failure.kind, disposition: input.failure.disposition, cause: Cause.pretty(input.cause), @@ -650,7 +749,7 @@ export abstract class McpAgentSessionDOBase< event: "mcp_execution_owner_directory_error", operation: input.operation, executionId: input.executionId, - sessionId: self.sessionId, + sessionId: self.sessionIdForTelemetry(), exceptionType: first?.name ?? "Error", exceptionMessage: first?.message ?? "unknown", cause: Cause.pretty(input.cause), @@ -675,7 +774,7 @@ export abstract class McpAgentSessionDOBase< JSON.stringify({ event: "mcp_model_resume_forward_error", executionId: input.executionId, - sessionId: self.sessionId, + sessionId: self.sessionIdForTelemetry(), ownerSessionId: input.owner.sessionId, exceptionType: first?.name ?? "Error", exceptionMessage: first?.message ?? "unknown", @@ -701,7 +800,7 @@ export abstract class McpAgentSessionDOBase< event: "mcp_model_resume_forward_error", reason: "timeout", executionId: input.executionId, - sessionId: self.sessionId, + sessionId: self.sessionIdForTelemetry(), ownerSessionId: input.owner.sessionId, timeoutMs: input.timeoutMs, }), @@ -1059,7 +1158,11 @@ export abstract class McpAgentSessionDOBase< } override async alarm(): Promise { - if (!(await this.hasPartyServerName())) { + // Resolve the name FIRST, from every source, and hold onto it. Everything + // below — the lease logs, the dispose log, `super.alarm()` — reads the + // session id off that same answer, so an alarm this guard lets through can + // no longer die on the next read of an id it just decided exists. + if ((await this.resolveSessionName()) === undefined) { await this.cleanupUnaddressableSessionAlarm(); return; } @@ -1086,7 +1189,7 @@ export abstract class McpAgentSessionDOBase< console.info( JSON.stringify( pausedLeaseExtensionLog({ - sessionId: this.sessionId, + sessionId: this.sessionIdForTelemetry(), pausedExecutionCount, idleMs, leaseMs: decision.leaseMs, @@ -1101,7 +1204,7 @@ export abstract class McpAgentSessionDOBase< console.info( JSON.stringify( runningLeaseExtensionLog({ - sessionId: this.sessionId, + sessionId: this.sessionIdForTelemetry(), runningExecutionCount, activeStreamCount, idleMs,