[pull] main from UsefulSoftwareCo:main

.changeset/describe-compile-scope.md

@@ -0,0 +1,5 @@
+---
+"executor": patch
+---
+
+Compile `describe.tool` previews against only the definitions a tool references, drop the compiler's per-call retained graph, and fall back to `unknown` for schemas over a node limit. Describing a tool from a large OpenAPI spec no longer burns seconds of CPU or leaks memory in the shared session isolate.

.changeset/lost-execution-visibility.md

@@ -0,0 +1,5 @@
+---
+"executor": patch
+---
+
+Report an MCP `execute` call that dies with a session reset as a JSON-RPC error instead of a silently closed stream. The front worker answers outstanding request ids when the session socket closes abnormally or a response deadline passes, and a rebuilt session answers ids stranded by a previous incarnation on the next stream. The plain memory-limit reset is now classified as transient.

apps/cloud/src/observability/observability.test.ts

@@ -385,15 +385,24 @@ describe("Durable Object platform reset noise", () => {
     expect(beforeSendWithOtelCorrelation(defect)).not.toBeNull();
   });
 
-  // The memory-limit reset is deliberately absent from the classifier: the
-  // runtime blames the application for it, so it is a defect, not noise.
-  it("keeps the memory-limit reset the classifier deliberately excludes", () => {
+  // The storage-cache memory-limit variant stays absent from the classifier:
+  // the runtime blames the application for it (un-awaited writes, an oversized
+  // read), so it is a defect, not noise. Its plain sibling is a platform reset
+  // and IS classified — the two are separated only by that qualifier.
+  it("keeps the memory-limit variant the classifier deliberately excludes", () => {
     const memory = doInstrumentationEvent(
       "Durable Object's isolate exceeded its memory limit due to overflowing the storage cache. All objects in the isolate were reset.",
     );
     expect(beforeSendWithOtelCorrelation(memory)).not.toBeNull();
   });
 
+  it("drops the plain memory-limit reset as platform noise", () => {
+    const memory = doInstrumentationEvent(
+      "Durable Object's isolate exceeded its memory limit and was reset.",
+    );
+    expect(beforeSendWithOtelCorrelation(memory)).toBeNull();
+  });
+
   it("the hook the worker and DOs install drops the deploy reset", () => {
     const options = cloudSentryOptions({ SENTRY_DSN: "https://public@example.invalid/1" } as Env);
     const event = doInstrumentationEvent("Durable Object reset because its code was updated.");

packages/core/sdk/src/executor.ts

@@ -5724,12 +5724,17 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         for (const def of definitionRows) defs.set(def.name, decodeJsonColumn(def.schema));
 
         const referenced = collectReferencedDefinitions([inputSchema, effectiveOutputSchema], defs);
+        // Compile against the referenced subgraph only. The compiler walks
+        // every definition it is handed (its parser scans the whole `$defs`
+        // map per node), so passing the connection's full component set made a
+        // single describe of a large spec cost seconds of CPU on the shared
+        // session isolate. Unreferenced definitions never appear in the output.
         const preview = yield* Effect.tryPromise({
           try: () =>
             buildToolTypeScriptPreview({
               inputSchema,
               outputSchema: effectiveOutputSchema,
-              defs,
+              defs: new Map(Object.entries(referenced)),
             }),
           catch: (cause) =>
             storageFailureFromUnknown("Failed to build tool TypeScript preview", cause),

packages/core/sdk/src/schema-types.test.ts

@@ -3,6 +3,7 @@ import { describe, expect, it } from "@effect/vitest";
 import { Schema } from "effect";
 
 import {
+  MAX_PREVIEW_SCHEMA_NODES,
   buildToolTypeScriptPreview,
   schemaToTypeScriptPreview,
   schemaToTypeScriptPreviewWithDefs,
@@ -385,6 +386,38 @@ describe("schema-types", () => {
     });
   });
 
+  it("falls back to unknown instead of compiling a schema over the node limit", async () => {
+    const defs = new Map<string, unknown>();
+    // One definition per node-limit slice, all referenced from the input, so the
+    // wrapped schema handed to the compiler is guaranteed to cross the cap.
+    const properties: Record<string, unknown> = {};
+    for (let i = 0; i < MAX_PREVIEW_SCHEMA_NODES / 2; i++) {
+      defs.set(`D${i}`, { type: "object", properties: { v: { type: "string" } } });
+      properties[`p${i}`] = { $ref: `#/$defs/D${i}` };
+    }
+    const preview = await buildToolTypeScriptPreview({
+      inputSchema: { type: "object", properties },
+      outputSchema: { type: "string" },
+      defs,
+    });
+    expect(preview).toEqual({ inputTypeScript: "unknown", outputTypeScript: "unknown" });
+  });
+
+  it("only compiles the definitions a tool references", async () => {
+    // A big pile of unrelated definitions must not change the output or the
+    // time it takes: the caller passes the referenced subgraph, and this pins
+    // the contract that the preview is a pure function of that subgraph.
+    const referenced = new Map<string, unknown>([
+      ["Person", { type: "object", properties: { name: { type: "string" } }, required: ["name"] }],
+    ]);
+    const preview = await buildToolTypeScriptPreview({
+      inputSchema: { type: "object", properties: { who: { $ref: "#/$defs/Person" } } },
+      defs: referenced,
+    });
+    expect(preview.inputTypeScript).toBe("{ who?: Person; }");
+    expect(preview.typeScriptDefinitions).toEqual({ Person: "{ name: string; }" });
+  });
+
   it("renders unconstrained schemas as unknown", async () => {
     await expect(
       buildToolTypeScriptPreview({

packages/core/sdk/src/schema-types.ts

@@ -789,6 +789,31 @@ export type ToolTypeScriptPreview = {
   typeScriptDefinitions?: Record<string, string>;
 };
 
+/**
+ * Upper bound on the number of schema nodes handed to the compiler for one
+ * tool preview. The compiler is fully synchronous, so nothing can interrupt
+ * it once it starts; a pre-check is the only protection a shared isolate has
+ * against a pathological schema. Sized well above any real operation (a
+ * PostHog or Stripe tool with its referenced definitions is a few thousand
+ * nodes) and well below where the compile would take seconds of CPU.
+ */
+export const MAX_PREVIEW_SCHEMA_NODES = 50_000;
+
+/** Count object/array nodes in a schema, stopping early once `limit` is hit. */
+const exceedsNodeLimit = (root: unknown, limit: number): boolean => {
+  let count = 0;
+  const stack: unknown[] = [root];
+  while (stack.length > 0) {
+    const node = stack.pop();
+    if (node === null || typeof node !== "object") continue;
+    if (++count > limit) return true;
+    for (const value of Array.isArray(node) ? node : Object.values(node as object)) {
+      if (value !== null && typeof value === "object") stack.push(value);
+    }
+  }
+  return false;
+};
+
 export const buildToolTypeScriptPreview = async (input: {
   inputSchema?: unknown;
   outputSchema?: unknown;
@@ -806,14 +831,19 @@ export const buildToolTypeScriptPreview = async (input: {
     return {};
   }
 
+  const unknownPreview: ToolTypeScriptPreview = {
+    ...(input.inputSchema !== undefined ? { inputTypeScript: "unknown" } : {}),
+    ...(input.outputSchema !== undefined ? { outputTypeScript: "unknown" } : {}),
+  };
+
   const wrappedSchema = buildWrappedObjectSchema(properties, input.defs);
+  if (exceedsNodeLimit(wrappedSchema, MAX_PREVIEW_SCHEMA_NODES)) {
+    return unknownPreview;
+  }
   return Promise.resolve()
     .then(() => compile(wrappedSchema, ROOT_WRAPPER_NAME, compilerOptionsFrom(input.options ?? {})))
     .then(
       (source) => previewToolFromCompiledTypeScript(source),
-      () => ({
-        ...(input.inputSchema !== undefined ? { inputTypeScript: "unknown" } : {}),
-        ...(input.outputSchema !== undefined ? { outputTypeScript: "unknown" } : {}),
-      }),
+      () => unknownPreview,
     );
 };

packages/core/sdk/src/vendor/json-schema-to-typescript/compat.ts

@@ -39,12 +39,21 @@ export const findKey = <T>(
   return undefined;
 };
 
+/**
+ * Memoize on object identity. The cache is a WeakMap so a memoized function
+ * installed at module scope (`generateType`, `getDefinitionsMemoized`) does
+ * not pin every AST and dereferenced schema it has ever seen for the lifetime
+ * of the isolate. Upstream uses a strong Map, which is harmless in a one-shot
+ * CLI but leaks the whole compiled graph per call inside a long-lived worker.
+ */
 export const memoize = <F extends (arg: any, ...rest: any[]) => any>(fn: F): F => {
-  const cache = new Map<Parameters<F>[0], ReturnType<F>>();
+  // Every caller keys on a schema or AST node, which is always an object.
+  const cache = new WeakMap<object, ReturnType<F>>();
   return ((arg: Parameters<F>[0], ...rest: unknown[]) => {
-    if (cache.has(arg)) return cache.get(arg);
+    const key = arg as object;
+    if (cache.has(key)) return cache.get(key);
     const value = fn(arg, ...rest);
-    cache.set(arg, value);
+    cache.set(key, value);
     return value;
   }) as F;
 };

packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts

@@ -153,6 +153,20 @@ type HarnessSession = {
   >;
   alarm: () => Promise<void>;
   ctx: MemoryStorage;
+  currentSessionEpoch: () => Promise<number>;
+  getStaleEpochStreamRequestIds: () => Promise<
+    ReadonlyArray<{
+      readonly streamId: string;
+      readonly requestIds: ReadonlyArray<string | number>;
+      readonly epoch: number;
+      readonly currentEpoch: number;
+    }>
+  >;
+  getStreamRequestIds: (streamId: string) => Promise<ReadonlyArray<string | number> | undefined>;
+  setStreamRequestIds: (
+    streamId: string,
+    requestIds: ReadonlyArray<string | number>,
+  ) => Promise<void>;
   dbHandle: { readonly end: () => void } | null;
   engine: ExecutionEngine<Cause.YieldableError> | null;
   getConnections?: () => Iterable<unknown>;
@@ -266,7 +280,14 @@ const approval = {
   content: { approved: true },
 } satisfies ResumeResponse;
 
-const makeHarnessSession = async (): Promise<HarnessSession> => {
+/**
+ * `storage` is a parameter so a test can build a SECOND session on the same
+ * durable storage — that is exactly what a Durable Object reset looks like from
+ * storage's point of view: same keys, brand new instance.
+ */
+const makeHarnessSession = async (
+  storage: MemoryStorage = new MemoryStorage(),
+): Promise<HarnessSession> => {
   const sessionId = "session-reconnect";
   const sessionMeta: SessionMeta = {
     organizationId: "org-1",
@@ -275,7 +296,6 @@ const makeHarnessSession = async (): Promise<HarnessSession> => {
     userId: "user-1",
     resource: defaultMcpResource,
   };
-  const storage = new MemoryStorage();
   const server = makeServer();
   await server.connect(new StaleCloseTransport());
 
@@ -1766,3 +1786,110 @@ describe("McpAgentSessionDOBase residency cap eviction", () => {
     });
   });
 });
+
+// The request-id ledger (`__mcp_stream_reqs__:<streamId>`, written by the
+// patched McpAgent — see patches/agents@0.17.3.patch) is the only durable
+// record that a POST is still owed a response. A row exists from the moment the
+// request is accepted until its final response is written, so a row that
+// outlives the incarnation which accepted it is a request nothing will ever
+// answer: the isolate was reset mid-execute. Each row carries the epoch of the
+// incarnation that wrote it, and that is what separates "stranded" from
+// "legitimately still running" — a browser-approval pause holds a row open for
+// minutes inside ONE incarnation and must never be swept.
+describe("McpAgentSessionDOBase stranded-request ledger", () => {
+  const ledgerKey = (streamId: string) => `__mcp_stream_reqs__:${streamId}`;
+
+  it("stamps the accepting incarnation on every ledger row", async () => {
+    const session = await makeHarnessSession();
+
+    await session.setStreamRequestIds("stream-a", [1, "two"]);
+
+    expect(await session.ctx.storage.get(ledgerKey("stream-a"))).toEqual({
+      epoch: await session.currentSessionEpoch(),
+      requestIds: [1, "two"],
+    });
+    expect(
+      await session.getStreamRequestIds("stream-a"),
+      "readers still see a plain request-id list",
+    ).toEqual([1, "two"]);
+  });
+
+  it("does not treat a row from the running incarnation as stranded", async () => {
+    const session = await makeHarnessSession();
+
+    // What a browser-approval pause looks like: accepted, unanswered, and
+    // legitimately going to stay that way for minutes.
+    await session.setStreamRequestIds("stream-paused", [9]);
+
+    expect(await session.getStaleEpochStreamRequestIds()).toEqual([]);
+  });
+
+  it("reports a row left by a previous incarnation as stranded", async () => {
+    const storage = new MemoryStorage();
+    const beforeReset = await makeHarnessSession(storage);
+    await beforeReset.setStreamRequestIds("stream-lost", [42]);
+    await beforeReset.setStreamRequestIds("stream-also-lost", ["abc"]);
+
+    // The reset: same durable storage, a brand new Durable Object instance.
+    const afterReset = await makeHarnessSession(storage);
+    await afterReset.setStreamRequestIds("stream-live", [100]);
+
+    const stranded = await afterReset.getStaleEpochStreamRequestIds();
+
+    // Order follows storage's key order, which this fake does not model, so
+    // the assertion is on the set.
+    expect(
+      [...stranded].sort((a, b) => a.streamId.localeCompare(b.streamId)),
+      "only the rows the dead incarnation accepted",
+    ).toMatchObject([
+      { streamId: "stream-also-lost", requestIds: ["abc"] },
+      { streamId: "stream-lost", requestIds: [42] },
+    ]);
+    for (const row of stranded) expect(row.epoch).toBeLessThan(row.currentEpoch);
+  });
+
+  it("reports a pre-epoch ledger row as stranded", async () => {
+    const session = await makeHarnessSession();
+
+    // The shape rows had before they carried an epoch. One can only have been
+    // written by an earlier deployment, so it reads as epoch 0 and is swept.
+    await session.ctx.storage.put(ledgerKey("stream-legacy"), [7]);
+
+    expect(await session.getStaleEpochStreamRequestIds()).toEqual([
+      {
+        currentEpoch: await session.currentSessionEpoch(),
+        epoch: 0,
+        requestIds: [7],
+        streamId: "stream-legacy",
+      },
+    ]);
+    expect(
+      await session.getStreamRequestIds("stream-legacy"),
+      "and it is still readable as a request-id list",
+    ).toEqual([7]);
+  });
+
+  it("holds the idle lease for a request the running incarnation still owes", async () => {
+    const session = await makeHarnessSession();
+    await session.setStreamRequestIds("stream-live", [1]);
+
+    await session.alarm();
+
+    expect(session.initialized, "live work keeps the runtime resident").toBe(true);
+    expect(session.ctx.alarm, "and re-arms the lease").toBeGreaterThan(0);
+  });
+
+  it("does not let a stranded row extend the idle lease", async () => {
+    const storage = new MemoryStorage();
+    const beforeReset = await makeHarnessSession(storage);
+    await beforeReset.setStreamRequestIds("stream-lost", [1]);
+
+    const afterReset = await makeHarnessSession(storage);
+    await afterReset.alarm();
+
+    expect(
+      afterReset.initialized,
+      "a request nothing will ever answer is dead work, not running work",
+    ).toBe(false);
+  });
+});

packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts

@@ -203,7 +203,6 @@ const AGENTS_DESTROY_PENDING_KEY = "cf_agents_destroy_pending";
 const MCP_HTTP_METHOD_HEADER = "cf-mcp-method";
 const MCP_MESSAGE_HEADER = "cf-mcp-message";
 const MODEL_RESUME_FORWARD_TIMEOUT_MS = 10_000;
-const MCP_STREAM_REQS_KEY_PREFIX = "__mcp_stream_reqs__:";
 const approvalResponseKey = (executionId: string) => `approval-response:${executionId}`;
 const BrowserApprovalDecisionStorage = Schema.Struct({
   response: ResumeResponsePayload,
@@ -752,13 +751,20 @@ export abstract class McpAgentSessionDOBase<
     // which survives disposeIdleRuntime, so a later reconnect GET re-inits the
     // DO and replays it. Counting them would make every delivered-but-unacked
     // POST response pin the runtime alive indefinitely.
-    const rows = await this.ctx.storage.list<readonly JsonRpcRequestId[]>({
-      prefix: MCP_STREAM_REQS_KEY_PREFIX,
-      limit: 1_000,
-    });
+    //
+    // Rows stamped with an epoch older than this incarnation's are dead work,
+    // not running work: the isolate that was going to produce their response
+    // was reset, so nothing will ever answer them and they must not hold the
+    // runtime open. The transport's orphan sweep tells the client and removes
+    // the row on the next GET; until then they simply do not count.
+    const [openStreams, currentEpoch] = await Promise.all([
+      this.getOpenStreamRequestIds(),
+      this.currentSessionEpoch(),
+    ]);
     let count = 0;
-    for (const requestIds of rows.values()) {
-      if (Array.isArray(requestIds)) count += requestIds.length;
+    for (const stream of openStreams) {
+      if (stream.epoch < currentEpoch) continue;
+      count += stream.requestIds.length;
     }
     return count;
   }