fix(agent-directory): isolate bulk reads per entry; verify stanza parity (CL-7592)

Author: TheGreatAxios

apps/hub/src/skills-mount.test.ts

@@ -0,0 +1,120 @@
+// Covers the `pinnedBy` resolver `mountSkills` hands the skill registry:
+// every definition's pins are read out of its own asset snapshot, so one
+// unreadable asset (a pre-cutover retired envelope, a missing blob) skips
+// its row — reported, never failing the whole resolve — and the fan-out
+// stays bounded no matter how many definitions a tenant carries.
+import { expect, test } from "bun:test";
+
+import type { DB } from "@intx/db";
+import type { AssetService, RepoStore } from "@intx/hub-sessions";
+import {
+  agentDefinitionSourceTree,
+  AGENT_DEFINITION_ENTRY_PATH,
+  buildAgentDefinitionWorkflow,
+  reindexPinnedSkills,
+  RetiredWorkflowEnvelopeError,
+  serializeAgentDefinitionWorkflow,
+} from "@corbits/agent-directory";
+
+import { mountSkills } from "./skills-mount";
+
+/** Entry-module bytes pinning `names` — the stanza `pinnedBy` reads. */
+function definitionBytesPinning(...names: string[]): Uint8Array {
+  const tree = agentDefinitionSourceTree({
+    handle: "research-buddy",
+    workflowJson: reindexPinnedSkills(
+      serializeAgentDefinitionWorkflow(
+        buildAgentDefinitionWorkflow({
+          handle: "research-buddy",
+          tenantDomain: "acme.example",
+          description: "",
+          systemPrompt: "You are a careful research assistant.",
+        }),
+      ),
+      names.map((name) => ({ name, description: `What ${name} does.` })),
+    ),
+  });
+  return new TextEncoder().encode(tree[AGENT_DEFINITION_ENTRY_PATH]);
+}
+
+function mountFor(
+  rows: readonly {
+    id: string;
+    tenantId: string;
+    assetId: string | null;
+    name: string;
+  }[],
+  readAssetBlob: AssetService["readAssetBlob"],
+) {
+  const db = {
+    query: {
+      workflowDefinition: { findMany: async () => rows },
+    },
+  } as unknown as DB["db"];
+  return mountSkills({
+    db,
+    assetService: { readAssetBlob } as AssetService,
+    repoStore: {} as RepoStore,
+  });
+}
+
+test("a row on the retired envelope skips while healthy rows still resolve", async () => {
+  const mount = mountFor(
+    [
+      {
+        id: "def_healthy",
+        tenantId: "tnt_1",
+        assetId: "ast_healthy",
+        name: "research-buddy",
+      },
+      {
+        id: "def_retired",
+        tenantId: "tnt_1",
+        assetId: "ast_retired",
+        name: "old-scout",
+      },
+    ],
+    (params) =>
+      params.assetId === "ast_healthy"
+        ? Promise.resolve(definitionBytesPinning("research"))
+        : Promise.reject(new RetiredWorkflowEnvelopeError(params.assetId)),
+  );
+  const pinning = await mount.pinnedBy.resolve("tnt_1", "research");
+  expect(pinning).toEqual([
+    { definitionId: "def_healthy", name: "research-buddy" },
+  ]);
+});
+
+test("the blob fan-out stays bounded no matter how many definitions pin", async () => {
+  const COUNT = 20;
+  const rows = Array.from({ length: COUNT }, (_, index) => ({
+    id: `def_${index}`,
+    tenantId: "tnt_1",
+    assetId: `ast_${index}`,
+    name: `buddy-${index}`,
+  }));
+  let release!: () => void;
+  const gate = new Promise<void>((resolve) => {
+    release = resolve;
+  });
+  let active = 0;
+  let peak = 0;
+  const mount = mountFor(rows, async () => {
+    active += 1;
+    peak = Math.max(peak, active);
+    try {
+      await gate;
+      return definitionBytesPinning("research");
+    } finally {
+      active -= 1;
+    }
+  });
+  const pending = mount.pinnedBy.resolve("tnt_1", "research");
+  // Every worker reaches the gate before any read can finish, so the
+  // peak observed here is the whole fan-out.
+  await new Promise((resolve) => setTimeout(resolve, 20));
+  expect(peak).toBeLessThanOrEqual(8);
+  release();
+  const pinning = await pending;
+  expect(pinning).toHaveLength(COUNT);
+});

