From b5cae2c3d925abdb019adb6502bdde0c3473c27a Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 00:55:41 -0700 Subject: [PATCH 1/6] Add tests for concurrent capability adds (CL-7216) Two overlapping POSTs against the same definition must both land their pins. The suite uses a live in-memory asset so a last-write-wins clobber is observable. --- packages/agent-directory/test/routes.test.ts | 72 ++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/packages/agent-directory/test/routes.test.ts b/packages/agent-directory/test/routes.test.ts index 293ec5b00..1d78c5a61 100644 --- a/packages/agent-directory/test/routes.test.ts +++ b/packages/agent-directory/test/routes.test.ts @@ -24,6 +24,7 @@ import { import { agentDefinitionSourceTree, AGENT_DEFINITION_ENTRY_PATH, + readAgentDefinitionWorkflowJson, } from "../src/definition-asset"; import { createAgentDefinitionRoutes } from "../src/routes"; import type { PinnedSkillIndexResolver } from "../src/routes"; @@ -111,6 +112,29 @@ function fakeAssetService(overrides: Partial = {}): AssetService { }; } +/** An asset whose reads observe writes — the race two concurrent + * capability-adds hit is only visible when the second read can see the + * first write, or when both reads share a snapshot the later write then + * clobbers. */ +function liveDefinitionAsset(initial: Uint8Array): AssetService { + let current = initial; + return fakeAssetService({ + readAssetBlob: async () => { + await Promise.resolve(); + return current; + }, + populateAsset: async (params) => { + const entry = params.tree.files[AGENT_DEFINITION_ENTRY_PATH]; + if (typeof entry !== "string") { + throw new Error("populateAsset wrote no entry module"); + } + await Promise.resolve(); + current = new TextEncoder().encode(entry); + return { commitSha: "deadbeef" }; + }, + }); +} + /** The entry module a stored definition's asset carries, so the PUT * path has something real to re-index. */ function storedDefinitionBytes( @@ -1806,3 +1830,51 @@ test("capabilities route 404s for an unknown definition", async () => { }); expect(response.status).toBe(404); }); + +test("two concurrent capability-adds on the same definition both land", async () => { + const inventory: CapabilityInventoryProvider = { + resolve: () => + Promise.resolve({ + toolPackages: [ + { name: "@corbits/github-tools" }, + { name: "@corbits/memory-tools" }, + ], + skills: [{ name: "research" }], + models: [{ canonicalName: "anthropic/claude-sonnet" }], + }), + }; + const assetService = liveDefinitionAsset(storedDefinitionBytes()); + const app = buildApp( + assetService, + fakeInstructionsDb({ + id: "def_1", + assetId: "ast_1", + name: "research-buddy", + }), + allowAllRequireGrant, + fakeHistory(), + inventory, + ); + + const [first, second] = await Promise.all([ + postTo(app, "/def_1/capabilities", { + kind: "toolPackage", + name: "@corbits/github-tools", + }), + postTo(app, "/def_1/capabilities", { + kind: "toolPackage", + name: "@corbits/memory-tools", + }), + ]); + expect(first.status).toBe(200); + expect(second.status).toBe(200); + + const workflowJson = await readAgentDefinitionWorkflowJson( + assetService, + "ast_1", + ); + const names = pinsFrom(workflowJson) + .map((pin) => pin.name) + .toSorted(); + expect(names).toEqual(["@corbits/github-tools", "@corbits/memory-tools"]); +}); From 2a17a231d2af2075309aa23bf85bd318200216b0 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 00:55:56 -0700 Subject: [PATCH 2/6] Retry concurrent capability adds on the latest snapshot (CL-7216) Capability-add was a lock-free read-modify-write of the definition asset, so two overlapping POSTs could each snapshot the same tree and the later write would drop the earlier pin. The add now re-reads under a per-asset lock and reapplies the losing edit on the winner tree instead of last-write-wins. --- .../agent-directory/src/capability-add.ts | 183 ++++++++++++++++++ packages/agent-directory/src/routes.ts | 59 +----- .../src/workflow-capability-routes.ts | 69 +------ 3 files changed, 195 insertions(+), 116 deletions(-) create mode 100644 packages/agent-directory/src/capability-add.ts diff --git a/packages/agent-directory/src/capability-add.ts b/packages/agent-directory/src/capability-add.ts new file mode 100644 index 000000000..4a1f0062f --- /dev/null +++ b/packages/agent-directory/src/capability-add.ts @@ -0,0 +1,183 @@ +// Capability-add is a read-modify-write of the definition's asset: two +// concurrent POSTs that both snapshot the same workflow.json and then +// each write their own pin would last-write-wins clobber the other +// (CL-7216). This module owns that RMW so both the tenant-session route +// and the workflow-run route retry the loser against the latest snapshot +// instead of silently dropping an add. + +import type { PinnedSkillIndexEntry } from "@corbits/skills"; +import type { AssetService } from "@intx/hub-sessions"; + +import { + readAgentCapabilities, + reindexPinnedSkills, + withAgentModel, + withAgentToolPackagePin, +} from "./agent-workflow"; +import type { AddCapabilityInput } from "./capability-inventory"; +import { + readAgentDefinitionWorkflowJson, + writeAndDeployAgentDefinition, + type AgentDefinitionDeployer, +} from "./definition-asset"; +import type { DefinitionSkillsStore } from "./skills-store"; + +const MAX_STALE_SNAPSHOT_RETRIES = 8; + +const writeChains = new Map>(); + +async function withAssetWriteLock( + assetId: string, + fn: () => Promise, +): Promise { + const previous = writeChains.get(assetId) ?? Promise.resolve(); + let release: () => void = () => undefined; + const held = new Promise((resolve) => { + release = resolve; + }); + writeChains.set( + assetId, + previous.then(() => held), + ); + try { + await previous; + return await fn(); + } finally { + release(); + } +} + +export type CommitAgentCapabilityAddArgs = { + assetService: AssetService; + deployer: AgentDefinitionDeployer; + skillsStore: DefinitionSkillsStore; + skillIndex: { + resolve( + tenantId: string, + principalId: string, + names: readonly string[], + ): Promise; + }; + tenantId: string; + principalId: string; + assetId: string; + handle: string; + body: AddCapabilityInput; +}; + +export type CommitAgentCapabilityAddResult = { + toolPackagePins: ReturnType["toolPackagePins"]; + skills: readonly string[]; + model?: string; +}; + +/** + * Applies one capability add against the definition's current asset, + * retrying when a concurrent writer moved the snapshot between this + * call's read and its write. The per-asset lock makes that stale check + * atomic so the loser reapplies on the winner's tree instead of + * clobbering it. + */ +export async function commitAgentCapabilityAdd( + args: CommitAgentCapabilityAddArgs, +): Promise { + for ( + let remaining = MAX_STALE_SNAPSHOT_RETRIES; + remaining > 0; + remaining -= 1 + ) { + const snapshot = await readAgentDefinitionWorkflowJson( + args.assetService, + args.assetId, + ); + const prepared = await prepareCapabilityAdd(snapshot, args); + const wrote = await withAssetWriteLock(args.assetId, async () => { + const latest = await readAgentDefinitionWorkflowJson( + args.assetService, + args.assetId, + ); + if (latest !== snapshot) return false; + await writeAndDeployAgentDefinition({ + assetService: args.assetService, + deployer: args.deployer, + tenantId: args.tenantId, + principalId: args.principalId, + assetId: args.assetId, + handle: args.handle, + workflowJson: prepared.workflowJson, + message: prepared.message, + }); + if (prepared.nextSkills !== null) { + await args.skillsStore.setSkills(args.assetId, prepared.nextSkills); + } + return true; + }); + if (wrote) return prepared.result; + } + throw new Error( + `capability add for asset ${args.assetId} conflicted after ${String(MAX_STALE_SNAPSHOT_RETRIES)} retries`, + ); +} + +type PreparedCapabilityAdd = { + workflowJson: string; + message: string; + nextSkills: readonly string[] | null; + result: CommitAgentCapabilityAddResult; +}; + +async function prepareCapabilityAdd( + workflowJson: string, + args: CommitAgentCapabilityAddArgs, +): Promise { + let nextWorkflowJson: string; + let message: string; + let skills = await args.skillsStore.getSkills(args.assetId); + let nextSkills: readonly string[] | null = null; + + switch (args.body.kind) { + case "toolPackage": { + nextWorkflowJson = withAgentToolPackagePin(workflowJson, { + name: args.body.name, + version: "*", + }); + message = `Add ${args.body.name} to ${args.handle}`; + break; + } + case "skill": { + nextSkills = skills.includes(args.body.name) + ? skills + : [...skills, args.body.name]; + nextWorkflowJson = reindexPinnedSkills( + workflowJson, + await args.skillIndex.resolve( + args.tenantId, + args.principalId, + nextSkills, + ), + ); + skills = nextSkills; + message = `Add ${args.body.name} skill to ${args.handle}`; + break; + } + case "model": { + nextWorkflowJson = withAgentModel(workflowJson, args.body.canonicalName); + message = `Set ${args.handle}'s model to ${args.body.canonicalName}`; + break; + } + } + + const capabilities = readAgentCapabilities(nextWorkflowJson); + return { + workflowJson: nextWorkflowJson, + message, + nextSkills, + result: { + toolPackagePins: capabilities.toolPackagePins, + skills, + ...(capabilities.model !== undefined + ? { model: capabilities.model } + : {}), + }, + }; +} diff --git a/packages/agent-directory/src/routes.ts b/packages/agent-directory/src/routes.ts index 9ce6e74fb..4e1250d2c 100644 --- a/packages/agent-directory/src/routes.ts +++ b/packages/agent-directory/src/routes.ts @@ -37,13 +37,12 @@ import { readAgentCapabilities, readAgentSystemPrompt, reindexPinnedSkills, - withAgentModel, withAgentSystemPrompt, - withAgentToolPackagePin, withoutAgentModel, type CreateAgentDefinitionCoreDeps, type CreateAgentDefinitionCoreInput, } from "./agent-workflow"; +import { commitAgentCapabilityAdd } from "./capability-add"; import { agentDefinitionSourceTree, AGENT_DEFINITION_ENTRY_PATH, @@ -542,64 +541,18 @@ export function createAgentDefinitionRoutes({ // fetched, never a stale or wider one. assertCapabilityInInventory(body, inventory); - const workflowJson = await readAgentDefinitionWorkflowJson( - assetService, - row.assetId, - ); - - let nextWorkflowJson: string; - let message: string; - let skills = await skillsStore.getSkills(row.assetId); - let nextSkills: readonly string[] | null = null; - - switch (body.kind) { - case "toolPackage": { - nextWorkflowJson = withAgentToolPackagePin(workflowJson, { - name: body.name, - version: "*", - }); - message = `Add ${body.name} to ${row.name}`; - break; - } - case "skill": { - nextSkills = skills.includes(body.name) - ? skills - : [...skills, body.name]; - nextWorkflowJson = reindexPinnedSkills( - workflowJson, - await skillIndex.resolve(tenant.id, principal.id, nextSkills), - ); - skills = nextSkills; - message = `Add ${body.name} skill to ${row.name}`; - break; - } - case "model": { - nextWorkflowJson = withAgentModel(workflowJson, body.canonicalName); - message = `Set ${row.name}'s model to ${body.canonicalName}`; - break; - } - } - - await writeAndDeployAgentDefinition({ + const added = await commitAgentCapabilityAdd({ assetService, deployer, + skillsStore, + skillIndex, tenantId: tenant.id, principalId: principal.id, assetId: row.assetId, handle: row.name, - workflowJson: nextWorkflowJson, - message, - }); - if (nextSkills !== null) { - await skillsStore.setSkills(row.assetId, nextSkills); - } - - const capabilities = readAgentCapabilities(nextWorkflowJson); - return c.json({ - toolPackagePins: capabilities.toolPackagePins, - skills, - model: capabilities.model, + body, }); + return c.json(added); }, ); diff --git a/packages/agent-directory/src/workflow-capability-routes.ts b/packages/agent-directory/src/workflow-capability-routes.ts index 70f8aecfa..49506b6e1 100644 --- a/packages/agent-directory/src/workflow-capability-routes.ts +++ b/packages/agent-directory/src/workflow-capability-routes.ts @@ -48,12 +48,7 @@ import type { AssetService } from "@intx/hub-sessions"; import { isWorkbenchHostDefinitionName } from "@corbits/chat/workbench-host-naming"; -import { - reindexPinnedSkills, - withAgentModel, - withAgentToolPackagePin, - readAgentCapabilities, -} from "./agent-workflow"; +import { commitAgentCapabilityAdd } from "./capability-add"; import { AddCapabilityInput, assertCapabilityInInventory, @@ -61,10 +56,8 @@ import { type CapabilityInventoryProvider, } from "./capability-inventory"; import { - readAgentDefinitionWorkflowJson, RetiredWorkflowEnvelopeError, statusForAgentDefinitionDeployError, - writeAndDeployAgentDefinition, WorkflowAuthorError, type AgentDefinitionDeployer, } from "./definition-asset"; @@ -242,68 +235,18 @@ export function createWorkflowCapabilityRoutes( // tenant-session route's own fail-closed check. assertCapabilityInInventory(body, inventory); - const workflowJson = await readAgentDefinitionWorkflowJson( - deps.assetService, - row.assetId, - ); - - let nextWorkflowJson: string; - let message: string; - let skills = await deps.skillsStore.getSkills(row.assetId); - let nextSkills: readonly string[] | null = null; - - switch (body.kind) { - case "toolPackage": { - nextWorkflowJson = withAgentToolPackagePin(workflowJson, { - name: body.name, - version: "*", - }); - message = `Add ${body.name} to ${row.name}`; - break; - } - case "skill": { - nextSkills = skills.includes(body.name) - ? skills - : [...skills, body.name]; - nextWorkflowJson = reindexPinnedSkills( - workflowJson, - await deps.skillIndex.resolve( - scope.tenantId, - scope.principalId, - nextSkills, - ), - ); - skills = nextSkills; - message = `Add ${body.name} skill to ${row.name}`; - break; - } - case "model": { - nextWorkflowJson = withAgentModel(workflowJson, body.canonicalName); - message = `Set ${row.name}'s model to ${body.canonicalName}`; - break; - } - } - - await writeAndDeployAgentDefinition({ + const added = await commitAgentCapabilityAdd({ assetService: deps.assetService, deployer: deps.deployer, + skillsStore: deps.skillsStore, + skillIndex: deps.skillIndex, tenantId: scope.tenantId, principalId: scope.principalId, assetId: row.assetId, handle: row.name, - workflowJson: nextWorkflowJson, - message, - }); - if (nextSkills !== null) { - await deps.skillsStore.setSkills(row.assetId, nextSkills); - } - - const capabilities = readAgentCapabilities(nextWorkflowJson); - return c.json({ - toolPackagePins: capabilities.toolPackagePins, - skills, - model: capabilities.model, + body, }); + return c.json(added); }); return app; From b0eb1cdac20886a09aca46b3ef3078c0eb3cf54e Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 15:02:36 -0700 Subject: [PATCH 3/6] Add tests for concurrent sibling definition writes (CL-7216) PUT instructions, PUT skills, and DELETE model still last-write-wins clobber a concurrent edit of the same asset. The suite races those routes so a lost prompt, skill pin, or model-clear is observable. --- packages/agent-directory/test/routes.test.ts | 100 +++++++++++++++++++ 1 file changed, 100 insertions(+) diff --git a/packages/agent-directory/test/routes.test.ts b/packages/agent-directory/test/routes.test.ts index 1d78c5a61..408dc70b9 100644 --- a/packages/agent-directory/test/routes.test.ts +++ b/packages/agent-directory/test/routes.test.ts @@ -1878,3 +1878,103 @@ test("two concurrent capability-adds on the same definition both land", async () .toSorted(); expect(names).toEqual(["@corbits/github-tools", "@corbits/memory-tools"]); }); + +test("concurrent PUT instructions and DELETE model both land", async () => { + const assetService = liveDefinitionAsset( + storedDefinitionBytesWithModel("anthropic/claude-sonnet"), + ); + const app = buildApp( + assetService, + fakeInstructionsDb({ + id: "def_1", + assetId: "ast_1", + name: "research-buddy", + }), + ); + + const [putRes, delRes] = await Promise.all([ + put(app, "/def_1", { + name: "Research Buddy", + systemPrompt: "You are now a blunt researcher.", + }), + app.request("/def_1/capabilities/model", { method: "DELETE" }), + ]); + expect(putRes.status).toBe(200); + expect(delRes.status).toBe(200); + + const workflowJson = await readAgentDefinitionWorkflowJson( + assetService, + "ast_1", + ); + expect(promptFrom(workflowJson)).toBe("You are now a blunt researcher."); + expect(modelFrom(workflowJson)).toBeUndefined(); +}); + +test("concurrent PUT skills and DELETE model both land", async () => { + const skillsStore = createInMemoryDefinitionSkillsStore(); + const assetService = liveDefinitionAsset( + storedDefinitionBytesWithModel("anthropic/claude-sonnet"), + ); + const app = buildApp( + assetService, + fakeInstructionsDb({ + id: "def_1", + assetId: "ast_1", + name: "research-buddy", + }), + allowAllRequireGrant, + fakeHistory(), + fakeCapabilityInventory, + skillsStore, + ); + + const [skillsRes, delRes] = await Promise.all([ + put(app, "/def_1/skills", { skills: ["long-form-write"] }), + app.request("/def_1/capabilities/model", { method: "DELETE" }), + ]); + expect(skillsRes.status).toBe(200); + expect(delRes.status).toBe(200); + + const workflowJson = await readAgentDefinitionWorkflowJson( + assetService, + "ast_1", + ); + expect(await skillsStore.getSkills("ast_1")).toEqual(["long-form-write"]); + expect(promptFrom(workflowJson)).toContain( + "- long-form-write: What long-form-write does.", + ); + expect(modelFrom(workflowJson)).toBeUndefined(); +}); + +test("concurrent DELETE model and a capability-add both land", async () => { + const assetService = liveDefinitionAsset( + storedDefinitionBytesWithModel("anthropic/claude-sonnet"), + ); + const app = buildApp( + assetService, + fakeInstructionsDb({ + id: "def_1", + assetId: "ast_1", + name: "research-buddy", + }), + ); + + const [delRes, capRes] = await Promise.all([ + app.request("/def_1/capabilities/model", { method: "DELETE" }), + postTo(app, "/def_1/capabilities", { + kind: "toolPackage", + name: "@corbits/github-tools", + }), + ]); + expect(delRes.status).toBe(200); + expect(capRes.status).toBe(200); + + const workflowJson = await readAgentDefinitionWorkflowJson( + assetService, + "ast_1", + ); + expect(modelFrom(workflowJson)).toBeUndefined(); + expect(pinsFrom(workflowJson).map((pin) => pin.name)).toContain( + "@corbits/github-tools", + ); +}); From 7e4f1fae69ccd1666d2e55175fc659435873b49a Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 15:02:41 -0700 Subject: [PATCH 4/6] Retry concurrent definition, skills, and model writes (CL-7216) PUT instructions, PUT skills, and DELETE model were still lock-free read-modify-writes of the same asset, so a concurrent edit could snapshot the old tree and clobber the other field. They now share the capability-add per-asset lock and reapply on the latest snapshot. Hub is a single replica; the lock is an in-process chain, not a distributed advisory lock. --- packages/agent-directory/src/asset-write.ts | 94 +++++++++++ .../agent-directory/src/capability-add.ts | 86 ++++------ packages/agent-directory/src/routes.ts | 147 ++++++++++-------- 3 files changed, 208 insertions(+), 119 deletions(-) create mode 100644 packages/agent-directory/src/asset-write.ts diff --git a/packages/agent-directory/src/asset-write.ts b/packages/agent-directory/src/asset-write.ts new file mode 100644 index 000000000..826ee0620 --- /dev/null +++ b/packages/agent-directory/src/asset-write.ts @@ -0,0 +1,94 @@ +// Same-asset definition edits are read-modify-write: two concurrent +// routes that snapshot the same workflow.json and then each write their +// own field would last-write-wins clobber the other (CL-7216). This +// module serializes those writers and retries the loser against the +// latest snapshot. +// +// The lock is an in-process promise chain keyed by asset id. Hub is a +// single replica, so that chain is the lock — not a Redis or Postgres +// advisory lock, and not safe across multiple hub replicas. + +import type { AssetService } from "@intx/hub-sessions"; + +import { readAgentDefinitionWorkflowJson } from "./definition-asset"; + +const MAX_STALE_SNAPSHOT_RETRIES = 8; + +const writeChains = new Map>(); + +async function withAssetWriteLock( + assetId: string, + fn: () => Promise, +): Promise { + const previous = writeChains.get(assetId) ?? Promise.resolve(); + let release: () => void = () => undefined; + const held = new Promise((resolve) => { + release = resolve; + }); + writeChains.set( + assetId, + previous.then(() => held), + ); + try { + await previous; + return await fn(); + } finally { + release(); + } +} + +export type PreparedAgentAssetWrite = { + workflowJson: string; + message: string; + result: T; + afterWrite?: () => Promise; +}; + +export type CommitLatestAgentAssetSnapshotArgs = { + assetService: AssetService; + assetId: string; + operation: string; + prepare: (snapshot: string) => Promise>; + write: (prepared: { workflowJson: string; message: string }) => Promise; +}; + +/** + * Applies one mutation against the definition's current asset, retrying + * when a concurrent writer moved the snapshot between this call's read + * and its write. The per-asset lock makes that stale check atomic so + * the loser reapplies on the winner's tree instead of clobbering it. + */ +export async function commitLatestAgentAssetSnapshot( + args: CommitLatestAgentAssetSnapshotArgs, +): Promise { + for ( + let remaining = MAX_STALE_SNAPSHOT_RETRIES; + remaining > 0; + remaining -= 1 + ) { + const snapshot = await readAgentDefinitionWorkflowJson( + args.assetService, + args.assetId, + ); + const prepared = await args.prepare(snapshot); + const wrote = await withAssetWriteLock(args.assetId, async () => { + const latest = await readAgentDefinitionWorkflowJson( + args.assetService, + args.assetId, + ); + if (latest !== snapshot) return false; + await args.write({ + workflowJson: prepared.workflowJson, + message: prepared.message, + }); + if (prepared.afterWrite !== undefined) { + await prepared.afterWrite(); + } + return true; + }); + if (wrote) return prepared.result; + } + throw new Error( + `${args.operation} for asset ${args.assetId} conflicted after ${String(MAX_STALE_SNAPSHOT_RETRIES)} retries`, + ); +} diff --git a/packages/agent-directory/src/capability-add.ts b/packages/agent-directory/src/capability-add.ts index 4a1f0062f..456e8abfd 100644 --- a/packages/agent-directory/src/capability-add.ts +++ b/packages/agent-directory/src/capability-add.ts @@ -1,9 +1,10 @@ // Capability-add is a read-modify-write of the definition's asset: two // concurrent POSTs that both snapshot the same workflow.json and then // each write their own pin would last-write-wins clobber the other -// (CL-7216). This module owns that RMW so both the tenant-session route -// and the workflow-run route retry the loser against the latest snapshot -// instead of silently dropping an add. +// (CL-7216). This module owns that mutation so both the tenant-session +// route and the workflow-run route retry the loser against the latest +// snapshot instead of silently dropping an add. The lock itself lives +// in `./asset-write.ts` so sibling RMW routes share it. import type { PinnedSkillIndexEntry } from "@corbits/skills"; import type { AssetService } from "@intx/hub-sessions"; @@ -14,39 +15,14 @@ import { withAgentModel, withAgentToolPackagePin, } from "./agent-workflow"; +import { commitLatestAgentAssetSnapshot } from "./asset-write"; import type { AddCapabilityInput } from "./capability-inventory"; import { - readAgentDefinitionWorkflowJson, writeAndDeployAgentDefinition, type AgentDefinitionDeployer, } from "./definition-asset"; import type { DefinitionSkillsStore } from "./skills-store"; -const MAX_STALE_SNAPSHOT_RETRIES = 8; - -const writeChains = new Map>(); - -async function withAssetWriteLock( - assetId: string, - fn: () => Promise, -): Promise { - const previous = writeChains.get(assetId) ?? Promise.resolve(); - let release: () => void = () => undefined; - const held = new Promise((resolve) => { - release = resolve; - }); - writeChains.set( - assetId, - previous.then(() => held), - ); - try { - await previous; - return await fn(); - } finally { - release(); - } -} - export type CommitAgentCapabilityAddArgs = { assetService: AssetService; deployer: AgentDefinitionDeployer; @@ -81,22 +57,26 @@ export type CommitAgentCapabilityAddResult = { export async function commitAgentCapabilityAdd( args: CommitAgentCapabilityAddArgs, ): Promise { - for ( - let remaining = MAX_STALE_SNAPSHOT_RETRIES; - remaining > 0; - remaining -= 1 - ) { - const snapshot = await readAgentDefinitionWorkflowJson( - args.assetService, - args.assetId, - ); - const prepared = await prepareCapabilityAdd(snapshot, args); - const wrote = await withAssetWriteLock(args.assetId, async () => { - const latest = await readAgentDefinitionWorkflowJson( - args.assetService, - args.assetId, - ); - if (latest !== snapshot) return false; + return commitLatestAgentAssetSnapshot({ + assetService: args.assetService, + assetId: args.assetId, + 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 }) => { await writeAndDeployAgentDefinition({ assetService: args.assetService, deployer: args.deployer, @@ -104,19 +84,11 @@ export async function commitAgentCapabilityAdd( principalId: args.principalId, assetId: args.assetId, handle: args.handle, - workflowJson: prepared.workflowJson, - message: prepared.message, + workflowJson, + message, }); - if (prepared.nextSkills !== null) { - await args.skillsStore.setSkills(args.assetId, prepared.nextSkills); - } - return true; - }); - if (wrote) return prepared.result; - } - throw new Error( - `capability add for asset ${args.assetId} conflicted after ${String(MAX_STALE_SNAPSHOT_RETRIES)} retries`, - ); + }, + }); } type PreparedCapabilityAdd = { diff --git a/packages/agent-directory/src/routes.ts b/packages/agent-directory/src/routes.ts index 4e1250d2c..be6ce1cea 100644 --- a/packages/agent-directory/src/routes.ts +++ b/packages/agent-directory/src/routes.ts @@ -42,6 +42,7 @@ import { type CreateAgentDefinitionCoreDeps, type CreateAgentDefinitionCoreInput, } from "./agent-workflow"; +import { commitLatestAgentAssetSnapshot } from "./asset-write"; import { commitAgentCapabilityAdd } from "./capability-add"; import { agentDefinitionSourceTree, @@ -586,28 +587,32 @@ export function createAgentDefinitionRoutes({ return c.json(definitionNotFound(definitionId), 404); } - const workflowJson = await readAgentDefinitionWorkflowJson( - assetService, - row.assetId, - ); - - // Git first: the row updates below are what can still be retried - // safely if they fail after this succeeds (see the catch below) — - // the reverse order would leave a renamed row pointing at - // instructions that were never actually written. - const nextWorkflowJson = withAgentSystemPrompt( - workflowJson, - body.systemPrompt, - ); - await writeAndDeployAgentDefinition({ + const next = await commitLatestAgentAssetSnapshot({ assetService, - deployer, - tenantId: tenant.id, - principalId: principal.id, assetId: row.assetId, - handle: row.name, - workflowJson: nextWorkflowJson, - message: `Update agent instructions for ${row.name}`, + operation: "update instructions", + prepare: (snapshot) => + Promise.resolve({ + workflowJson: withAgentSystemPrompt(snapshot, body.systemPrompt), + message: `Update agent instructions for ${row.name}`, + result: { name: body.name, systemPrompt: body.systemPrompt }, + }), + write: async ({ workflowJson, message }) => { + // Git first: the row updates below are what can still be retried + // safely if they fail after this succeeds (see the catch below) — + // the reverse order would leave a renamed row pointing at + // instructions that were never actually written. + await writeAndDeployAgentDefinition({ + assetService, + deployer, + tenantId: tenant.id, + principalId: principal.id, + assetId: row.assetId, + handle: row.name, + workflowJson, + message, + }); + }, }); const now = new Date(); @@ -645,7 +650,7 @@ export function createAgentDefinitionRoutes({ ); } - return c.json({ name: body.name, systemPrompt: body.systemPrompt }); + return c.json(next); }, ); @@ -672,31 +677,40 @@ export function createAgentDefinitionRoutes({ return c.json(definitionNotFound(definitionId), 404); } - const workflowJson = await readAgentDefinitionWorkflowJson( + const cleared = await commitLatestAgentAssetSnapshot({ assetService, - row.assetId, - ); - const nextWorkflowJson = withoutAgentModel(workflowJson); - await assetService.populateAsset({ assetId: row.assetId, - ref: DEFAULT_ASSET_REF, - principal: { kind: "hub" }, - tree: { - files: agentDefinitionSourceTree({ - handle: row.name, - workflowJson: nextWorkflowJson, - }), - message: `Clear ${row.name}'s model`, + operation: "clear model", + prepare: async (snapshot) => { + const workflowJson = withoutAgentModel(snapshot); + const capabilities = readAgentCapabilities(workflowJson); + const skills = await skillsStore.getSkills(row.assetId); + return { + workflowJson, + message: `Clear ${row.name}'s model`, + result: { + toolPackagePins: capabilities.toolPackagePins, + skills, + model: capabilities.model, + }, + }; + }, + write: async ({ workflowJson, message }) => { + await assetService.populateAsset({ + assetId: row.assetId, + ref: DEFAULT_ASSET_REF, + principal: { kind: "hub" }, + tree: { + files: agentDefinitionSourceTree({ + handle: row.name, + workflowJson, + }), + message, + }, + }); }, }); - - const capabilities = readAgentCapabilities(nextWorkflowJson); - const skills = await skillsStore.getSkills(row.assetId); - return c.json({ - toolPackagePins: capabilities.toolPackagePins, - skills, - model: capabilities.model, - }); + return c.json(cleared); }, ); @@ -806,29 +820,38 @@ export function createAgentDefinitionRoutes({ ); } + const assetId = row.assetId; const principal = c.get("principal"); - const workflowJson = await readAgentDefinitionWorkflowJson( + const updated = await commitLatestAgentAssetSnapshot({ assetService, - row.assetId, - ); - - const nextWorkflowJson = reindexPinnedSkills( - workflowJson, - await skillIndex.resolve(tenant.id, principal.id, body.skills), - ); - await writeAndDeployAgentDefinition({ - assetService, - deployer, - tenantId: tenant.id, - principalId: principal.id, - assetId: row.assetId, - handle: row.name, - workflowJson: nextWorkflowJson, - message: `Update agent skills for ${row.name}`, + assetId, + operation: "update skills", + prepare: async (snapshot) => { + const workflowJson = reindexPinnedSkills( + snapshot, + await skillIndex.resolve(tenant.id, principal.id, body.skills), + ); + return { + workflowJson, + message: `Update agent skills for ${row.name}`, + afterWrite: () => skillsStore.setSkills(assetId, body.skills), + result: { skills: body.skills }, + }; + }, + write: async ({ workflowJson, message }) => { + await writeAndDeployAgentDefinition({ + assetService, + deployer, + tenantId: tenant.id, + principalId: principal.id, + assetId, + handle: row.name, + workflowJson, + message, + }); + }, }); - await skillsStore.setSkills(row.assetId, body.skills); - - return c.json({ skills: body.skills }); + return c.json(updated); }, ); From 1828d8608626db5023e20559c88b2484ae048a8b Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 15:44:26 -0700 Subject: [PATCH 5/6] Add tests for concurrent pin_skill and DELETE model (CL-7216) POST pin was still a lock-free read-modify-write of the same asset, so a concurrent model-clear could last-write-wins drop the pin (or the pin could restore the model). The suite races those two routes so a lost skill or model-clear is observable. --- packages/agent-directory/test/routes.test.ts | 70 ++++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/packages/agent-directory/test/routes.test.ts b/packages/agent-directory/test/routes.test.ts index 408dc70b9..48c0f8608 100644 --- a/packages/agent-directory/test/routes.test.ts +++ b/packages/agent-directory/test/routes.test.ts @@ -28,6 +28,10 @@ import { } from "../src/definition-asset"; import { createAgentDefinitionRoutes } from "../src/routes"; import type { PinnedSkillIndexResolver } from "../src/routes"; +import { + createWorkflowSkillPinRoutes, + type WorkflowRunAuthenticator, +} from "../src/workflow-skill-pin-routes"; import { createInMemoryDefinitionSkillsStore, type DefinitionSkillsStore, @@ -1978,3 +1982,69 @@ test("concurrent DELETE model and a capability-add both land", async () => { "@corbits/github-tools", ); }); + +test("concurrent pin_skill and DELETE model both land", async () => { + const skillsStore = createInMemoryDefinitionSkillsStore(); + const assetService = liveDefinitionAsset( + storedDefinitionBytesWithModel("anthropic/claude-sonnet"), + ); + const db = fakeInstructionsDb({ + id: "def_1", + assetId: "ast_1", + name: "research-buddy", + }); + const definitionApp = buildApp( + assetService, + db, + allowAllRequireGrant, + fakeHistory(), + fakeCapabilityInventory, + skillsStore, + ); + const pinAuthenticator: WorkflowRunAuthenticator = { + resolve: (token, address) => + Promise.resolve( + token === "sidecar-token" && address === "run_1@example.com" + ? { + tenantId: TENANT.id, + principalId: PRINCIPAL.id, + runId: "run_1", + } + : null, + ), + }; + const pinApp = createWorkflowSkillPinRoutes({ + db, + assetService, + skillIndex: fakeSkillIndex, + skillsStore, + authenticator: pinAuthenticator, + deployer: recordingAgentDefinitionDeployer(), + }); + + const [pinRes, delRes] = await Promise.all([ + pinApp.request("/pin", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer sidecar-token", + "x-workflow-run-address": "run_1@example.com", + }, + body: JSON.stringify({ + definitionId: "def_1", + skillName: "research", + }), + }), + definitionApp.request("/def_1/capabilities/model", { method: "DELETE" }), + ]); + expect(pinRes.status).toBe(200); + expect(delRes.status).toBe(200); + + const workflowJson = await readAgentDefinitionWorkflowJson( + assetService, + "ast_1", + ); + expect(await skillsStore.getSkills("ast_1")).toEqual(["research"]); + expect(promptFrom(workflowJson)).toContain("- research: What research does."); + expect(modelFrom(workflowJson)).toBeUndefined(); +}); From 3ad9d7d7aa9ae90bc4bb916955d21605aa2d7ad3 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 15:44:29 -0700 Subject: [PATCH 6/6] Retry concurrent pin_skill on the latest snapshot (CL-7216) POST /pin was still a lock-free read-modify-write of the definition asset, so a concurrent DELETE model could snapshot the old tree and clobber the pin (or the pin could restore the model). It now shares the capability-add per-asset lock and reapplies on the latest snapshot. --- .../src/workflow-skill-pin-routes.ts | 63 ++++++++++--------- 1 file changed, 35 insertions(+), 28 deletions(-) diff --git a/packages/agent-directory/src/workflow-skill-pin-routes.ts b/packages/agent-directory/src/workflow-skill-pin-routes.ts index a0f7f7358..20a59cca4 100644 --- a/packages/agent-directory/src/workflow-skill-pin-routes.ts +++ b/packages/agent-directory/src/workflow-skill-pin-routes.ts @@ -32,8 +32,8 @@ import type { AssetService } from "@intx/hub-sessions"; import { isWorkbenchHostDefinitionName } from "@corbits/chat/workbench-host-naming"; import { reindexPinnedSkills } from "./agent-workflow"; +import { commitLatestAgentAssetSnapshot } from "./asset-write"; import { - readAgentDefinitionWorkflowJson, RetiredWorkflowEnvelopeError, statusForAgentDefinitionDeployError, writeAndDeployAgentDefinition, @@ -167,37 +167,44 @@ export function createWorkflowSkillPinRoutes( return c.json(definitionNotFound(body.definitionId), 404); } - const workflowJson = await readAgentDefinitionWorkflowJson( - deps.assetService, - row.assetId, - ); - - const skills = await deps.skillsStore.getSkills(row.assetId); - const nextSkills = skills.includes(body.skillName) - ? skills - : [...skills, body.skillName]; - const nextWorkflowJson = reindexPinnedSkills( - workflowJson, - await deps.skillIndex.resolve( - scope.tenantId, - scope.principalId, - nextSkills, - ), - ); - - await writeAndDeployAgentDefinition({ + const next = await commitLatestAgentAssetSnapshot({ assetService: deps.assetService, - deployer: deps.deployer, - tenantId: scope.tenantId, - principalId: scope.principalId, assetId: row.assetId, - handle: row.name, - workflowJson: nextWorkflowJson, - message: `Pin ${body.skillName} skill to ${row.name}`, + operation: "pin skill", + prepare: async (snapshot) => { + const skills = await deps.skillsStore.getSkills(row.assetId); + const nextSkills = skills.includes(body.skillName) + ? skills + : [...skills, body.skillName]; + return { + workflowJson: reindexPinnedSkills( + snapshot, + await deps.skillIndex.resolve( + scope.tenantId, + scope.principalId, + nextSkills, + ), + ), + message: `Pin ${body.skillName} skill to ${row.name}`, + afterWrite: () => deps.skillsStore.setSkills(row.assetId, nextSkills), + result: { skills: nextSkills }, + }; + }, + write: async ({ workflowJson, message }) => { + await writeAndDeployAgentDefinition({ + assetService: deps.assetService, + deployer: deps.deployer, + tenantId: scope.tenantId, + principalId: scope.principalId, + assetId: row.assetId, + handle: row.name, + workflowJson, + message, + }); + }, }); - await deps.skillsStore.setSkills(row.assetId, nextSkills); - return c.json({ skills: nextSkills }); + return c.json(next); }); return app;