diff --git a/VENDORED.md b/VENDORED.md index e5e9dd2a4..d6e9be24f 100644 --- a/VENDORED.md +++ b/VENDORED.md @@ -25,7 +25,7 @@ never a convenience. | Vendored path | What was copied | Upstream repo @ commit | Why not a published package | Owner | Kill date | Kill-date test | | ----------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------ | ---------- | ----------------- | | `apps/sidecar` | Derived from upstream's own `apps/sidecar`: 11 shared modules, of which `signing-keypair.ts` is near-verbatim and the rest (`index.ts`, `config.ts`, `tool-materialization.ts`, `workflow-run-pack-client.ts`, …) are substantially rewritten, plus workbench-only modules. A living fork, not a frozen copy, so this row carries no tree hash. | [faremeter/interchange](https://github.com/faremeter/interchange) @ `b5580a02` (v0.3.0) | An app is never npm-published, so no publish can cover the execution host; retired by consuming an upstream-published host, or by renewing this row deliberately | sawyer | 2026-09-19 | `check:killdates` | -| `vendor/intx/db` | `@intx/db` source (`src/`, `migrations/`, drizzle config, manifest, tsconfigs) | [faremeter/interchange](https://github.com/faremeter/interchange) @ `b5580a02` (v0.3.0) | npm 0.3.0 covers the base package but not the `wire_projection` column/loader delta (CL-6324); retired when upstream absorbs the delta | sawyer | 2026-09-19 | `check:killdates` | +| `vendor/intx/db` | `@intx/db` source (`src/`, `migrations/`, drizzle config, manifest, tsconfigs) | [faremeter/interchange](https://github.com/faremeter/interchange) @ `b5580a02` (v0.3.0) | npm 0.3.0 covers the base package but not the `wire_projection` column/loader delta (CL-6324) or the `workflow_definition.origin` column separating a definition from the per-run record of one folded run's deploy (CL-6452); retired when upstream absorbs the deltas | sawyer | 2026-09-19 | `check:killdates` | | `vendor/intx/hub-api` | `@intx/hub-api` source (`src/`, manifest, tsconfig) | [faremeter/interchange](https://github.com/faremeter/interchange) @ `b5580a02` (v0.3.0) | npm 0.3.0 covers the base package but not the `needs-you` approval-route reservation or the exported null-principal `resolveApproval` (CL-6345); retired when upstream absorbs the deltas | sawyer | 2026-09-19 | `check:killdates` | | `vendor/intx/hub-sessions` | `@intx/hub-sessions` source (`src/`, manifest, tsconfig) | [faremeter/interchange](https://github.com/faremeter/interchange) @ `b5580a02` (v0.3.0) | npm 0.3.0 covers the base package but not the usage forward (CL-5879), pack-acceptance fixes, adopted deploy front, wire-projection writer, event-collector serialization, or anchor ordering | sawyer | 2026-09-19 | `check:killdates` | | `vendor/intx/workflow` | `@intx/workflow` source (`src/`, manifest, tsconfig) | [faremeter/interchange](https://github.com/faremeter/interchange) @ `b5580a02` (v0.3.0) | npm 0.3.0 covers the base package but not the `onBodyFailure` trigger policy and its projection (CL-6326, CL-6324); retired when upstream absorbs the delta | sawyer | 2026-09-19 | `check:killdates` | diff --git a/packages/chat/src/platform-adapter.ts b/packages/chat/src/platform-adapter.ts index 325c51cfb..70a1a8591 100644 --- a/packages/chat/src/platform-adapter.ts +++ b/packages/chat/src/platform-adapter.ts @@ -10,11 +10,12 @@ import { and, desc, eq } from "drizzle-orm"; import { createAgentLifecycle } from "@corbits/agent-lifecycle"; import { + authoredDefinitionCandidates, createCryptoProviderCache, + DefinitionProjectionMissingError, domainOf, launchFoldedRun, mintFoldedRun, - readDefinitionProjection, readFoldedBody, resolveFoldedRunSessionId, resolveNewestProjectedDefinition, @@ -502,6 +503,48 @@ export function createHubChatPlatform( } } + // CL-6452: every run deploy ensures a same-named sibling definition + // over the agent's asset under its per-run wire hash — a frozen + // deploy record carrying whatever projection was current at that + // deploy. Launch bodies resolve only from the hub-authored row(s) of + // the asset, so a skill pin or instructions save (which refreezes + // the authored row in place) reaches every later launch instead of + // being shadowed by the newest clone. Raises the named + // `DefinitionProjectionMissingError` — mapped to a 4xx at the route + // boundary, never an unhandled 500 — when the asset has no authored + // definition or none of its authored rows carries a projection. + async function resolveAuthoredProjectedDefinition( + tenantId: string, + definitionAsset: { assetId: string; name: string }, + ) { + const assetSiblingRows = await deps.db.query.workflowDefinition.findMany({ + where: and( + eq(workflowDefinition.tenantId, tenantId), + eq(workflowDefinition.assetId, definitionAsset.assetId), + eq(workflowDefinition.status, "deployed"), + ), + orderBy: desc(workflowDefinition.createdAt), + }); + const candidates = authoredDefinitionCandidates(assetSiblingRows); + if (candidates.length === 0) { + throw new DefinitionProjectionMissingError(definitionAsset.name); + } + const resolved = await resolveNewestProjectedDefinition( + deps.db, + candidates, + ); + const row = candidates.find( + (candidate) => candidate.id === resolved.definitionId, + ); + if (row === undefined) { + throw new Error( + `resolved definition "${resolved.definitionId}" is not among the ` + + `authored candidates for asset "${definitionAsset.assetId}"`, + ); + } + return { row, projection: resolved.projection }; + } + const platform: ChatPlatform = { async launchInvite(input): Promise { const definitionRow = await deps.db.query.workflowDefinition.findFirst({ @@ -534,37 +577,14 @@ export function createHubChatPlatform( throw new Error(`No tenant "${input.tenantId}"`); } - // CL-6357: a long-lived DB can carry a definition row whose asset - // repo has gone unresolvable (DB/blob drift — a `.data` reset - // that never touched Postgres) while a newer, healthy sibling - // under the same name already exists (a re-seed, typically). - // Resolution tries every deployed sibling under this name - // newest-first and uses the first one that actually reads — the - // specifically requested (possibly stale) row never wins over a - // healthy newer one. `resolveNewestProjectedDefinition` - // raises the named `DefinitionProjectionMissingError` — mapped - // to a 4xx at the route boundary, never an unhandled 500 — only - // once every sibling has failed to resolve. - const siblingRows = await deps.db.query.workflowDefinition.findMany({ - where: and( - eq(workflowDefinition.tenantId, input.tenantId), - eq(workflowDefinition.name, definitionRow.name), - eq(workflowDefinition.status, "deployed"), - ), - orderBy: desc(workflowDefinition.createdAt), - }); - const candidates = siblingRows.length > 0 ? siblingRows : [definitionRow]; - - const resolved = await resolveNewestProjectedDefinition( - deps.db, - candidates, - ); - const resolvedDefinitionRow = - candidates.find((row) => row.id === resolved.definitionId) ?? - definitionRow; + const { row: resolvedDefinitionRow, projection } = + await resolveAuthoredProjectedDefinition(input.tenantId, { + assetId: definitionRow.assetId, + name: definitionRow.name, + }); const foldedBody = readFoldedBody( - resolved.projection, + projection, resolvedDefinitionRow.grantRequirements, ); if (foldedBody.systemPrompt === "") { @@ -619,7 +639,10 @@ export function createHubChatPlatform( ), orderBy: desc(workflowDefinition.createdAt), }); - return rows + // Only hub-authored definitions are invitable: the run-deploy + // clones sharing an agent's name are deploy records, and listing + // them would offer N stale copies of every agent that has run. + return authoredDefinitionCandidates(rows) .filter((row) => !isWorkbenchHostDefinitionName(row.name)) .map((row) => { const base = { id: row.id, name: row.name }; @@ -664,10 +687,18 @@ export function createHubChatPlatform( return; } - const projection = await readDefinitionProjection(deps.db, definitionRow); + // The run's own definition row is the per-run clone the deploy + // repointed it to; the refresh recomputes from the hub-authored + // sibling so the saved edit — not the clone's frozen snapshot — + // is what the next wake replays. + const { row: authoredRow, projection } = + await resolveAuthoredProjectedDefinition(tenantId, { + assetId: definitionRow.assetId, + name: definitionRow.name, + }); const foldedBody = readFoldedBody( projection, - definitionRow.grantRequirements, + authoredRow.grantRequirements, ); await deps.db diff --git a/packages/chat/test/platform-adapter.test.ts b/packages/chat/test/platform-adapter.test.ts index dd62a48d3..e8030b21d 100644 --- a/packages/chat/test/platform-adapter.test.ts +++ b/packages/chat/test/platform-adapter.test.ts @@ -89,6 +89,23 @@ function selectChain(rows: unknown[]): SelectChain { return chain; } +/** + * The string parameters bound into a drizzle SQL expression, in query + * order: `eq(column, value)` wraps each value in a `Param` whose + * `value` is the bound string, and `and(...)` nests sub-expressions + * under `queryChunks` — column and operator chunks carry no bare + * string `value`, so the walk collects exactly the bound parameters. + */ +function boundStringValues(expression: unknown, out: string[] = []): string[] { + if (expression === null || typeof expression !== "object") return out; + const chunk = expression as { value?: unknown; queryChunks?: unknown[] }; + if (typeof chunk.value === "string") out.push(chunk.value); + if (Array.isArray(chunk.queryChunks)) { + for (const nested of chunk.queryChunks) boundStringValues(nested, out); + } + return out; +} + type InsertChain = { onConflictDoNothing(...args: unknown[]): InsertChain; returning(...args: unknown[]): Promise; @@ -151,6 +168,7 @@ function createFakeDb(opts: { status: string; assetId: string | null; name?: string; + origin?: "authored" | "run"; grantRequirements?: unknown; } | undefined; @@ -162,6 +180,7 @@ function createFakeDb(opts: { name: string; description?: string; assetId?: string | null; + origin?: "authored" | "run"; grantRequirements?: unknown; }[] | undefined; @@ -186,11 +205,9 @@ function createFakeDb(opts: { * via `select().from(workflowDefinitionVersion)`) returns for each * definition id, keyed by id. An id with no entry (or an explicit * `null`) mirrors a pre-cutover row that carries no stored - * projection. Call order mirrors the real resolution order: - * `launchInvite`'s candidates (siblings newest-first, or the single - * requested row when no siblings are configured) in order, then - * `refreshAgentInstanceFromDefinition`'s single lookup on the run's - * own definition row. + * projection. Lookups resolve by the definition id actually bound + * into the query's `where`, so the record answers exactly the ids + * production code asks for, in any order. */ wireProjectionsByDefinitionId?: Record | undefined; }) { @@ -198,19 +215,6 @@ function createFakeDb(opts: { const updated: { table: unknown; values: unknown }[] = []; const deleted: { table: unknown }[] = []; - // Mirrors the real resolution order for `loadFrozenWireProjection` - // calls: `launchInvite`'s candidates (siblings newest-first, falling - // back to the single requested row) or, absent any siblings/requested - // row config, `refreshAgentInstanceFromDefinition`'s single lookup - // against the run's own definition row. - const wireProjectionCandidateIds = ( - opts.workflowDefinitionRows && opts.workflowDefinitionRows.length > 0 - ? opts.workflowDefinitionRows - : opts.workflowDefinitionRow !== undefined - ? [opts.workflowDefinitionRow] - : [] - ).map((row) => row.id); - let wireProjectionCallIndex = 0; const wireProjectionCalls: string[] = []; function updateOn(table: unknown): UpdateChain { @@ -261,7 +265,14 @@ function createFakeDb(opts: { }, workflowDefinition: { findFirst: async () => opts.workflowDefinitionRow, - findMany: async () => opts.workflowDefinitionRows ?? [], + // The requested definition row is itself a deployed row of its + // asset, so the real asset-sibling query always returns at + // least it — the single-row default mirrors that. + findMany: async () => + opts.workflowDefinitionRows ?? + (opts.workflowDefinitionRow !== undefined + ? [opts.workflowDefinitionRow] + : []), }, tenant: { findFirst: async () => opts.tenantRow, @@ -279,14 +290,19 @@ function createFakeDb(opts: { } if (table === asset) return selectChain([opts.assetRow]); if (table === workflowDefinitionVersion) { - const definitionId = - wireProjectionCandidateIds[wireProjectionCallIndex]; - wireProjectionCallIndex += 1; - if (definitionId === undefined) return selectChain([]); - wireProjectionCalls.push(definitionId); - const projection = - opts.wireProjectionsByDefinitionId?.[definitionId] ?? null; - return selectChain([{ wireProjection: projection }]); + // `loadFrozenWireProjection` filters on + // `and(eq(definitionId, id), eq(version, "1"))`; the first + // bound string in that expression is the definition id. + return { + where: (expression: unknown) => { + const [definitionId] = boundStringValues(expression); + if (definitionId === undefined) return selectChain([]); + wireProjectionCalls.push(definitionId); + const projection = + opts.wireProjectionsByDefinitionId?.[definitionId] ?? null; + return selectChain([{ wireProjection: projection }]); + }, + }; } if (table === workbenchLaunch) { const insertedLaunch = inserted.findLast( @@ -940,6 +956,7 @@ describe("createHubChatPlatform", () => { id: "wfd_echo", tenantId: "ten_1", status: "deployed", + origin: "authored", assetId: "asst_echo", }, tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, @@ -1042,6 +1059,7 @@ describe("createHubChatPlatform", () => { id: "wfd_echo", tenantId: "ten_1", status: "deployed", + origin: "authored", assetId: "asst_echo", }, tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, @@ -1117,12 +1135,11 @@ describe("createHubChatPlatform", () => { ).rejects.toThrow(/not in a launchable state/); }); - // CL-6357: a long-lived dev DB can carry a definition row with no - // frozen wire projection stored on it (a pre-cutover row) alongside a - // fresher, healthy sibling under the same name — a re-seed, say. - // Resolution must prefer that newest-healthy sibling rather than - // dying on the specific (possibly stale) row the caller asked for. - test("launchInvite resolves the newest healthy sibling definition over the requested definition's own stale one", async () => { + // A dev DB whose authored definition has drifted (no stored + // projection) must answer a named error a caller can map to a 4xx, + // never let the raw lookup failure escape as an unhandled 500 — and + // never fall back to a run-deploy clone's frozen snapshot. + test("launchInvite raises DefinitionProjectionMissingError, not a raw 500, when no sibling definition resolves", async () => { const db = createFakeDb({ assetRow: { tenantId: "ten_1", @@ -1132,33 +1149,97 @@ describe("createHubChatPlatform", () => { }, definitionId: "wfd_workbench1", workflowDefinitionRow: { - id: "wfd_stale", + id: "wfd_dead", tenantId: "ten_1", status: "deployed", - assetId: "asst_stale", + origin: "authored", + assetId: "asst_dead", }, workflowDefinitionRows: [ - // Newest first, matching `orderBy: desc(createdAt)`. { - id: "wfd_fresh", + id: "wfd_dead", tenantId: "ten_1", status: "deployed", + origin: "authored", name: "assistant", - assetId: "asst_fresh", + assetId: "asst_dead", }, + ], + tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, + // No entry for "wfd_dead" — mirrors a definition with no stored + // projection, and there is no other sibling to fall back to. + }); + + const platform = createHubChatPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService: createFakeSessionService(), + assetService: createFakeAssetService(), + sidecarRouter: createFakeSidecarRouter({ routableAddresses: [] }), + eventCollectors: createFakeEventCollectors(), + }); + + await expect( + platform.launchInvite({ + tenantId: "ten_1", + creatorPrincipalId: "prin_creator", + definitionId: "wfd_dead", + }), + ).rejects.toThrow(DefinitionProjectionMissingError); + }); + + // CL-6452: every run deploy mints a same-named, same-asset sibling + // definition row frozen with the projection current at that deploy. + // A later invite must launch the hub-authored row's CURRENT + // projection — the one a skill pin or instructions save refroze in + // place — never a newer run clone's stale snapshot. + test("launchInvite launches the hub-authored projection, not a newer run-deploy clone's stale one", async () => { + const db = createFakeDb({ + assetRow: { + tenantId: "ten_1", + creatorPrincipalId: "prin_creator", + name: "fact-checker", + displayName: null, + }, + definitionId: "wfd_workbench1", + workflowDefinitionRow: { + id: "wfd_authored", + tenantId: "ten_1", + status: "deployed", + assetId: "asst_agent", + name: "fact-checker", + origin: "authored", + }, + workflowDefinitionRows: [ + // Newest first: the clone the last run deploy minted, frozen + // before the skill pin landed. { - id: "wfd_stale", + id: "wfd_run_clone", tenantId: "ten_1", status: "deployed", - name: "assistant", - assetId: "asst_stale", + name: "fact-checker", + assetId: "asst_agent", + origin: "run", + }, + { + id: "wfd_authored", + tenantId: "ten_1", + status: "deployed", + name: "fact-checker", + assetId: "asst_agent", + origin: "authored", }, ], tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, - // The stale sibling carries no stored projection at all; the - // fresh one does — the newer definition must win. wireProjectionsByDefinitionId: { - wfd_fresh: inertProjection({ id: "wfd_fresh" }), + wfd_run_clone: inertProjection({ + id: "wfd_run_clone", + systemPrompt: "pre-pin instructions frozen at the run deploy", + }), + wfd_authored: inertProjection({ + id: "wfd_authored", + systemPrompt: "post-pin instructions", + }), }, }); @@ -1171,56 +1252,66 @@ describe("createHubChatPlatform", () => { eventCollectors: createFakeEventCollectors(), }); - const launched = await platform.launchInvite({ + await platform.launchInvite({ tenantId: "ten_1", creatorPrincipalId: "prin_creator", - definitionId: "wfd_stale", + definitionId: "wfd_authored", }); - expect(launched.instanceId).toMatch(/^run_/); - // The minted run's definitionId is the resolved healthy sibling, - // not the stale requested id — every later wake reads the - // definition's projection through this row, so it must be one that - // actually resolves. const runInsert = db.inserted.find((row) => row.table === workflowRun); - expect(runInsert?.values).toMatchObject({ definitionId: "wfd_fresh" }); - // Resolution walked every deployed sibling under the name - // newest-first and used the fresh one — never fell back to - // re-reading the specifically requested (stale) row. - expect(db.wireProjectionCalls).toEqual(["wfd_fresh"]); + expect(runInsert?.values).toMatchObject({ definitionId: "wfd_authored" }); + const launchInsert = db.inserted.find( + (row) => row.table === workbenchLaunch, + ); + expect( + (launchInsert?.values as { foldedBody: { systemPrompt: string } }) + .foldedBody.systemPrompt, + ).toBe("post-pin instructions"); }); - // A dev DB whose definition rows have all drifted (no sibling under - // the name carries a stored projection) must answer a named error a - // caller can map to a 4xx, never let the raw lookup failure escape as - // an unhandled 500. - test("launchInvite raises DefinitionProjectionMissingError, not a raw 500, when no sibling definition resolves", async () => { + // The candidate set a launch resolves over is the authored row alone: + // N runs mint N clones, and none of them may ever be consulted. + test("run-deploy clones never grow the authoritative candidate set", async () => { + const cloneRows = Array.from({ length: 5 }, (_, index) => ({ + id: `wfd_run_${String(5 - index)}`, + tenantId: "ten_1", + status: "deployed", + name: "fact-checker", + assetId: "asst_agent", + origin: "run" as const, + })); const db = createFakeDb({ assetRow: { tenantId: "ten_1", creatorPrincipalId: "prin_creator", - name: "workbench-1", + name: "fact-checker", displayName: null, }, definitionId: "wfd_workbench1", workflowDefinitionRow: { - id: "wfd_dead", + id: "wfd_authored", tenantId: "ten_1", status: "deployed", - assetId: "asst_dead", + assetId: "asst_agent", + name: "fact-checker", + origin: "authored", }, workflowDefinitionRows: [ + ...cloneRows, { - id: "wfd_dead", + id: "wfd_authored", tenantId: "ten_1", status: "deployed", - name: "assistant", - assetId: "asst_dead", + name: "fact-checker", + assetId: "asst_agent", + origin: "authored", }, ], tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, - // No entry for "wfd_dead" — mirrors a definition with no stored - // projection, and there is no other sibling to fall back to. + wireProjectionsByDefinitionId: { + wfd_run_5: inertProjection({ id: "wfd_run_5" }), + wfd_authored: inertProjection({ id: "wfd_authored" }), + }, }); const platform = createHubChatPlatform({ @@ -1232,13 +1323,13 @@ describe("createHubChatPlatform", () => { eventCollectors: createFakeEventCollectors(), }); - await expect( - platform.launchInvite({ - tenantId: "ten_1", - creatorPrincipalId: "prin_creator", - definitionId: "wfd_dead", - }), - ).rejects.toThrow(DefinitionProjectionMissingError); + await platform.launchInvite({ + tenantId: "ten_1", + creatorPrincipalId: "prin_creator", + definitionId: "wfd_authored", + }); + + expect(db.wireProjectionCalls).toEqual(["wfd_authored"]); }); test("launchInvite fails loud when no such definition exists for the tenant", async () => { @@ -1293,6 +1384,7 @@ describe("createHubChatPlatform", () => { id: "wfd_echo", tenantId: "ten_1", status: "deployed", + origin: "authored", assetId: "asst_echo", }, tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, @@ -1360,6 +1452,7 @@ describe("createHubChatPlatform", () => { id: "wfd_echo", tenantId: "ten_1", status: "deployed", + origin: "authored", assetId: "asst_echo", }, tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, @@ -1413,6 +1506,7 @@ describe("createHubChatPlatform", () => { id: "wfd_echo", tenantId: "ten_1", status: "deployed", + origin: "authored", assetId: "asst_echo", }, tenantRow: { id: "ten_1", domain: "ten1.workbench.test" }, @@ -1458,6 +1552,7 @@ describe("createHubChatPlatform", () => { id: "wfd_echo", tenantId: "ten_1", status: "deployed", + origin: "authored", name: "echo", description: "Echo", }, @@ -1465,12 +1560,14 @@ describe("createHubChatPlatform", () => { id: "wfd_host1", tenantId: "ten_1", status: "deployed", + origin: "authored", name: "ins-0f1e2d3c4b5a69788796a5b4c3d2e1f0", }, { id: "wfd_host2", tenantId: "ten_1", status: "deployed", + origin: "authored", name: "run-682bf127e22124c01b4b0996aabaab5f", }, ], @@ -1579,6 +1676,7 @@ describe("createHubChatPlatform", () => { id: "wfd_workbench1", tenantId: "ten_1", status: "deployed", + origin: "authored", assetId: "asst_workbench1", }, workbenchLaunchRow: { @@ -1674,6 +1772,7 @@ describe("createHubChatPlatform", () => { id: "wfd_workbench1", tenantId: "ten_1", status: "deployed", + origin: "authored", assetId: "asst_workbench1", }, workbenchLaunchRow: { @@ -2224,6 +2323,7 @@ describe("createHubChatPlatform", () => { id: "wfd_agent1", tenantId: "ten_1", status: "deployed", + origin: "authored", assetId: "asst_agent1", }, workbenchLaunchRow: { @@ -2322,6 +2422,95 @@ describe("createHubChatPlatform", () => { "You are now a blunt, no-nonsense assistant.", ); }); + + // CL-6452: the deploy repoints `workflow_run.definitionId` at the + // per-run clone it minted, so the run's own definition row carries + // the projection frozen at that deploy — a refresh reading it would + // replay the stale body forever. The refresh must recompute from + // the hub-authored sibling of the run's asset instead. + test("recomputes from the hub-authored definition, not the run's own deploy clone", async () => { + const db = createFakeDb({ + assetRow: { + tenantId: "ten_1", + creatorPrincipalId: null, + name: "unused", + displayName: null, + }, + definitionId: "wfd_unused", + workflowRunRow: { + id: "run_agent1", + address: "agent1@ten1.workbench.test", + principalId: "prin_agent1", + definitionId: "wfd_run_clone", + }, + workflowDefinitionRow: { + id: "wfd_run_clone", + tenantId: "ten_1", + status: "deployed", + assetId: "asst_agent1", + name: "fact-checker", + origin: "run", + }, + workflowDefinitionRows: [ + { + id: "wfd_run_clone", + tenantId: "ten_1", + status: "deployed", + name: "fact-checker", + assetId: "asst_agent1", + origin: "run", + }, + { + id: "wfd_agent1", + tenantId: "ten_1", + status: "deployed", + name: "fact-checker", + assetId: "asst_agent1", + origin: "authored", + }, + ], + workbenchLaunchRow: { + tenantId: "ten_1", + instanceId: "run_agent1", + foldedBody: { + systemPrompt: "You are a careful research assistant.", + model: "claude-sonnet-5", + toolPackagePins: [], + grantRequirements: [], + credentialBindings: [], + }, + }, + wireProjectionsByDefinitionId: { + wfd_run_clone: inertProjection({ + id: "wfd_run_clone", + systemPrompt: "You are a careful research assistant.", + }), + wfd_agent1: NEW_PROJECTION, + }, + }); + const platform = createHubChatPlatform({ + toolGrantsForPins: () => [], + db: db as never, + sessionService: createFakeSessionService(), + assetService: createFakeAssetService(), + sidecarRouter: createFakeSidecarRouter(), + eventCollectors: createFakeEventCollectors(), + }); + + await platform.refreshAgentInstanceFromDefinition( + "ten_1", + "ch_1", + "agent1@ten1.workbench.test", + ); + + const launchUpdate = db.updated.find( + (row) => row.table === workbenchLaunch, + ); + expect( + (launchUpdate?.values as { foldedBody: { systemPrompt: string } }) + .foldedBody.systemPrompt, + ).toBe("You are now a blunt, no-nonsense assistant."); + }); }); }); diff --git a/packages/folded-runs/src/definition.ts b/packages/folded-runs/src/definition.ts index f850492fa..459e77500 100644 --- a/packages/folded-runs/src/definition.ts +++ b/packages/folded-runs/src/definition.ts @@ -62,13 +62,27 @@ export async function readDefinitionProjection( return projection; } -/** One definition candidate for a name, ordered newest-first by the - * caller (typically `createdAt desc`). */ +/** One definition candidate, ordered newest-first by the caller + * (typically `createdAt desc`). */ export type DefinitionCandidate = { readonly id: string; readonly name: string; }; +/** + * The launch-authoritative subset of an asset's definition rows: only + * the hub-authored row(s), whose projection a skill pin or + * instructions save refreezes in place. Every code-sourced run deploy + * ensures a same-named sibling over the same asset under its per-run + * wire hash — a frozen deploy record carrying whatever projection was + * current at that deploy, which must never resolve a launch (CL-6452). + */ +export function authoredDefinitionCandidates< + T extends { readonly origin: "authored" | "run" }, +>(rows: readonly T[]): T[] { + return rows.filter((row) => row.origin === "authored"); +} + /** * Resolves a definition's launch body by trying its candidates * newest-first and returning the first one that actually carries a diff --git a/packages/folded-runs/src/index.ts b/packages/folded-runs/src/index.ts index 076df2011..f08a0b20e 100644 --- a/packages/folded-runs/src/index.ts +++ b/packages/folded-runs/src/index.ts @@ -8,6 +8,7 @@ export type { ListedFoldedMailItem, } from "./types"; export { + authoredDefinitionCandidates, readDefinitionProjection, readFoldedBody, readLiveFoldedBody, diff --git a/packages/folded-runs/src/launch.ts b/packages/folded-runs/src/launch.ts index 01f52bcdd..519ec84f8 100644 --- a/packages/folded-runs/src/launch.ts +++ b/packages/folded-runs/src/launch.ts @@ -135,12 +135,15 @@ function domainOfAddress(address: string): string { * per-run source tree is committed INTO that asset on its own ref * rather than into a second asset minted per deploy. */ -async function resolveRunDefinitionAssetId( +async function resolveRunDefinition( db: FoldedRunsDeps["db"], instanceId: string, -): Promise { +): Promise<{ definitionId: string; assetId: string }> { const row = await db - .select({ assetId: workflowDefinition.assetId }) + .select({ + definitionId: workflowDefinition.id, + assetId: workflowDefinition.assetId, + }) .from(workflowRun) .innerJoin( workflowDefinition, @@ -154,7 +157,36 @@ async function resolveRunDefinitionAssetId( `folded run ${instanceId} has no workflow-kind definition asset to commit its per-run source tree into`, ); } - return row.assetId; + return { definitionId: row.definitionId, assetId: row.assetId }; +} + +/** + * Mark the definition row a code-sourced deploy just minted for THIS + * run as a per-run clone (CL-6452). + * + * A folded run's deployed bytes carry per-run values (`wf_`, the + * run's trigger address), so their wire hash is unique to the run and + * the deploy's freeze ensures a fresh `workflow_definition` over the + * agent's asset, then repoints the run at it. That row is a frozen + * deploy record, not a definition anyone may launch from: left + * unmarked it would shadow the hub-authored row every agent edit + * refreezes, and no invite after the agent's first run would ever carry + * an updated prompt or skill pin. + * + * Only a row the deploy actually repointed to is marked, so a deploy + * that left the run on its original definition can never demote it. + */ +async function markRunDeployClone( + db: FoldedRunsDeps["db"], + instanceId: string, + definitionIdBeforeDeploy: string, +): Promise { + const { definitionId } = await resolveRunDefinition(db, instanceId); + if (definitionId === definitionIdBeforeDeploy) return; + await db + .update(workflowDefinition) + .set({ origin: "run" }) + .where(eq(workflowDefinition.id, definitionId)); } /** @@ -369,10 +401,8 @@ export async function deployAtHead( credentialBindings, mode: params.mode ?? { kind: "step" }, }; - const definitionAssetId = await resolveRunDefinitionAssetId( - deps.db, - params.instanceId, - ); + const { definitionId: definitionIdBeforeDeploy, assetId: definitionAssetId } = + await resolveRunDefinition(deps.db, params.instanceId); const { commitSha } = await deps.assetService.populateAsset({ assetId: definitionAssetId, ref: foldedRunSourceRef(params.instanceId), @@ -441,6 +471,12 @@ export async function deployAtHead( : {}), }); + await markRunDeployClone( + deps.db, + params.instanceId, + definitionIdBeforeDeploy, + ); + // Produce the run's `run.grants` frame, the same contract upstream's hub // fires on every run birth: the sidecar writes it to // `runs//grants.json` in the deployment's workflow-run repo, and diff --git a/packages/folded-runs/test/definition.test.ts b/packages/folded-runs/test/definition.test.ts index ac86fa08e..264c6de17 100644 --- a/packages/folded-runs/test/definition.test.ts +++ b/packages/folded-runs/test/definition.test.ts @@ -20,6 +20,7 @@ mock.module("@intx/db", () => ({ })); const { + authoredDefinitionCandidates, readFoldedBody, readLiveFoldedBody, readDefinitionProjection, @@ -220,6 +221,45 @@ describe("readLiveFoldedBody", () => { }); }); +// CL-6452: every run deploy mints a same-named, same-asset sibling +// definition row (its per-run rendered bytes carry a per-run wire hash), +// frozen with the projection current AT THAT DEPLOY. Those run clones +// are deploy records, never launch candidates — only the hub-authored +// row, whose projection a skill pin or instructions save refreezes in +// place, may resolve a launch. +describe("authoredDefinitionCandidates", () => { + test("keeps only the hub-authored row, dropping run-deploy clones", () => { + const authored = { + id: "wfd_authored", + name: "fact-checker", + origin: "authored", + } as const; + expect( + authoredDefinitionCandidates([ + // Newest first: the clones every run deploy minted after the + // agent was authored. + { id: "wfd_run_2", name: "fact-checker", origin: "run" }, + { id: "wfd_run_1", name: "fact-checker", origin: "run" }, + authored, + ]), + ).toEqual([authored]); + }); + + test("N run deploys never grow the authoritative candidate set", () => { + const authored = { + id: "wfd_authored", + name: "fact-checker", + origin: "authored", + } as const; + const clones = Array.from({ length: 5 }, (_, index) => ({ + id: `wfd_run_${String(index + 1)}`, + name: "fact-checker", + origin: "run" as const, + })); + expect(authoredDefinitionCandidates([...clones, authored])).toHaveLength(1); + }); +}); + describe("resolveNewestProjectedDefinition", () => { test("prefers the newest definition that actually carries a projection over a pre-cutover one", async () => { const healthy = inertProjection({ id: "wfd_new" }); diff --git a/packages/folded-runs/test/launch.test.ts b/packages/folded-runs/test/launch.test.ts index ee006eb34..991849f1f 100644 --- a/packages/folded-runs/test/launch.test.ts +++ b/packages/folded-runs/test/launch.test.ts @@ -80,10 +80,21 @@ type InsertChain = { values(values: unknown): Promise; }; -function createFakeDb(assetId: string | null = "ast_definition1") { +/** + * `deployAtHead` reads the run's definition row twice: once before the + * deploy (for the asset its per-run tree is committed into) and once + * after, to see which row the deploy repointed the run at. `definitionIds` + * answers those reads in order — a second id different from the first is + * the per-run clone every real code-sourced deploy mints. + */ +function createFakeDb( + assetId: string | null = "ast_definition1", + definitionIds: readonly string[] = ["wfd_definition1", "wfd_definition1"], +) { const inserted: { table: unknown; values: unknown }[] = []; const updated: { table: unknown; values: unknown }[] = []; const deleted: { table: unknown }[] = []; + let definitionReadIndex = 0; function insertOn(table: unknown): InsertChain { return { @@ -94,14 +105,17 @@ function createFakeDb(assetId: string | null = "ast_definition1") { } return { - // The one read `deployAtHead` does: the run's definition asset, the - // asset its per-run source tree is committed into. select() { + const definitionId = + definitionIds[definitionReadIndex] ?? + definitionIds[definitionIds.length - 1]; + definitionReadIndex += 1; return { from: () => ({ innerJoin: () => ({ where: () => ({ - limit: async () => (assetId === null ? [] : [{ assetId }]), + limit: async () => + assetId === null ? [] : [{ definitionId, assetId }], }), }), }), @@ -1187,6 +1201,77 @@ describe("deployAtHead — mcp credential bindings", () => { }); }); +// CL-6452: a folded run's deployed bytes carry per-run values, so their +// wire hash is unique to the run and the deploy's freeze ensures a fresh +// definition row over the agent's asset, then repoints the run at it. +// That row is a frozen deploy record — marking it keeps it out of the +// candidate set a launch resolves over, so the agent's own definition +// (the one an instructions save or skill pin refreezes in place) is what +// every later launch reads. +describe("deployAtHead — per-run definition records", () => { + const SOURCES = { + ok: true as const, + sources: [ + { + id: "off_1", + provider: "anthropic", + baseURL: "https://inference.invalid", + apiKey: "placeholder", + model: "claude-sonnet-5", + }, + ], + defaultSource: "off_1", + }; + + const PARAMS = { + tenantId: "ten_1", + instanceId: "run_origin1", + triggerAddress: "run_origin1@ten1.workbench.test", + principalId: "prn_1", + sessionId: "ses_1", + foldedBody: FOLDED_BODY, + launchLabel: "the invited agent", + }; + + function makeDeps(db: ReturnType) { + return { + db: db as never, + sessionService: createFakeSessionService(), + assetService: createFakeAssetService(), + sidecarRouter: createFakeSidecarRouter(), + eventCollectors: createFakeEventCollectors(), + toolGrantsForPins: () => [], + }; + } + + test("marks the row the deploy repointed the run at as a per-run record", async () => { + resolveDefinitionSourcesResult = SOURCES; + // The deploy minted a fresh definition for this run's bytes and + // repointed the run at it. + const db = createFakeDb("ast_agent", ["wfd_authored", "wfd_run_clone"]); + + await deployAtHead(makeDeps(db), PARAMS); + + const originUpdate = db.updated.find( + (row) => (row.values as { origin?: string }).origin !== undefined, + ); + expect(originUpdate?.values).toEqual({ origin: "run" }); + }); + + test("never demotes the agent's own definition when the deploy left the run on it", async () => { + resolveDefinitionSourcesResult = SOURCES; + const db = createFakeDb("ast_agent", ["wfd_authored", "wfd_authored"]); + + await deployAtHead(makeDeps(db), PARAMS); + + expect( + db.updated.find( + (row) => (row.values as { origin?: string }).origin !== undefined, + ), + ).toBeUndefined(); + }); +}); + describe("deployAtHead — run.grants production", () => { const SOURCES = { ok: true as const, diff --git a/packages/workflow-freeze/src/index.ts b/packages/workflow-freeze/src/index.ts index 31403e01a..6a1683e68 100644 --- a/packages/workflow-freeze/src/index.ts +++ b/packages/workflow-freeze/src/index.ts @@ -136,6 +136,9 @@ export async function projectAndWalkInertDefinition( * path. Persists through `createDbFrozenApprovalWriter`, so the ensure * and the stamp are one transaction and the row can never exist in the * half-frozen state a bare `ensureWorkflowDefinitionForAsset` leaves. + * The row keeps the schema's `origin: "authored"` default — only a + * folded run's own deploy demotes the sibling it mints to a per-run + * record (CL-6452, `@corbits/folded-runs`' `markRunDeployClone`). */ export async function freezeInertWorkflowDefinition( db: DBExecutor, diff --git a/packages/workflow-freeze/test/freeze.drizzle.test.ts b/packages/workflow-freeze/test/freeze.drizzle.test.ts index f13dbd9d7..1c7fddc68 100644 --- a/packages/workflow-freeze/test/freeze.drizzle.test.ts +++ b/packages/workflow-freeze/test/freeze.drizzle.test.ts @@ -177,4 +177,30 @@ describeIfDb("freezeInertWorkflowDefinition against Postgres", () => { expect(await loadFrozenWireProjection(db, definitionId)).not.toBeNull(); expect(await loadFrozenGrantSnapshot(db, definitionId)).not.toBeNull(); }); + + // CL-6452: a freeze produces a launch-authoritative definition. Only + // a folded run's own deploy demotes the sibling it mints to a per-run + // record (`@corbits/folded-runs`' `markRunDeployClone`), so a freeze — + // and any other deploy that ensures a definition — stays authored. + test("a freeze produces a launch-authoritative definition", async () => { + const assetId = await insertAsset("freeze-origin"); + const { definitionId } = await freezeInertWorkflowDefinition(db, { + assetId, + workflowJson: agentWorkflowJson("Authored instructions."), + }); + const authoredRow = await db.query.workflowDefinition.findFirst({ + where: eq(workflowDefinition.id, definitionId), + }); + expect(authoredRow?.origin).toBe("authored"); + + const sibling = await ensureWorkflowDefinitionForAsset(db, { + assetId, + wireHash: "another-deploy-hash", + }); + expect(sibling.definitionId).not.toBe(definitionId); + const siblingRow = await db.query.workflowDefinition.findFirst({ + where: eq(workflowDefinition.id, sibling.definitionId), + }); + expect(siblingRow?.origin).toBe("authored"); + }); }); diff --git a/vendor/intx/db/migrations/0086_workflow_definition_origin.sql b/vendor/intx/db/migrations/0086_workflow_definition_origin.sql new file mode 100644 index 000000000..1b08eaeca --- /dev/null +++ b/vendor/intx/db/migrations/0086_workflow_definition_origin.sql @@ -0,0 +1,27 @@ +-- WORKBENCH DELTA (see VENDORED.md, CL-6452): separate a workflow_definition +-- that is a definition in its own right from the per-run record of one deploy. +-- A folded run's deployed bytes carry per-run values, so their wire hash is +-- unique to the run and the deploy's freeze ensures a fresh same-named sibling +-- over the agent's asset. Launch resolution used to walk deployed siblings +-- newest-first by NAME, so the newest such record's stale frozen projection +-- shadowed every hub-authored edit (skill pins, instruction saves) after an +-- agent's first run. +-- +-- Backfill: an agent's own definition is always the first row minted for its +-- asset -- the create-path freeze (or a native deploy) precedes any run that +-- could deploy from it -- so the earliest row per (tenant_id, asset_id) keeps +-- the "authored" default and every later sibling becomes a per-run record. +-- Rows with no asset predate the asset-keyed model and have no per-run +-- siblings, so they are all authored. +ALTER TABLE "workflow_definition" ADD COLUMN "origin" text DEFAULT 'authored' NOT NULL;--> statement-breakpoint +UPDATE "workflow_definition" AS d +SET "origin" = 'run' +WHERE d."asset_id" IS NOT NULL + AND d."id" <> ( + SELECT first_row."id" + FROM "workflow_definition" AS first_row + WHERE first_row."tenant_id" = d."tenant_id" + AND first_row."asset_id" = d."asset_id" + ORDER BY first_row."created_at" ASC, first_row."id" ASC + LIMIT 1 + ); diff --git a/vendor/intx/db/migrations/meta/_journal.json b/vendor/intx/db/migrations/meta/_journal.json index 2636440ef..ff5dc847b 100644 --- a/vendor/intx/db/migrations/meta/_journal.json +++ b/vendor/intx/db/migrations/meta/_journal.json @@ -596,6 +596,13 @@ "when": 1787696000000, "tag": "0085_workflow_definition_version_wire_projection", "breakpoints": true + }, + { + "idx": 86, + "version": "7", + "when": 1787810000000, + "tag": "0086_workflow_definition_origin", + "breakpoints": true } ] } \ No newline at end of file diff --git a/vendor/intx/db/src/schema/workflow-definitions.ts b/vendor/intx/db/src/schema/workflow-definitions.ts index 1431081c1..24131abf2 100644 --- a/vendor/intx/db/src/schema/workflow-definitions.ts +++ b/vendor/intx/db/src/schema/workflow-definitions.ts @@ -52,6 +52,18 @@ export const workflowDefinition = pgTable( wireHash: text("wire_hash"), name: text("name").notNull(), description: text("description"), + // WORKBENCH DELTA (see VENDORED.md): whether this row is a + // definition in its own right or the per-run record of one deploy. + // "authored" (the default: a definition someone deployed or the hub + // froze) is launch-authoritative — the row an agent's edits refreeze + // in place. "run" is the sibling a folded run's deploy ensures over + // that same asset under the wire hash of its per-run rendered bytes: + // a frozen deploy record, never a launch candidate. Without the + // distinction, resolution fell back to matching on `name` and every + // run's clone shadowed the agent's own definition (CL-6452). + origin: text("origin", { enum: ["authored", "run"] }) + .notNull() + .default("authored"), // Grant requirements manifest, resolved at launch into materialized grants. // Validated as GrantRequirement[] at parse time. grantRequirements: jsonb("grant_requirements"),