apps/hub/src/skills-mount.ts

@@ -14,6 +14,7 @@ import {
   readPinnedSkillNames,
   type PinnedSkillIndexResolver,
 } from "@corbits/agent-directory";
+import { reportError } from "@corbits/error-sink";
 import {
   createDrizzleSkillAccessStore,
   createHubSkillAssetStore,
@@ -29,6 +30,37 @@ export type SkillsMount = {
   skillIndex: PinnedSkillIndexResolver;
 };
 
+/** At most this many concurrent asset-blob reads while resolving who
+ * pins a skill — a tenant's definition count is unbounded, and one
+ * `readAssetBlob` per definition with no cap is a self-inflicted load
+ * spike against the asset store. */
+const PINNED_BY_READ_CONCURRENCY = 8;
+
+/** Runs `fn` over `items` with at most `limit` in flight, preserving
+ * order — the same bounded fan-out every per-row asset read in this
+ * composition root needs, so a many-definition tenant cannot open a
+ * blob read per definition at once. */
+async function mapWithConcurrencyLimit<T, R>(
+  items: readonly T[],
+  limit: number,
+  fn: (item: T) => Promise<R>,
+): Promise<readonly R[]> {
+  const results = new Array<R>(items.length);
+  let next = 0;
+  const workers = Array.from(
+    { length: Math.min(limit, items.length) },
+    async () => {
+      while (next < items.length) {
+        const index = next;
+        next += 1;
+        results[index] = await fn(items[index] as T);
+      }
+    },
+  );
+  await Promise.all(workers);
+  return results;
+}
+
 export function mountSkills(deps: {
   db: DB["db"];
   assetService: AssetService;
@@ -48,20 +80,37 @@ export function mountSkills(deps: {
       const rows = await deps.db.query.workflowDefinition.findMany({
         where: and(eq(workflowDefinition.tenantId, tenantId)),
       });
-      const pinning: { definitionId: string; name: string }[] = [];
-      for (const row of rows) {
-        if (row.assetId === null) continue;
-        // Pins live in the definition's own asset snapshot — the same
-        // stanza every agent-directory read goes through.
-        const workflowJson = await readAgentDefinitionWorkflowJson(
-          deps.assetService,
-          row.assetId,
-        );
-        if (readPinnedSkillNames(workflowJson).includes(skillName)) {
-          pinning.push({ definitionId: row.id, name: row.name });
-        }
-      }
-      return pinning;
+      const candidates = rows.filter(
+        (row): row is typeof row & { assetId: string } => row.assetId !== null,
+      );
+      // Pins live in each definition's own asset snapshot — the same
+      // stanza every agent-directory read goes through. Bounded fan-out
+      // instead of the old sequential N+1, and one unreadable asset
+      // (a pre-cutover retired envelope, a missing blob) skips its
+      // row — reported, never failing the whole resolve.
+      const matches = await mapWithConcurrencyLimit(
+        candidates,
+        PINNED_BY_READ_CONCURRENCY,
+        async (row) => {
+          try {
+            const workflowJson = await readAgentDefinitionWorkflowJson(
+              deps.assetService,
+              row.assetId,
+            );
+            return readPinnedSkillNames(workflowJson).includes(skillName)
+              ? { definitionId: row.id, name: row.name }
+              : null;
+          } catch (err) {
+            reportError(err, {
+              operation: "skills.pinnedBy.resolve",
+              tenantId,
+              extra: { definitionId: row.id, skillName },
+            });
+            return null;
+          }
+        },
+      );
+      return matches.filter((match) => match !== null);
     },
   };
 

packages/agent-directory/src/migrations.ts

@@ -29,6 +29,19 @@ export const agentDirectoryMigrations: readonly AgentDirectoryMigration[] = [
   // definition assets' own pinned-skills stanzas: the table 0001 created
   // is dropped, never read again. Append-only like every ledger entry
   // before it — history is not rewritten, the store is deleted forward.
+  //
+  // Parity (stanza ⊇ store, so the drop loses nothing): from the
+  // table's introduction (baabe260) to this cutover, every `setSkills`
+  // writer dual-wrote the identical skill set into the asset stanza
+  // first — create (`createAgentDefinitionCore`), `PUT
+  // /:definitionId/skills`, the skill-pin route, and the
+  // capability-add skill path all `reindexPinnedSkills` the written
+  // workflow and persist the store only in `afterWrite`, which runs
+  // after the asset write succeeds. The store has no delete path and
+  // reads a missing row as []. A crash between the two writes leaves
+  // the stanza ahead (safe: reads now come from the stanza); no path
+  // writes the store without first writing the stanza, so no dropped
+  // row can name a skill its asset's stanza lacks.
   {
     name: "0002_drop_definition_skills",
     sql: `DROP TABLE IF EXISTS "agent_directory"."definition_skills";`,

packages/agent-directory/src/routes.ts

@@ -274,22 +274,34 @@ export function createAgentDefinitionRoutes({
 
       const entries = await Promise.all(
         ids.map(async (definitionId) => {
-          const row = await db.query.workflowDefinition.findFirst({
-            where: and(
-              eq(workflowDefinition.id, definitionId),
-              eq(workflowDefinition.tenantId, tenant.id),
-            ),
-          });
-          if (row === undefined || row.assetId === null) return null;
-          // Pins read out of the asset's own stanza: the bulk read
-          // survives the side table's deletion by going to the same
-          // source `GET /:definitionId` reads.
-          const workflowJson = await readAgentDefinitionWorkflowJson(
-            assetService,
-            row.assetId,
-          );
-          const skills = readPinnedSkillNames(workflowJson);
-          return [definitionId, skills] as const;
+          try {
+            const row = await db.query.workflowDefinition.findFirst({
+              where: and(
+                eq(workflowDefinition.id, definitionId),
+                eq(workflowDefinition.tenantId, tenant.id),
+              ),
+            });
+            if (row === undefined || row.assetId === null) return null;
+            // Pins read out of the asset's own stanza: the bulk read
+            // survives the side table's deletion by going to the same
+            // source `GET /:definitionId` reads. One unreadable asset
+            // (a pre-cutover retired envelope, a missing blob) must not
+            // fail the whole batch — skip that id, report it, serve the
+            // healthy ones.
+            const workflowJson = await readAgentDefinitionWorkflowJson(
+              assetService,
+              row.assetId,
+            );
+            const skills = readPinnedSkillNames(workflowJson);
+            return [definitionId, skills] as const;
+          } catch (err) {
+            reportError(err, {
+              operation: "agentDirectory.bulkSkills",
+              tenantId: tenant.id,
+              extra: { definitionId },
+            });
+            return null;
+          }
         }),
       );
 

packages/agent-directory/test/routes.test.ts

@@ -23,12 +23,12 @@ import {
   serializeAgentDefinitionWorkflow,
   withAgentToolPackagePin,
   readPinnedSkillNames,
-  reindexPinnedSkills,
 } from "../src/agent-workflow";
 import {
   agentDefinitionSourceTree,
   AGENT_DEFINITION_ENTRY_PATH,
   readAgentDefinitionWorkflowJson,
+  RetiredWorkflowEnvelopeError,
 } from "../src/definition-asset";
 import { createAgentDefinitionRoutes } from "../src/routes";
 import type { PinnedSkillIndexResolver } from "../src/routes";
@@ -38,7 +38,11 @@ import {
 } from "../src/workflow-skill-pin-routes";
 import type { DefinitionAssetHistory } from "../src/definition-history";
 import type { CapabilityInventoryProvider } from "../src/capability-inventory";
-import { definitionFrom, SOURCE_TREE_PATHS } from "./source-tree";
+import {
+  definitionFrom,
+  SOURCE_TREE_PATHS,
+  storedDefinitionBytesWithSkills,
+} from "./source-tree";
 
 /** A `readAssetBlob` that always answers the definition's entry module
  * with `workflowBytes` — pins live in the asset's own stanza, so a test
@@ -213,27 +217,6 @@ function storedDefinitionBytesWithModel(model: string): Uint8Array {
   return new TextEncoder().encode(tree[AGENT_DEFINITION_ENTRY_PATH]);
 }
 
-/** A stored definition that already pins skills — the state every
- * pin-reading route observes. The stanza is the seed: no side table to
- * write, the bytes carry the pins like a real asset would. */
-function storedDefinitionBytesWithSkills(...names: string[]): Uint8Array {
-  const tree = agentDefinitionSourceTree({
-    handle: "research-buddy",
-    workflowJson: reindexPinnedSkills(
-      serializeAgentDefinitionWorkflow(
-        buildAgentDefinitionWorkflow({
-          handle: "research-buddy",
-          tenantDomain: TENANT.domain,
-          description: "",
-          systemPrompt: "You are a careful research assistant.",
-        }),
-      ),
-      names.map((name) => ({ name, description: `What ${name} does.` })),
-    ),
-  });
-  return new TextEncoder().encode(tree[AGENT_DEFINITION_ENTRY_PATH]);
-}
-
 /** The model the one step agent resolves against, or `undefined` when it
  * pins none. */
 function modelFrom(workflowJson: string): string | undefined {
@@ -763,6 +746,49 @@ test("GET /skills omits unknown definition ids from the map rather than erroring
   expect(body.skills).toEqual({});
 });
 
+test("GET /skills serves the healthy ids when one asset is on the retired envelope", async () => {
+  // The route issues one `findFirst` per requested id, in request order
+  // (each `map` callback runs synchronously to its first await), so the
+  // fake answers each call from this queue — drizzle's `where`
+  // expression tree isn't inspectable without a real query builder.
+  const rows = [
+    { id: "def_healthy", assetId: "ast_healthy" },
+    { id: "def_retired", assetId: "ast_retired" },
+  ];
+  const db = {
+    query: {
+      workflowDefinition: {
+        findFirst: async () => {
+          const row = rows.shift();
+          return row === undefined
+            ? undefined
+            : {
+                id: row.id,
+                tenantId: TENANT.id,
+                assetId: row.assetId,
+                name: "Research Buddy",
+              };
+        },
+      },
+    },
+  } as unknown as DB["db"];
+  const app = buildApp(
+    fakeAssetService({
+      readAssetBlob: (params) =>
+        params.assetId === "ast_healthy"
+          ? Promise.resolve(storedDefinitionBytesWithSkills("web-research"))
+          : Promise.reject(new RetiredWorkflowEnvelopeError(params.assetId)),
+    }),
+    db,
+  );
+  const response = await app.request("/skills?ids=def_healthy,def_retired");
+  expect(response.status).toBe(200);
+  const body = (await response.json()) as {
+    skills: Record<string, readonly string[]>;
+  };
+  expect(body.skills).toEqual({ def_healthy: ["web-research"] });
+});
+
 test("PUT /:definitionId/skills replaces the skill set, writing the definition source tree to the asset", async () => {
   let writtenFiles: Record<string, string | Uint8Array> | undefined;
   const app = buildApp(