fix(mcp): share cold catalog search projections

.github/workflows/fork-ci.yml

@@ -67,6 +67,24 @@ jobs:
       - name: Security scenarios
         run: |
           bun run e2e:self-host -- --test-name 'application_type|fetch without a User-Agent names Executor|cannot create an all-apps connection|cannot widen a connection or mint|fails to load fails only its own reads'
+      # Run the large cold catalog alone, not beside the functional scenarios' servers.
+      - name: Cold MCP discovery
+        run: bun run e2e:self-host --test-name 'Cold MCP discovery serves 8549'
+      - name: Cold discovery phase timings
+        if: always()
+        run: |
+          node --input-type=module <<'JS'
+          import { globSync, readFileSync } from "node:fs";
+          for (const file of globSync(".local/e2e/**/telemetry.json")) {
+            for (const trace of JSON.parse(readFileSync(file, "utf8"))) {
+              const spans = (trace.data ?? []).map(row => row.span).filter(span =>
+                /app\.|sdk\.tools|mcp\.|MCP|build.*load|runtime/.test(span.operationName));
+              console.log(JSON.stringify({file, id: trace.id, spans: spans.map(s => ({
+                phase: s.operationName, ms: s.durationMs, start: s.startTime
+              }))}));
+            }
+          }
+          JS
       - name: MCP search scenarios
         run: |
           bun run e2e:local --test-name 'MCP search resolves namespace-relative dotted paths|A broad MCP tool search returns concise pages'

e2e/ci-selection.ts

