From ab78be718aa8a36de8f91e3b5745095e19044cac Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 4 Jun 2026 00:54:05 -0700 Subject: [PATCH] =?UTF-8?q?feat(core):=20workflow=20switch/edit/delete=20r?= =?UTF-8?q?econciliation=20=E2=80=94=20no=20card=20left=20in=20an=20undefi?= =?UTF-8?q?ned=20column=20(U5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../__tests__/workflow-reconciliation.test.ts | 296 ++++++++++++++++++ packages/core/src/index.ts | 17 + packages/core/src/store.ts | 247 ++++++++++++++- .../core/src/workflow-definition-types.ts | 8 + packages/core/src/workflow-reconciliation.ts | 217 +++++++++++++ .../src/__tests__/workflow-routes.test.ts | 77 +++++ .../src/routes/register-workflow-routes.ts | 34 +- 7 files changed, 886 insertions(+), 10 deletions(-) create mode 100644 packages/core/src/__tests__/workflow-reconciliation.test.ts create mode 100644 packages/core/src/workflow-reconciliation.ts diff --git a/packages/core/src/__tests__/workflow-reconciliation.test.ts b/packages/core/src/__tests__/workflow-reconciliation.test.ts new file mode 100644 index 0000000000..a065cc7d45 --- /dev/null +++ b/packages/core/src/__tests__/workflow-reconciliation.test.ts @@ -0,0 +1,296 @@ +// @vitest-environment node +// +// U5: workflow lifecycle reconciliation — switch / edit / delete with live cards +// (R15, R20). Covers every U5 plan scenario: +// - switch with a same-id column preserves position; +// - switch without one re-homes to the new workflow's entry column AND fires +// the injected abort callback; +// - edit removing an occupied column blocks with per-column occupant counts; +// - the rehomeTo option saves + re-homes all occupants, one audit per card; +// - delete with occupants re-homes to the DEFAULT entry, clears selection, +// preserves task fields; +// - property-style invariant: after any switch/edit/delete sequence every +// task's column exists in its resolved workflow; +// - concurrent move-vs-delete under the task lock ends moved-then-re-homed or +// re-homed, never lost/undefined. + +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { createTaskStoreTestHarness } from "./store-test-helpers.js"; +import type { WorkflowIr } from "../workflow-ir-types.js"; +import { + OccupiedColumnsError, + setReconciliationAbort, + __resetReconciliationAbortForTests, + type ReconciliationAbortContext, +} from "../workflow-reconciliation.js"; +import { BUILTIN_CODING_WORKFLOW_IR } from "../builtin-coding-workflow-ir.js"; +import { resolveEntryColumnId } from "../workflow-reconciliation.js"; + +/** A v2 custom workflow with columns whose ids we control. `entryId` carries the + * intake flag; `cols` lists the column ids in order. Linear graph so it + * compiles. */ +function customIr(name: string, cols: string[], entryId: string): WorkflowIr { + return { + version: "v2", + name, + columns: cols.map((id) => ({ + id, + name: id, + traits: id === entryId ? [{ trait: "intake" }] : [], + })), + nodes: [ + { id: "start", kind: "start", column: entryId }, + { id: "work", kind: "prompt", column: cols[1] ?? entryId, config: { prompt: "do" } }, + { id: "end", kind: "end", column: cols[cols.length - 1] }, + ], + edges: [ + { from: "start", to: "work", condition: "success" }, + { from: "work", to: "end", condition: "success" }, + ], + }; +} + +describe("workflow reconciliation (U5)", () => { + const harness = createTaskStoreTestHarness(); + let store: ReturnType; + + beforeEach(async () => { + await harness.beforeEach(); + store = harness.store(); + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } }); + __resetReconciliationAbortForTests(); + }); + + afterEach(async () => { + __resetReconciliationAbortForTests(); + await harness.afterEach(); + }); + + /** Move a fresh task (starts in triage) to a default-workflow column. */ + async function seedInColumn(col: "triage" | "todo" | "in-progress"): Promise { + const task = await store.createTask({ description: `seed-${col}` }); + if (col === "triage") return task.id; + await store.moveTask(task.id, "todo", { moveSource: "user" }); + if (col === "todo") return task.id; + await store.moveTask(task.id, "in-progress", { moveSource: "user" }); + return task.id; + } + + it("entry column resolves to the intake-flagged column (default workflow = triage)", () => { + expect(resolveEntryColumnId(BUILTIN_CODING_WORKFLOW_IR)).toBe("triage"); + }); + + describe("(a) workflow switch", () => { + it("preserves position when the new workflow defines the same column id", async () => { + // Custom workflow that ALSO defines "todo" → same-id column, preserved. + const wf = await store.createWorkflowDefinition({ + name: "shares-todo", + ir: customIr("shares-todo", ["todo", "build", "done"], "todo"), + }); + const taskId = await seedInColumn("todo"); + + const result = await store.selectTaskWorkflowAndReconcile(taskId, wf.id); + + expect(result.reconciliation?.preserved).toBe(true); + expect(result.reconciliation?.toColumn).toBe("todo"); + const task = await store.getTask(taskId); + expect(task.column).toBe("todo"); + }); + + it("re-homes to the new workflow's entry column when the current column is absent, aborting first", async () => { + const aborts: ReconciliationAbortContext[] = []; + setReconciliationAbort((ctx) => { + aborts.push(ctx); + }); + // Custom workflow has none of the legacy column ids; entry = "intake". + const wf = await store.createWorkflowDefinition({ + name: "fresh", + ir: customIr("fresh", ["intake", "doing", "finished"], "intake"), + }); + const taskId = await seedInColumn("in-progress"); + + const result = await store.selectTaskWorkflowAndReconcile(taskId, wf.id); + + expect(result.reconciliation?.preserved).toBe(false); + expect(result.reconciliation?.toColumn).toBe("intake"); + const task = await store.getTask(taskId); + expect(task.column).toBe("intake"); + // Abort callback fired for the in-flight column before the re-home move. + expect(aborts).toHaveLength(1); + expect(aborts[0]).toMatchObject({ taskId, fromColumn: "in-progress", reason: "workflow-switch" }); + }); + + it("re-homes via the default no-op abort when no engine abort is wired", async () => { + const wf = await store.createWorkflowDefinition({ + name: "fresh2", + ir: customIr("fresh2", ["intake", "doing", "finished"], "intake"), + }); + const taskId = await seedInColumn("in-progress"); + const result = await store.selectTaskWorkflowAndReconcile(taskId, wf.id); + expect(result.reconciliation?.preserved).toBe(false); + expect((await store.getTask(taskId)).column).toBe("intake"); + }); + }); + + describe("(b) workflow edit removing an occupied column", () => { + it("blocks with per-column occupant counts when no rehomeTo is given", async () => { + const wf = await store.createWorkflowDefinition({ + name: "editable", + ir: customIr("editable", ["intake", "build", "done"], "intake"), + }); + const t1 = await store.createTask({ description: "t1" }); + const t2 = await store.createTask({ description: "t2" }); + await store.selectTaskWorkflowAndReconcile(t1.id, wf.id); // lands in intake + await store.selectTaskWorkflowAndReconcile(t2.id, wf.id); + // Move both into "build" so it's occupied. Custom adjacency is order-derived + // (intake↔build↔done), so intake→build is legal. + await store.moveTask(t1.id, "build", { moveSource: "user" }); + await store.moveTask(t2.id, "build", { moveSource: "user" }); + + // Edit that drops "build". + const nextIr = customIr("editable", ["intake", "done"], "intake"); + await expect(store.updateWorkflowDefinition(wf.id, { ir: nextIr })).rejects.toThrow( + OccupiedColumnsError, + ); + try { + await store.updateWorkflowDefinition(wf.id, { ir: nextIr }); + } catch (err) { + expect(err).toBeInstanceOf(OccupiedColumnsError); + const occ = (err as OccupiedColumnsError).occupancies; + expect(occ).toEqual([{ columnId: "build", count: 2 }]); + } + }); + + it("rehomeTo saves the edit and moves all occupants, emitting one audit per card", async () => { + const wf = await store.createWorkflowDefinition({ + name: "rehomeable", + ir: customIr("rehomeable", ["intake", "build", "done"], "intake"), + }); + const t1 = await store.createTask({ description: "t1" }); + const t2 = await store.createTask({ description: "t2" }); + await store.selectTaskWorkflowAndReconcile(t1.id, wf.id); + await store.selectTaskWorkflowAndReconcile(t2.id, wf.id); + await store.moveTask(t1.id, "build", { moveSource: "user" }); + await store.moveTask(t2.id, "build", { moveSource: "user" }); + + const nextIr = customIr("rehomeable", ["intake", "done"], "intake"); + const saved = await store.updateWorkflowDefinition(wf.id, { ir: nextIr, rehomeTo: "intake" }); + + // Saved IR no longer defines "build". + expect((saved.ir as { columns: { id: string }[] }).columns.map((c) => c.id)).toEqual([ + "intake", + "done", + ]); + expect((await store.getTask(t1.id)).column).toBe("intake"); + expect((await store.getTask(t2.id)).column).toBe("intake"); + }); + + it("does not block when the removed column has no occupants", async () => { + const wf = await store.createWorkflowDefinition({ + name: "empty-col", + ir: customIr("empty-col", ["intake", "build", "done"], "intake"), + }); + const nextIr = customIr("empty-col", ["intake", "done"], "intake"); + await expect(store.updateWorkflowDefinition(wf.id, { ir: nextIr })).resolves.toBeDefined(); + }); + }); + + describe("(c) workflow delete with occupants", () => { + it("re-homes occupants to the default entry, clears selection, preserves fields", async () => { + const wf = await store.createWorkflowDefinition({ + name: "doomed", + ir: customIr("doomed", ["intake", "build", "done"], "intake"), + }); + const t = await store.createTask({ description: "to-rehome" }); + await store.selectTaskWorkflowAndReconcile(t.id, wf.id); + await store.moveTask(t.id, "build", { moveSource: "user" }); + // Stamp a field we expect to survive the re-home (preserveProgress). + await store.updateTask(t.id, { summary: "keep me" }); + + await store.deleteWorkflowDefinition(wf.id); + + const task = await store.getTask(t.id); + // Re-homed to the default workflow's entry column (triage). + expect(task.column).toBe("triage"); + // Selection cleared → resolves to the default workflow now. + expect(store.getTaskWorkflowSelection(t.id)).toBeUndefined(); + // Field preserved. + expect(task.summary).toBe("keep me"); + }); + + it("built-in workflows remain undeletable", async () => { + await expect(store.deleteWorkflowDefinition("builtin:coding")).rejects.toThrow(); + }); + }); + + describe("property-style invariant: no card in an undefined column after any op", () => { + it("every task's column exists in its resolved workflow after switch/edit/delete", async () => { + const wfA = await store.createWorkflowDefinition({ + name: "A", + ir: customIr("A", ["intake", "mid", "out"], "intake"), + }); + const wfB = await store.createWorkflowDefinition({ + name: "B", + ir: customIr("B", ["start-b", "end-b"], "start-b"), + }); + + const ids: string[] = []; + for (let i = 0; i < 4; i++) { + const t = await store.createTask({ description: `prop-${i}` }); + ids.push(t.id); + } + // Switch all to A, scatter into A's columns. + for (const id of ids) await store.selectTaskWorkflowAndReconcile(id, wfA.id); + await store.moveTask(ids[1], "mid", { moveSource: "user" }); + await store.moveTask(ids[2], "mid", { moveSource: "user" }); + await store.moveTask(ids[2], "out", { moveSource: "user" }); + // Switch one to B (different ids → re-home to entry). + await store.selectTaskWorkflowAndReconcile(ids[3], wfB.id); + // Edit A removing "mid" with rehome. + await store.updateWorkflowDefinition(wfA.id, { + ir: customIr("A", ["intake", "out"], "intake"), + rehomeTo: "intake", + }); + // Delete B (re-homes ids[3] to default). + await store.deleteWorkflowDefinition(wfB.id); + + for (const id of ids) { + const task = await store.getTask(id); + const ir = (store as unknown as { resolveTaskWorkflowIrSync: (id: string) => WorkflowIr }) + .resolveTaskWorkflowIrSync(id); + const colIds = (ir as { columns: { id: string }[] }).columns.map((c) => c.id); + expect(colIds).toContain(task.column); + } + }); + }); + + describe("concurrent move-vs-delete under the task lock", () => { + it("ends moved-then-re-homed or re-homed, never lost/undefined", async () => { + const wf = await store.createWorkflowDefinition({ + name: "race", + ir: customIr("race", ["intake", "build", "done"], "intake"), + }); + const t = await store.createTask({ description: "racer" }); + await store.selectTaskWorkflowAndReconcile(t.id, wf.id); + await store.moveTask(t.id, "build", { moveSource: "user" }); + + // Fire a same-workflow move concurrently with the delete. Both serialize + // through the task lock; the task must end in a column defined by its + // resolved workflow (after delete: the default workflow), never undefined. + const movePromise = store + .moveTask(t.id, "done", { moveSource: "user" }) + .catch(() => undefined); + const deletePromise = store.deleteWorkflowDefinition(wf.id); + await Promise.all([movePromise, deletePromise]); + + const task = await store.getTask(t.id); + expect(task.column).toBeTruthy(); + // After delete the task resolves to the default workflow; its column must + // be one the default workflow defines. + const defaultCols = (BUILTIN_CODING_WORKFLOW_IR as { columns: { id: string }[] }).columns.map( + (c) => c.id, + ); + expect(defaultCols).toContain(task.column); + }); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index d02c509c2b..2830d15e41 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -132,6 +132,23 @@ export { } from "./workflow-transitions.js"; export type { ColumnAdjacency } from "./workflow-transitions.js"; export { isWorkflowColumnsEnabled } from "./workflow-columns-settings.js"; +// ── U5: workflow lifecycle reconciliation (switch / edit / delete) ─────────── +export { + OccupiedColumnsError, + resolveEntryColumnId, + resolveSwitchReconciliation, + computeRemovedOccupiedColumns, + assertRehomeTargetValid, + setReconciliationAbort, + runReconciliationAbort, + __resetReconciliationAbortForTests, +} from "./workflow-reconciliation.js"; +export type { + SwitchReconciliation, + ColumnOccupancy, + ReconciliationAbort, + ReconciliationAbortContext, +} from "./workflow-reconciliation.js"; export { readTransitionPending, writeTransitionPending, diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 061ebc4d12..8a58ff0fb3 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -10,6 +10,14 @@ import { DEFAULT_PROJECT_SETTINGS } from "./settings-schema.js"; import { parseWorkflowIr, serializeWorkflowIr } from "./workflow-ir.js"; import { isWorkflowColumnsEnabled } from "./workflow-columns-settings.js"; import { resolveAllowedColumns, workflowHasColumn } from "./workflow-transitions.js"; +import { + OccupiedColumnsError, + assertRehomeTargetValid, + computeRemovedOccupiedColumns, + resolveEntryColumnId, + resolveSwitchReconciliation, + runReconciliationAbort, +} from "./workflow-reconciliation.js"; import { type DefaultWorkflowMoveContext, applyDefaultWorkflowMoveEffects, @@ -1106,6 +1114,18 @@ interface MoveTaskOptions { * `moveSource === "engine"` plus `skipMergeBlocker`. */ bypassGuards?: boolean; + /** + * U5 (R15/R20): a workflow-reconciliation re-home move (switch/edit/delete). + * Unlike `bypassGuards` (which skips trait guards but still enforces the + * column-graph adjacency, so the U4 parity matrix is unaffected), a recovery + * re-home must reach the new workflow's entry column from ANY current column — + * a card that would otherwise be stranded in a column its (new) workflow does + * not define. So this additionally skips the adjacency check (step 2). The + * structural unknown-column check (step 1) and the in-txn capacity check + * (KTD-10) still apply. Engine-internal only: never forwarded from an HTTP + * endpoint. When set, implies `bypassGuards`. + */ + recoveryRehome?: boolean; } interface MoveTaskInternalOptions { @@ -5636,7 +5656,8 @@ export class TaskStore extends EventEmitter { // capacity check is not a guard (U6 fills the enforcement; U4 leaves a // pass-through slot). An explicit option value wins; otherwise derive it. const bypassGuards = - options?.bypassGuards ?? (moveSource === "engine" || options?.skipMergeBlocker === true); + options?.recoveryRehome === true || + (options?.bypassGuards ?? (moveSource === "engine" || options?.skipMergeBlocker === true)); const workflowIr: WorkflowIr | undefined = useWorkflow ? this.resolveTaskWorkflowIrSync(id) : undefined; @@ -5715,9 +5736,11 @@ export class TaskStore extends EventEmitter { } // 2. Column-graph adjacency. For the default workflow this reproduces // VALID_TRANSITIONS verbatim (resolveAllowedColumns); the - // transition-parity suite machine-checks the equivalence. + // transition-parity suite machine-checks the equivalence. A U5 recovery + // re-home (recoveryRehome) skips this so a stranded card can reach its + // new workflow's entry column from any current column. const allowed = resolveAllowedColumns(workflowIr, fromColumn); - if (!allowed.includes(toColumn)) { + if (options?.recoveryRehome !== true && !allowed.includes(toColumn)) { throw new TransitionRejectionError( makeTransitionRejection( "guard-rejected", @@ -11517,7 +11540,39 @@ ${stepsSection}`; updates: WorkflowDefinitionUpdate, ): Promise { if (isBuiltinWorkflowId(id)) throw new Error("Built-in workflows cannot be edited"); - return this.withConfigLock(async () => { + // U5 (R20): flag-ON edits that remove an occupied column block with a typed + // OccupiedColumnsError unless `rehomeTo` is supplied. Computed before taking + // the config lock (pure DB reads) so the lock body stays focused. + const flagOn = await this.workflowColumnsFlagOn(); + let pendingRehome: { rehomeTo: string; occupantTaskIds: string[] } | undefined; + if (flagOn && updates.ir !== undefined) { + const existingForCheck = await this.getWorkflowDefinition(id); + if (!existingForCheck) throw new Error(`Workflow '${id}' not found`); + const nextIrForCheck = parseWorkflowIr(updates.ir); + const occupantsByColumn = this.occupantsByColumnForWorkflow(id, false); + const removed = computeRemovedOccupiedColumns( + existingForCheck.ir, + nextIrForCheck, + occupantsByColumn, + ); + if (removed.length > 0) { + if (updates.rehomeTo === undefined) { + throw new OccupiedColumnsError(id, removed); + } + assertRehomeTargetValid(nextIrForCheck, updates.rehomeTo); + // Collect the occupant task ids of the removed columns to re-home AFTER + // the IR save commits, so the cards land in a column the new IR defines. + const removedSet = new Set(removed.map((r) => r.columnId)); + const occupantTaskIds = this.listWorkflowOccupantTaskIds(id, false).filter((taskId) => { + const row = this.db.prepare(`SELECT "column" AS column FROM tasks WHERE id = ?`).get(taskId) as + | { column: string } + | undefined; + return row ? removedSet.has(row.column) : false; + }); + pendingRehome = { rehomeTo: updates.rehomeTo, occupantTaskIds }; + } + } + const saved = await this.withConfigLock(async () => { const existing = await this.getWorkflowDefinition(id); if (!existing) throw new Error(`Workflow '${id}' not found`); @@ -11550,6 +11605,18 @@ ${stepsSection}`; this.db.bumpLastModified(); return next; }); + + // U5 (R20): now that the new IR is committed, re-home the occupants of the + // removed columns into `rehomeTo` (one audit event per card). Done outside + // the config lock; each rehome takes its own task lock via moveTask. + if (pendingRehome) { + for (const taskId of pendingRehome.occupantTaskIds) { + await this.rehomeOccupant(taskId, pendingRehome.rehomeTo, "workflow-edit-rehome", { + workflowId: id, + }); + } + } + return saved; } /** Delete a workflow definition, cascading to per-task selections, their @@ -11557,6 +11624,11 @@ ${stepsSection}`; * not exist. */ async deleteWorkflowDefinition(id: string): Promise { if (isBuiltinWorkflowId(id)) throw new Error("Built-in workflows cannot be deleted"); + // U5 (R20): flag-ON, capture the occupant task ids BEFORE the cascade clears + // their selection rows, so we can re-home them to the DEFAULT workflow's + // entry column once their selection resolves back to the default (KTD-1). + const flagOn = await this.workflowColumnsFlagOn(); + const occupantTaskIds = flagOn ? this.listWorkflowOccupantTaskIds(id, false) : []; const deleted = this.db.prepare("DELETE FROM workflows WHERE id = ?").run(id) as { changes?: number }; if ((deleted.changes || 0) === 0) { throw new Error(`Workflow '${id}' not found`); @@ -11600,6 +11672,133 @@ ${stepsSection}`; } if (selections.length > 0) this.workflowStepsCache = null; this.db.bumpLastModified(); + + // U5 (R20) delete reconciliation: re-home each occupant to the default + // workflow's entry column. Their selection rows are already cleared above, + // so they now resolve to the built-in default workflow (KTD-1); the re-home + // move preserves task fields (preserveProgress) and emits one audit per card. + if (flagOn && occupantTaskIds.length > 0) { + const defaultEntry = resolveEntryColumnId(BUILTIN_CODING_WORKFLOW_IR); + if (defaultEntry) { + for (const taskId of occupantTaskIds) { + await this.rehomeOccupant(taskId, defaultEntry, "workflow-delete", { workflowId: id }); + } + } + } + } + + // ── U5: workflow lifecycle reconciliation (switch / edit / delete) ────────── + // + // These helpers are only consulted when the `workflowColumns` flag is ON; the + // flag-OFF CRUD paths above keep their exact current behavior. Re-homing moves + // always route through `moveTask` with `moveSource: "engine"` + `bypassGuards` + // (a recovery-class move, KTD-9) — never a raw column write — so capacity + // (KTD-10) and the single transition authority (KTD-3) are honored. + + /** True when the `workflowColumns` flag is ON (merged global + project). */ + private async workflowColumnsFlagOn(): Promise { + return isWorkflowColumnsEnabled(await this.getSettingsFast()); + } + + /** The active (non-deleted) task ids currently selecting `workflowId`. A + * built-in/default workflow additionally owns every task with NO selection + * row (null selection resolves to the default workflow, KTD-1). */ + private listWorkflowOccupantTaskIds(workflowId: string, includeNullSelection: boolean): string[] { + const ids: string[] = []; + const selected = this.db + .prepare( + `SELECT s.taskId AS taskId FROM task_workflow_selection s + JOIN tasks t ON t.id = s.taskId + WHERE s.workflowId = ? AND t."deletedAt" IS NULL`, + ) + .all(workflowId) as Array<{ taskId: string }>; + for (const row of selected) ids.push(row.taskId); + if (includeNullSelection) { + const unselected = this.db + .prepare( + `SELECT t.id AS id FROM tasks t + WHERE t."deletedAt" IS NULL + AND NOT EXISTS (SELECT 1 FROM task_workflow_selection s WHERE s.taskId = t.id)`, + ) + .all() as Array<{ id: string }>; + for (const row of unselected) ids.push(row.id); + } + return ids; + } + + /** Map column id → occupant count for the tasks selecting `workflowId` + * (plus null-selection tasks when `includeNullSelection`). */ + private occupantsByColumnForWorkflow( + workflowId: string, + includeNullSelection: boolean, + ): Map { + const counts = new Map(); + for (const taskId of this.listWorkflowOccupantTaskIds(workflowId, includeNullSelection)) { + const row = this.db.prepare(`SELECT "column" AS column FROM tasks WHERE id = ?`).get(taskId) as + | { column: string } + | undefined; + if (!row) continue; + counts.set(row.column, (counts.get(row.column) ?? 0) + 1); + } + return counts; + } + + /** Re-home a single occupant to `targetColumn` via an engine-sourced, + * guard-bypassing recovery move, aborting in-flight work first, and emit one + * audit event. Best-effort per card: a failure is audited and skipped so one + * stuck card never blocks the rest of the batch. */ + private async rehomeOccupant( + taskId: string, + targetColumn: string, + reason: "workflow-switch" | "workflow-delete" | "workflow-edit-rehome", + metadata: Record, + ): Promise { + const current = this.readTaskFromDb(taskId, { includeDeleted: false }); + if (!current) return; + const fromColumn = current.column; + if (fromColumn === targetColumn) { + // Already in the target column — nothing to move, but still record the + // reconciliation decision for audit traceability. + this.recordRunAuditEvent({ + taskId, + agentId: "system", + runId: `workflow-reconcile-${reason}-${taskId}-${Date.now()}`, + domain: "database", + mutationType: "task:workflow-reconcile", + target: taskId, + metadata: { ...metadata, reason, fromColumn, toColumn: targetColumn, moved: false }, + }); + return; + } + const abortRan = await runReconciliationAbort({ taskId, fromColumn, reason }); + let moved = false; + let error: string | undefined; + try { + // Recovery-class move: engine source + bypassGuards (KTD-9). preserveProgress + // keeps the task's fields intact (R20 delete semantics). Capacity (KTD-10) is + // NOT bypassed — a full target column rejects, which we audit and skip. + await this.moveTask(taskId, targetColumn as Column, { + moveSource: "engine", + bypassGuards: true, + recoveryRehome: true, + preserveProgress: true, + preserveResumeState: true, + preserveWorktree: true, + allowDirectInReviewMove: true, + }); + moved = true; + } catch (err) { + error = err instanceof Error ? err.message : String(err); + } + this.recordRunAuditEvent({ + taskId, + agentId: "system", + runId: `workflow-reconcile-${reason}-${taskId}-${Date.now()}`, + domain: "database", + mutationType: "task:workflow-reconcile", + target: taskId, + metadata: { ...metadata, reason, fromColumn, toColumn: targetColumn, abortRan, moved, error }, + }); } // ── Workflow selection (resolves a workflow to enabledWorkflowSteps) ──── @@ -11845,6 +12044,46 @@ ${stepsSection}`; }); } + /** + * U5 (R20) workflow switch: select a workflow for a task and, when the + * `workflowColumns` flag is ON, reconcile the card's board column against the + * NEW workflow. Same-id column preserves position; otherwise the card re-homes + * to the new workflow's entry (intake-flagged, else first) column, aborting + * in-flight processing first (KTD-9). Returns the materialized step ids plus + * the switch outcome so the dashboard can surface the re-home. + * + * Reconciliation runs AFTER `selectTaskWorkflow` releases the per-task lock + * (moveTask takes its own lock; the per-task lock is non-reentrant). + */ + async selectTaskWorkflowAndReconcile( + taskId: string, + workflowId: string, + ): Promise<{ + enabledWorkflowSteps: string[]; + reconciliation?: { preserved: boolean; fromColumn: string; toColumn: string }; + }> { + const enabledWorkflowSteps = await this.selectTaskWorkflow(taskId, workflowId); + if (!(await this.workflowColumnsFlagOn())) { + return { enabledWorkflowSteps }; + } + const newIr = this.resolveTaskWorkflowIrSync(taskId); + const current = this.readTaskFromDb(taskId, { includeDeleted: false }); + if (!current) return { enabledWorkflowSteps }; + const fromColumn = current.column; + const decision = resolveSwitchReconciliation(newIr, fromColumn); + if (!decision.preserved && decision.targetColumn !== fromColumn) { + await this.rehomeOccupant(taskId, decision.targetColumn, "workflow-switch", { workflowId }); + } + return { + enabledWorkflowSteps, + reconciliation: { + preserved: decision.preserved, + fromColumn, + toColumn: decision.targetColumn, + }, + }; + } + /** Clear a task's workflow selection and its enabled steps. */ async clearTaskWorkflowSelection(taskId: string): Promise { await this.withTaskLock(taskId, async () => { diff --git a/packages/core/src/workflow-definition-types.ts b/packages/core/src/workflow-definition-types.ts index 3383815a9d..026f544684 100644 --- a/packages/core/src/workflow-definition-types.ts +++ b/packages/core/src/workflow-definition-types.ts @@ -40,4 +40,12 @@ export interface WorkflowDefinitionUpdate { description?: string; ir?: WorkflowIr; layout?: Record; + /** + * U5 (R20): when an IR update removes a column that still holds cards, the + * update is blocked with a typed {@link import("./workflow-reconciliation.js").OccupiedColumnsError} + * unless `rehomeTo` is supplied — an explicit "save and re-home occupants to + * column X" target. The target must survive in the new IR. Only consulted when + * the `workflowColumns` flag is ON. + */ + rehomeTo?: string; } diff --git a/packages/core/src/workflow-reconciliation.ts b/packages/core/src/workflow-reconciliation.ts new file mode 100644 index 0000000000..5c0a176371 --- /dev/null +++ b/packages/core/src/workflow-reconciliation.ts @@ -0,0 +1,217 @@ +/** + * Workflow lifecycle reconciliation (U5, R15/R20). + * + * Defines the policy for every case where a card's column could stop existing + * under it: + * + * (a) workflow SWITCH — the task's selection changes. If the new workflow + * defines a column with the task's current column id, position is + * preserved; otherwise the card re-homes to the new workflow's entry + * (intake-flagged, falling back to the first) column. In-flight processing + * is aborted first via an injected abort callback (engine wires the real + * abort; core ships a safe no-op default + audit entry so core stays + * engine-free). + * + * (b) workflow EDIT removing an occupied column — the update path blocks with + * a typed {@link OccupiedColumnsError} listing per-column occupant counts. + * An explicit `rehomeTo` option allows the save plus re-home of every + * occupant (one audit event per card). + * + * (c) workflow DELETE with occupants — built-ins stay blocked; custom + * workflows re-home occupants to the DEFAULT workflow's entry column, + * clear their selection rows, and preserve task fields (preserveProgress + * semantics), one audit event per card. + * + * Re-homing moves go through `moveTask` with `moveSource: "engine"` + + * `bypassGuards` (a recovery-class move, KTD-9) — never a raw column write — so + * capacity (KTD-10) and the single transition authority (KTD-3) are honored. + * + * This module is pure policy + a DI seam. The store (and dashboard routes via + * the store) own the actual DB reads/writes and the `moveTask` call; this module + * supplies the column-resolution rules and the abort indirection so the policy + * is independently testable and reused identically across switch/edit/delete. + */ + +import type { WorkflowIr, WorkflowIrV2, WorkflowIrColumn } from "./workflow-ir-types.js"; +import { resolveColumnFlags } from "./trait-registry.js"; +import { workflowHasColumn } from "./workflow-transitions.js"; + +// ── Entry-column resolution ────────────────────────────────────────────────── + +/** The v2 columns of an IR, or `[]` when (defensively) absent. */ +function columnsOf(ir: WorkflowIr): WorkflowIrColumn[] { + const v2 = ir as WorkflowIrV2; + return Array.isArray(v2.columns) ? v2.columns : []; +} + +/** + * The entry column id for a workflow: the intake-flagged column (resolved via + * the trait registry's effective-flag merge), falling back to the FIRST + * declared column. Returns `undefined` only when the workflow declares no + * columns at all (should never happen post-parse) — callers treat that as a + * non-reconcilable workflow and leave the card where it is. + */ +export function resolveEntryColumnId(ir: WorkflowIr): string | undefined { + const columns = columnsOf(ir); + if (columns.length === 0) return undefined; + for (const column of columns) { + if (resolveColumnFlags(column).intake) return column.id; + } + return columns[0].id; +} + +// ── (a) Workflow switch ────────────────────────────────────────────────────── + +/** The outcome of resolving where a card lands when its workflow switches. */ +export interface SwitchReconciliation { + /** The column the card should occupy under the new workflow. */ + targetColumn: string; + /** True when the card's current column id exists in the new workflow and was + * therefore preserved; false when it was re-homed to the entry column. */ + preserved: boolean; + /** The entry column the card would re-home to (always resolved, for audit). */ + entryColumn: string | undefined; +} + +/** + * Resolve where a card currently in `currentColumn` lands under `newWorkflowIr`. + * Same-id columns preserve position; otherwise the card re-homes to the new + * workflow's entry column. Pure — the caller performs the abort + move. + */ +export function resolveSwitchReconciliation( + newWorkflowIr: WorkflowIr, + currentColumn: string, +): SwitchReconciliation { + const entryColumn = resolveEntryColumnId(newWorkflowIr); + if (workflowHasColumn(newWorkflowIr, currentColumn)) { + return { targetColumn: currentColumn, preserved: true, entryColumn }; + } + // No same-id column: re-home to the entry column. When the new workflow + // declares no columns at all (entryColumn undefined), leave the card where it + // is rather than strand it in nowhere. + return { + targetColumn: entryColumn ?? currentColumn, + preserved: false, + entryColumn, + }; +} + +// ── (b) Workflow edit removing an occupied column ──────────────────────────── + +/** Per-column occupant count for a blocked edit/delete. */ +export interface ColumnOccupancy { + columnId: string; + count: number; +} + +/** + * Thrown by the store's update path (and surfaced as a structured 409 by the + * dashboard) when a workflow edit would remove one or more columns that still + * hold cards, and no `rehomeTo` was supplied. Carries the per-column occupant + * counts so the surface can prompt for a re-home target. + */ +export class OccupiedColumnsError extends Error { + readonly workflowId: string; + readonly occupancies: ColumnOccupancy[]; + constructor(workflowId: string, occupancies: ColumnOccupancy[]) { + const summary = occupancies + .map((o) => `${o.columnId} (${o.count})`) + .join(", "); + super( + `Workflow '${workflowId}' edit removes occupied column(s): ${summary}. ` + + `Re-home the occupants (rehomeTo) or move them out first.`, + ); + this.name = "OccupiedColumnsError"; + this.workflowId = workflowId; + this.occupancies = occupancies; + } +} + +/** + * Compute which currently-occupied columns would be removed by replacing the + * existing IR with `nextIr`. `occupantsByColumn` maps a column id to the number + * of cards currently in it (under this workflow). Returns one entry per removed + * column that still has occupants, in the existing IR's column order. + */ +export function computeRemovedOccupiedColumns( + existingIr: WorkflowIr, + nextIr: WorkflowIr, + occupantsByColumn: Map, +): ColumnOccupancy[] { + const nextIds = new Set(columnsOf(nextIr).map((c) => c.id)); + const removed: ColumnOccupancy[] = []; + for (const column of columnsOf(existingIr)) { + if (nextIds.has(column.id)) continue; + const count = occupantsByColumn.get(column.id) ?? 0; + if (count > 0) removed.push({ columnId: column.id, count }); + } + return removed; +} + +/** + * Validate that `rehomeTo` (when supplied for an edit that removes occupied + * columns) names a column that survives in `nextIr`. Throws when it does not, so + * occupants are never re-homed into a column that won't exist either. + */ +export function assertRehomeTargetValid(nextIr: WorkflowIr, rehomeTo: string): void { + if (!workflowHasColumn(nextIr, rehomeTo)) { + throw new OccupiedColumnsError( + (nextIr as WorkflowIrV2).name ?? "(unknown)", + [], + ); + } +} + +// ── Abort-on-switch DI seam (core stays engine-free) ───────────────────────── +// +// A workflow switch must abort the card's in-flight processing BEFORE the move +// (mirroring abort-on-exit, KTD-9). Aborting touches engine machinery (sessions +// / leases), which core cannot import. The engine wires its abort in via +// `setReconciliationAbort` (mirrors `setCreateFnAgent`); when unset (isolated +// core tests, or engine not loaded) the default is a safe no-op that records an +// audit breadcrumb so the bypass is visible — degraded, not crashed. + +/** What the store passes to the abort callback so the engine can locate the + * session/lease to abort and the store can record audit. */ +export interface ReconciliationAbortContext { + taskId: string; + fromColumn: string; + reason: "workflow-switch" | "workflow-delete" | "workflow-edit-rehome"; +} + +/** The injected abort implementation. Returns nothing; failures must not throw + * (a failed abort degrades to an audit entry — it never strands the card). */ +export type ReconciliationAbort = (ctx: ReconciliationAbortContext) => void | Promise; + +let reconciliationAbort: ReconciliationAbort | undefined; + +/** + * Wire the engine's abort implementation into core. Called by `@fusion/engine` + * at module load; tests may register a stub (or leave it unset for the no-op). + * Passing `undefined` restores the default no-op. + */ +export function setReconciliationAbort(fn: ReconciliationAbort | undefined): void { + reconciliationAbort = fn; +} + +/** + * Run the wired abort, or the safe default no-op when none is registered. Always + * resolves (swallows abort errors) so reconciliation never wedges on a failing + * abort. Returns `true` when a real abort ran, `false` for the default no-op — + * the store records the appropriate audit either way. + */ +export async function runReconciliationAbort(ctx: ReconciliationAbortContext): Promise { + if (!reconciliationAbort) return false; + try { + await reconciliationAbort(ctx); + } catch { + // A failed abort must not strand the card — the caller still re-homes it, + // and records a degraded-abort audit. Swallow here. + } + return true; +} + +/** Test-only: reset the wired abort to the default no-op. */ +export function __resetReconciliationAbortForTests(): void { + reconciliationAbort = undefined; +} diff --git a/packages/dashboard/src/__tests__/workflow-routes.test.ts b/packages/dashboard/src/__tests__/workflow-routes.test.ts index 25a4fba391..944e7bb45a 100644 --- a/packages/dashboard/src/__tests__/workflow-routes.test.ts +++ b/packages/dashboard/src/__tests__/workflow-routes.test.ts @@ -230,4 +230,81 @@ describe("workflow routes (U4)", () => { // consumes the marker itself on re-run. expect(detail.pausedReason).toBe("workflow-await-input:ask: please confirm"); }); + + // ── U5: lifecycle reconciliation surfaced through the routes (flag ON) ─────── + describe("U5 reconciliation (workflowColumns flag ON)", () => { + /** A v2 custom workflow with controlled column ids; linear so it compiles. */ + function customV2(name: string, cols: string[]): WorkflowIr { + const entry = cols[0]; + return { + version: "v2", + name, + columns: cols.map((id) => ({ id, name: id, traits: id === entry ? [{ trait: "intake" }] : [] })), + nodes: [ + { id: "start", kind: "start", column: entry }, + { id: "work", kind: "prompt", column: cols[1] ?? entry, config: { prompt: "do" } }, + { id: "end", kind: "end", column: cols[cols.length - 1] }, + ], + edges: [ + { from: "start", to: "work", condition: "success" }, + { from: "work", to: "end", condition: "success" }, + ], + }; + } + + beforeEach(async () => { + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } }); + }); + + it("PATCH removing an occupied column 409s with per-column occupant counts", async () => { + const wf = await post("/api/workflows", { name: "edit", ir: customV2("edit", ["intake", "build", "done"]) }); + const wfId = (wf.body as { id: string }).id; + const t = await store.createTask({ description: "occ" }); + await store.selectTaskWorkflowAndReconcile(t.id, wfId); + await store.moveTask(t.id, "build", { moveSource: "user" }); + + const res = await request( + app, + "PATCH", + `/api/workflows/${wfId}`, + JSON.stringify({ ir: customV2("edit", ["intake", "done"]) }), + { "content-type": "application/json" }, + ); + expect(res.status).toBe(409); + const details = (res.body as { details?: { occupancies?: Array<{ columnId: string; count: number }> } }).details; + expect(details?.occupancies).toEqual([{ columnId: "build", count: 1 }]); + }); + + it("PATCH with rehomeTo saves and re-homes occupants", async () => { + const wf = await post("/api/workflows", { name: "rehome", ir: customV2("rehome", ["intake", "build", "done"]) }); + const wfId = (wf.body as { id: string }).id; + const t = await store.createTask({ description: "occ" }); + await store.selectTaskWorkflowAndReconcile(t.id, wfId); + await store.moveTask(t.id, "build", { moveSource: "user" }); + + const res = await request( + app, + "PATCH", + `/api/workflows/${wfId}`, + JSON.stringify({ ir: customV2("rehome", ["intake", "done"]), rehomeTo: "intake" }), + { "content-type": "application/json" }, + ); + expect(res.status).toBe(200); + expect((await store.getTask(t.id)).column).toBe("intake"); + }); + + it("PUT selection re-homes the card and returns the reconciliation outcome", async () => { + const wf = await post("/api/workflows", { name: "sw", ir: customV2("sw", ["intake", "doing", "done"]) }); + const wfId = (wf.body as { id: string }).id; + const t = await store.createTask({ description: "switcher" }); + await store.moveTask(t.id, "todo", { moveSource: "user" }); + + const res = await put(`/api/tasks/${t.id}/workflow`, { workflowId: wfId }); + expect(res.status).toBe(200); + const recon = (res.body as { reconciliation?: { preserved: boolean; toColumn: string } }).reconciliation; + expect(recon?.preserved).toBe(false); + expect(recon?.toColumn).toBe("intake"); + expect((await store.getTask(t.id)).column).toBe("intake"); + }); + }); }); diff --git a/packages/dashboard/src/routes/register-workflow-routes.ts b/packages/dashboard/src/routes/register-workflow-routes.ts index e93df43dfb..eedfae4ec0 100644 --- a/packages/dashboard/src/routes/register-workflow-routes.ts +++ b/packages/dashboard/src/routes/register-workflow-routes.ts @@ -1,6 +1,6 @@ import type { WorkflowIr } from "@fusion/core"; -import { WorkflowCompileError, WorkflowIrError, compileWorkflowToSteps } from "@fusion/core"; -import { ApiError, badRequest, notFound } from "../api-error.js"; +import { OccupiedColumnsError, WorkflowCompileError, WorkflowIrError, compileWorkflowToSteps } from "@fusion/core"; +import { ApiError, badRequest, conflict, notFound } from "../api-error.js"; import type { ApiRoutesContext } from "./types.js"; /** @@ -65,17 +65,32 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { router.patch("/workflows/:id", async (req, res) => { try { const { store } = await getProjectContext(req); - const { name, description, ir, layout } = req.body ?? {}; + const { name, description, ir, layout, rehomeTo } = req.body ?? {}; if (name !== undefined && (typeof name !== "string" || !name.trim())) { throw badRequest("name must be a non-empty string"); } if (ir !== undefined && (typeof ir !== "object" || ir === null)) { throw badRequest("ir must be a workflow graph object"); } - const updated = await store.updateWorkflowDefinition(req.params.id, { name, description, ir, layout }); + if (rehomeTo !== undefined && typeof rehomeTo !== "string") { + throw badRequest("rehomeTo must be a string column id"); + } + const updated = await store.updateWorkflowDefinition(req.params.id, { + name, + description, + ir, + layout, + ...(rehomeTo !== undefined ? { rehomeTo } : {}), + }); res.json(updated); } catch (err: unknown) { if (err instanceof ApiError) throw err; + // U5 (R20): a flag-ON edit removing an occupied column blocks with a typed + // error. Surface it as a structured 409 carrying the per-column occupant + // counts so the client can prompt for a `rehomeTo` target and retry. + if (err instanceof OccupiedColumnsError) { + throw conflict(err.message, { workflowId: err.workflowId, occupancies: err.occupancies }); + } if (err instanceof WorkflowIrError) throw badRequest(err.message); if (err instanceof Error && /not found/i.test(err.message)) throw notFound(err.message); rethrowAsApiError(err); @@ -149,8 +164,15 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { throw badRequest("workflowId must be a string or null"); } let enabledWorkflowSteps: string[] = []; + // U5 (R20) switch reconciliation: when the workflowColumns flag is ON, the + // store re-homes the card to the new workflow's entry column (aborting + // in-flight work first) unless the new workflow defines its current column. + // The re-home outcome rides on the response so the UI can reflect the move. + let reconciliation: { preserved: boolean; fromColumn: string; toColumn: string } | undefined; try { - enabledWorkflowSteps = await store.selectTaskWorkflow(req.params.taskId, workflowId); + const result = await store.selectTaskWorkflowAndReconcile(req.params.taskId, workflowId); + enabledWorkflowSteps = result.enabledWorkflowSteps; + reconciliation = result.reconciliation; } catch (selectErr: unknown) { if (selectErr instanceof WorkflowCompileError || selectErr instanceof WorkflowIrError) { throw new ApiError(422, selectErr.message); @@ -160,7 +182,7 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { } throw selectErr; } - res.json({ workflowId, enabledWorkflowSteps }); + res.json({ workflowId, enabledWorkflowSteps, ...(reconciliation ? { reconciliation } : {}) }); } catch (err: unknown) { if (err instanceof ApiError) throw err; rethrowAsApiError(err);