feat(agent-directory)!: read pinned skills from the asset, drop definition_skills

apps/hub/src/index.ts

@@ -55,7 +55,6 @@ import {
   createAgentDefinitionDraftRoutes,
   createAgentDefinitionRoutes,
   createDefinitionAssetHistory,
-  createDrizzleDefinitionSkillsStore,
   createWorkflowAgentCreateRoutes,
   createWorkflowCapabilityRoutes,
   createWorkflowSkillPinRoutes,
@@ -2183,7 +2182,6 @@ export async function createHub(config: HubConfig) {
     assetService,
     repoStore: agentRepoStore.repoStore,
   });
-  const definitionSkillsStore = createDrizzleDefinitionSkillsStore(db);
   app.route(
     `${TENANT_PREFIX}/skills`,
     createSkillRoutes({
@@ -2354,7 +2352,6 @@ export async function createHub(config: HubConfig) {
       assetService,
       deployer: workflowDeployer,
       skillIndex: skills.skillIndex,
-      skillsStore: definitionSkillsStore,
       history: createDefinitionAssetHistory({
         repoStore: agentRepoStore.repoStore,
       }),
@@ -2384,7 +2381,6 @@ export async function createHub(config: HubConfig) {
       assetService,
       deployer: workflowDeployer,
       skillIndex: skills.skillIndex,
-      skillsStore: definitionSkillsStore,
       capabilityInventory,
       authenticator: createWorkflowRunAuthenticator({ db }),
       tenantDefaultModel: async (tenantId) =>
@@ -2406,7 +2402,6 @@ export async function createHub(config: HubConfig) {
       assetService,
       deployer: workflowDeployer,
       skillIndex: skills.skillIndex,
-      skillsStore: definitionSkillsStore,
       capabilityInventory,
       authenticator: createWorkflowRunAuthenticator({ db }),
     }),
@@ -2423,7 +2418,6 @@ export async function createHub(config: HubConfig) {
       assetService,
       deployer: workflowDeployer,
       skillIndex: skills.skillIndex,
-      skillsStore: definitionSkillsStore,
       authenticator: createWorkflowRunAuthenticator({ db }),
     }),
   );

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

@@ -1,18 +1,20 @@
 // Composition for `@corbits/skills`: the registry itself plus the two
 // adapters that only this composition root can supply — "which agent
-// definitions pin this skill" (read from `@corbits/agent-directory`'s
-// own `definition_skills` table, keyed by each definition's asset id)
-// and "what index does a definition's pinned names resolve to" (read
-// from the registry, on behalf of the pushing principal).
+// definitions pin this skill" (read from each definition's own asset
+// snapshot, where its pinned-skills stanza lives) and "what index does
+// a definition's pinned names resolve to" (read from the registry, on
+// behalf of the pushing principal).
 import { and, eq } from "drizzle-orm";
 
 import type { DB } from "@intx/db";
 import { workflowDefinition } from "@intx/db/schema";
 import { type AssetService, type RepoStore } from "@intx/hub-sessions";
 import {
-  createDrizzleDefinitionSkillsStore,
+  readAgentDefinitionWorkflowJson,
+  readPinnedSkillNames,
   type PinnedSkillIndexResolver,
 } from "@corbits/agent-directory";
+import { reportError } from "@corbits/error-sink";
 import {
   createDrizzleSkillAccessStore,
   createHubSkillAssetStore,
@@ -28,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;
@@ -42,22 +75,42 @@ export function mountSkills(deps: {
     access: createDrizzleSkillAccessStore(deps.db),
   });
 
-  const definitionSkills = createDrizzleDefinitionSkillsStore(deps.db);
-
   const pinnedBy: PinnedByResolver = {
     async resolve(tenantId, skillName) {
       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;
-        const skills = await definitionSkills.getSkills(row.assetId);
-        if (skills.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/agent-workflow.test.ts

@@ -14,11 +14,12 @@ import {
 import {
   buildAgentDefinitionWorkflow,
   createAgentDefinitionCore,
+  readPinnedSkillNames,
+  reindexPinnedSkills,
   serializeAgentDefinitionWorkflow,
   SKILLS_TOOL_PACKAGE_PIN,
   withAgentToolPackagePin,
 } from "./agent-workflow";
-import { createInMemoryDefinitionSkillsStore } from "./skills-store";
 
 describe("SKILLS_TOOL_PACKAGE_PIN", () => {
   test("resolves through the corbits-tools registry", async () => {
@@ -91,6 +92,45 @@ describe("withAgentToolPackagePin", () => {
   });
 });
 
+// CL-7592: pinned skill names read back out of the definition's own
+// serialized `workflow.json` (the `<available_skills>` stanza
+// `reindexPinnedSkills` writes) — the asset is the source of truth for
+// pins now that the Workbench-owned `definition_skills` store is gone.
+describe("readPinnedSkillNames", () => {
+  function freshWorkflowJson(): string {
+    return serializeAgentDefinitionWorkflow(
+      buildAgentDefinitionWorkflow({
+        handle: "pin-test",
+        tenantDomain: "example.test",
+        description: "",
+        systemPrompt: "You are a test agent.",
+      }),
+    );
+  }
+
+  test("a definition with no pins reads back no names", () => {
+    expect(readPinnedSkillNames(freshWorkflowJson())).toEqual([]);
+  });
+
+  test("round-trips the names `reindexPinnedSkills` writes", () => {
+    const workflowJson = reindexPinnedSkills(freshWorkflowJson(), [
+      { name: "web-research", description: "Researches the web." },
+      { name: "long-form-write", description: "Writes long documents." },
+    ]);
+    expect(readPinnedSkillNames(workflowJson)).toEqual([
+      "web-research",
+      "long-form-write",
+    ]);
+  });
+
+  test("unpinning everything reads back no names", () => {
+    const pinned = reindexPinnedSkills(freshWorkflowJson(), [
+      { name: "web-research", description: "Researches the web." },
+    ]);
+    expect(readPinnedSkillNames(reindexPinnedSkills(pinned, []))).toEqual([]);
+  });
+});
+
 // CL-7389: a `create_agent`/`POST /agent-definitions` call pinning several
 // tool packages by name shares one registry resolver across all of them
 // (`createPinnedVersionResolver`), so it costs one ancestor walk and one
@@ -164,7 +204,6 @@ describe("createAgentDefinitionCore: shared registry resolution across pins", ()
         db,
         assetService,
         skillIndex: { resolve: () => Promise.resolve([]) },
-        skillsStore: createInMemoryDefinitionSkillsStore(),
         deployer: {
           deploy: () =>
             Promise.resolve({