diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 959398695..53f75a1bb 100644 --- a/apps/hub/src/index.ts +++ b/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 }), }), ); diff --git a/apps/hub/src/skills-mount.test.ts b/apps/hub/src/skills-mount.test.ts new file mode 100644 index 000000000..d8f246a7f --- /dev/null +++ b/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((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); +}); diff --git a/apps/hub/src/skills-mount.ts b/apps/hub/src/skills-mount.ts index e60fe76ce..ecb57d57a 100644 --- a/apps/hub/src/skills-mount.ts +++ b/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( + items: readonly T[], + limit: number, + fn: (item: T) => Promise, +): Promise { + const results = new Array(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); }, }; diff --git a/packages/agent-directory/src/agent-workflow.test.ts b/packages/agent-directory/src/agent-workflow.test.ts index 247e8c9dd..9106f837f 100644 --- a/packages/agent-directory/src/agent-workflow.test.ts +++ b/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 `` 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({ diff --git a/packages/agent-directory/src/agent-workflow.ts b/packages/agent-directory/src/agent-workflow.ts index 197e838da..2d9fec191 100644 --- a/packages/agent-directory/src/agent-workflow.ts +++ b/packages/agent-directory/src/agent-workflow.ts @@ -22,6 +22,8 @@ import { asset, workflowDefinition } from "@intx/db/schema"; import { AssetServiceError } from "@intx/hub-sessions"; import type { AssetService } from "@intx/hub-sessions"; import { + AVAILABLE_SKILLS_CLOSE_TAG, + AVAILABLE_SKILLS_OPEN_TAG, withAvailableSkills, type PinnedSkillIndexEntry, } from "@corbits/skills"; @@ -32,7 +34,6 @@ import { writeAndDeployAgentDefinition, type AgentDefinitionDeployer, } from "./definition-asset"; -import type { DefinitionSkillsStore } from "./skills-store"; import { createPinnedVersionResolver } from "./tool-package-version"; export const AGENT_DEFINITION_STEP_ID = "agent"; @@ -125,6 +126,48 @@ export function reindexPinnedSkills( return JSON.stringify(definition); } +/** Reads a definition's pinned skill names back out of its serialized + * `workflow.json` — the `` stanza `reindexPinnedSkills` + * writes into the step agent's system prompt. The asset is the source of + * truth for pins: every writer of pins (create, skills edit, capability + * add, run pin) reindexes the stanza in the same commit it deploys, so a + * reader never needs side state beside the definition row. Every builder + * in this codebase produces exactly one step, so the definition's one + * step is unambiguous regardless of the step's own key. */ +export function readPinnedSkillNames(workflowJson: string): readonly string[] { + const raw: unknown = JSON.parse(workflowJson); + const definition = DefinitionWithAgentSteps(raw); + if (definition instanceof type.errors) { + throw new Error( + `workflow.json does not carry step agents to read pinned skills from: ${definition.summary}`, + ); + } + const [step] = Object.values(definition.steps); + if (step === undefined) { + throw new Error("workflow.json has no steps"); + } + const open = step.agent.systemPrompt.indexOf(AVAILABLE_SKILLS_OPEN_TAG); + const close = step.agent.systemPrompt.indexOf(AVAILABLE_SKILLS_CLOSE_TAG); + if (open === -1 || close === -1 || close < open) return []; + // Stanza lines render as `- name: description`, and skill names can + // never contain a space or a colon, so the first colon on a `- ` line + // always ends the name — even when the description itself holds colons. + const body = step.agent.systemPrompt.slice( + open + AVAILABLE_SKILLS_OPEN_TAG.length, + close, + ); + const names: string[] = []; + for (const line of body.split("\n")) { + const trimmed = line.trim(); + if (!trimmed.startsWith("- ")) continue; + const colon = trimmed.indexOf(":", 2); + if (colon === -1) continue; + const name = trimmed.slice(2, colon).trim(); + if (name !== "") names.push(name); + } + return names; +} + /** Reads a definition's system prompt back out of its serialized * `workflow.json` — the raw text a person edits in the Assistant * settings section, before `reindexPinnedSkills` appends the @@ -435,7 +478,6 @@ export type CreateAgentDefinitionCoreDeps = { names: readonly string[], ): Promise; }; - readonly skillsStore: DefinitionSkillsStore; /** * Resolves the tenant's current catalog default model — the same * first-connected-provider model `@corbits/chat`'s @@ -500,8 +542,9 @@ export class DuplicateAgentHandleError extends Error { /** * The full create-agent-definition sequence: resolve the tenant's mail * domain, build and pin the definition's serialized workflow, materialize - * it as a `workflow`-kind asset, persist its pinned skills, and - * project it onto a first-class `workflow_definition` row. Factored out + * it as a `workflow`-kind asset carrying its pinned-skills index in its + * own stanza, and project it onto a first-class `workflow_definition` + * row. Factored out * of `./routes.ts`'s `POST /` handler so `./workflow-create-routes.ts` * (a workflow-run-authenticated surface a tool call reaches, never a * person through a form) can create a definition through the exact @@ -604,7 +647,6 @@ export async function createAgentDefinitionCore( workflowJson, message: `Define agent ${input.name}`, }); - await deps.skillsStore.setSkills(assetId, input.skills); // The deploy above projects, walks, and stamps the definition row in // one transaction — the same machinery the sidecar probe deploy diff --git a/packages/agent-directory/src/asset-write.ts b/packages/agent-directory/src/asset-write.ts index 826ee0620..5196ae090 100644 --- a/packages/agent-directory/src/asset-write.ts +++ b/packages/agent-directory/src/asset-write.ts @@ -41,7 +41,6 @@ export type PreparedAgentAssetWrite = { workflowJson: string; message: string; result: T; - afterWrite?: () => Promise; }; export type CommitLatestAgentAssetSnapshotArgs = { @@ -81,9 +80,6 @@ export async function commitLatestAgentAssetSnapshot( workflowJson: prepared.workflowJson, message: prepared.message, }); - if (prepared.afterWrite !== undefined) { - await prepared.afterWrite(); - } return true; }); if (wrote) return prepared.result; diff --git a/packages/agent-directory/src/capability-add.ts b/packages/agent-directory/src/capability-add.ts index 3ecf06f90..b2628bef4 100644 --- a/packages/agent-directory/src/capability-add.ts +++ b/packages/agent-directory/src/capability-add.ts @@ -12,6 +12,7 @@ import type { AssetService } from "@intx/hub-sessions"; import { readAgentCapabilities, + readPinnedSkillNames, reindexPinnedSkills, withAgentModel, withAgentToolPackagePin, @@ -22,14 +23,12 @@ import { writeAndDeployAgentDefinition, type AgentDefinitionDeployer, } from "./definition-asset"; -import type { DefinitionSkillsStore } from "./skills-store"; import { resolvePinnedVersion } from "./tool-package-version"; export type CommitAgentCapabilityAddArgs = { db: DB["db"]; assetService: AssetService; deployer: AgentDefinitionDeployer; - skillsStore: DefinitionSkillsStore; skillIndex: { resolve( tenantId: string, @@ -66,17 +65,10 @@ export async function commitAgentCapabilityAdd( operation: "capability add", prepare: async (snapshot) => { const prepared = await prepareCapabilityAdd(snapshot, args); - const nextSkills = prepared.nextSkills; return { workflowJson: prepared.workflowJson, message: prepared.message, result: prepared.result, - ...(nextSkills !== null - ? { - afterWrite: () => - args.skillsStore.setSkills(args.assetId, nextSkills), - } - : {}), }; }, write: async ({ workflowJson, message }) => { @@ -97,7 +89,6 @@ export async function commitAgentCapabilityAdd( type PreparedCapabilityAdd = { workflowJson: string; message: string; - nextSkills: readonly string[] | null; result: CommitAgentCapabilityAddResult; }; @@ -107,8 +98,11 @@ async function prepareCapabilityAdd( ): Promise { let nextWorkflowJson: string; let message: string; - let skills = await args.skillsStore.getSkills(args.assetId); - let nextSkills: readonly string[] | null = null; + // Pins read out of the snapshot itself — the asset's stanza is the + // source of truth, so a concurrent writer's pins survive a retry + // against the latest snapshot instead of being clobbered by a stale + // side-table read. + let skills = readPinnedSkillNames(workflowJson); switch (args.body.kind) { case "toolPackage": { @@ -137,7 +131,7 @@ async function prepareCapabilityAdd( break; } case "skill": { - nextSkills = skills.includes(args.body.name) + const nextSkills = skills.includes(args.body.name) ? skills : [...skills, args.body.name]; nextWorkflowJson = reindexPinnedSkills( @@ -163,7 +157,6 @@ async function prepareCapabilityAdd( return { workflowJson: nextWorkflowJson, message, - nextSkills, result: { toolPackagePins: capabilities.toolPackagePins, skills, diff --git a/packages/agent-directory/src/index.ts b/packages/agent-directory/src/index.ts index cba857b86..08abd0a11 100644 --- a/packages/agent-directory/src/index.ts +++ b/packages/agent-directory/src/index.ts @@ -2,6 +2,7 @@ export { buildAgentDefinitionWorkflow, serializeAgentDefinitionWorkflow, readAgentCapabilities, + readPinnedSkillNames, reindexPinnedSkills, withAgentModel, withAgentToolPackagePin, @@ -29,11 +30,6 @@ export { AGENT_DEFINITION_ENTRY_PATH, RetiredWorkflowEnvelopeError, } from "./definition-asset"; -export { - createDrizzleDefinitionSkillsStore, - createInMemoryDefinitionSkillsStore, - type DefinitionSkillsStore, -} from "./skills-store"; export { agentDirectoryMigrations, applyAgentDirectoryMigrations, diff --git a/packages/agent-directory/src/migrations.ts b/packages/agent-directory/src/migrations.ts index d37733fd0..d2b48450d 100644 --- a/packages/agent-directory/src/migrations.ts +++ b/packages/agent-directory/src/migrations.ts @@ -25,6 +25,27 @@ export const agentDirectoryMigrations: readonly AgentDirectoryMigration[] = [ ); `, }, + // CL-7592 cuts the Workbench-owned directory store over to the + // 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";`, + }, ]; const LEDGER_TABLE = "agent_directory_migrations"; diff --git a/packages/agent-directory/src/routes.ts b/packages/agent-directory/src/routes.ts index e9ece5f6b..7a315b75a 100644 --- a/packages/agent-directory/src/routes.ts +++ b/packages/agent-directory/src/routes.ts @@ -36,6 +36,7 @@ import { DuplicateAgentHandleError, readAgentCapabilities, readAgentSystemPrompt, + readPinnedSkillNames, reindexPinnedSkills, withAgentSystemPrompt, withoutAgentModel, @@ -55,7 +56,6 @@ import { WorkflowAuthorError, type AgentDefinitionDeployer, } from "./definition-asset"; -import type { DefinitionSkillsStore } from "./skills-store"; import { CreateAgentDefinitionInput, RestoreDefinitionInput, @@ -92,7 +92,6 @@ export type CreateAgentDefinitionRoutesDeps = { db: DB["db"]; assetService: AssetService; skillIndex: PinnedSkillIndexResolver; - skillsStore: DefinitionSkillsStore; history: DefinitionAssetHistory; capabilityInventory: CapabilityInventoryProvider; requireGrant: RequireGrant; @@ -137,7 +136,6 @@ export function createAgentDefinitionRoutes({ db, assetService, skillIndex, - skillsStore, history, capabilityInventory, requireGrant, @@ -228,7 +226,6 @@ export function createAgentDefinitionRoutes({ db, assetService, skillIndex, - skillsStore, deployer, ...(tenantDefaultModel !== undefined ? { tenantDefaultModel } : {}), }, @@ -277,15 +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; - const skills = await skillsStore.getSkills(row.assetId); - 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; + } }), ); @@ -386,7 +402,10 @@ export function createAgentDefinitionRoutes({ row.assetId, ); const capabilities = readAgentCapabilities(workflowJson); - const skills = await skillsStore.getSkills(row.assetId); + // Pins read out of the asset's own stanza: deleting the side + // table leaves this surface's only skills source the snapshot + // every write reindexes. + const skills = readPinnedSkillNames(workflowJson); return c.json({ id: row.id, @@ -473,11 +492,9 @@ export function createAgentDefinitionRoutes({ row.assetId, ); - // Pinned skills live outside the asset tree (see - // `DefinitionSkillsStore`), so restoring a prior commit only ever - // rewrites the definition's source tree — the definition's - // currently pinned skills are untouched by restoring an earlier - // instructions revision. + // Pins live in the asset's own stanza (reindexed on every write), + // so restoring a prior commit restores that revision's pins with + // the source tree — there is no side table left to stay behind. await writeAndDeployAgentDefinition({ assetService, deployer, @@ -490,7 +507,7 @@ export function createAgentDefinitionRoutes({ }); const capabilities = readAgentCapabilities(restoredWorkflowJson); - const skills = await skillsStore.getSkills(row.assetId); + const skills = readPinnedSkillNames(restoredWorkflowJson); return c.json({ id: row.id, @@ -546,7 +563,6 @@ export function createAgentDefinitionRoutes({ db, assetService, deployer, - skillsStore, skillIndex, tenantId: tenant.id, principalId: principal.id, @@ -685,7 +701,9 @@ export function createAgentDefinitionRoutes({ prepare: async (snapshot) => { const workflowJson = withoutAgentModel(snapshot); const capabilities = readAgentCapabilities(workflowJson); - const skills = await skillsStore.getSkills(row.assetId); + // The snapshot carries the pins in its stanza — read them from + // the commit being prepared, not a deleted side table. + const skills = readPinnedSkillNames(workflowJson); return { workflowJson, message: `Clear ${row.name}'s model`, @@ -832,10 +850,11 @@ export function createAgentDefinitionRoutes({ snapshot, await skillIndex.resolve(tenant.id, principal.id, body.skills), ); + // The reindexed stanza is the pins: the commit above writes + // the source of truth, so no post-write side-table sync. return { workflowJson, message: `Update agent skills for ${row.name}`, - afterWrite: () => skillsStore.setSkills(assetId, body.skills), result: { skills: body.skills }, }; }, diff --git a/packages/agent-directory/src/schema.ts b/packages/agent-directory/src/schema.ts deleted file mode 100644 index d4f9c5aab..000000000 --- a/packages/agent-directory/src/schema.ts +++ /dev/null @@ -1,31 +0,0 @@ -// The one table `@corbits/agent-directory` owns: which skill names a -// hand-authored agent definition has pinned. Lives in its own -// `agent_directory` Postgres schema, never `public` — see -// docs/package-migrations.md. Moved off a `skills.json` asset-tree -// sidecar (CL-6135): vendor/intx/hub-sessions' `workflow-kind.ts` -// validates a `workflow`-kind asset tree against a hard allowlist of -// `workflow.json`, `capability-declarations.json`, and `.gitignore` — a -// package-owned sidecar file can never be a fourth entry there without -// forking read-only vendor code, so pinned skills are product-owned -// state instead, keyed by the definition's stable asset id. -import { jsonb, pgSchema, text, timestamp } from "drizzle-orm/pg-core"; - -export const agentDirectorySchema = pgSchema("agent_directory"); - -/** - * `skills` is jsonb — a flat array of skill names, record-as-truth like - * `@corbits/config-profiles`' `profile.entries` — so the pinned-skill - * list never requires a migration to evolve. - */ -export const definitionSkills = agentDirectorySchema.table( - "definition_skills", - { - assetId: text("asset_id").primaryKey(), - skills: jsonb("skills").notNull().default([]), - updatedAt: timestamp("updated_at", { withTimezone: true }) - .notNull() - .defaultNow(), - }, -); - -export type DefinitionSkillsTableRow = typeof definitionSkills.$inferSelect; diff --git a/packages/agent-directory/src/skills-store.ts b/packages/agent-directory/src/skills-store.ts deleted file mode 100644 index f7556967e..000000000 --- a/packages/agent-directory/src/skills-store.ts +++ /dev/null @@ -1,84 +0,0 @@ -// Persistence for the one table this package owns: which skill names a -// hand-authored agent definition has pinned, keyed by the definition's -// stable asset id. Kept apart from route wiring the same way -// `@corbits/config-profiles`' `store.ts` separates persistence from -// `routes.ts`. `DefinitionSkillsStore` is the seam `./routes.ts` and -// `./workflow-capability-routes.ts` depend on; `createDrizzleDefinitionSkillsStore` -// is its one production implementation, over the table in `./schema.ts`. -import { eq } from "drizzle-orm"; -import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; -import { type } from "arktype"; - -import { definitionSkills } from "./schema"; - -export type DefinitionSkillsDb< - TSchema extends Record = Record, -> = PostgresJsDatabase; - -/** Parses `skills` jsonb read back out of the - * `agent_directory.definition_skills` table (see `./schema.ts`) — the DB - * is untrusted the same as any other external boundary, so a row's - * `skills` column is arktype-parsed on the way out rather than `as`-cast, - * and a malformed row fails loud instead of silently masquerading as a - * well-formed `string[]`. */ -const SkillNamesSchema = type("string[]"); - -function parseSkills(raw: unknown): readonly string[] { - const parsed = SkillNamesSchema(raw); - if (parsed instanceof type.errors) { - throw new Error( - `definition_skills row has malformed skills: ${parsed.summary}`, - ); - } - return parsed; -} - -export interface DefinitionSkillsStore { - /** A definition with no row yet — one that has never had skills - * attached — reads as "no skills attached", not an error. */ - getSkills(assetId: string): Promise; - setSkills(assetId: string, skills: readonly string[]): Promise; -} - -export function createDrizzleDefinitionSkillsStore< - TSchema extends Record, ->(db: DefinitionSkillsDb): DefinitionSkillsStore { - return { - async getSkills(assetId) { - const [row] = await db - .select() - .from(definitionSkills) - .where(eq(definitionSkills.assetId, assetId)) - .limit(1); - return row === undefined ? [] : parseSkills(row.skills); - }, - - async setSkills(assetId, skills) { - const now = new Date(); - await db - .insert(definitionSkills) - .values({ assetId, skills: [...skills], updatedAt: now }) - .onConflictDoUpdate({ - target: definitionSkills.assetId, - set: { skills: [...skills], updatedAt: now }, - }); - }, - }; -} - -/** - * An in-memory `DefinitionSkillsStore`, for tests that want the seam - * without a database. Not a supported deployment target. - */ -export function createInMemoryDefinitionSkillsStore(): DefinitionSkillsStore { - const rows = new Map(); - - return { - async getSkills(assetId) { - return rows.get(assetId) ?? []; - }, - async setSkills(assetId, skills) { - rows.set(assetId, [...skills]); - }, - }; -} diff --git a/packages/agent-directory/src/workflow-capability-routes.ts b/packages/agent-directory/src/workflow-capability-routes.ts index bd2cc748d..41c764516 100644 --- a/packages/agent-directory/src/workflow-capability-routes.ts +++ b/packages/agent-directory/src/workflow-capability-routes.ts @@ -62,7 +62,6 @@ import { type AgentDefinitionDeployer, } from "./definition-asset"; import type { PinnedSkillIndexResolver } from "./routes"; -import type { DefinitionSkillsStore } from "./skills-store"; import { makeErrorEnvelope } from "@corbits/error-sink"; /** @@ -114,7 +113,6 @@ export type CreateWorkflowCapabilityRoutesDeps = { db: DB["db"]; assetService: AssetService; skillIndex: PinnedSkillIndexResolver; - skillsStore: DefinitionSkillsStore; capabilityInventory: CapabilityInventoryProvider; authenticator: WorkflowRunAuthenticator; /** Deploys the definition's commit through the native source pipeline @@ -239,7 +237,6 @@ export function createWorkflowCapabilityRoutes( db: deps.db, assetService: deps.assetService, deployer: deps.deployer, - skillsStore: deps.skillsStore, skillIndex: deps.skillIndex, tenantId: scope.tenantId, principalId: scope.principalId, diff --git a/packages/agent-directory/src/workflow-create-routes.ts b/packages/agent-directory/src/workflow-create-routes.ts index 1b60e95f3..a260967e1 100644 --- a/packages/agent-directory/src/workflow-create-routes.ts +++ b/packages/agent-directory/src/workflow-create-routes.ts @@ -102,7 +102,6 @@ export type CreateWorkflowAgentCreateRoutesDeps = { readonly db: DB["db"]; readonly assetService: AssetService; readonly skillIndex: CreateAgentDefinitionCoreDeps["skillIndex"]; - readonly skillsStore: CreateAgentDefinitionCoreDeps["skillsStore"]; readonly capabilityInventory: CapabilityInventoryProvider; readonly authenticator: WorkflowRunAuthenticator; readonly deployer: CreateAgentDefinitionCoreDeps["deployer"]; @@ -248,7 +247,6 @@ export function createWorkflowAgentCreateRoutes( db: deps.db, assetService: deps.assetService, skillIndex: deps.skillIndex, - skillsStore: deps.skillsStore, deployer: deps.deployer, ...(deps.tenantDefaultModel !== undefined ? { tenantDefaultModel: deps.tenantDefaultModel } diff --git a/packages/agent-directory/src/workflow-skill-pin-routes.ts b/packages/agent-directory/src/workflow-skill-pin-routes.ts index 20a59cca4..3b6196ffc 100644 --- a/packages/agent-directory/src/workflow-skill-pin-routes.ts +++ b/packages/agent-directory/src/workflow-skill-pin-routes.ts @@ -31,7 +31,7 @@ import type { AssetService } from "@intx/hub-sessions"; import { isWorkbenchHostDefinitionName } from "@corbits/chat/workbench-host-naming"; -import { reindexPinnedSkills } from "./agent-workflow"; +import { readPinnedSkillNames, reindexPinnedSkills } from "./agent-workflow"; import { commitLatestAgentAssetSnapshot } from "./asset-write"; import { RetiredWorkflowEnvelopeError, @@ -41,7 +41,6 @@ import { type AgentDefinitionDeployer, } from "./definition-asset"; import type { PinnedSkillIndexResolver } from "./routes"; -import type { DefinitionSkillsStore } from "./skills-store"; import { makeErrorEnvelope } from "@corbits/error-sink"; import type { WorkflowCapabilityRunScope, @@ -90,7 +89,6 @@ export type CreateWorkflowSkillPinRoutesDeps = { db: DB["db"]; assetService: AssetService; skillIndex: PinnedSkillIndexResolver; - skillsStore: DefinitionSkillsStore; authenticator: WorkflowRunAuthenticator; /** Deploys the definition's commit through the native source pipeline * after the rewrite; the composition root injects the SAME @@ -172,7 +170,10 @@ export function createWorkflowSkillPinRoutes( assetId: row.assetId, operation: "pin skill", prepare: async (snapshot) => { - const skills = await deps.skillsStore.getSkills(row.assetId); + // Pins read out of the commit being prepared: the asset's + // stanza is the source of truth, so a concurrent writer's pins + // survive the retry instead of being clobbered by a stale read. + const skills = readPinnedSkillNames(snapshot); const nextSkills = skills.includes(body.skillName) ? skills : [...skills, body.skillName]; @@ -186,7 +187,6 @@ export function createWorkflowSkillPinRoutes( ), ), message: `Pin ${body.skillName} skill to ${row.name}`, - afterWrite: () => deps.skillsStore.setSkills(row.assetId, nextSkills), result: { skills: nextSkills }, }; }, diff --git a/packages/agent-directory/test/migrations.test.ts b/packages/agent-directory/test/migrations.test.ts index ee14bffbf..521121f38 100644 --- a/packages/agent-directory/test/migrations.test.ts +++ b/packages/agent-directory/test/migrations.test.ts @@ -19,7 +19,10 @@ function scratchUrlFor(e2eUrl: string): string { const databaseUrl = e2eDatabaseUrl(); const describeIfDb = dbGate(databaseUrl, import.meta.path); -const migrationNames = ["0001_definition_skills"]; +const migrationNames = [ + "0001_definition_skills", + "0002_drop_definition_skills", +]; describeIfDb("applyAgentDirectoryMigrations", () => { const scratchUrl = scratchUrlFor( @@ -56,7 +59,7 @@ describeIfDb("applyAgentDirectoryMigrations", () => { } }, 20000); - test("applies the table into its own schema and is idempotent on a second run", async () => { + test("drops the store table forward and is idempotent on a second run", async () => { const first = await applyAgentDirectoryMigrations(scratchUrl); expect(first.applied).toEqual(migrationNames); @@ -66,13 +69,13 @@ describeIfDb("applyAgentDirectoryMigrations", () => { const sql = postgres(scratchUrl, { max: 1, onnotice: () => undefined }); try { + // 0002 dropped what 0001 created: the cutover deletes the + // Workbench-owned store forward, never by rewriting history. const tables = await sql.unsafe( `SELECT table_name FROM information_schema.tables ` + `WHERE table_schema = 'agent_directory' AND table_name = 'definition_skills'`, ); - expect(tables.map((row) => String(row["table_name"]))).toEqual([ - "definition_skills", - ]); + expect(tables).toHaveLength(0); const inPublic = await sql.unsafe( `SELECT table_name FROM information_schema.tables ` + diff --git a/packages/agent-directory/test/routes.integration.test.ts b/packages/agent-directory/test/routes.integration.test.ts index 5af223f33..90cc55588 100644 --- a/packages/agent-directory/test/routes.integration.test.ts +++ b/packages/agent-directory/test/routes.integration.test.ts @@ -6,12 +6,13 @@ // `AssetService` never exercises that validator, so it could not have // caught this. A definition created WITH skills used to write // `skills.json` into the asset tree beside its definition, which that -// validator rejected. Pinned skills now live in this package's own -// `agent_directory.definition_skills` table (see -// `../src/skills-store.ts`), so the asset tree only ever carries the -// source codebase `agentDefinitionSourceTree` renders — the one shape -// the validator now accepts, the retired `workflow.json` envelope -// having been refused at the push boundary. +// validator rejected. Pinned skills now live in the definition's own +// entry-module stanza (read back out of the asset, never a side table — +// CL-7592 cut the Workbench-owned `definition_skills` store), so the +// asset tree only ever carries the source codebase +// `agentDefinitionSourceTree` renders — the one shape the validator now +// accepts, the retired `workflow.json` envelope having been refused at +// the push boundary. // // DB-gated: skipped when DATABASE_URL is unset, so a fresh checkout // still runs the unit gates. Run with e.g. @@ -21,13 +22,12 @@ import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import { randomUUID } from "node:crypto"; -import { eq, inArray } from "drizzle-orm"; +import { eq } from "drizzle-orm"; import { Hono } from "hono"; import type { MiddlewareHandler } from "hono"; import { createDB } from "@intx/db"; import { - asset as assetTable, principal as principalTable, tenant as tenantTable, workflowDefinition, @@ -38,10 +38,8 @@ import type { RequireGrant, TenantEnv } from "@intx/hub-api"; import { dbTargetFromUrl } from "../../../scripts/db-setup"; import { applyAgentDirectoryMigrations } from "../src/migrations"; -import { definitionSkills } from "../src/schema"; import { createAgentDefinitionRoutes } from "../src/routes"; import type { PinnedSkillIndexResolver } from "../src/routes"; -import { createDrizzleDefinitionSkillsStore } from "../src/skills-store"; import type { DefinitionAssetHistory } from "../src/definition-history"; import type { CapabilityInventoryProvider } from "../src/capability-inventory"; import { dbGate } from "../../../scripts/e2e/db-gate"; @@ -180,14 +178,12 @@ describeIfDb("agent-directory routes against a real assetService", () => { db, repoStore: agentRepoStore.repoStore, }); - const skillsStore = createDrizzleDefinitionSkillsStore(db); deployer = recordingAgentDefinitionDeployer(db); const routes = createAgentDefinitionRoutes({ db, assetService, skillIndex: fakeSkillIndex, - skillsStore, history: fakeHistory, capabilityInventory: fakeCapabilityInventory, requireGrant: allowAllRequireGrant, @@ -204,16 +200,6 @@ describeIfDb("agent-directory routes against a real assetService", () => { }, 30000); afterAll(async () => { - const assetRows = await db - .select({ id: assetTable.id }) - .from(assetTable) - .where(eq(assetTable.tenantId, TENANT.id)); - const assetIds = assetRows.map((row) => row.id); - if (assetIds.length > 0) { - await db - .delete(definitionSkills) - .where(inArray(definitionSkills.assetId, assetIds)); - } await db.delete(tenantTable).where(eq(tenantTable.id, TENANT.id)); await close(); await rm(dataDir, { recursive: true, force: true }); diff --git a/packages/agent-directory/test/routes.test.ts b/packages/agent-directory/test/routes.test.ts index f04d59123..45e2de706 100644 --- a/packages/agent-directory/test/routes.test.ts +++ b/packages/agent-directory/test/routes.test.ts @@ -22,11 +22,13 @@ import { buildAgentDefinitionWorkflow, serializeAgentDefinitionWorkflow, withAgentToolPackagePin, + readPinnedSkillNames, } 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"; @@ -34,19 +36,18 @@ import { createWorkflowSkillPinRoutes, type WorkflowRunAuthenticator, } from "../src/workflow-skill-pin-routes"; -import { - createInMemoryDefinitionSkillsStore, - type DefinitionSkillsStore, -} from "../src/skills-store"; 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` — pinned skills no longer live in the asset tree - * (see `../src/skills-store.ts`), so a test that needs a definition's - * skills seeds a `DefinitionSkillsStore` directly instead of stubbing a - * second path here. */ + * with `workflowBytes` — pins live in the asset's own stanza, so a test + * that needs a definition's skills stubs the bytes with + * `storedDefinitionBytesWithSkills` instead of seeding side state. */ function readAssetBlobFor( workflowBytes: Uint8Array, ): AssetService["readAssetBlob"] { @@ -369,7 +370,6 @@ function buildApp( requireGrant: RequireGrant = allowAllRequireGrant, history: DefinitionAssetHistory = fakeHistory(), capabilityInventory: CapabilityInventoryProvider = fakeCapabilityInventory, - skillsStore: DefinitionSkillsStore = createInMemoryDefinitionSkillsStore(), deployer: ReturnType< typeof recordingAgentDefinitionDeployer > = recordingAgentDefinitionDeployer(), @@ -378,7 +378,6 @@ function buildApp( db, assetService, skillIndex: fakeSkillIndex, - skillsStore, history, capabilityInventory, requireGrant, @@ -548,9 +547,8 @@ function fakeCreateDb(): DB["db"] { } as unknown as DB["db"]; } -test("a create request with skills writes the definition source tree to the asset and records skills in the skills store", async () => { +test("a create request with skills writes the definition source tree to the asset with the pins indexed in its stanza", async () => { let writtenFiles: Record | undefined; - const skillsStore = createInMemoryDefinitionSkillsStore(); const deployer = recordingAgentDefinitionDeployer(); const app = buildApp( fakeAssetService({ @@ -574,7 +572,6 @@ test("a create request with skills writes the definition source tree to the asse allowAllRequireGrant, fakeHistory(), fakeCapabilityInventory, - skillsStore, deployer, ); const response = await post(app, { @@ -593,7 +590,8 @@ test("a create request with skills writes the definition source tree to the asse expect(deployer.deploys).toHaveLength(1); expect(deployer.deploys[0]?.assetId).toBe("ast_1"); expect(deployer.deploys[0]?.commitSha).toBe("deadbeef"); - expect(await skillsStore.getSkills("ast_1")).toEqual([ + // The stanza the write indexed is the pins — no side table to consult. + expect(readPinnedSkillNames(definitionFrom(writtenFiles))).toEqual([ "web-research", "long-form-write", ]); @@ -601,9 +599,8 @@ test("a create request with skills writes the definition source tree to the asse expect(body.skills).toEqual(["web-research", "long-form-write"]); }); -test("a create request without skills records an empty skills list", async () => { +test("a create request without skills writes a stanza with no pins", async () => { let writtenFiles: Record | undefined; - const skillsStore = createInMemoryDefinitionSkillsStore(); const app = buildApp( fakeAssetService({ createAsset: () => @@ -626,7 +623,6 @@ test("a create request without skills records an empty skills list", async () => allowAllRequireGrant, fakeHistory(), fakeCapabilityInventory, - skillsStore, ); const response = await post(app, { name: "Research Buddy", @@ -635,7 +631,7 @@ test("a create request without skills records an empty skills list", async () => }); expect(response.status).toBe(201); expect(Object.keys(writtenFiles ?? {})).toEqual(SOURCE_TREE_PATHS); - expect(await skillsStore.getSkills("ast_1")).toEqual([]); + expect(readPinnedSkillNames(definitionFrom(writtenFiles))).toEqual([]); }); test("a create request with toolPackagePins pins each named package at its highest published version", async () => { @@ -710,9 +706,11 @@ function fakeSkillsDb( } as unknown as DB["db"]; } -test("GET /skills returns an empty list for a definition with no skills store row", async () => { +test("GET /skills returns an empty list for a definition with no pinned skills", async () => { const app = buildApp( - fakeAssetService(), + fakeAssetService({ + readAssetBlob: readAssetBlobFor(storedDefinitionBytes()), + }), fakeSkillsDb({ id: "def_1", assetId: "ast_1" }), ); const response = await app.request("/skills?ids=def_1"); @@ -723,16 +721,14 @@ test("GET /skills returns an empty list for a definition with no skills store ro expect(body.skills).toEqual({ def_1: [] }); }); -test("GET /skills returns the stored skill list", async () => { - const skillsStore = createInMemoryDefinitionSkillsStore(); - await skillsStore.setSkills("ast_1", ["web-research"]); +test("GET /skills returns the stanza's pinned skills", async () => { const app = buildApp( - fakeAssetService(), + fakeAssetService({ + readAssetBlob: readAssetBlobFor( + storedDefinitionBytesWithSkills("web-research"), + ), + }), fakeSkillsDb({ id: "def_1", assetId: "ast_1" }), - allowAllRequireGrant, - fakeHistory(), - fakeCapabilityInventory, - skillsStore, ); const response = await app.request("/skills?ids=def_1"); const body = (await response.json()) as { @@ -750,9 +746,51 @@ 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; + }; + 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 | undefined; - const skillsStore = createInMemoryDefinitionSkillsStore(); const app = buildApp( fakeAssetService({ readAssetBlob: () => Promise.resolve(storedDefinitionBytes()), @@ -765,14 +803,16 @@ test("PUT /:definitionId/skills replaces the skill set, writing the definition s allowAllRequireGrant, fakeHistory(), fakeCapabilityInventory, - skillsStore, ); const response = await put(app, "/def_1/skills", { skills: ["long-form-write"], }); expect(response.status).toBe(200); expect(Object.keys(writtenFiles ?? {})).toEqual(SOURCE_TREE_PATHS); - expect(await skillsStore.getSkills("ast_1")).toEqual(["long-form-write"]); + // The written tree's stanza is the pins — no side table to consult. + expect(readPinnedSkillNames(definitionFrom(writtenFiles))).toEqual([ + "long-form-write", + ]); const body = (await response.json()) as { skills: readonly string[] }; expect(body.skills).toEqual(["long-form-write"]); }); @@ -1032,7 +1072,6 @@ test("PUT /:definitionId writes the new system prompt in a single source-tree co allowAllRequireGrant, fakeHistory(), fakeCapabilityInventory, - createInMemoryDefinitionSkillsStore(), deployer, ); const response = await put(app, "/def_1", { @@ -1529,7 +1568,6 @@ test("pinning a skill the registry cannot resolve is a 400, not a 500", async () new SkillRegistryError("not_found", 'cannot pin skill "ghost"'), ), }, - skillsStore: createInMemoryDefinitionSkillsStore(), history: fakeHistory(), capabilityInventory: fakeCapabilityInventory, requireGrant: () => async (_c, next) => { @@ -1902,9 +1940,8 @@ test("adding a skill the inventory doesn't offer is a 400, never written", async expect(populateCalled).toBe(false); }); -test("adding a skill merges it additively into the skills store and re-indexes the prompt", async () => { +test("adding a skill merges it additively into the definition stanza and re-indexes the prompt", async () => { let writtenFiles: Record | undefined; - const skillsStore = createInMemoryDefinitionSkillsStore(); const app = buildApp( fakeAssetService({ readAssetBlob: readAssetBlobFor(storedDefinitionBytes()), @@ -1921,7 +1958,6 @@ test("adding a skill merges it additively into the skills store and re-indexes t allowAllRequireGrant, fakeHistory(), fakeCapabilityInventory, - skillsStore, ); const response = await postTo(app, "/def_1/capabilities", { kind: "skill", @@ -1929,7 +1965,11 @@ test("adding a skill merges it additively into the skills store and re-indexes t }); expect(response.status).toBe(200); expect(Object.keys(writtenFiles ?? {})).toEqual(SOURCE_TREE_PATHS); - expect(await skillsStore.getSkills("ast_1")).toEqual(["research"]); + // Pins live in the definition's own workflow stanza now — no store to + // consult, the written source tree is the assertion. + expect(readPinnedSkillNames(definitionFrom(writtenFiles))).toEqual([ + "research", + ]); expect(definitionFrom(writtenFiles).includes("research")).toBe(true); const body = (await response.json()) as { skills: string[] }; expect(body.skills).toEqual(["research"]); @@ -2100,7 +2140,6 @@ test("concurrent PUT instructions and DELETE model both land", async () => { }); test("concurrent PUT skills and DELETE model both land", async () => { - const skillsStore = createInMemoryDefinitionSkillsStore(); const assetService = liveDefinitionAsset( storedDefinitionBytesWithModel("anthropic/claude-sonnet"), ); @@ -2114,7 +2153,6 @@ test("concurrent PUT skills and DELETE model both land", async () => { allowAllRequireGrant, fakeHistory(), fakeCapabilityInventory, - skillsStore, ); const [skillsRes, delRes] = await Promise.all([ @@ -2128,7 +2166,7 @@ test("concurrent PUT skills and DELETE model both land", async () => { assetService, "ast_1", ); - expect(await skillsStore.getSkills("ast_1")).toEqual(["long-form-write"]); + expect(readPinnedSkillNames(workflowJson)).toEqual(["long-form-write"]); expect(promptFrom(workflowJson)).toContain( "- long-form-write: What long-form-write does.", ); @@ -2171,7 +2209,6 @@ test("concurrent DELETE model and a capability-add both land", async () => { }); test("concurrent pin_skill and DELETE model both land", async () => { - const skillsStore = createInMemoryDefinitionSkillsStore(); const assetService = liveDefinitionAsset( storedDefinitionBytesWithModel("anthropic/claude-sonnet"), ); @@ -2186,7 +2223,6 @@ test("concurrent pin_skill and DELETE model both land", async () => { allowAllRequireGrant, fakeHistory(), fakeCapabilityInventory, - skillsStore, ); const pinAuthenticator: WorkflowRunAuthenticator = { resolve: (token, address) => @@ -2204,7 +2240,6 @@ test("concurrent pin_skill and DELETE model both land", async () => { db, assetService, skillIndex: fakeSkillIndex, - skillsStore, authenticator: pinAuthenticator, deployer: recordingAgentDefinitionDeployer(), }); @@ -2231,7 +2266,7 @@ test("concurrent pin_skill and DELETE model both land", async () => { assetService, "ast_1", ); - expect(await skillsStore.getSkills("ast_1")).toEqual(["research"]); + expect(readPinnedSkillNames(workflowJson)).toEqual(["research"]); expect(promptFrom(workflowJson)).toContain("- research: What research does."); expect(modelFrom(workflowJson)).toBeUndefined(); }); diff --git a/packages/agent-directory/test/skills-store.drizzle.test.ts b/packages/agent-directory/test/skills-store.drizzle.test.ts deleted file mode 100644 index 5cb1d5243..000000000 --- a/packages/agent-directory/test/skills-store.drizzle.test.ts +++ /dev/null @@ -1,95 +0,0 @@ -// DB-gated: skipped when no DATABASE_URL is reachable (a fresh checkout -// still runs the unit gates), mirroring @corbits/config-profiles' own -// store.drizzle.test.ts. Exercises the real -// `createDrizzleDefinitionSkillsStore` path against Postgres. -import { afterAll, beforeAll, expect, test } from "bun:test"; -import { drizzle } from "drizzle-orm/postgres-js"; -import postgres from "postgres"; - -import { e2eDatabaseUrl } from "../../../scripts/e2e/database-url"; -import { applyAgentDirectoryMigrations } from "../src/migrations"; -import { createDrizzleDefinitionSkillsStore } from "../src/skills-store"; -import { dbGate } from "../../../scripts/e2e/db-gate"; - -function scratchUrlFor(e2eUrl: string): string { - const url = new URL(e2eUrl); - const database = url.pathname.replace(/^\//, ""); - url.pathname = `/${database}_agent_directory_skills_store_drizzle_test`; - return url.toString(); -} - -const databaseUrl = e2eDatabaseUrl(); -const describeIfDb = dbGate(databaseUrl, import.meta.path); - -describeIfDb("createDrizzleDefinitionSkillsStore", () => { - const scratchUrl = scratchUrlFor( - databaseUrl ?? "postgres://localhost:5432/unused", - ); - const scratchDatabase = new URL(scratchUrl).pathname.replace(/^\//, ""); - - beforeAll(async () => { - const maintenanceUrl = new URL(scratchUrl); - maintenanceUrl.pathname = "/postgres"; - const maintenance = postgres(maintenanceUrl.toString(), { - max: 1, - onnotice: () => undefined, - }); - try { - await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); - await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`); - } finally { - await maintenance.end(); - } - await applyAgentDirectoryMigrations(scratchUrl); - }, 20000); - - afterAll(async () => { - const maintenanceUrl = new URL(scratchUrl); - maintenanceUrl.pathname = "/postgres"; - const maintenance = postgres(maintenanceUrl.toString(), { - max: 1, - onnotice: () => undefined, - }); - try { - await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); - } finally { - await maintenance.end(); - } - }, 20000); - - test("a definition with no row yet reads as no skills attached", async () => { - const sql = postgres(scratchUrl, { max: 1, onnotice: () => undefined }); - try { - const store = createDrizzleDefinitionSkillsStore(drizzle(sql)); - expect(await store.getSkills("asset_never_written")).toEqual([]); - } finally { - await sql.end(); - } - }); - - test("setSkills then getSkills round-trips through real Postgres", async () => { - const sql = postgres(scratchUrl, { max: 1, onnotice: () => undefined }); - try { - const store = createDrizzleDefinitionSkillsStore(drizzle(sql)); - await store.setSkills("asset_1", ["web-research", "long-form-write"]); - expect(await store.getSkills("asset_1")).toEqual([ - "web-research", - "long-form-write", - ]); - } finally { - await sql.end(); - } - }); - - test("setSkills on an existing asset id upserts rather than duplicating", async () => { - const sql = postgres(scratchUrl, { max: 1, onnotice: () => undefined }); - try { - const store = createDrizzleDefinitionSkillsStore(drizzle(sql)); - await store.setSkills("asset_2", ["research"]); - await store.setSkills("asset_2", []); - expect(await store.getSkills("asset_2")).toEqual([]); - } finally { - await sql.end(); - } - }); -}); diff --git a/packages/agent-directory/test/source-tree.ts b/packages/agent-directory/test/source-tree.ts index 2f1fd8b17..b475d82a5 100644 --- a/packages/agent-directory/test/source-tree.ts +++ b/packages/agent-directory/test/source-tree.ts @@ -5,8 +5,14 @@ import { AGENT_DEFINITION_ENTRY_PATH, + agentDefinitionSourceTree, parseAgentDefinitionEntry, } from "../src/definition-asset"; +import { + buildAgentDefinitionWorkflow, + reindexPinnedSkills, + serializeAgentDefinitionWorkflow, +} from "../src/agent-workflow"; /** The two files a definition's asset tree carries, in render order. */ export const SOURCE_TREE_PATHS = ["package.json", AGENT_DEFINITION_ENTRY_PATH]; @@ -21,3 +27,26 @@ export function definitionFrom( } return parseAgentDefinitionEntry(new TextEncoder().encode(entry), "ast_1"); } + +/** 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. */ +export function storedDefinitionBytesWithSkills( + ...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]); +} diff --git a/packages/agent-directory/test/workflow-capability-routes.test.ts b/packages/agent-directory/test/workflow-capability-routes.test.ts index a11f63d3a..fd736871d 100644 --- a/packages/agent-directory/test/workflow-capability-routes.test.ts +++ b/packages/agent-directory/test/workflow-capability-routes.test.ts @@ -14,6 +14,7 @@ import type { DB } from "@intx/db"; import { buildAgentDefinitionWorkflow, serializeAgentDefinitionWorkflow, + readPinnedSkillNames, } from "../src/agent-workflow"; import { createWorkflowCapabilityRoutes, @@ -25,11 +26,7 @@ import { AGENT_DEFINITION_ENTRY_PATH, } from "../src/definition-asset"; import type { PinnedSkillIndexResolver } from "../src/routes"; -import { - createInMemoryDefinitionSkillsStore, - type DefinitionSkillsStore, -} from "../src/skills-store"; -import { SOURCE_TREE_PATHS } from "./source-tree"; +import { definitionFrom, SOURCE_TREE_PATHS } from "./source-tree"; import type { CapabilityInventoryProvider } from "../src/capability-inventory"; import { CORBITS_TOOLS_REGISTRY } from "@corbits/tool-registry-publish"; @@ -72,9 +69,9 @@ function storedDefinitionBytes(): Uint8Array { } /** A `readAssetBlob` that always answers the definition's entry module - * — pinned skills no longer live in the asset tree, so a test that needs - * a definition's skills seeds a `DefinitionSkillsStore` directly - * instead. */ + * with `workflowBytes` — pins live in the asset's own stanza, so a test + * that needs a definition's skills reads them back out of the written + * tree instead of seeding side state. */ function readAssetBlobFor( workflowBytes: Uint8Array, ): AssetService["readAssetBlob"] { @@ -184,14 +181,12 @@ function buildApp(opts: { db?: DB["db"]; authenticator?: WorkflowRunAuthenticator; capabilityInventory?: CapabilityInventoryProvider; - skillsStore?: DefinitionSkillsStore; deployer?: ReturnType; }): Hono { return createWorkflowCapabilityRoutes({ db: opts.db ?? fakeDb(), assetService: opts.assetService ?? fakeAssetService(), skillIndex: fakeSkillIndex, - skillsStore: opts.skillsStore ?? createInMemoryDefinitionSkillsStore(), capabilityInventory: opts.capabilityInventory ?? fakeCapabilityInventory, authenticator: opts.authenticator ?? authenticateAsOwnRun, deployer: opts.deployer ?? recordingAgentDefinitionDeployer(), @@ -317,9 +312,8 @@ test("adding a capability the tenant's inventory doesn't offer is a 400, never w expect(populateCalled).toBe(false); }); -test("adding a skill merges it additively into the skills store and re-indexes the prompt", async () => { +test("adding a skill merges it additively into the definition stanza and re-indexes the prompt", async () => { let writtenFiles: Record | undefined; - const skillsStore = createInMemoryDefinitionSkillsStore(); const app = buildApp({ assetService: fakeAssetService({ readAssetBlob: readAssetBlobFor(storedDefinitionBytes()), @@ -328,7 +322,6 @@ test("adding a skill merges it additively into the skills store and re-indexes t return Promise.resolve({ commitSha: "deadbeef" }); }, }), - skillsStore, }); const response = await postCapability(app, OWN_DEFINITION_ID, { kind: "skill", @@ -336,7 +329,10 @@ test("adding a skill merges it additively into the skills store and re-indexes t }); expect(response.status).toBe(200); expect(Object.keys(writtenFiles ?? {})).toEqual(SOURCE_TREE_PATHS); - expect(await skillsStore.getSkills("ast_1")).toEqual(["research"]); + // The written tree's stanza is the pins — no side table to consult. + expect(readPinnedSkillNames(definitionFrom(writtenFiles))).toEqual([ + "research", + ]); const body = (await response.json()) as { skills: string[] }; expect(body.skills).toEqual(["research"]); }); diff --git a/packages/agent-directory/test/workflow-create-routes.test.ts b/packages/agent-directory/test/workflow-create-routes.test.ts index f19ab18d4..4947fa7b7 100644 --- a/packages/agent-directory/test/workflow-create-routes.test.ts +++ b/packages/agent-directory/test/workflow-create-routes.test.ts @@ -19,7 +19,6 @@ import type { WorkflowRunAuthenticator, } from "../src/workflow-capability-routes"; import type { PinnedSkillIndexResolver } from "../src/routes"; -import { createInMemoryDefinitionSkillsStore } from "../src/skills-store"; import type { CapabilityInventoryProvider } from "../src/capability-inventory"; import { CORBITS_TOOLS_REGISTRY } from "@corbits/tool-registry-publish"; import { definitionFrom, SOURCE_TREE_PATHS } from "./source-tree"; @@ -197,7 +196,6 @@ function buildApp( db: opts.db ?? fakeDb(), assetService: opts.assetService ?? fakeAssetService(), skillIndex: opts.skillIndex ?? fakeSkillIndex, - skillsStore: opts.skillsStore ?? createInMemoryDefinitionSkillsStore(), capabilityInventory: opts.capabilityInventory ?? fakeCapabilityInventory, authenticator: opts.authenticator ?? authenticateAsRun, deployer: opts.deployer ?? recordingAgentDefinitionDeployer(), diff --git a/packages/agent-directory/test/workflow-skill-pin-routes.test.ts b/packages/agent-directory/test/workflow-skill-pin-routes.test.ts index 96ab0aa50..9da542583 100644 --- a/packages/agent-directory/test/workflow-skill-pin-routes.test.ts +++ b/packages/agent-directory/test/workflow-skill-pin-routes.test.ts @@ -13,6 +13,7 @@ import type { DB } from "@intx/db"; import { buildAgentDefinitionWorkflow, serializeAgentDefinitionWorkflow, + readPinnedSkillNames, } from "../src/agent-workflow"; import { createWorkflowSkillPinRoutes, @@ -25,10 +26,10 @@ import { } from "../src/definition-asset"; import type { PinnedSkillIndexResolver } from "../src/routes"; import { - createInMemoryDefinitionSkillsStore, - type DefinitionSkillsStore, -} from "../src/skills-store"; -import { SOURCE_TREE_PATHS } from "./source-tree"; + definitionFrom, + SOURCE_TREE_PATHS, + storedDefinitionBytesWithSkills, +} from "./source-tree"; const TENANT_ID = "tnt_1"; const OTHER_TENANT_ID = "tnt_2"; @@ -174,14 +175,12 @@ function buildApp(opts: { assetService?: AssetService; db?: DB["db"]; authenticator?: WorkflowRunAuthenticator; - skillsStore?: DefinitionSkillsStore; deployer?: ReturnType; }): Hono { return createWorkflowSkillPinRoutes({ db: opts.db ?? fakeDbWithRows([]), assetService: opts.assetService ?? fakeAssetService(), skillIndex: fakeSkillIndex, - skillsStore: opts.skillsStore ?? createInMemoryDefinitionSkillsStore(), authenticator: opts.authenticator ?? authenticateAsTenant1, deployer: opts.deployer ?? recordingAgentDefinitionDeployer(), }) as unknown as Hono; @@ -270,7 +269,6 @@ test("pins a skill onto another definition in the same tenant and re-indexes its lookupTenant = TENANT_ID; let writtenFiles: Record | undefined; let writtenMessage: string | undefined; - const skillsStore = createInMemoryDefinitionSkillsStore(); const deployer = recordingAgentDefinitionDeployer(); const app = buildApp({ db: fakeDbWithRows([ @@ -288,7 +286,6 @@ test("pins a skill onto another definition in the same tenant and re-indexes its return Promise.resolve({ commitSha: "deadbeef" }); }, }), - skillsStore, deployer, }); const response = await postPin(app, { @@ -303,7 +300,10 @@ test("pins a skill onto another definition in the same tenant and re-indexes its expect(deployer.deploys[0]?.commitSha).toBe("deadbeef"); expect(Object.keys(writtenFiles ?? {})).toEqual(SOURCE_TREE_PATHS); expect(writtenMessage).toBe("Pin research skill to research-buddy"); - expect(await skillsStore.getSkills("ast_1")).toEqual(["research"]); + // The written tree's stanza is the pins — no side table to consult. + expect(readPinnedSkillNames(definitionFrom(writtenFiles))).toEqual([ + "research", + ]); const body = (await response.json()) as { skills: string[] }; expect(body.skills).toEqual(["research"]); }); @@ -346,8 +346,6 @@ test("a definition still on the retired envelope is a 409, never a 500, and writ test("pinning the same skill twice is idempotent, never duplicated", async () => { lookupId = TARGET_DEFINITION_ID; lookupTenant = TENANT_ID; - const skillsStore = createInMemoryDefinitionSkillsStore(); - await skillsStore.setSkills("ast_1", ["research"]); const app = buildApp({ db: fakeDbWithRows([ { @@ -357,7 +355,11 @@ test("pinning the same skill twice is idempotent, never duplicated", async () => name: "research-buddy", }, ]), - skillsStore, + assetService: fakeAssetService({ + readAssetBlob: readAssetBlobFor( + storedDefinitionBytesWithSkills("research"), + ), + }), }); const response = await postPin(app, { definitionId: TARGET_DEFINITION_ID, @@ -366,7 +368,6 @@ test("pinning the same skill twice is idempotent, never duplicated", async () => expect(response.status).toBe(200); const body = (await response.json()) as { skills: string[] }; expect(body.skills).toEqual(["research"]); - expect(await skillsStore.getSkills("ast_1")).toEqual(["research"]); }); test("a malformed body is a 400", async () => { diff --git a/scripts/checks/no-product-tenancy.ts b/scripts/checks/no-product-tenancy.ts index 0866b84df..a29adcdf5 100644 --- a/scripts/checks/no-product-tenancy.ts +++ b/scripts/checks/no-product-tenancy.ts @@ -164,14 +164,6 @@ const ALLOWLIST: readonly { maxOccurrences: 1, tables: ["user_preferences"], }, - { - // Skills pinned to a workbench's agent definition (CL-6135): the - // workflow-kind asset tree forbids skills.json, so this pin list is - // package-owned state beside the native definition row. - relPath: "packages/agent-directory/src/schema.ts", - maxOccurrences: 1, - tables: ["agent_directory.definition_skills"], - }, { // Eval-run history (CL-6143): one row per (eval, config) scored // run, product-owned scoring data, never tenancy.