fix(core): shared workflow-IR resolver, prod branch persistence + pruning, latest-run JOIN, sweep reservation-leak fix (#1402 #1407 #1412 #1413)

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<typeof harness.store>;
+
+  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);

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<string, { ir: string | WorkflowIr } | undefined>;
+}) {
+  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<string, WorkflowIr>();
+    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<string, WorkflowIr>();
+    await resolveWorkflowIrById(store, "wf-custom", cache);
+    await resolveWorkflowIrById(store, "wf-custom", cache);
+    expect(store.getWorkflowDefinition).toHaveBeenCalledTimes(1);
+  });
+});

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 {

packages/core/src/store.ts

@@ -5071,17 +5071,37 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
       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<TaskStoreEvents> {
           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<string, { runId: string; updatedAt: string }>();
       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<TaskStoreEvents> {
     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);

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<WorkflowIrResolverStore, "getWorkflowDefinition">,
+  workflowId: string,
+  irCache?: Map<string, WorkflowIr>,
+): Promise<WorkflowIr> {
+  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<string, WorkflowIr>,
+): Promise<WorkflowIr> {
+  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);
+}

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<TaskStore, "getWorkflowDefinition">,
-  workflowId: string,
-): Promise<WorkflowIr> {
-  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<TaskStore, "getWorkflowDefinition">,
   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) };
   }

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 () => {

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);
+  });
 });

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

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<string, WorkflowIr>,
-): Promise<WorkflowIr> {
-  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<boolean> {
-  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<object>();
+  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") {

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<WorkflowIr> {
-  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<string, unknown> | 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;

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<WorkflowIr> {
+    return resolveWorkflowIrForTask(this.options.taskStore, taskId);
   }
 
   getPluginWorkflowStepTemplates(): Array<{ pluginId: string; template: WorkflowStepTemplate }> {

packages/engine/src/plugin-trait-adapter.ts

@@ -210,8 +210,9 @@ export class PluginTraitHasDependentsError extends Error {
  */
 export async function findLivePluginTraitDependents(params: {
   store: Pick<TaskStore, "listTasks">;
-  /** 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<WorkflowIr | undefined>;
   /** The registry ids of the plugin's traits to check for. */
   pluginTraitIds: string[];
 }): Promise<PluginTraitDependent[]> {
@@ -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;

packages/engine/src/workflow-graph-branches.ts

@@ -39,6 +39,12 @@ export interface WorkflowBranchPersistence {
   saveBranchState?(state: WorkflowBranchRunState): void | Promise<void>;
   /** Load any persisted branch states for a run (used on resume). */
   loadBranchStates?(taskId: string, runId: string): WorkflowBranchRunState[] | Promise<WorkflowBranchRunState[]>;
+  /**
+   * 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<void>;
 }
 
 /**

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<void> {
+    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";