@@ -62,7 +62,7 @@ class SelectionFailed extends Schema.TaggedError<SelectionFailed>()("SelectionFa
 const claude = "Claude Code connects";
 const inventoryLoad = "concurrent owners and admins save every account";
 const catalogScale =
-  "MCP execute over 7,000 tools|MCP execute pays for a slow app|MCP execute remembers a stalled|MCP execute serves a search from rows past|MCP execute reports an app whose aged";
+  "Cold MCP discovery serves 8549|MCP execute over 7,000 tools|MCP execute pays for a slow app|MCP execute remembers a stalled|MCP execute serves a search from rows past|MCP execute reports an app whose aged";
 
 /**
  * Each output feeds one `--test-name` in checks.yml. Full runs use these patterns unchanged.

e2e/test-plan.ts

@@ -6245,6 +6245,16 @@ export const scenarios = plan({
       local: na("The module counts are read from Cloud's delivered traces."),
     },
   },
+  coldOpenapiDiscovery: {
+    fixtures: "actors",
+    file: "cold-openapi-discovery.spec.ts",
+    title: "Cold MCP discovery serves 8549 generated OpenAPI tools after activation",
+    targets: {
+      "self-host": scheduled,
+      cloud: na("Exercises the shared listing and OpenAPI runtime on an isolated self-host."),
+      local: na("Uses the shared SDK through the hosted API."),
+    },
+  },
   largeOpenapiApp: {
     fixtures: "actors",
     file: "large-openapi-app.spec.ts",

e2e/tests/cold-openapi-discovery.spec.ts

@@ -0,0 +1,156 @@
+/** Cold discovery must serve a newly activated generated catalog, including concurrent readers. */
+import { expect, layer } from "@effect/vitest";
+import { Clock, Console, Effect, Schema } from "effect";
+import { randomUUID } from "node:crypto";
+import { scenarios } from "../test-plan.ts";
+import { Actors } from "../support/actors.ts";
+import { Api, body } from "../support/api.ts";
+import { HostedLive, withHostedCase } from "../support/case.ts";
+import { App } from "../support/contracts.ts";
+import { Evidence } from "../support/evidence.ts";
+import { appsManifest } from "../support/apps-release.ts";
+import { saveAndDeploy } from "../support/app-authoring.ts";
+import { McpClient } from "../support/mcp-client.ts";
+
+/** Graph-sized operation count; cyclic shared definitions make duplicate projections costly. */
+const count = 8_549;
+const source = (revision: string) => `import { defineApp } from "apps";
+import { liveOpenapiRouter } from "apps/openapi";
+const paths = {};
+for (let i = 0; i < ${count}; i++) paths["/records/" + i] = { get: {
+  operationId: "getRecord" + i, tags: ["records"], summary: "${revision} record " + i,
+  parameters: [{ name: "id", in: "query", schema: { type: "string" } }],
+  responses: { "200": { description: "OK", content: { "application/json": { schema: { $ref: "#/components/schemas/Record0" } } } } }
+} };
+const schemas = Object.fromEntries(Array.from({length: 600}, (_, i) => ["Record" + i, {type: "object", properties: { parent: { $ref: "#/components/schemas/Root" }, value: { type: "string" } }}]));
+schemas.Root = {type: "object", properties: Object.fromEntries(Array.from({length: 600}, (_, i) => ["record" + i, {$ref: "#/components/schemas/Record" + i}]))};
+const document = { components: {schemas}, openapi: "3.0.3", info: { title: "Generated catalog", version: "1" }, servers: [{ url: "https://example.com" }], paths };
+export default defineApp({}, async ({ cache, fetch }) => ({ tools: liveOpenapiRouter({ source: { document }, allowedOrigin: "https://example.com", securitySchemes: {}, methods: {}, oauth: [], cache, fetch }) }));`;
+const Search = Schema.Struct({
+  items: Schema.Array(Schema.Struct({ path: Schema.String, description: Schema.String })),
+  unavailableApps: Schema.optional(Schema.Array(Schema.Unknown)),
+});
+
+layer(HostedLive, { excludeTestServices: true })("Cold OpenAPI discovery", (it) => {
+  it.effect(scenarios.coldOpenapiDiscovery.title, (context) =>
+    withHostedCase(
+      context,
+      Effect.gen(function* () {
+        const api = yield* Api,
+          actors = yield* Actors,
+          mcp = yield* McpClient,
+          evidence = yield* Evidence;
+        const prefix = `/api/organizations/${actors.organization.id}/apps`;
+        const files = (revision: string) => [
+          { path: "index.ts", content: source(revision) },
+          appsManifest,
+        ];
+        const key = yield* body(
+          Schema.Struct({ id: Schema.String, key: Schema.RedactedFromValue(Schema.String) }),
+          yield* api.request(actors.owner, "POST", "/api/auth/api-key/create", {
+            name: "Cold discovery",
+          }),
+        );
+        yield* Effect.addFinalizer(() =>
+          api
+            .request(actors.owner, "POST", "/api/auth/api-key/delete", { keyId: key.id })
+            .pipe(Effect.orDie),
+        );
+        const client = yield* mcp.connect(key.key, "cold-discovery", {
+          organization: actors.organization.id,
+        });
+        const deployed = yield* api.request(actors.owner, "POST", `${prefix}/deploy`, {
+          name: `Cold catalog ${randomUUID().slice(0, 8)}`,
+          files: files("Initial"),
+        });
+        expect(deployed.status, JSON.stringify(deployed.body)).toBe(200);
+        const app = yield* body(App, deployed);
+        const path = `${prefix}/${app.id}`;
+        yield* Effect.addFinalizer(() =>
+          api.request(actors.owner, "DELETE", path).pipe(Effect.orDie),
+        );
+        const search = (label: string, revision: string) =>
+          Effect.gen(function* () {
+            const start = yield* Clock.currentTimeMillis;
+            const result = yield* client.use(label, (client, signal) =>
+              client.callTool(
+                {
+                  name: "search",
+                  arguments: { namespace: app.slug, query: "records.getRecord8548", limit: 5 },
+                },
+                undefined,
+                { signal },
+              ),
+            );
+            const elapsed = (yield* Clock.currentTimeMillis) - start;
+            yield* Console.log(
+              JSON.stringify({ label, elapsed, result: result.structuredContent }),
+            );
+            yield* evidence.json(`${label}.json`, { elapsed, result: result.structuredContent });
+            expect(result.isError).not.toBe(true);
+            const found = yield* Schema.decodeUnknownEffect(Search)(result.structuredContent);
+            expect(found.unavailableApps ?? []).toEqual([]);
+            expect(found.items).toHaveLength(1);
+            expect(found.items[0]!.path).toContain("getRecord8548");
+            expect(found.items[0]!.description).toBe(`${revision} record 8548`);
+          });
+        yield* search("initial-cold", "Initial");
+        const changed = yield* saveAndDeploy(actors.owner, path, { files: files("Updated") });
+        expect(changed.status).toBe(200);
+        yield* Effect.forEach([0, 1, 2, 3], (i) => search(`concurrent-cold-${i}`, "Updated"), {
+          concurrency: "unbounded",
+        });
+        yield* search("warm", "Updated");
+        // Faster discovery must not omit the shared cyclic schemas. Read through the public
+        // tool endpoint only after the cold searches, so this cannot prewarm their catalog.
+        const described = yield* body(
+          Schema.Struct({
+            description: Schema.String,
+            inputSchema: Schema.Record(Schema.String, Schema.Json),
+            outputSchema: Schema.Struct({
+              anyOf: Schema.Array(Schema.Struct({ $ref: Schema.String })),
+              $defs: Schema.Record(Schema.String, Schema.Json),
+            }),
+          }),
+          yield* api.request(actors.owner, "GET", `${path}/tools/records.getRecord8548`),
+        );
+        expect(described.description).toBe("Updated record 8548");
+        expect(described.inputSchema).toMatchObject({
+          properties: { query: { properties: { id: { type: "string" } } } },
+        });
+        expect(described.outputSchema.anyOf).toEqual([{ $ref: "#/$defs/Record0" }]);
+        expect(Object.keys(described.outputSchema.$defs)).toHaveLength(601);
+        expect(described.outputSchema.$defs.Record0).toMatchObject({
+          properties: { parent: { $ref: "#/$defs/Root" }, value: { type: "string" } },
+        });
+        expect(described.outputSchema.$defs.Root).toMatchObject({
+          properties: { record599: { $ref: "#/$defs/Record599" } },
+        });
+        // Walk public search pages, not the management index, so no prewarm can hide the cold path.
+        const paged = yield* client.use("Walk the updated catalog", (client, signal) =>
+          client.callTool(
+            {
+              name: "execute",
+              arguments: {
+                code: `const items = [];
+for (let page = await tools.search({ namespace: ${JSON.stringify(app.slug)}, limit: 100 }); ; page = await tools.search(page.next)) {
+  items.push(...page.items);
+  if (page.next === null) break;
+}
+return { count: items.length, unique: new Set(items.map(item => item.path)).size, updated: items.every(item => item.description.startsWith("Updated record ")) };`,
+              },
+            },
+            undefined,
+            { signal },
+          ),
+        );
+        yield* evidence.json("updated-pages.json", paged.structuredContent);
+        expect(paged.structuredContent).toMatchObject({
+          status: "completed",
+          execution: { ok: true, value: { count, unique: count, updated: true } },
+          unavailableApps: [],
+        });
+      }).pipe(Effect.provide(McpClient.layer)),
+    ),
+  );
+});

packages/apps/src/implementation/openapi-source.ts

@@ -604,7 +604,9 @@ export const liveOpenapiRouter = (options: OpenapiSourceOptions): RouterDeclarat
   // is released before the parts are written.
   const serialize = (context: CacheLoadContext) =>
     Effect.gen(function* () {
-      const { revision, meta, compiled } = yield* compile(context);
+      const { revision, meta, compiled } = yield* compile(context).pipe(
+        Effect.withSpan("app.openapi.compile"),
+      );
       // oxlint-disable-next-line executor/authored-code-through-adapter -- scheduler yield
       yield* Effect.promise(yieldToRuntime);
       const summaries = paginate(compiled.operations.map(summaryOf));
@@ -664,7 +666,9 @@ export const liveOpenapiRouter = (options: OpenapiSourceOptions): RouterDeclarat
     });
   const refresh = (context: CacheLoadContext) =>
     Effect.gen(function* () {
-      const { manifest, batches, retained, retainedBytes } = yield* serialize(context);
+      const { manifest, batches, retained, retainedBytes } = yield* serialize(context).pipe(
+        Effect.withSpan("app.openapi.serialize"),
+      );
       // Revision keys are immutable and independent. Await every write before
       // publishing the manifest, without serializing their network round trips.
       yield* Effect.forEach(
@@ -1015,17 +1019,21 @@ export const liveOpenapiRouter = (options: OpenapiSourceOptions): RouterDeclarat
     list: () =>
       withRevision((manifest) =>
         Effect.gen(function* () {
-          const all = yield* operationsFor(manifest);
+          const all = yield* operationsFor(manifest).pipe(
+            Effect.withSpan("app.openapi.operations"),
+          );
           if (all === undefined) return undefined;
-          const definitions = yield* definitionsFor(manifest, all);
+          const definitions = yield* definitionsFor(manifest, all).pipe(
+            Effect.withSpan("app.openapi.definitions"),
+          );
           if (definitions === undefined) return undefined;
           const bundle = bundler(definitions);
           const request = createRequest(options);
-          return {
+          return yield* Effect.sync(() => ({
             value: all
               .filter((op) => request.available(op, options.account))
               .map((operation) => describe(operation, bundle)),
-          };
+          })).pipe(Effect.withSpan("app.openapi.schemas"));
         }),
       ),
     summaries: () =>

packages/apps/src/implementation/openapi.ts

@@ -34,25 +34,36 @@ export function references(value: JsonValue, found = new Set<string>()): Set<str
  */
 export const bundler = (definitions: Readonly<Record<string, JsonObject>>) => {
   const direct = new Map<string, ReadonlySet<string>>();
+  // Operations often reach the same graph. Walk it once per ordered set of roots, and share
+  // the resulting $defs so catalog validation and encoding also visit it only once. This cache
+  // belongs to this bundler's immutable definitions, never to another revision or account.
+  const closures = new Map<string, JsonObject | null>();
   return (schema: JsonObject): JsonObject => {
-    const reached = new Map<string, JsonObject>();
     const pending = [...references(schema)];
-    for (let name = pending.pop(); name !== undefined; name = pending.pop()) {
-      const definition = Object.hasOwn(definitions, name) ? definitions[name] : undefined;
-      if (definition === undefined || reached.has(name)) continue;
-      reached.set(name, definition);
-      let names = direct.get(name);
-      if (names === undefined) direct.set(name, (names = references(definition)));
-      pending.push(...names);
+    if (pending.length === 0) return schema;
+    const key = JSON.stringify(pending);
+    let bundled = closures.get(key);
+    if (bundled === undefined) {
+      const reached = new Map<string, JsonObject>();
+      for (let name = pending.pop(); name !== undefined; name = pending.pop()) {
+        const definition = Object.hasOwn(definitions, name) ? definitions[name] : undefined;
+        if (definition === undefined || reached.has(name)) continue;
+        reached.set(name, definition);
+        let names = direct.get(name);
+        if (names === undefined) direct.set(name, (names = references(definition)));
+        pending.push(...names);
+      }
+      bundled = reached.size === 0 ? null : Object.fromEntries(reached);
+      closures.set(key, bundled);
     }
-    if (reached.size === 0) return schema;
+    if (bundled === null) return schema;
     const own = schema.$defs;
     return {
       ...schema,
-      $defs: {
-        ...Object.fromEntries(reached),
-        ...(own !== null && typeof own === "object" && !Array.isArray(own) ? own : {}),
-      },
+      $defs:
+        own !== null && typeof own === "object" && !Array.isArray(own)
+          ? { ...bundled, ...own }
+          : bundled,
     };
   };
 };