diff --git a/packages/core/src/__tests__/migration-workflow-columns.test.ts b/packages/core/src/__tests__/migration-workflow-columns.test.ts index c92259fcf7..84669edf16 100644 --- a/packages/core/src/__tests__/migration-workflow-columns.test.ts +++ b/packages/core/src/__tests__/migration-workflow-columns.test.ts @@ -272,6 +272,102 @@ describe("Residual B: getBranchProgressByTask reads workflow_run_branches", () = }); }); +describe("#1407/#1412/#1413: workflow_run_branches persistence + latest-run JOIN + prune", () => { + const harness = createTaskStoreTestHarness(); + let store: ReturnType; + + beforeEach(async () => { + await harness.beforeEach(); + store = harness.store(); + }); + afterEach(async () => { + await harness.afterEach(); + }); + + type BranchStore = { + saveWorkflowRunBranch(state: { + taskId: string; runId: string; branchId: string; currentNodeId: string; status: string; + }): void; + loadWorkflowRunBranches(taskId: string, runId: string): Array<{ + taskId: string; runId: string; branchId: string; currentNodeId: string; status: string; + }>; + clearWorkflowRunBranches(taskId: string, keepRunId: string): void; + }; + const bs = (): BranchStore => store as unknown as BranchStore; + + function rawCount(taskId: string): number { + const db = (store as unknown as { db: { prepare: (s: string) => { get: (...a: unknown[]) => unknown } } }).db; + const row = db + .prepare("SELECT COUNT(*) AS c FROM workflow_run_branches WHERE taskId = ?") + .get(taskId) as { c: number }; + return row.c; + } + + it("saveWorkflowRunBranch upserts one row per (taskId, runId, branchId) keyed by currentNodeId", async () => { + const t = await store.createTask({ description: "fanout" }); + bs().saveWorkflowRunBranch({ taskId: t.id, runId: "r1", branchId: "b1", currentNodeId: "n1", status: "running" }); + bs().saveWorkflowRunBranch({ taskId: t.id, runId: "r1", branchId: "b1", currentNodeId: "n2", status: "completed" }); + bs().saveWorkflowRunBranch({ taskId: t.id, runId: "r1", branchId: "b2", currentNodeId: "n3", status: "running" }); + + // b1 overwrote in place (still one row), b2 added — 2 rows total. + expect(rawCount(t.id)).toBe(2); + const loaded = bs().loadWorkflowRunBranches(t.id, "r1"); + const b1 = loaded.find((s) => s.branchId === "b1"); + expect(b1?.currentNodeId).toBe("n2"); + expect(b1?.status).toBe("completed"); + }); + + it("loadWorkflowRunBranches returns only the requested run", async () => { + const t = await store.createTask({ description: "fanout" }); + bs().saveWorkflowRunBranch({ taskId: t.id, runId: "r1", branchId: "b1", currentNodeId: "n1", status: "completed" }); + bs().saveWorkflowRunBranch({ taskId: t.id, runId: "r2", branchId: "b1", currentNodeId: "n9", status: "running" }); + expect(bs().loadWorkflowRunBranches(t.id, "r1").length).toBe(1); + expect(bs().loadWorkflowRunBranches(t.id, "r1")[0]?.currentNodeId).toBe("n1"); + }); + + it("clearWorkflowRunBranches prunes all runs except the kept one (#1412)", async () => { + const t = await store.createTask({ description: "fanout" }); + bs().saveWorkflowRunBranch({ taskId: t.id, runId: "old-1", branchId: "b1", currentNodeId: "n1", status: "completed" }); + bs().saveWorkflowRunBranch({ taskId: t.id, runId: "old-2", branchId: "b1", currentNodeId: "n1", status: "completed" }); + bs().saveWorkflowRunBranch({ taskId: t.id, runId: "keep", branchId: "b1", currentNodeId: "n5", status: "running" }); + expect(rawCount(t.id)).toBe(3); + + bs().clearWorkflowRunBranches(t.id, "keep"); + expect(rawCount(t.id)).toBe(1); + expect(bs().loadWorkflowRunBranches(t.id, "keep").length).toBe(1); + }); + + it("getBranchProgressByTask returns only the latest run's branches across multiple runs (#1413)", async () => { + const t = await store.createTask({ description: "fanout" }); + const ins = `INSERT INTO workflow_run_branches (taskId, runId, branchId, currentNodeId, status, updatedAt) VALUES (?, ?, ?, ?, ?, ?)`; + const db = (store as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } }).db; + // Older run. + db.prepare(ins).run(t.id, "run-1", "b1", "n1", "completed", "2026-06-01T00:00:00.000Z"); + db.prepare(ins).run(t.id, "run-1", "b2", "n2", "completed", "2026-06-01T00:00:01.000Z"); + // Latest run, two branches with staggered updatedAt (both must be returned). + db.prepare(ins).run(t.id, "run-2", "b1", "n3", "running", "2026-06-03T00:00:00.000Z"); + db.prepare(ins).run(t.id, "run-2", "b2", "n4", "completed", "2026-06-03T00:00:01.000Z"); + + const entries = store.getBranchProgressByTask([t.id]).get(t.id) ?? []; + expect(entries.length).toBe(2); + expect(entries.map((e) => e.nodeId).sort()).toEqual(["n3", "n4"]); + }); + + it("getBranchProgressByTask breaks updatedAt ties deterministically by runId (#1413)", async () => { + const t = await store.createTask({ description: "fanout" }); + const ins = `INSERT INTO workflow_run_branches (taskId, runId, branchId, currentNodeId, status, updatedAt) VALUES (?, ?, ?, ?, ?, ?)`; + const db = (store as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } }).db; + const ts = "2026-06-03T00:00:00.000Z"; + // Two runs with identical updatedAt — runId DESC ("run-b" > "run-a") wins. + db.prepare(ins).run(t.id, "run-a", "b1", "nA", "running", ts); + db.prepare(ins).run(t.id, "run-b", "b1", "nB", "running", ts); + + const entries = store.getBranchProgressByTask([t.id]).get(t.id) ?? []; + expect(entries.length).toBe(1); + expect(entries[0]?.nodeId).toBe("nB"); + }); +}); + describe("U12 graduation report — parity drift is caught", () => { it("transition-parity holds for the unmodified default workflow", () => { expect(checkTransitionParity(BUILTIN_CODING_WORKFLOW_IR).agree).toBe(true); diff --git a/packages/core/src/__tests__/workflow-ir-resolver.test.ts b/packages/core/src/__tests__/workflow-ir-resolver.test.ts new file mode 100644 index 0000000000..e4d96cf635 --- /dev/null +++ b/packages/core/src/__tests__/workflow-ir-resolver.test.ts @@ -0,0 +1,107 @@ +import { describe, it, expect, vi } from "vitest"; +import { getBuiltinWorkflow } from "../builtin-workflows.js"; +import { BUILTIN_CODING_WORKFLOW_IR } from "../builtin-coding-workflow-ir.js"; +import type { WorkflowIr } from "../workflow-ir-types.js"; +import { + resolveWorkflowIrForTask, + resolveWorkflowIrById, +} from "../workflow-ir-resolver.js"; + +/** A minimal custom IR distinguishable from the built-in default. */ +const CUSTOM_IR: WorkflowIr = { + version: "v2", + name: "custom-flow", + nodes: [ + { id: "start", kind: "start" }, + { id: "end", kind: "end" }, + ], + edges: [{ from: "start", to: "end" }], + columns: [{ id: "todo", name: "Todo", traits: [] }], +} as unknown as WorkflowIr; + +function makeStore(opts: { + selection?: { workflowId: string; stepIds: string[] }; + selectionThrows?: boolean; + defs?: Record; +}) { + const getWorkflowDefinition = vi.fn(async (id: string) => opts.defs?.[id]); + const getTaskWorkflowSelection = vi.fn((_taskId: string) => { + if (opts.selectionThrows) throw new Error("boom"); + return opts.selection; + }); + return { getWorkflowDefinition, getTaskWorkflowSelection }; +} + +describe("resolveWorkflowIrForTask", () => { + it("resolves a selection pointing at a custom definition", async () => { + const store = makeStore({ + selection: { workflowId: "wf-custom", stepIds: [] }, + defs: { "wf-custom": { ir: CUSTOM_IR } }, + }); + const ir = await resolveWorkflowIrForTask(store, "t1"); + expect(ir).toBe(CUSTOM_IR); + expect(store.getWorkflowDefinition).toHaveBeenCalledWith("wf-custom"); + }); + + it("resolves a built-in workflow id without touching getWorkflowDefinition", async () => { + const store = makeStore({ + selection: { workflowId: "builtin:quick-fix", stepIds: [] }, + }); + const ir = await resolveWorkflowIrForTask(store, "t1"); + expect(ir).toEqual(getBuiltinWorkflow("builtin:quick-fix")!.ir); + expect(store.getWorkflowDefinition).not.toHaveBeenCalled(); + }); + + it("falls back to the built-in default when the definition is missing", async () => { + const store = makeStore({ + selection: { workflowId: "wf-gone", stepIds: [] }, + defs: { "wf-gone": undefined }, + }); + const ir = await resolveWorkflowIrForTask(store, "t1"); + expect(ir).toBe(BUILTIN_CODING_WORKFLOW_IR); + }); + + it("falls back to the default when there is no selection", async () => { + const store = makeStore({ selection: undefined }); + const ir = await resolveWorkflowIrForTask(store, "t1"); + expect(ir).toBe(BUILTIN_CODING_WORKFLOW_IR); + expect(store.getWorkflowDefinition).not.toHaveBeenCalled(); + }); + + it("degrades to the default when the selection lookup throws", async () => { + const store = makeStore({ selectionThrows: true }); + const ir = await resolveWorkflowIrForTask(store, "t1"); + expect(ir).toBe(BUILTIN_CODING_WORKFLOW_IR); + }); + + it("caches by workflowId so the definition is fetched once across calls", async () => { + const store = makeStore({ + selection: { workflowId: "wf-custom", stepIds: [] }, + defs: { "wf-custom": { ir: CUSTOM_IR } }, + }); + const cache = new Map(); + const a = await resolveWorkflowIrForTask(store, "t1", cache); + const b = await resolveWorkflowIrForTask(store, "t2", cache); + expect(a).toBe(CUSTOM_IR); + expect(b).toBe(CUSTOM_IR); + expect(store.getWorkflowDefinition).toHaveBeenCalledTimes(1); + }); +}); + +describe("resolveWorkflowIrById", () => { + it("parses a raw-string IR from the definition", async () => { + const raw = JSON.stringify(CUSTOM_IR); + const store = makeStore({ defs: { "wf-raw": { ir: raw } } }); + const ir = await resolveWorkflowIrById(store, "wf-raw"); + expect(ir.version).toBe("v2"); + expect(ir.name).toBe("custom-flow"); + }); + + it("returns a cache hit without re-fetching the definition", async () => { + const store = makeStore({ defs: { "wf-custom": { ir: CUSTOM_IR } } }); + const cache = new Map(); + await resolveWorkflowIrById(store, "wf-custom", cache); + await resolveWorkflowIrById(store, "wf-custom", cache); + expect(store.getWorkflowDefinition).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 3a5a88962d..dac2393945 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -188,6 +188,11 @@ export { getBuiltinWorkflow, isBuiltinWorkflowId, } from "./builtin-workflows.js"; +export { + resolveWorkflowIrForTask, + resolveWorkflowIrById, + type WorkflowIrResolverStore, +} from "./workflow-ir-resolver.js"; // ── Engine wiring (set by @fusion/engine at module load) ──────────── export { diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index da180873cc..8cd91e8b16 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -5071,17 +5071,37 @@ export class TaskStore extends EventEmitter { if (!any) return result; const placeholders = taskIds.map(() => "?").join(", "); + // Filter to the latest run per task entirely in SQL (#1413): the + // correlated subquery resolves the winning (updatedAt, runId) pair per + // task — MAX(updatedAt) with a deterministic MAX(runId) tie-break — and + // the JOIN matches both columns so only the latest run's rows are read. + // The runId tie-break makes ties on updatedAt deterministic instead of + // letting an arbitrary historical run win. const rows = this.db .prepare( `SELECT b.taskId AS taskId, b.runId AS runId, b.branchId AS branchId, b.currentNodeId AS nodeId, b.status AS status, b.updatedAt AS updatedAt FROM workflow_run_branches b JOIN ( - SELECT taskId, MAX(updatedAt) AS latest - FROM workflow_run_branches - WHERE taskId IN (${placeholders}) - GROUP BY taskId - ) latest_run ON latest_run.taskId = b.taskId + -- Resolve the winning run per task: the run owning the row with + -- the greatest updatedAt, with runId as a deterministic + -- tie-break when two runs share an updatedAt. Returns the whole + -- run's rows (all its branches), not just the single max row. + SELECT taskId, runId AS latestRunId + FROM ( + SELECT taskId, runId, + ROW_NUMBER() OVER ( + PARTITION BY taskId + ORDER BY MAX(updatedAt) DESC, runId DESC + ) AS rn + FROM workflow_run_branches + WHERE taskId IN (${placeholders}) + GROUP BY taskId, runId + ) + WHERE rn = 1 + ) latest_run + ON latest_run.taskId = b.taskId + AND latest_run.latestRunId = b.runId WHERE b.taskId IN (${placeholders})`, ) .all(...taskIds, ...taskIds) as Array<{ @@ -5093,18 +5113,7 @@ export class TaskStore extends EventEmitter { updatedAt: string; }>; - // Group by task; for each task keep only the branches of its most-recent - // run (the runId of the row with the latest updatedAt). - const maxByTask = new Map(); for (const row of rows) { - const cur = maxByTask.get(row.taskId); - if (!cur || row.updatedAt > cur.updatedAt) { - maxByTask.set(row.taskId, { runId: row.runId, updatedAt: row.updatedAt }); - } - } - for (const row of rows) { - const latest = maxByTask.get(row.taskId); - if (!latest || row.runId !== latest.runId) continue; const list = result.get(row.taskId) ?? []; list.push({ branchId: row.branchId, nodeId: row.nodeId, status: row.status }); result.set(row.taskId, list); @@ -5116,6 +5125,94 @@ export class TaskStore extends EventEmitter { return result; } + /** + * Persist (idempotent upsert) one branch's progress for a fan-out run (#1407). + * Keyed by (taskId, runId, branchId) — the table PK — so re-running the same + * branch overwrites its single row with the latest currentNodeId/status. The + * executor's crash-resume reads only `status = 'completed'` rows and skips + * those nodes, so resume granularity is keyed by the persisted currentNodeId. + * Additive: silently no-ops on a legacy/missing table. + */ + saveWorkflowRunBranch(state: { + taskId: string; + runId: string; + branchId: string; + currentNodeId: string; + status: string; + }): void { + try { + this.db + .prepare( + `INSERT INTO workflow_run_branches + (taskId, runId, branchId, currentNodeId, status, updatedAt) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(taskId, runId, branchId) DO UPDATE SET + currentNodeId = excluded.currentNodeId, + status = excluded.status, + updatedAt = excluded.updatedAt`, + ) + .run( + state.taskId, + state.runId, + state.branchId, + state.currentNodeId, + state.status, + new Date().toISOString(), + ); + } catch { + // Legacy/missing table — persistence is additive, so degrade silently. + } + } + + /** Load persisted branch states for a run (crash-resume; #1407). */ + loadWorkflowRunBranches( + taskId: string, + runId: string, + ): Array<{ + taskId: string; + runId: string; + branchId: string; + currentNodeId: string; + status: "running" | "completed" | "failed" | "aborted"; + }> { + try { + const rows = this.db + .prepare( + `SELECT taskId, runId, branchId, currentNodeId, status + FROM workflow_run_branches + WHERE taskId = ? AND runId = ?`, + ) + .all(taskId, runId) as Array<{ + taskId: string; + runId: string; + branchId: string; + currentNodeId: string; + status: "running" | "completed" | "failed" | "aborted"; + }>; + return rows; + } catch { + return []; + } + } + + /** + * Prune stale branch rows for a task (#1412). Deletes every row for `taskId` + * whose runId differs from the supplied `keepRunId`, bounding growth across a + * long-lived task's repeated runs. Called on run start and run completion. + * Additive: silently no-ops on a legacy/missing table. + */ + clearWorkflowRunBranches(taskId: string, keepRunId: string): void { + try { + this.db + .prepare( + `DELETE FROM workflow_run_branches WHERE taskId = ? AND runId != ?`, + ) + .run(taskId, keepRunId); + } catch { + // Legacy/missing table — pruning is additive, so degrade silently. + } + } + async listTasksForGithubTrackingReconcile(options?: { offset?: number; limit?: number }): Promise<{ tasks: Task[]; hasMore: boolean }> { const reconcileScanLimit = 200; const offset = Math.max(0, options?.offset ?? 0); diff --git a/packages/core/src/workflow-ir-resolver.ts b/packages/core/src/workflow-ir-resolver.ts new file mode 100644 index 0000000000..c2c703aa29 --- /dev/null +++ b/packages/core/src/workflow-ir-resolver.ts @@ -0,0 +1,79 @@ +/** + * Single source of truth for the workflow-IR resolution rule. + * + * The selection → builtin/custom → default-fallback rule was independently + * reimplemented in engine/hold-release.ts, engine/merge-trait.ts, + * engine/plugin-runner.ts (which bypassed the public API via getDatabase()), + * and dashboard/board-workflows.ts, with behavioral divergence already creeping + * in (GitHub #1402). This module consolidates the read-only resolution into one + * pair of helpers built on the *public* store surface so every call site shares + * one implementation. + * + * A missing/corrupt definition degrades to the built-in default workflow so + * resolution never throws. The store-private, txn-hot `resolveTaskWorkflowIrSync` + * stays separate by design. + */ + +import { getBuiltinWorkflow, isBuiltinWorkflowId } from "./builtin-workflows.js"; +import { BUILTIN_CODING_WORKFLOW_IR } from "./builtin-coding-workflow-ir.js"; +import { parseWorkflowIr } from "./workflow-ir.js"; +import type { WorkflowIr } from "./workflow-ir-types.js"; + +/** Minimal store surface the resolver needs (public APIs only). */ +export interface WorkflowIrResolverStore { + getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined; + getWorkflowDefinition(id: string): Promise<{ ir: string | WorkflowIr } | undefined>; +} + +/** + * Resolve a workflow IR by its id (built-in or custom). + * + * @param irCache optional cache keyed by workflowId so each distinct workflow's + * IR (and its definition fetch) is resolved at most once per caller-scoped + * sweep. Hits short-circuit before any builtin/db lookup. + */ +export async function resolveWorkflowIrById( + store: Pick, + workflowId: string, + irCache?: Map, +): Promise { + const cached = irCache?.get(workflowId); + if (cached) return cached; + + if (isBuiltinWorkflowId(workflowId)) { + const builtin = getBuiltinWorkflow(workflowId); + const ir = builtin?.ir ?? BUILTIN_CODING_WORKFLOW_IR; + const resolved = typeof ir === "string" ? parseWorkflowIr(ir) : ir; + irCache?.set(workflowId, resolved); + return resolved; + } + + try { + const def = await store.getWorkflowDefinition(workflowId); + if (!def) return BUILTIN_CODING_WORKFLOW_IR; + const ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir; + irCache?.set(workflowId, ir); + return ir; + } catch { + return BUILTIN_CODING_WORKFLOW_IR; + } +} + +/** + * Resolve a task's workflow IR via its selection. A null/absent selection or any + * lookup failure degrades to the built-in default workflow. + */ +export async function resolveWorkflowIrForTask( + store: WorkflowIrResolverStore, + taskId: string, + irCache?: Map, +): Promise { + let workflowId: string | undefined; + try { + workflowId = store.getTaskWorkflowSelection(taskId)?.workflowId; + } catch { + return BUILTIN_CODING_WORKFLOW_IR; + } + if (!workflowId) return BUILTIN_CODING_WORKFLOW_IR; + return resolveWorkflowIrById(store, workflowId, irCache); +} diff --git a/packages/dashboard/src/routes/board-workflows.ts b/packages/dashboard/src/routes/board-workflows.ts index 7ab6a4b637..c1933229ab 100644 --- a/packages/dashboard/src/routes/board-workflows.ts +++ b/packages/dashboard/src/routes/board-workflows.ts @@ -22,6 +22,7 @@ import { isWorkflowColumnsEnabled, parseWorkflowIr, resolveColumnFlags, + resolveWorkflowIrById, type Settings, type TaskStore, type TraitFlags, @@ -72,25 +73,6 @@ function describeColumns(ir: WorkflowIr): BoardWorkflowColumn[] { })); } -async function resolveWorkflowIr( - store: Pick, - workflowId: string, -): Promise { - if (isBuiltinWorkflowId(workflowId)) { - const builtin = getBuiltinWorkflow(workflowId); - const ir = builtin?.ir; - if (!ir) return BUILTIN_CODING_WORKFLOW_IR; - return typeof ir === "string" ? parseWorkflowIr(ir) : ir; - } - try { - const def = await store.getWorkflowDefinition(workflowId); - if (!def) return BUILTIN_CODING_WORKFLOW_IR; - return typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir; - } catch { - return BUILTIN_CODING_WORKFLOW_IR; - } -} - async function describeWorkflow( store: Pick, workflowId: string, @@ -98,7 +80,7 @@ async function describeWorkflow( // The display name comes from the persisted definition when available, // otherwise the IR's own name (default workflow). if (isBuiltinWorkflowId(workflowId)) { - const ir = await resolveWorkflowIr(store, workflowId); + const ir = await resolveWorkflowIrById(store, workflowId); const name = getBuiltinWorkflow(workflowId)?.name ?? ir.name; return { id: workflowId, name, columns: describeColumns(ir) }; } diff --git a/packages/engine/src/__tests__/hold-release.test.ts b/packages/engine/src/__tests__/hold-release.test.ts index 3db74f010f..dd4de0cdb1 100644 --- a/packages/engine/src/__tests__/hold-release.test.ts +++ b/packages/engine/src/__tests__/hold-release.test.ts @@ -164,27 +164,18 @@ describe("hold-release sweep (U6)", () => { // (b) Reservation accounting across the racing sweeps. // - // PRODUCTION BUG CAPTURED HERE (report only — prod is owned by another agent): - // The desired safety invariant is `reserveCount - releaseCount <= 1` (at most - // one live reservation, backing the single occupant). Under two overlapping - // sweeps with one held card + one slot, that invariant is VIOLATED: both - // sweeps read the same snapshot, both pass the pre-check, both reserve a slot - // (reserveCount === 2), and BOTH moveTask calls succeed — the second is an - // idempotent same-column move (todo→in-progress on an already-released card) - // whose in-txn capacity count includes the card as its own occupant, so it - // never throws capacity-exhausted and `issueRelease` never calls - // reservation.release(). Result: releaseCount === 0, leaking the loser's - // semaphore/worktree reservation. - // - // We assert the OBSERVED (leaking) behavior so the suite stays green while the - // leak is documented. Tighten this to `<= 1` once the prod fix lands (e.g. - // re-read the card's column inside issueRelease and skip/release when it is - // already at target). + // Both sweeps read the same snapshot, both pass the pre-check, and both + // reserve a slot (reserveCount === 2). The winning sweep commits the move; + // the losing sweep, after acquiring its reservation, re-reads the card's + // current column inside `issueRelease`, sees it already at the target (the + // winner moved it), and releases its reservation without issuing a redundant + // same-column move. The safety invariant therefore holds: at most one live + // reservation backs the single occupant. expect(reserveCount).toBe(2); - expect(releaseCount).toBe(0); - // The net leaked reservations (2) is the bug; single board occupancy (asserted - // above) is still preserved, so no double card placement occurs. - expect(reserveCount - releaseCount).toBe(2); + // The loser releases its reservation, so the net live reservations is exactly + // one (the winner's), backing the single in-progress occupant — no leak. + expect(releaseCount).toBe(1); + expect(reserveCount - releaseCount).toBe(1); }); it("sweep release into a full column is rejected by the in-txn check (capacity is not a guard, scheduler bypasses guards)", async () => { diff --git a/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts b/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts index 2d2f5ed4d7..9e865bf9d3 100644 --- a/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts +++ b/packages/engine/src/__tests__/workflow-graph-task-runner.test.ts @@ -242,4 +242,66 @@ describe("WorkflowGraphTaskRunner (CU-U2)", () => { const result = await runner.run(task, flagOn); expect(result.disposition).toBe("completed"); }); + + // #1407/#1412: the runner forwards its injected branchPersistence into the + // WorkflowGraphExecutor, which writes per-branch state and prunes stale runs. + // Uses a real in-memory persistence whose method shape matches the store- + // backed adapter the production executor builds (saveBranchState / + // loadBranchStates / clearStaleBranchStates) — no mock of a nonexistent API. + function fanoutIr(): WorkflowIr { + return { + version: "v1", + name: "fanout", + nodes: [ + { id: "start", kind: "start" }, + { id: "split", kind: "split" }, + { id: "a", kind: "prompt", config: { prompt: "a" } }, + { id: "b", kind: "prompt", config: { prompt: "b" } }, + { id: "join", kind: "join", config: { mode: "all" } }, + { id: "zend", kind: "end" }, + ], + edges: [ + { from: "start", to: "split" }, + { from: "split", to: "a" }, + { from: "split", to: "b" }, + { from: "a", to: "join" }, + { from: "b", to: "join" }, + { from: "join", to: "zend", condition: "success" }, + ], + }; + } + + it("forwards branchPersistence to the executor: writes branch state and prunes stale runs", async () => { + const saved: Array<{ branchId: string; currentNodeId: string; status: string }> = []; + const pruneCalls: Array<{ taskId: string; keepRunId: string }> = []; + const persistence = { + saveBranchState: (s: { branchId: string; currentNodeId: string; status: string }) => { + saved.push({ branchId: s.branchId, currentNodeId: s.currentNodeId, status: s.status }); + }, + loadBranchStates: () => [], + clearStaleBranchStates: (taskId: string, keepRunId: string) => { + pruneCalls.push({ taskId, keepRunId }); + }, + }; + + const runner = new WorkflowGraphTaskRunner({ + store: storeWith(definition(fanoutIr())), + seams: recordingSeams([]), + runCustomNode: async () => ({ outcome: "success" }), + branchPersistence: persistence, + }); + + const result = await runner.run(task, flagOn); + expect(result.disposition).toBe("completed"); + + // Both branches persisted, and each reached "completed" at the join. + expect(saved.some((s) => s.branchId === "a")).toBe(true); + expect(saved.some((s) => s.branchId === "b")).toBe(true); + expect(saved.some((s) => s.status === "completed")).toBe(true); + + // Prune ran (on start AND completion) keyed by the runner's runId. + expect(pruneCalls.length).toBeGreaterThanOrEqual(2); + expect(pruneCalls.every((c) => c.taskId === task.id)).toBe(true); + expect(pruneCalls.every((c) => c.keepRunId === `${task.id}:WF-001`)).toBe(true); + }); }); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 22cdcf226f..c0013945d0 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -17,6 +17,7 @@ import { type WorkflowRunObservation, } from "@fusion/core"; import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js"; +import type { WorkflowBranchPersistence, WorkflowBranchRunState } from "./workflow-graph-branches.js"; import { observeWorkflowParity, WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from "./workflow-parity-observer.js"; import type { WorkflowLegacySeams } from "./workflow-node-handlers.js"; import type { WorkflowNodeResult } from "./workflow-graph-executor.js"; @@ -3259,6 +3260,12 @@ export class TaskExecutor { seams: this.createGraphSeams(settings), runCustomNode: (node, nodeTask) => this.runGraphCustomNode(node, nodeTask, settings), onEvent: (event) => executorLog.log(`[workflow-graph] ${event.type} ${event.taskId}: ${event.detail}`), + // Wire SQLite-backed per-branch persistence in production (#1407): the + // executor writes each branch's currentNodeId/status to + // workflow_run_branches so fan-out crash-resume and the U9 badges have + // real data, and prunes stale runs (#1412). Adapter degrades to no-op + // when the store predates these methods (additive guard). + branchPersistence: this.buildBranchPersistence(), }); let result: WorkflowGraphTaskRunResult; try { @@ -3285,6 +3292,27 @@ export class TaskExecutor { } } + /** + * Build the store-backed WorkflowBranchPersistence wired into production + * fan-out runs (#1407/#1412). Returns undefined when the store predates the + * persistence methods (older embedded DBs) so the runner stays fully + * in-memory — purely additive. Each adapter method is itself guarded so a + * mixed/partial store never throws into the run. + */ + private buildBranchPersistence(): WorkflowBranchPersistence | undefined { + const store = this.store as unknown as { + saveWorkflowRunBranch?: (state: WorkflowBranchRunState) => void; + loadWorkflowRunBranches?: (taskId: string, runId: string) => WorkflowBranchRunState[]; + clearWorkflowRunBranches?: (taskId: string, keepRunId: string) => void; + }; + if (typeof store.saveWorkflowRunBranch !== "function") return undefined; + return { + saveBranchState: (state) => store.saveWorkflowRunBranch?.(state), + loadBranchStates: (taskId, runId) => store.loadWorkflowRunBranches?.(taskId, runId) ?? [], + clearStaleBranchStates: (taskId, keepRunId) => store.clearWorkflowRunBranches?.(taskId, keepRunId), + }; + } + /** * Dual-observe parity (CU-U5): for a workflow-selected task, compare the * selected graph's routing against the legacy authoritative run for the SAME diff --git a/packages/engine/src/hold-release.ts b/packages/engine/src/hold-release.ts index 307eeaebc1..a265cb923f 100644 --- a/packages/engine/src/hold-release.ts +++ b/packages/engine/src/hold-release.ts @@ -43,10 +43,7 @@ import { resolveColumnAdjacency, DEFAULT_WORKFLOW_POOL_ID, TransitionRejectionError, - BUILTIN_CODING_WORKFLOW_IR, - getBuiltinWorkflow, - isBuiltinWorkflowId, - parseWorkflowIr, + resolveWorkflowIrForTask, type TaskStore, type Task, type WorkflowIr, @@ -92,40 +89,10 @@ export interface HoldReleaseResult { held: Array<{ taskId: string; reason: string }>; } -// ── Workflow IR resolution (read-only, mirrors store + merge-trait) ─────────── - -async function resolveTaskWorkflowIr( - store: TaskStore, - taskId: string, - // Optional per-sweep cache keyed by workflowId so each distinct workflow's IR - // is resolved (and its definition fetched) at most once per sweep. - irCache?: Map, -): Promise { - let workflowId: string | undefined; - try { - workflowId = store.getTaskWorkflowSelection(taskId)?.workflowId; - } catch { - workflowId = undefined; - } - if (!workflowId) return BUILTIN_CODING_WORKFLOW_IR; - const cached = irCache?.get(workflowId); - if (cached) return cached; - if (isBuiltinWorkflowId(workflowId)) { - const builtin = getBuiltinWorkflow(workflowId); - const ir = builtin?.ir ?? BUILTIN_CODING_WORKFLOW_IR; - irCache?.set(workflowId, ir); - return ir; - } - try { - const def = await store.getWorkflowDefinition(workflowId); - if (!def) return BUILTIN_CODING_WORKFLOW_IR; - const ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir; - irCache?.set(workflowId, ir); - return ir; - } catch { - return BUILTIN_CODING_WORKFLOW_IR; - } -} +// ── Workflow IR resolution (read-only) ──────────────────────────────────────── +// The selection → builtin/custom → default rule lives in @fusion/core's +// resolveWorkflowIrForTask (GitHub #1402); the optional per-sweep irCache Map is +// threaded straight through. function effectiveWorkflowId(store: TaskStore, taskId: string): string { try { @@ -220,7 +187,7 @@ function legacyDependencySatisfied(dep: Task): boolean { * audit-diff event is logged. */ async function dependencySatisfied(store: TaskStore, dep: Task): Promise { - const ir = await resolveTaskWorkflowIr(store, dep.id); + const ir = await resolveWorkflowIrForTask(store, dep.id); const column = findColumn(ir, dep.column); const completeFlag = column ? resolveColumnFlags(column).complete === true : false; @@ -358,7 +325,7 @@ export async function runHoldReleaseSweep( continue; } - const ir = await resolveTaskWorkflowIr(store, task.id, irCache); + const ir = await resolveWorkflowIrForTask(store, task.id, irCache); if (!isHeldTask(ir, task)) continue; const column = findColumn(ir, task.column); @@ -452,14 +419,37 @@ async function issueRelease( } } + // A concurrent sweep (or explicit promote) can win the move for this same card + // while we hold a reservation. The store serializes the move under a per-task + // lock and resolves a redundant same-column move to a silent no-op: it returns + // the card already at the target WITHOUT re-allocating a slot or emitting a + // `task:moved`. A snapshot/pre-read can't tell winner from loser (both reads + // race ahead of either commit on the per-task lock). Instead we attribute the + // transition by OBJECT IDENTITY: a real move emits `task:moved` with the very + // Task object it then returns, whereas a no-op returns a freshly-read object + // and emits nothing. So the call whose `moveTask` result IS the emitted task is + // the real mover; any other call that reserved performed a redundant no-op and + // must release the slot it grabbed (FN-1415). + const movedTaskObjects = new Set(); + const onMoved = (data: { task: object; to: string }): void => { + if (data.to === target) movedTaskObjects.add(data.task); + }; + store.on("task:moved", onMoved); + try { - await store.moveTask(task.id, target, { + const result = await store.moveTask(task.id, target, { moveSource: "scheduler", allocateWorktree: targetIsProcessing && deps.allocateWorktree ? (reservedNames) => deps.allocateWorktree!(task, reservedNames) : undefined, }); + if (reservation && !movedTaskObjects.has(result)) { + // Same-column no-op: a racing sweep already moved this card to the target. + reservation.release(); + schedulerLog.log(`Hold release for ${task.id} skipped — already at ${target} (racing sweep won)`); + return false; + } return true; } catch (error) { if (error instanceof TransitionRejectionError && error.rejection.code === "capacity-exhausted") { @@ -474,6 +464,8 @@ async function issueRelease( `Hold release for ${task.id} into ${target} failed: ${error instanceof Error ? error.message : String(error)}`, ); return false; + } finally { + store.off("task:moved", onMoved); } } @@ -495,7 +487,7 @@ export async function promoteHeldTask( const task = await store.getTask(taskId); if (!task) return { released: false, rejection: "task-not-found" }; - const ir = await resolveTaskWorkflowIr(store, taskId); + const ir = await resolveWorkflowIrForTask(store, taskId); if (!isHeldTask(ir, task)) { return { released: false, rejection: "not-held" }; } @@ -527,7 +519,7 @@ export async function releaseHeldTaskByEvent( const task = await store.getTask(taskId); if (!task) return { released: false, rejection: "task-not-found" }; - const ir = await resolveTaskWorkflowIr(store, taskId); + const ir = await resolveWorkflowIrForTask(store, taskId); const column = findColumn(ir, task.column); const holdConfig = column ? resolveHoldConfig(column) : undefined; if (!column || !holdConfig || holdConfig.release !== "external-event") { diff --git a/packages/engine/src/merge-trait.ts b/packages/engine/src/merge-trait.ts index 5b2a8cec5c..a300d64590 100644 --- a/packages/engine/src/merge-trait.ts +++ b/packages/engine/src/merge-trait.ts @@ -38,12 +38,9 @@ */ import { - BUILTIN_CODING_WORKFLOW_IR, - getBuiltinWorkflow, - isBuiltinWorkflowId, isWorkflowColumnsEnabled, - parseWorkflowIr, registerTraitHookImpl, + resolveWorkflowIrForTask, type DirectMergeCommitStrategy, type Settings, type Task, @@ -83,37 +80,9 @@ export interface ResolvedMergePolicy { } // ── Workflow IR resolution (read-only, flag-gated) ─────────────────────────── - -/** - * Resolve the task's workflow IR. Mirrors the store's private - * `resolveTaskWorkflowIrSync` resolution rule (selection → builtin/custom → - * default) but stays read-only and engine-side. A missing/corrupt definition - * degrades to the default workflow so policy resolution never throws. - */ -async function resolveTaskWorkflowIr(store: TaskStore, taskId: string): Promise { - let workflowId: string | undefined; - try { - workflowId = store.getTaskWorkflowSelection(taskId)?.workflowId; - } catch { - workflowId = undefined; - } - if (!workflowId) return BUILTIN_CODING_WORKFLOW_IR; - - if (isBuiltinWorkflowId(workflowId)) { - const builtin = getBuiltinWorkflow(workflowId); - return builtin?.ir ?? BUILTIN_CODING_WORKFLOW_IR; - } - - try { - const def = await store.getWorkflowDefinition(workflowId); - if (!def) return BUILTIN_CODING_WORKFLOW_IR; - // `def.ir` is already a parsed WorkflowIr; reparse defensively only if a - // raw string ever slips through. - return typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir; - } catch { - return BUILTIN_CODING_WORKFLOW_IR; - } -} +// The selection → builtin/custom → default rule is shared via @fusion/core's +// resolveWorkflowIrForTask (GitHub #1402); a missing/corrupt definition degrades +// to the default workflow so policy resolution never throws. /** Find the column the task currently sits in (by id). */ function findColumn(ir: WorkflowIr, columnId: string): WorkflowIrColumn | undefined { @@ -178,7 +147,7 @@ export async function resolveMergePolicy( let config: Record | undefined; try { - const ir = await resolveTaskWorkflowIr(store, task.id); + const ir = await resolveWorkflowIrForTask(store, task.id); config = readMergeTraitConfig(findColumn(ir, task.column)); } catch { config = undefined; diff --git a/packages/engine/src/plugin-runner.ts b/packages/engine/src/plugin-runner.ts index cad8ceb27e..e00648812e 100644 --- a/packages/engine/src/plugin-runner.ts +++ b/packages/engine/src/plugin-runner.ts @@ -36,10 +36,7 @@ import { Type } from "@earendil-works/pi-ai"; import { isAbsolute } from "node:path"; import { getTraitRegistry, - parseWorkflowIr, - BUILTIN_CODING_WORKFLOW_IR, - getBuiltinWorkflow, - isBuiltinWorkflowId, + resolveWorkflowIrForTask, } from "@fusion/core"; import { createLogger, executorLog } from "./logger.js"; import type { WorkflowCustomNodeRunner } from "./workflow-node-handlers.js"; @@ -539,33 +536,13 @@ export class PluginRunner { } /** - * Resolve a task's workflow IR through the public store API (selection + - * workflow definition). Mirrors the store's private resolver but stays on the - * public surface so the adapter never reaches into store internals. Falls back - * to the built-in default workflow on any miss. + * Resolve a task's workflow IR through the shared @fusion/core resolver + * (selection → builtin/custom → default fallback) on the public store surface + * — the adapter never reaches into store internals (GitHub #1402; previously a + * divergent raw-SQL copy via getDatabase()). */ - private resolveTaskWorkflowIr(taskId: string): WorkflowIr | undefined { - const store = this.options.taskStore; - let workflowId: string | undefined; - try { - workflowId = store.getTaskWorkflowSelection?.(taskId)?.workflowId; - } catch { - return BUILTIN_CODING_WORKFLOW_IR; - } - if (!workflowId) return BUILTIN_CODING_WORKFLOW_IR; - if (isBuiltinWorkflowId(workflowId)) { - return getBuiltinWorkflow(workflowId)?.ir ?? BUILTIN_CODING_WORKFLOW_IR; - } - try { - const db = store.getDatabase(); - const row = db.prepare("SELECT ir FROM workflows WHERE id = ?").get(workflowId) as - | { ir: string } - | undefined; - if (!row) return BUILTIN_CODING_WORKFLOW_IR; - return parseWorkflowIr(row.ir); - } catch { - return BUILTIN_CODING_WORKFLOW_IR; - } + private resolveTaskWorkflowIr(taskId: string): Promise { + return resolveWorkflowIrForTask(this.options.taskStore, taskId); } getPluginWorkflowStepTemplates(): Array<{ pluginId: string; template: WorkflowStepTemplate }> { diff --git a/packages/engine/src/plugin-trait-adapter.ts b/packages/engine/src/plugin-trait-adapter.ts index 79488192e3..42c9dcb57c 100644 --- a/packages/engine/src/plugin-trait-adapter.ts +++ b/packages/engine/src/plugin-trait-adapter.ts @@ -210,8 +210,9 @@ export class PluginTraitHasDependentsError extends Error { */ export async function findLivePluginTraitDependents(params: { store: Pick; - /** Resolve the (already-parsed) workflow IR for a task id. */ - resolveTaskWorkflowIr: (taskId: string) => WorkflowIr | undefined; + /** Resolve the (already-parsed) workflow IR for a task id. May resolve + * asynchronously (the shared @fusion/core resolver awaits the definition). */ + resolveTaskWorkflowIr: (taskId: string) => WorkflowIr | undefined | Promise; /** The registry ids of the plugin's traits to check for. */ pluginTraitIds: string[]; }): Promise { @@ -222,7 +223,7 @@ export async function findLivePluginTraitDependents(params: { const dependents: PluginTraitDependent[] = []; const tasks = await store.listTasks({ slim: true, includeArchived: false }); for (const task of tasks) { - const ir = resolveTaskWorkflowIr(task.id); + const ir = await resolveTaskWorkflowIr(task.id); if (!ir) continue; const column = findWorkflowColumn(ir, task.column); if (!column) continue; diff --git a/packages/engine/src/workflow-graph-branches.ts b/packages/engine/src/workflow-graph-branches.ts index 2103803ac9..7c6194852e 100644 --- a/packages/engine/src/workflow-graph-branches.ts +++ b/packages/engine/src/workflow-graph-branches.ts @@ -39,6 +39,12 @@ export interface WorkflowBranchPersistence { saveBranchState?(state: WorkflowBranchRunState): void | Promise; /** Load any persisted branch states for a run (used on resume). */ loadBranchStates?(taskId: string, runId: string): WorkflowBranchRunState[] | Promise; + /** + * Prune stale branch rows for a task, keeping only `keepRunId` (#1412). + * Called on run start and run completion to bound unbounded growth across a + * long-lived task's repeated runs. + */ + clearStaleBranchStates?(taskId: string, keepRunId: string): void | Promise; } /** diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 91c40b473c..fe64000cbc 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -119,6 +119,11 @@ export class WorkflowGraphExecutor { ); } + // Prune prior-run branch rows on run start (#1412). Done after the resume + // load so this run's own (taskId, runId) rows survive while every stale run + // is removed. Never throws into the run. + await this.pruneStaleBranches(task.id, runId); + // Shared branch environment: built lazily so the sequential path pays nothing. const branchEnv = (): BranchEnvironment => ({ task, @@ -206,6 +211,9 @@ export class WorkflowGraphExecutor { }; const terminal = await walk(startNode.id); + // Prune again on run completion (#1412): keeps only this run's rows so the + // table does not accumulate historical runs for a long-lived task. + await this.pruneStaleBranches(task.id, runId); return { executed: true, outcome: terminal.outcome, @@ -214,6 +222,15 @@ export class WorkflowGraphExecutor { }; } + /** Best-effort prune of stale-run branch rows; never throws into the run. */ + private async pruneStaleBranches(taskId: string, keepRunId: string): Promise { + try { + await this.deps.branchPersistence?.clearStaleBranchStates?.(taskId, keepRunId); + } catch { + // Pruning is additive bookkeeping — a failure must not affect the run. + } + } + private shouldTraverseEdge(edge: WorkflowIrEdge, sourceResult: WorkflowNodeResult): boolean { if (!edge.condition) return sourceResult.outcome === "success"; if (edge.condition === "success") return sourceResult.outcome === "success";