feat(core): workflowColumns integrity pass, graduation report, server-side trait validation, plugin post-commit hooks, docs + changeset (U12)

.changeset/workflow-custom-columns-traits.md

@@ -0,0 +1,7 @@
+---
+"@runfusion/fusion": minor
+---
+
+Add workflow-defined custom columns with composable traits, behind the `experimentalFeatures.workflowColumns` flag (off by default).
+
+Workflows can now define their own columns, each carrying composable traits (declarative flags plus lifecycle hooks) instead of the fixed `triage → todo → in-progress → in-review → done → archived` pipeline. The dashboard board renders one lane per workflow in use, and graphs gain `hold`, `split`, and `join` nodes for passive dwell and parallel fan-out/join branches. The built-in default workflow reproduces today's pipeline verbatim, and migration rewrites zero task rows — a null workflow selection resolves to the default workflow at read time. With the flag off, the legacy board, transitions, and engine behavior are unchanged.

CONCEPTS.md

@@ -134,6 +134,31 @@ The user's mid-stage feedback channel: free-text guidance attached to an answer,
 ### Rehydration
 Re-establishing a live agent handle for a paused CE Session by replaying its recorded conversation against the model. Replay is side-effect-suppressed: it reconstructs the agent's context without re-emitting events, re-streaming Live activity, or re-writing artifacts.
 
+## Workflow columns & traits
+
+*Behind the `experimentalFeatures.workflowColumns` flag. With the flag off, the legacy fixed pipeline (the closed column enum + `VALID_TRANSITIONS`) is authoritative and unchanged.*
+
+### Column (workflow-defined)
+A first-class, workflow-defined unit of task state: an id, a display name, and a set of Trait configurations. A Task's board position is its current column, persisted in `tasks."column"`. Column validity is workflow-scoped — the legacy closed enum widens to a string validated against the Task's resolved workflow. The Default workflow's column ids are byte-identical to the legacy enum values, so no task row is ever rewritten.
+
+### Trait
+Composable column configuration: declarative flags (e.g. `complete`, `archived`, `countsTowardWip`) plus optional lifecycle hooks (`guard`, `gate`, `onEnter`, `onExit`, `releaseCondition`). Built-in and plugin-contributed traits register through one registry. Sync `guard` hooks and the `complete`/`archived` flags are built-in-only; plugin traits get async hook points only. A column's effective flags are the merged flags of its traits; conflicting compositions are rejected at save (server-side and in the editor).
+
+### Lane
+A horizontal row on the multi-lane board, one per workflow in use by visible cards. Each lane renders its own workflow's columns. Tasks with no workflow selection appear in the Default workflow's lane; every card appears in exactly one lane. Zero-card lanes are hidden; lanes are collapsible with persisted state.
+
+### Hold node
+A workflow node kind expressing passive dwell — a card rests in its column until a release condition fires: manual promote, timer, downstream capacity available, dependency satisfied, or external event. Hold release is evaluated by a substrate sweep (the generalized scheduler), which reserves worktree + semaphore slots before issuing the release move.
+
+### Split / Join
+Parallel-branch node kinds. A `split` launches its outgoing edges concurrently; a `join` synchronizes them with `mode: all | any | quorum(n)` and `onBranchFailure: fail-fast | collect`. During the parallel window the card stays in the split's column (its board position never forks); on join resolution it advances to the join's column. `execute`/`merge` seam nodes are forbidden inside branches (one worktree/session per task; merge is exclusive). Per-branch run state persists in SQLite so a crashed branch resumes where it died.
+
+### Default workflow
+The built-in workflow (`builtin:coding`) that reproduces the legacy pipeline verbatim: six columns whose ids equal the legacy enum values, with traits matching legacy semantics (`triage`=intake, `todo`=hold+reset-on-entry, `in-progress`=wip+abort-on-exit+timing, `in-review`=merge-blocker+stall-detection+merge, `done`=complete, `archived`=archived). A null workflow selection resolves to it at read time. Non-editable, non-deletable.
+
+### transitionPending
+A persisted crash-safe marker (`tasks.transitionPending`) written in the same transaction as a column change, recording the post-commit hooks (`hooksRemaining`) that still owe idempotent execution. Cleared once they complete. Recovery reads it exclusively from SQLite (the authoritative store); a crash mid-transition re-runs the idempotent hooks. A throwing or missing hook degrades (audit) and clears its entry — it never strands the card or wedges the task lock.
+
 ## Flagged ambiguities
 
 - "Merging" a shared-branch-group Task had been used for both member integration and group promotion — these are distinct steps with independent gating and must not be conflated.

docs/architecture.md

@@ -1230,6 +1230,23 @@ Detection is visibility-only: no scheduler/self-healing actions are triggered by
 
 Tune sensitivity by adjusting the exported constants in `stalled-review-detector.ts`. Increase thresholds to reduce noise; decrease thresholds only with incident evidence, because lower values can over-flag transient recovery bursts.
 
+---
+
+### Workflow-defined columns & traits (`experimentalFeatures.workflowColumns`)
+
+*Behind the `workflowColumns` flag (accessor: `packages/core/src/workflow-columns-settings.ts`). With the flag off the legacy pipeline above is authoritative and untouched. The flag default-flips only when the graduation report (below) shows zero drift — a field decision, not yet taken.*
+
+**Engine as substrate, workflows as policy.** The flag inverts the architecture: the engine becomes a **capability substrate** (worktree/git/session mechanics, persistence, crash recovery, audit, machine resource ceilings — non-configurable) and **workflows carry the operating logic** as composable column traits. The mechanism/policy line (KTD-4):
+
+- **Substrate (engine-owned, never workflow-configurable):** `AgentSemaphore`, checkout leases, worktree/git/session ops, SQLite + WAL, the crash-recovery machinery, the audit trail, the global max-sessions cap, and the three non-configurable lost-work merge guards (no sibling `fusion/fn-*` target, line-anchored attribution, no `modifiedFiles` clear on a no-op finalize).
+- **Policy (workflow/trait-owned):** transition validity, WIP/capacity, hold/release, drag meaning, retries, merge strategy, squash posture, file-scope enforcement mode.
+
+**Transition authority.** `moveTaskInternal` remains the single transition authority. Flag-on, it swaps the `VALID_TRANSITIONS` lookup for workflow-resolved column-graph validation (`resolveAllowedColumns`/`workflowHasColumn` in `workflow-transitions.ts`) plus sync trait guards run in-lock; rejections are typed `TransitionRejection`s. `VALID_TRANSITIONS` and the closed `Column`/`COLUMNS` helpers in `types.ts` are `@deprecated` while the flag exists — retained as the flag-off authority and the parity oracle, not yet removed.
+
+**Trait model.** A trait is declarative flags + optional lifecycle hooks (`guard`, `gate`, `onEnter`, `onExit`, `releaseCondition`), resolved through one registry (`trait-registry.ts`, built-ins in `builtin-traits.ts`). Sync `guard` and the `complete`/`archived` flags are built-in-only; plugin traits (KTD-7) get async hook points only and route through the prompt-session/script machinery. Composition conflicts are rejected at save both in the editor and server-side (`assertColumnTraitsValid` in `createWorkflowDefinition`/`updateWorkflowDefinition`, surfaced as a 400). Capacity is enforced in-txn (KTD-10), never bypassable — not a guard. Enter/exit effects run post-commit, idempotent, guarded by the `transitionPending` marker; a throwing/missing plugin hook degrades (audit) and never strands the card or wedges the lock.
+
+**Graduation.** The flag default-flip is gated by `computeWorkflowColumnsGraduationReport()` (`workflow-parity.ts`; store method `TaskStore.computeWorkflowColumnsGraduationReport`), aggregating: five-invariant dual-observe parity, default-workflow transition parity vs `VALID_TRANSITIONS` (`checkTransitionParity`), and the U6 dual-accept marker/column disagreement count. `ready` is true only when all gates pass over a non-empty observation window. The report is the gate; it does not flip the flag.
+
 ## 10) Agent System
 
 Fusion has two complementary agent models:

docs/workflow-steps.md

@@ -57,6 +57,10 @@ Current reconciliation in v1:
 
 FN-5769 evaluated whether those conventions required a `1.1.0` schema bump and recorded the answer as **no**: the current `prompt` + `config` and canonical `edge.condition` token conventions are sufficient for the parity-critical interpreter rollout, so they remain the canonical v1 contract until a future consumer needs stronger schema-level validation or discoverability.
 
+### Workflow IR v2 — columns, traits, hold & split/join nodes
+
+The `workflowColumns` track introduces **IR v2** (`version: "v2"`), where a workflow additionally defines its own **columns** (`{ id, name, traits: [{ trait, config }] }`), places nodes in columns (`node.column`), and gains `hold`, `split`, and `join` node kinds. Columns become first-class, workflow-defined task state carrying composable **traits** (declarative flags + lifecycle hooks); this generalizes the fixed pipeline + the `gateMode` semantics documented below into per-column trait configuration. v1 graphs still parse and upgrade by synthesizing default-workflow columns. The column/trait model — the trait vocabulary, the substrate/policy line, the transition authority, and the graduation gate — is documented in **`docs/architecture.md` § 9 "Workflow-defined columns & traits"** and the **Concepts** glossary (column, trait, lane, hold node, split/join, default workflow, `transitionPending`). The whole v2 model is gated behind `experimentalFeatures.workflowColumns`; with the flag off, the v1 IR and the quality-gate `WorkflowStep` model below are unchanged.
+
 ## What They Are
 
 A workflow step is a reusable check (AI prompt or script) that can be enabled on tasks.

packages/core/src/__tests__/migration-workflow-columns.test.ts

@@ -0,0 +1,317 @@
+// @vitest-environment node
+//
+// U12: workflow-columns migration / integrity / graduation + rollback safety.
+//
+// Proves the U12 plan scenarios:
+//   - Migration rewrites ZERO task rows (KTD-1): fresh DB and an aged fixture DB
+//     (tasks in every legacy column, some with workflow selections) resolve every
+//     task to a valid (workflow, column) pair.
+//   - The integrity pass re-homes a task whose stored column is invalid in its
+//     resolved workflow, and is IDEMPOTENT (a second run is a no-op).
+//   - done/archived (terminal) cards are left untouched by the integrity pass.
+//   - Flag OFF after running flag-ON: legacy board + engine behavior intact.
+//   - Deliberate parity-drift injection (altered default-workflow adjacency) is
+//     CAUGHT by the graduation report's transition-parity gate.
+
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import { createTaskStoreTestHarness } from "./store-test-helpers.js";
+import type { WorkflowIr } from "../workflow-ir-types.js";
+import { BUILTIN_CODING_WORKFLOW_IR } from "../builtin-coding-workflow-ir.js";
+import { workflowHasColumn } from "../workflow-transitions.js";
+import {
+  checkTransitionParity,
+  computeWorkflowColumnsGraduationReport,
+  countDualAcceptDisagreements,
+} from "../workflow-parity.js";
+import type { Column } from "../types.js";
+
+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("U12 migration — zero task-row rewrites (KTD-1)", () => {
+  const harness = createTaskStoreTestHarness();
+  let store: ReturnType<typeof harness.store>;
+
+  beforeEach(async () => {
+    await harness.beforeEach();
+    store = harness.store();
+    await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } });
+  });
+  afterEach(async () => {
+    await harness.afterEach();
+  });
+
+  async function seedInColumn(column: Column): Promise<string> {
+    const task = await store.createTask({ description: `seed-${column}` });
+    const u = { moveSource: "user" as const };
+    if (column === "triage") return task.id;
+    await store.moveTask(task.id, "todo", u);
+    if (column === "todo") return task.id;
+    await store.moveTask(task.id, "in-progress", u);
+    if (column === "in-progress") return task.id;
+    await store.moveTask(task.id, "in-review", { ...u, allowDirectInReviewMove: true });
+    if (column === "in-review") return task.id;
+    await store.moveTask(task.id, "done", { moveSource: "engine", skipMergeBlocker: true });
+    if (column === "done") return task.id;
+    await store.moveTask(task.id, "archived", u);
+    return task.id;
+  }
+
+  it("fresh DB: a default-workflow task resolves to a valid (workflow, column) pair", async () => {
+    const id = await seedInColumn("todo");
+    const task = await store.getTask(id);
+    expect(workflowHasColumn(BUILTIN_CODING_WORKFLOW_IR, task.column)).toBe(true);
+  });
+
+  it("aged fixture: tasks in every legacy column all resolve to a valid column; integrity pass touches none", async () => {
+    const ids: string[] = [];
+    for (const col of ["triage", "todo", "in-progress", "in-review", "done", "archived"] as Column[]) {
+      ids.push(await seedInColumn(col));
+    }
+    // A task with a custom-workflow selection whose column IS valid in it.
+    const wf = await store.createWorkflowDefinition({
+      name: "valid-custom",
+      ir: customIr("valid-custom", ["todo", "build", "done"], "todo"),
+    });
+    const customTask = await store.createTask({ description: "custom" });
+    await store.moveTask(customTask.id, "todo", { moveSource: "user" });
+    await store.selectTaskWorkflowAndReconcile(customTask.id, wf.id);
+
+    const before = await Promise.all(ids.map((id) => store.getTask(id)));
+    const result = await store.runWorkflowColumnsIntegrityPass();
+    // No row was invalid → nothing re-homed.
+    expect(result.rehomed).toBe(0);
+
+    const after = await Promise.all(ids.map((id) => store.getTask(id)));
+    for (let i = 0; i < ids.length; i += 1) {
+      expect(after[i].column).toBe(before[i].column);
+    }
+  });
+});
+
+describe("U12 integrity pass — invalid column re-home + idempotency + terminal-untouched", () => {
+  const harness = createTaskStoreTestHarness();
+  let store: ReturnType<typeof harness.store>;
+
+  beforeEach(async () => {
+    await harness.beforeEach();
+    store = harness.store();
+    await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } });
+  });
+  afterEach(async () => {
+    await harness.afterEach();
+  });
+
+  function rawDb(): { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } {
+    return (store as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } }).db;
+  }
+
+  it("re-homes a task whose stored column is invalid in its resolved workflow, and is idempotent", async () => {
+    // Select a custom workflow that defines [stage-a, stage-b, finished], then
+    // force the stored column to one that workflow never defines.
+    const wf = await store.createWorkflowDefinition({
+      name: "drifted",
+      ir: customIr("drifted", ["stage-a", "stage-b", "finished"], "stage-a"),
+    });
+    const task = await store.createTask({ description: "drifter" });
+    await store.selectTaskWorkflowAndReconcile(task.id, wf.id);
+    // Out-of-band corruption: stored column not in the workflow.
+    rawDb().prepare(`UPDATE tasks SET "column" = ? WHERE id = ?`).run("ghost-column", task.id);
+
+    const first = await store.runWorkflowColumnsIntegrityPass();
+    expect(first.rehomed).toBe(1);
+    const afterFirst = await store.getTask(task.id);
+    expect(afterFirst.column).toBe("stage-a"); // entry (intake) column
+
+    // Idempotent: a second run finds nothing out of place.
+    const second = await store.runWorkflowColumnsIntegrityPass();
+    expect(second.rehomed).toBe(0);
+    expect((await store.getTask(task.id)).column).toBe("stage-a");
+  });
+
+  it("leaves done/archived (terminal) cards untouched even if their column were invalid", async () => {
+    // A task selecting a custom workflow that lacks "done" but the task sits in
+    // "done" — terminal cards are never re-homed.
+    const wf = await store.createWorkflowDefinition({
+      name: "no-done",
+      ir: customIr("no-done", ["start-col", "mid-col", "fin-col"], "start-col"),
+    });
+    const task = await store.createTask({ description: "terminal" });
+    await store.selectTaskWorkflowAndReconcile(task.id, wf.id);
+    rawDb().prepare(`UPDATE tasks SET "column" = ? WHERE id = ?`).run("done", task.id);
+
+    const result = await store.runWorkflowColumnsIntegrityPass();
+    expect(result.skippedTerminal).toBeGreaterThanOrEqual(1);
+    expect((await store.getTask(task.id)).column).toBe("done");
+  });
+});
+
+describe("U12 rollback safety — flag OFF after flag ON keeps legacy behavior", () => {
+  const harness = createTaskStoreTestHarness();
+  let store: ReturnType<typeof harness.store>;
+
+  beforeEach(async () => {
+    await harness.beforeEach();
+    store = harness.store();
+  });
+  afterEach(async () => {
+    await harness.afterEach();
+  });
+
+  it("a board built under flag-ON resolves identically and moves legacy-style under flag-OFF", async () => {
+    // Build a board under flag-ON.
+    await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } });
+    const t = await store.createTask({ description: "rollback" });
+    await store.moveTask(t.id, "todo", { moveSource: "user" });
+    await store.moveTask(t.id, "in-progress", { moveSource: "user" });
+
+    // Flip the flag OFF.
+    await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: false } });
+
+    // Legacy board intact: the task is still in in-progress.
+    expect((await store.getTask(t.id)).column).toBe("in-progress");
+
+    // Legacy engine behavior: an illegal move throws the legacy string (not a
+    // typed rejection), and a legal move works exactly as before.
+    const archived = await store.createTask({ description: "legacy" });
+    await store.moveTask(archived.id, "todo", { moveSource: "user" });
+    await store.moveTask(archived.id, "in-progress", { moveSource: "user" });
+    await store.moveTask(archived.id, "in-review", { moveSource: "user", allowDirectInReviewMove: true });
+    await store.moveTask(archived.id, "done", { moveSource: "engine", skipMergeBlocker: true });
+    await store.moveTask(archived.id, "archived", { moveSource: "user" });
+    let caught: unknown;
+    try {
+      await store.moveTask(archived.id, "todo", { moveSource: "user" });
+    } catch (e) {
+      caught = e;
+    }
+    expect(caught).toBeInstanceOf(Error);
+    expect((caught as Error).message).toMatch(/Invalid transition/);
+  });
+});
+
+describe("Residual B: getBranchProgressByTask reads workflow_run_branches", () => {
+  const harness = createTaskStoreTestHarness();
+  let store: ReturnType<typeof harness.store>;
+
+  beforeEach(async () => {
+    await harness.beforeEach();
+    store = harness.store();
+  });
+  afterEach(async () => {
+    await harness.afterEach();
+  });
+
+  function db(): { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } {
+    return (store as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } }).db;
+  }
+
+  it("returns an empty map when the table is empty (cheap short-circuit)", async () => {
+    const t = await store.createTask({ description: "x" });
+    expect(store.getBranchProgressByTask([t.id]).size).toBe(0);
+  });
+
+  it("returns the latest run's branches for a task with rows", async () => {
+    const t = await store.createTask({ description: "fanout" });
+    const ins = `INSERT INTO workflow_run_branches (taskId, runId, branchId, currentNodeId, status, updatedAt) VALUES (?, ?, ?, ?, ?, ?)`;
+    // Older run (should be ignored).
+    db().prepare(ins).run(t.id, "run-1", "b1", "n1", "completed", "2026-06-01T00:00:00.000Z");
+    // Latest run with two branches.
+    db().prepare(ins).run(t.id, "run-2", "b1", "n2", "running", "2026-06-03T00:00:00.000Z");
+    db().prepare(ins).run(t.id, "run-2", "b2", "n3", "completed", "2026-06-03T00:00:01.000Z");
+
+    const byTask = store.getBranchProgressByTask([t.id]);
+    const entries = byTask.get(t.id) ?? [];
+    expect(entries.length).toBe(2);
+    expect(entries.map((e) => e.branchId).sort()).toEqual(["b1", "b2"]);
+    expect(entries.find((e) => e.branchId === "b2")?.status).toBe("completed");
+  });
+});
+
+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);
+  });
+
+  it("a deliberately drifted default-workflow adjacency is caught by transition parity", () => {
+    // Clone the default IR and remove a legal edge target from in-progress's
+    // adjacency by dropping the "todo" backward column from its outgoing edges.
+    const drifted = JSON.parse(JSON.stringify(BUILTIN_CODING_WORKFLOW_IR)) as WorkflowIr & {
+      edges: Array<{ from: string; to: string }>;
+      columns: Array<{ id: string }>;
+    };
+    // Remove ALL columns named "archived" so the column set itself diverges —
+    // a coarse but unambiguous drift the gate must catch.
+    drifted.columns = drifted.columns.filter((c) => c.id !== "archived");
+    const report = checkTransitionParity(drifted as unknown as WorkflowIr);
+    expect(report.agree).toBe(false);
+    expect(report.diffs.some((d) => d.from === "archived" || d.from === "done")).toBe(true);
+  });
+
+  it("graduation report is NOT ready with zero observations and is gated by every signal", () => {
+    const report = computeWorkflowColumnsGraduationReport({
+      parity: { observed: 0, agreed: 0, drift: 0, agreeRate: 0, driftFieldCounts: {}, recentDrift: [] },
+      defaultWorkflowIr: BUILTIN_CODING_WORKFLOW_IR,
+      dualAcceptEvents: [],
+    });
+    expect(report.ready).toBe(false);
+    expect(report.blockers.some((b) => /observation window empty/.test(b))).toBe(true);
+  });
+
+  it("graduation report is ready only when parity clean, transitions match, and zero dual-accept disagreement", () => {
+    const report = computeWorkflowColumnsGraduationReport({
+      parity: { observed: 100, agreed: 100, drift: 0, agreeRate: 1, driftFieldCounts: {}, recentDrift: [] },
+      defaultWorkflowIr: BUILTIN_CODING_WORKFLOW_IR,
+      dualAcceptEvents: [],
+    });
+    expect(report.transitionParity.agree).toBe(true);
+    expect(report.dualAccept.total).toBe(0);
+    expect(report.ready).toBe(true);
+    expect(report.blockers).toEqual([]);
+  });
+
+  it("dual-accept disagreements above zero block graduation", () => {
+    const events = [
+      {
+        domain: "database",
+        mutationType: "merge:dependency-parity-diff",
+        target: "FN-1",
+        timestamp: "2026-06-03T00:00:00.000Z",
+      },
+      {
+        domain: "database",
+        mutationType: "merge:lease-parity-diff",
+        target: "FN-2",
+        timestamp: "2026-06-03T00:00:01.000Z",
+      },
+    ] as unknown as Parameters<typeof countDualAcceptDisagreements>[0];
+    const counted = countDualAcceptDisagreements(events);
+    expect(counted.total).toBe(2);
+
+    const report = computeWorkflowColumnsGraduationReport({
+      parity: { observed: 50, agreed: 50, drift: 0, agreeRate: 1, driftFieldCounts: {}, recentDrift: [] },
+      defaultWorkflowIr: BUILTIN_CODING_WORKFLOW_IR,
+      dualAcceptEvents: events,
+    });
+    expect(report.ready).toBe(false);
+    expect(report.blockers.some((b) => /dual-accept/.test(b))).toBe(true);
+  });
+});

packages/core/src/index.ts

@@ -86,6 +86,8 @@ export {
   listTraits,
   resolveColumnFlags,
   validateColumnTraits,
+  assertColumnTraitsValid,
+  ColumnTraitValidationError,
   registerTraitHookImpl,
   __resetTraitRegistryForTests,
 } from "./trait-registry.js";
@@ -1266,6 +1268,10 @@ export {
   deriveStageTransitions,
   buildWorkflowObservationFromTask,
   buildWorkflowObservation,
+  checkTransitionParity,
+  countDualAcceptDisagreements,
+  computeWorkflowColumnsGraduationReport,
+  DUAL_ACCEPT_PARITY_MUTATIONS,
 } from "./workflow-parity.js";
 export type {
   WorkflowAuditObservation,
@@ -1280,6 +1286,11 @@ export type {
   WorkflowObservationBuildOptions,
   WorkflowObservationParts,
   WorkflowParitySummary,
+  TransitionParityDiff,
+  TransitionParityReport,
+  DualAcceptDisagreementReport,
+  WorkflowColumnsGraduationReport,
+  GraduationReportInputs,
 } from "./workflow-parity.js";
 export { isResearchExperimentalEnabled, resolveResearchSettings } from "./research-settings.js";
 export type { ResolvedResearchSettings } from "./research-settings.js";

packages/core/src/store.ts

@@ -15,7 +15,7 @@ import {
   findWorkflowColumn,
   resolveColumnPluginGates,
 } from "./plugin-gate-verdict.js";
-import { getTraitRegistry } from "./trait-registry.js";
+import { getTraitRegistry, assertColumnTraitsValid } from "./trait-registry.js";
 import { resolveColumnCapacity } from "./workflow-capacity.js";
 import {
   OccupiedColumnsError,
@@ -38,7 +38,7 @@ import {
 } from "./transition-types.js";
 import { writeTransitionPending, clearTransitionPending } from "./transition-pending.js";
 import { BUILTIN_CODING_WORKFLOW_IR } from "./builtin-coding-workflow-ir.js";
-import type { WorkflowIr } from "./workflow-ir-types.js";
+import type { WorkflowIr, WorkflowIrColumn } from "./workflow-ir-types.js";
 // Side-effect import: registers the 14 built-in trait DEFINITIONS into the
 // shared trait registry on load (the flag-ON path resolves traits by id).
 import "./builtin-traits.js";
@@ -53,8 +53,11 @@ import { BUILTIN_WORKFLOWS, getBuiltinWorkflow, isBuiltinWorkflowId } from "./bu
 import {
   WORKFLOW_PARITY_OBSERVED_MUTATION,
   WORKFLOW_PARITY_DRIFT_MUTATION,
+  DUAL_ACCEPT_PARITY_MUTATIONS,
+  computeWorkflowColumnsGraduationReport,
   type WorkflowParityDiff,
   type WorkflowParitySummary,
+  type WorkflowColumnsGraduationReport,
 } from "./workflow-parity.js";
 
 /** Tags WorkflowStep rows materialized by compiling a workflow so they can be
@@ -1590,6 +1593,22 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
         error: err instanceof Error ? err.message : String(err),
       });
     }
+
+    // U12: workflow-columns integrity pass. When the flag is ON, audit + re-home
+    // any task whose stored column is no longer valid in its resolved workflow
+    // (KTD-1 guarantees zero rewrites for healthy legacy rows, so this is a
+    // no-op for the common case). Idempotent; non-fatal — never blocks startup.
+    try {
+      const settings = await this.getSettingsFast();
+      if (isWorkflowColumnsEnabled(settings)) {
+        await this.runWorkflowColumnsIntegrityPass();
+      }
+    } catch (err) {
+      storeLog.warn("workflowColumns integrity pass failed during init", {
+        phase: "init:workflow-columns-integrity",
+        error: err instanceof Error ? err.message : String(err),
+      });
+    }
   }
 
   // ── Row <-> Task Conversion ────────────────────────────────────────
@@ -4982,6 +5001,82 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     return sorted.slice(offset, offset + Math.max(0, limit));
   }
 
+  /**
+   * Residual B (U13/U9): per-branch progress snapshots for the given tasks,
+   * read from the `workflow_run_branches` table. Used to populate the optional
+   * additive `branchProgress` field on the board task payload so U9's parallel-
+   * window badge can render. Cheap and additive:
+   *   - returns an empty map immediately when the table is empty (the common
+   *     case — no fan-out runs in flight);
+   *   - one query for the whole task batch (no per-card N+1);
+   *   - returns only the LATEST run's branches per task (a card is in exactly
+   *     one parallel window at a time — KTD-11 one-card-one-position).
+   * Never throws on a missing/legacy table (additive guard).
+   */
+  getBranchProgressByTask(
+    taskIds: readonly string[],
+  ): Map<string, Array<{ branchId: string; nodeId: string; status: string }>> {
+    const result = new Map<string, Array<{ branchId: string; nodeId: string; status: string }>>();
+    if (taskIds.length === 0) return result;
+    try {
+      // Skip entirely when the table has no rows (cheap existence probe).
+      const any = this.db
+        .prepare("SELECT 1 FROM workflow_run_branches LIMIT 1")
+        .get();
+      if (!any) return result;
+
+      const placeholders = taskIds.map(() => "?").join(", ");
+      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
+            WHERE b.taskId IN (${placeholders})`,
+        )
+        .all(...taskIds, ...taskIds) as Array<{
+          taskId: string;
+          runId: string;
+          branchId: string;
+          nodeId: string;
+          status: string;
+          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 latestRunByTask = new Map<string, string>();
+      for (const row of rows) {
+        const known = latestRunByTask.get(row.taskId);
+        if (!known) latestRunByTask.set(row.taskId, row.runId);
+      }
+      // Re-derive the latest runId precisely from the max-updatedAt row.
+      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);
+      }
+    } catch {
+      // Legacy/missing table or query failure — degrade to no branch progress.
+      return new Map();
+    }
+    return result;
+  }
+
   async listTasksForGithubTrackingReconcile(options?: { offset?: number; limit?: number }): Promise<{ tasks: Task[]; hasMore: boolean }> {
     const reconcileScanLimit = 200;
     const offset = Math.max(0, options?.offset ?? 0);
@@ -6184,11 +6279,29 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     // the fire-and-forget hook runner per KTD-2. It is idempotent and clears the
     // transitionPending marker once done. A crash before this point leaves the
     // marker for the recovery sweep to re-run (re-running is a no-op for the
-    // default workflow's already-committed field effects). We clear it
+    // default workflow's already-committed field effects).
-    // synchronously here because the default workflow has no async post-commit
+    //
-    // hook bodies in U4 (merge enqueue is in-txn via the handoff path);
+    // Residual C (U8): AFTER the built-in effects, invoke registered PLUGIN
-    // plugin/async post-commit hooks land in U7/U8 and will defer the clear.
+    // onExit (from column) / onEnter (to column) trait hook impls, recording
+    // per-hook completion in the marker's hooksRemaining. A throwing plugin hook
+    // DEGRADES (audit) and never wedges the lock or strands the marker — the
+    // marker is always cleared at the end regardless of hook failures.
     if (useWorkflow) {
+      // Plugin hooks are skipped on engine/recovery-sourced moves (KTD-9 — those
+      // bypass trait effects) and on same-column no-ops.
+      if (!bypassGuards && fromColumn !== toColumn && workflowIr) {
+        try {
+          await this.runPluginColumnTransitionHooks(id, workflowIr, fromColumn, toColumn);
+        } catch (err) {
+          // The runner itself swallows per-hook failures; this is a final guard
+          // so a runner-level fault never strands the marker.
+          storeLog.warn("Plugin column transition hook runner faulted (degraded)", {
+            phase: "moveTaskInternal:plugin-hooks",
+            taskId: id,
+            error: err instanceof Error ? err.message : String(err),
+          });
+        }
+      }
       try {
         clearTransitionPending(this.db, id);
       } catch {
@@ -6202,6 +6315,111 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     return task;
   }
 
+  /**
+   * Residual C (U8): run registered PLUGIN onExit (from column) / onEnter (to
+   * column) trait hook impls AFTER the built-in default-workflow effects, on the
+   * post-commit path. Plugin hooks are async-only (KTD-7) and route through the
+   * registry's resolved impl (the engine wires `runCustomNode` in via the trait
+   * adapter; an unregistered/degraded hook resolves to a no-op + audit warning).
+   *
+   * Per-hook completion is recorded in the `transitionPending` marker's
+   * `hooksRemaining` so a crash mid-hook is recoverable. A hook that THROWS is
+   * audited (`plugin:trait-hook-degraded`) and treated as completed (removed
+   * from `hooksRemaining`) — a misbehaving plugin never wedges the task lock or
+   * strands the card (KTD-2 degraded-not-stranded posture). The caller clears
+   * the marker after this returns.
+   */
+  private async runPluginColumnTransitionHooks(
+    taskId: string,
+    workflowIr: WorkflowIr,
+    fromColumn: string,
+    toColumn: string,
+  ): Promise<void> {
+    const registry = getTraitRegistry();
+    // Collect (traitId, hookKind) pairs: onExit for from-column plugin traits,
+    // onEnter for to-column plugin traits. Only plugin-namespaced traits (KTD-7).
+    const pending: Array<{ traitId: string; hookKind: "onEnter" | "onExit" }> = [];
+    const fromCol = findWorkflowColumn(workflowIr, fromColumn);
+    for (const ct of fromCol?.traits ?? []) {
+      if (!ct.trait.startsWith("plugin:")) continue;
+      const def = registry.getTrait(ct.trait);
+      if (def?.hooks?.onExit) pending.push({ traitId: ct.trait, hookKind: "onExit" });
+    }
+    const toCol = findWorkflowColumn(workflowIr, toColumn);
+    for (const ct of toCol?.traits ?? []) {
+      if (!ct.trait.startsWith("plugin:")) continue;
+      const def = registry.getTrait(ct.trait);
+      if (def?.hooks?.onEnter) pending.push({ traitId: ct.trait, hookKind: "onEnter" });
+    }
+    if (pending.length === 0) return;
+
+    // Record the plugin hooks in the marker's hooksRemaining (alongside the
+    // default-workflow:postCommit marker already written in-txn) so a crash
+    // mid-hook is recoverable.
+    const hookIds = pending.map((p) => `${p.traitId}:${p.hookKind}`);
+    const startedAt = Date.now();
+    try {
+      writeTransitionPending(
+        this.db,
+        taskId,
+        makeTransitionPending(toColumn, ["default-workflow:postCommit", ...hookIds], startedAt),
+      );
+    } catch {
+      // Marker bookkeeping is best-effort; proceed to run the hooks regardless.
+    }
+
+    // Read the task once for hook context. MUST be a non-locking read — this
+    // runs inside `withTaskLock`, so `getTask` (which re-acquires the lock)
+    // would deadlock. `readTaskFromDb` is the in-lock-safe read.
+    const taskRow = this.readTaskFromDb(taskId, { includeDeleted: false });
+    const taskDetail = taskRow as unknown as TaskDetail | undefined;
+
+    const remaining = ["default-workflow:postCommit", ...hookIds];
+    for (const { traitId, hookKind } of pending) {
+      const resolved = registry.resolveTraitHook(traitId, hookKind);
+      if (resolved.warning) {
+        // Degraded (no impl / force-disabled) → passive no-op, audit the warning.
+        this.recordRunAuditEvent({
+          taskId,
+          agentId: "system",
+          runId: `plugin-trait-hook-${traitId}-${taskId}-${Date.now()}`,
+          domain: "database",
+          mutationType: "plugin:trait-hook-degraded",
+          target: taskId,
+          metadata: { traitId, hookKind, reason: "no-impl", message: resolved.warning.message },
+        });
+      } else if (resolved.impl) {
+        try {
+          await resolved.impl({ task: taskDetail, context: { fromColumn, toColumn, hookKind } });
+        } catch (err) {
+          // A throwing plugin hook DEGRADES — audited, never wedges the lock.
+          this.recordRunAuditEvent({
+            taskId,
+            agentId: "system",
+            runId: `plugin-trait-hook-${traitId}-${taskId}-${Date.now()}`,
+            domain: "database",
+            mutationType: "plugin:trait-hook-degraded",
+            target: taskId,
+            metadata: {
+              traitId,
+              hookKind,
+              reason: "threw",
+              error: err instanceof Error ? err.message : String(err),
+            },
+          });
+        }
+      }
+      // Mark this hook complete in the marker (whether it ran, degraded, or threw).
+      const idx = remaining.indexOf(`${traitId}:${hookKind}`);
+      if (idx >= 0) remaining.splice(idx, 1);
+      try {
+        writeTransitionPending(this.db, taskId, makeTransitionPending(toColumn, remaining, startedAt));
+      } catch {
+        // Best-effort progress bookkeeping; the final clear is the backstop.
+      }
+    }
+  }
+
   private resetAllStepsToPending(task: Task): void {
     if (task.steps.length === 0) {
       return;
@@ -7729,6 +7947,37 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
     };
   }
 
+  /**
+   * Aggregate the `workflowColumns` flag default-flip criteria (U12, KTD-8) into
+   * a single graduation report: five-invariant dual-observe parity, the default
+   * workflow's transition parity vs VALID_TRANSITIONS, and the dual-accept
+   * marker/column disagreement count (U6, FN-5719). The flip is a FIELD decision
+   * — this report is the GATE. Does NOT flip the flag; callers inspect `ready`
+   * and `blockers`.
+   */
+  computeWorkflowColumnsGraduationReport(
+    options: { since?: string; limit?: number } = {},
+  ): WorkflowColumnsGraduationReport {
+    const limit = options.limit ?? 1000;
+    const parity = this.getWorkflowParitySummary(options);
+    const dualAcceptEvents: RunAuditEvent[] = [];
+    for (const mutationType of DUAL_ACCEPT_PARITY_MUTATIONS) {
+      dualAcceptEvents.push(
+        ...this.getRunAuditEvents({
+          domain: "database",
+          mutationType: mutationType as unknown as RunAuditEvent["mutationType"],
+          startTime: options.since,
+          limit,
+        }),
+      );
+    }
+    return computeWorkflowColumnsGraduationReport({
+      parity,
+      defaultWorkflowIr: BUILTIN_CODING_WORKFLOW_IR,
+      dualAcceptEvents,
+    });
+  }
+
   enqueueMergeQueue(taskId: string, opts: MergeQueueEnqueueOptions = {}): MergeQueueEntry {
     let invalidColumn: Column | null = null;
     const entry = this.db.transactionImmediate(() => {
@@ -11556,6 +11805,17 @@ ${stepsSection}`;
     return {};
   }
 
+  /** Server-side trait-composition validation (residual A). Throws a typed
+   *  ColumnTraitValidationError when the IR's columns have save-blocking trait
+   *  conflicts, so conflicts reject server-side and not only in the editor. A
+   *  v1 IR (no columns) is a no-op. */
+  private assertWorkflowIrTraitsValid(ir: WorkflowIr): void {
+    const columns = (ir as { columns?: WorkflowIrColumn[] }).columns;
+    if (Array.isArray(columns) && columns.length > 0) {
+      assertColumnTraitsValid(columns);
+    }
+  }
+
   /** Create a named workflow definition. The IR is validated via parseWorkflowIr. */
   async createWorkflowDefinition(
     input: WorkflowDefinitionInput,
@@ -11565,6 +11825,9 @@ ${stepsSection}`;
       if (!name) throw new Error("Workflow name is required");
       // Validate the IR shape up front so we never persist a malformed graph.
       const ir = parseWorkflowIr(input.ir);
+      // Residual A: also reject save-blocking trait composition conflicts here,
+      // not only in the editor's client-side validation.
+      this.assertWorkflowIrTraitsValid(ir);
       const layout = input.layout ?? {};
       const now = new Date().toISOString();
       const id = this.nextWorkflowDefinitionId();
@@ -11681,6 +11944,9 @@ ${stepsSection}`;
       const name = updates.name !== undefined ? updates.name.trim() : existing.name;
       if (!name) throw new Error("Workflow name is required");
       const ir = updates.ir !== undefined ? parseWorkflowIr(updates.ir) : existing.ir;
+      // Residual A: reject save-blocking trait composition conflicts server-side
+      // when the IR is being changed.
+      if (updates.ir !== undefined) this.assertWorkflowIrTraitsValid(ir);
       const next: WorkflowDefinition = {
         ...existing,
         name,
@@ -11903,6 +12169,87 @@ ${stepsSection}`;
     });
   }
 
+  // ── U12: workflow-columns integrity pass ──────────────────────────────────
+  //
+  // Migration rewrites ZERO task rows (KTD-1): a null selection resolves to the
+  // built-in default workflow at read time, and the default workflow's column
+  // IDs are byte-identical to the legacy enum values, so every legacy row is
+  // already valid. The only residual risk is a task whose stored column is not a
+  // valid column in its RESOLVED workflow — e.g. a custom workflow was edited to
+  // drop a column out-of-band, or a legacy row references a column the selected
+  // custom workflow never defined. The integrity pass audits those and re-homes
+  // them via the U5 reconciliation path (`recoveryRehome`, guard-bypassing,
+  // capacity-honoring), one audit event per card.
+  //
+  // Idempotent: a second run finds nothing out-of-place (the re-home lands the
+  // card in a valid column) and is a pure no-op. Tasks in complete- or
+  // archived-flagged columns are left UNTOUCHED (done/archived cards are terminal
+  // — re-homing them would corrupt the board) even if (defensively) their column
+  // were somehow not in the resolved IR; we never disturb terminal cards.
+  //
+  // Runs only when the `workflowColumns` flag is ON (flag-OFF keeps the legacy
+  // enum path, where every column is valid by construction).
+  async runWorkflowColumnsIntegrityPass(): Promise<{ scanned: number; rehomed: number; skippedTerminal: number }> {
+    let scanned = 0;
+    let rehomed = 0;
+    let skippedTerminal = 0;
+
+    const rows = this.db
+      .prepare(`SELECT id FROM tasks WHERE "deletedAt" IS NULL`)
+      .all() as Array<{ id: string }>;
+
+    const registry = getTraitRegistry();
+
+    for (const { id } of rows) {
+      scanned += 1;
+      const task = this.readTaskFromDb(id, { includeDeleted: false });
+      if (!task) continue;
+      const ir = this.resolveTaskWorkflowIrSync(id);
+      const currentColumn = task.column;
+
+      // Already valid in its resolved workflow — nothing to do (the common case;
+      // this is why the pass is idempotent and a no-op for healthy DBs).
+      if (workflowHasColumn(ir, currentColumn)) continue;
+
+      // The stored column is not in the resolved workflow. Before re-homing,
+      // never disturb a terminal card: if the column the card sits in carries a
+      // complete/archived flag in its workflow it is terminal — but since the
+      // column is NOT in the IR we cannot read its flags there. Fall back to the
+      // legacy terminal semantics (done/archived) so terminal cards are never
+      // re-homed, matching the plan's "done/archived untouched" rule.
+      const column = findWorkflowColumn(ir, currentColumn);
+      const flags = column ? registry.resolveColumnFlags(column) : undefined;
+      const isTerminal =
+        flags?.complete === true ||
+        flags?.archived === true ||
+        currentColumn === "done" ||
+        currentColumn === "archived";
+      if (isTerminal) {
+        skippedTerminal += 1;
+        continue;
+      }
+
+      const targetColumn = resolveEntryColumnId(ir);
+      if (!targetColumn) continue; // non-reconcilable IR — leave the card put.
+
+      await this.rehomeOccupant(id, targetColumn, "workflow-edit-rehome", {
+        integrityPass: true,
+        invalidColumn: currentColumn,
+      });
+      rehomed += 1;
+    }
+
+    if (rehomed > 0 || skippedTerminal > 0) {
+      storeLog.log("workflowColumns integrity pass completed", {
+        phase: "init:workflow-columns-integrity",
+        scanned,
+        rehomed,
+        skippedTerminal,
+      });
+    }
+    return { scanned, rehomed, skippedTerminal };
+  }
+
   // ── Workflow selection (resolves a workflow to enabledWorkflowSteps) ────
   //
   // Selection never touches the engine's scheduler/executor/merger. It compiles

packages/core/src/trait-registry.ts

@@ -71,6 +71,23 @@ export interface TraitViolation {
   message: string;
 }
 
+/**
+ * Thrown by the store's create/update workflow paths (residual A) when a
+ * workflow's trait composition has `error`-severity violations under `save`
+ * mode — so trait conflicts reject server-side, not only in the editor. Carries
+ * the structured violations so the surface can render them per-column. The
+ * dashboard routes map this to a 400 (consistent with `WorkflowIrError`).
+ */
+export class ColumnTraitValidationError extends Error {
+  readonly violations: TraitViolation[];
+  constructor(violations: TraitViolation[]) {
+    const summary = violations.map((v) => v.message).join("; ");
+    super(`Workflow trait composition invalid: ${summary}`);
+    this.name = "ColumnTraitValidationError";
+    this.violations = violations;
+  }
+}
+
 /** A simple audit-warning record returned by hook resolution / load-time
  *  re-validation. Modeled as a returned value (not a thrown error and not an
  *  engine logger) so core stays engine-free; callers may forward it to audit. */
@@ -390,6 +407,19 @@ export function validateColumnTraits(
   return getTraitRegistry().validateColumnTraits(columns, mode);
 }
 
+/**
+ * Save-mode composition validation that THROWS (residual A). Runs the registry's
+ * `validateColumnTraits` in `save` mode and throws a {@link ColumnTraitValidationError}
+ * if any `error`-severity violations are present. `degraded` advisories are
+ * ignored (they never block a save). A no-op for `[]`/no-error columns.
+ */
+export function assertColumnTraitsValid(columns: WorkflowIrColumn[]): void {
+  const violations = getTraitRegistry()
+    .validateColumnTraits(columns, "save")
+    .filter((v) => v.severity === "error");
+  if (violations.length > 0) throw new ColumnTraitValidationError(violations);
+}
+
 export function registerTraitHookImpl(
   traitId: string,
   hookKind: TraitHookKind,

packages/core/src/types.ts

@@ -15,7 +15,22 @@ export type { CapacityRiskSignal } from "./capacity.js";
 export const THINKING_LEVELS = ["off", "minimal", "low", "medium", "high"] as const;
 export type ThinkingLevel = (typeof THINKING_LEVELS)[number];
 
+/**
+ * The legacy default-workflow column set. Under
+ * `experimentalFeatures.workflowColumns` a task's valid columns are resolved
+ * from its workflow definition (the default workflow's column IDs are
+ * byte-identical to these — KTD-1). New flag-aware code should prefer the
+ * workflow-resolved path (`resolveAllowedColumns` / `workflowHasColumn` in
+ * `workflow-transitions.ts`) and trait-flag predicates over string equality;
+ * this enum remains the canonical id set for the built-in default workflow.
+ */
 export const COLUMNS = ["triage", "todo", "in-progress", "in-review", "done", "archived"] as const;
+/**
+ * The closed legacy column union — still the correct type for default-workflow
+ * column ids and the flag-OFF path. Movement entry points accept the wider
+ * {@link ColumnId}; flag-ON code validates ids against the task's resolved
+ * workflow at runtime.
+ */
 export type Column = (typeof COLUMNS)[number];
 
 /**
@@ -29,10 +44,22 @@ export type ColumnId = Column | (string & {});
 
 export const DEFAULT_COLUMN: Column = "triage";
 
+/**
+ * Tests membership against the closed legacy column enum. Note: under the
+ * workflowColumns flag, column validity is workflow-scoped — flag-aware code
+ * should use `workflowHasColumn(ir, columnId)` (`workflow-transitions.ts`);
+ * this remains correct for the flag-OFF path and default-workflow ids.
+ */
 export function isColumn(value: unknown): value is Column {
   return typeof value === "string" && (COLUMNS as readonly string[]).includes(value);
 }
 
+/**
+ * @deprecated (workflowColumns, U12) Coerces an arbitrary value to a legacy
+ * column, DISCARDING workflow-defined custom column ids — lossy under the
+ * flag. Resolve and validate against the task's workflow instead. Retained
+ * for the legacy flag-OFF path while the flag exists.
+ */
 export function normalizeColumn(value: unknown, fallback: Column = DEFAULT_COLUMN): Column {
   return isColumn(value) ? value : fallback;
 }
@@ -3959,6 +3986,15 @@ export const COLUMN_DESCRIPTIONS: Record<Column, string> = {
   archived: "Completed and archived",
 };
 
+/**
+ * @deprecated (workflowColumns, U12) The hardcoded legacy transition graph.
+ * Under `experimentalFeatures.workflowColumns`, transition validity is resolved
+ * from the task's workflow column graph (`resolveAllowedColumns` in
+ * `workflow-transitions.ts`) plus trait guards in `moveTaskInternal` — this
+ * constant is now only the flag-OFF authority and the parity oracle the default
+ * workflow is machine-checked against (transition-parity suite). Retained while
+ * the flag exists; do NOT remove until graduation + legacy-path deletion.
+ */
 export const VALID_TRANSITIONS: Record<Column, Column[]> = {
   // FN-4892: intake-side heuristics may cold-archive tasks before execution starts.
   triage: ["todo", "archived"],

packages/core/src/workflow-parity.ts

@@ -1,4 +1,7 @@
-import type { RunAuditEvent } from "./types.js";
+import type { Column, RunAuditEvent } from "./types.js";
+import { VALID_TRANSITIONS } from "./types.js";
+import type { WorkflowIr } from "./workflow-ir-types.js";
+import { resolveAllowedColumns, workflowHasColumn } from "./workflow-transitions.js";
 
 export const WORKFLOW_PARITY_OBSERVED_MUTATION = "workflow:parity-observed" as const;
 export const WORKFLOW_PARITY_DRIFT_MUTATION = "workflow:parity-drift" as const;
@@ -315,3 +318,168 @@ export function buildWorkflowObservation(parts: WorkflowObservationParts): Workf
     invariants: { ...DEFAULT_WORKFLOW_INVARIANTS, ...parts.invariants },
   };
 }
+
+// ── Transition parity (U12) ──────────────────────────────────────────────────
+//
+// The transition-parity suite (U4) proves, as a unit test, that the default
+// workflow's resolved column adjacency equals the legacy VALID_TRANSITIONS
+// graph. U12 surfaces the SAME comparison as a runtime check so the graduation
+// gate can re-evaluate it against whatever IR is actually resolved for the
+// default workflow in the field (not just the static fixture), catching a
+// deliberately or accidentally drifted default-workflow adjacency.
+
+/** One adjacency disagreement between the legacy graph and the resolved IR. */
+export interface TransitionParityDiff {
+  /** The `from` column whose allowed-set diverged. */
+  from: string;
+  /** Allowed targets per the legacy VALID_TRANSITIONS graph. */
+  legacyAllowed: string[];
+  /** Allowed targets per the resolved workflow IR column graph. */
+  resolvedAllowed: string[];
+}
+
+export interface TransitionParityReport {
+  /** True when every legacy column's allowed-set matches the resolved IR's. */
+  agree: boolean;
+  /** Per-column adjacency disagreements (empty when `agree`). */
+  diffs: TransitionParityDiff[];
+}
+
+const LEGACY_COLUMNS = Object.keys(VALID_TRANSITIONS) as Column[];
+
+function sortedUnique(values: readonly string[]): string[] {
+  return [...new Set(values)].sort();
+}
+
+/**
+ * Compare the default-workflow IR's resolved column adjacency against the legacy
+ * VALID_TRANSITIONS graph (R12 transition parity, machine-checked). For every
+ * legacy column, the resolved allowed-set must equal the legacy allowed-set
+ * exactly (allowed AND rejected). The IR must also recognize every legacy
+ * column. Any divergence is a graduation blocker.
+ */
+export function checkTransitionParity(ir: WorkflowIr): TransitionParityReport {
+  const diffs: TransitionParityDiff[] = [];
+  for (const from of LEGACY_COLUMNS) {
+    const legacyAllowed = sortedUnique(VALID_TRANSITIONS[from]);
+    // A column the resolved IR doesn't even define diverges by construction.
+    const resolvedAllowed = workflowHasColumn(ir, from)
+      ? sortedUnique(resolveAllowedColumns(ir, from))
+      : [];
+    const equal =
+      legacyAllowed.length === resolvedAllowed.length &&
+      legacyAllowed.every((value, index) => value === resolvedAllowed[index]);
+    if (!equal) diffs.push({ from, legacyAllowed, resolvedAllowed });
+  }
+  return { agree: diffs.length === 0, diffs };
+}
+
+// ── Dual-accept disagreement counter (U12) ───────────────────────────────────
+//
+// U6 logs `merge:dependency-parity-diff` audits whenever the explicit handoff
+// marker and the complete-flag column disagree during the FN-5719 dual-accept
+// window. The window CLOSES at graduation, so any disagreement above zero over
+// the observation period blocks the flip. This surfaces the count (and the
+// lease-parity counterpart) from the audit trail as a graduation signal.
+
+export const DUAL_ACCEPT_PARITY_MUTATIONS = [
+  "merge:dependency-parity-diff",
+  "merge:lease-parity-diff",
+] as const;
+
+const DUAL_ACCEPT_PARITY_MUTATION_SET = new Set<string>(DUAL_ACCEPT_PARITY_MUTATIONS);
+
+export interface DualAcceptDisagreementReport {
+  /** Total dual-accept disagreement audit events in scope. */
+  total: number;
+  /** Count per mutation type (dependency vs lease parity diff). */
+  byMutationType: Record<string, number>;
+}
+
+/**
+ * Count the dual-accept marker/column disagreement audits (U6) in scope. Pure
+ * over the supplied events so the store can feed it whatever audit window the
+ * graduation report observes.
+ */
+export function countDualAcceptDisagreements(
+  events: readonly RunAuditEvent[],
+): DualAcceptDisagreementReport {
+  const byMutationType: Record<string, number> = {};
+  let total = 0;
+  for (const event of events) {
+    const type = String(event.mutationType);
+    if (event.domain !== "database" || !DUAL_ACCEPT_PARITY_MUTATION_SET.has(type)) continue;
+    byMutationType[type] = (byMutationType[type] ?? 0) + 1;
+    total += 1;
+  }
+  return { total, byMutationType };
+}
+
+// ── Graduation report (U12) ──────────────────────────────────────────────────
+//
+// The flag default-flip criteria, aggregated into one report (KTD-8). The flip
+// is a FIELD decision — this report is the GATE, not the trigger. `ready` is
+// true only when ALL of:
+//   - the five-invariant dual-observe parity shows zero drift (drift === 0) over
+//     a non-empty observation window;
+//   - the default workflow's transition parity holds (no adjacency drift);
+//   - zero dual-accept marker/column disagreements over the window.
+
+export interface WorkflowColumnsGraduationReport {
+  /** Five-invariant dual-observe parity (from the audit trail). */
+  parity: WorkflowParitySummary;
+  /** Default-workflow transition-graph parity vs VALID_TRANSITIONS. */
+  transitionParity: TransitionParityReport;
+  /** Dual-accept marker/column disagreement count (U6). */
+  dualAccept: DualAcceptDisagreementReport;
+  /** True only when every gate passes — the flag is eligible to default on. */
+  ready: boolean;
+  /** Human-readable blockers when not ready (empty when ready). */
+  blockers: string[];
+}
+
+export interface GraduationReportInputs {
+  /** Dual-observe parity summary (e.g. `store.getWorkflowParitySummary()`). */
+  parity: WorkflowParitySummary;
+  /** The resolved default-workflow IR to transition-parity-check. */
+  defaultWorkflowIr: WorkflowIr;
+  /** Audit events in the observation window for dual-accept counting. */
+  dualAcceptEvents: readonly RunAuditEvent[];
+}
+
+/**
+ * Aggregate the flag default-flip criteria into a single graduation report
+ * (U12, absorbing plan 002's M-D). Pure: the caller assembles the inputs from
+ * the store's audit trail and resolved default workflow, and decides whether to
+ * flip the flag — this function only computes the gate.
+ */
+export function computeWorkflowColumnsGraduationReport(
+  inputs: GraduationReportInputs,
+): WorkflowColumnsGraduationReport {
+  const { parity, defaultWorkflowIr, dualAcceptEvents } = inputs;
+  const transitionParity = checkTransitionParity(defaultWorkflowIr);
+  const dualAccept = countDualAcceptDisagreements(dualAcceptEvents);
+
+  const blockers: string[] = [];
+  if (parity.observed === 0) {
+    blockers.push("no parity observations recorded yet (observation window empty)");
+  }
+  if (parity.drift > 0) {
+    blockers.push(`five-invariant parity drift observed (${parity.drift} drift events)`);
+  }
+  if (!transitionParity.agree) {
+    const cols = transitionParity.diffs.map((d) => d.from).join(", ");
+    blockers.push(`default-workflow transition parity drifted (columns: ${cols})`);
+  }
+  if (dualAccept.total > 0) {
+    blockers.push(`dual-accept marker/column disagreements above zero (${dualAccept.total})`);
+  }
+
+  return {
+    parity,
+    transitionParity,
+    dualAccept,
+    ready: blockers.length === 0,
+    blockers,
+  };
+}

packages/dashboard/src/__tests__/workflow-routes.test.ts

@@ -96,6 +96,68 @@ describe("workflow routes (U4)", () => {
     expect(bad.status).toBe(400);
   });
 
+  it("Residual A: POST /workflows rejects a server-side trait composition conflict with 400 + violations", async () => {
+    // A v2 column carrying BOTH `complete` and `wip` (countsTowardWip) — a
+    // terminal column cannot also hold a capacity slot. parseWorkflowIr accepts
+    // the shape; the save-mode composition validator must reject it.
+    const conflictIr: WorkflowIr = {
+      version: "v2",
+      name: "conflict",
+      columns: [
+        { id: "intake-col", name: "Intake", traits: [{ trait: "intake" }] },
+        { id: "bad-col", name: "Bad", traits: [{ trait: "complete" }, { trait: "wip", config: { limit: 1 } }] },
+      ],
+      nodes: [
+        { id: "start", kind: "start", column: "intake-col" },
+        { id: "end", kind: "end", column: "bad-col" },
+      ],
+      edges: [{ from: "start", to: "end" }],
+    } as WorkflowIr;
+    const res = await post("/api/workflows", { name: "Conflict", ir: conflictIr });
+    expect(res.status).toBe(400);
+    const details = (res.body as { details?: { violations?: unknown[] } }).details;
+    expect(Array.isArray(details?.violations)).toBe(true);
+    expect((details?.violations?.length ?? 0)).toBeGreaterThan(0);
+  });
+
+  it("Residual A: PATCH /workflows/:id rejects a trait composition conflict server-side", async () => {
+    const created = await post("/api/workflows", {
+      name: "Editable",
+      ir: {
+        version: "v2",
+        name: "editable",
+        columns: [
+          { id: "intake-col", name: "Intake", traits: [{ trait: "intake" }] },
+          { id: "work-col", name: "Work", traits: [] },
+        ],
+        nodes: [
+          { id: "start", kind: "start", column: "intake-col" },
+          { id: "end", kind: "end", column: "work-col" },
+        ],
+        edges: [{ from: "start", to: "end" }],
+      },
+    });
+    expect(created.status).toBe(201);
+    const id = (created.body as { id: string }).id;
+    const conflictIr: WorkflowIr = {
+      version: "v2",
+      name: "editable",
+      columns: [
+        { id: "intake-col", name: "Intake", traits: [{ trait: "intake" }] },
+        { id: "work-col", name: "Work", traits: [{ trait: "complete" }, { trait: "wip", config: { limit: 2 } }] },
+      ],
+      nodes: [
+        { id: "start", kind: "start", column: "intake-col" },
+        { id: "end", kind: "end", column: "work-col" },
+      ],
+      edges: [{ from: "start", to: "end" }],
+    } as WorkflowIr;
+    const res = await request(app, "PATCH", `/api/workflows/${id}`, JSON.stringify({ ir: conflictIr }), {
+      "content-type": "application/json",
+    });
+    expect(res.status).toBe(400);
+  });
+
   it("GET /workflows lists created workflows (ahead of read-only built-ins)", async () => {
     await post("/api/workflows", { name: "A", ir: linearIr() });
     const res = await get("/api/workflows");

packages/dashboard/src/routes/register-task-workflow-routes.ts

@@ -776,6 +776,28 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
         const listOptions = { limit, offset, slim: true, includeArchived, ...(column ? { column } : {}) };
         tasks = await scopedStore.listTasks(listOptions);
       }
+
+      // Residual B (U9/U13): additively populate `branchProgress` when the
+      // workflowColumns flag is ON and the fan-out branch table has rows for
+      // any of these tasks. One batched query (cheap; short-circuits when the
+      // table is empty). The payload is otherwise byte-identical.
+      try {
+        const settings = await scopedStore.getSettingsFast();
+        if (isWorkflowColumnsEnabled(settings) && tasks.length > 0) {
+          const byTask = scopedStore.getBranchProgressByTask(tasks.map((t) => t.id));
+          if (byTask.size > 0) {
+            tasks = tasks.map((task) => {
+              const branchProgress = byTask.get(task.id);
+              return branchProgress && branchProgress.length > 0
+                ? { ...task, branchProgress }
+                : task;
+            });
+          }
+        }
+      } catch {
+        // Branch-progress enrichment is best-effort and must never fail the
+        // board load — fall through with the un-enriched task list.
+      }
       res.json(tasks);
     } catch (err: unknown) {
       if (err instanceof ApiError) {

packages/dashboard/src/routes/register-workflow-routes.ts

@@ -1,5 +1,5 @@
 import type { WorkflowIr } from "@fusion/core";
-import { OccupiedColumnsError, WorkflowCompileError, WorkflowIrError, compileWorkflowToSteps, listTraits } from "@fusion/core";
+import { ColumnTraitValidationError, OccupiedColumnsError, WorkflowCompileError, WorkflowIrError, compileWorkflowToSteps, listTraits } from "@fusion/core";
 import { ApiError, badRequest, conflict, notFound } from "../api-error.js";
 import type { ApiRoutesContext } from "./types.js";
 
@@ -70,6 +70,11 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void {
     } catch (err: unknown) {
       if (err instanceof ApiError) throw err;
       if (err instanceof WorkflowIrError) throw badRequest(err.message);
+      // Residual A: server-side trait composition conflict → 400 with the
+      // structured violations (consistent with the IR-error 4xx mapping).
+      if (err instanceof ColumnTraitValidationError) {
+        throw badRequest(err.message, { violations: err.violations });
+      }
       rethrowAsApiError(err);
     }
   });
@@ -118,6 +123,9 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void {
         throw conflict(err.message, { workflowId: err.workflowId, occupancies: err.occupancies });
       }
       if (err instanceof WorkflowIrError) throw badRequest(err.message);
+      if (err instanceof ColumnTraitValidationError) {
+        throw badRequest(err.message, { violations: err.violations });
+      }
       if (err instanceof Error && /not found/i.test(err.message)) throw notFound(err.message);
       rethrowAsApiError(err);
     }

packages/engine/src/__tests__/plugin-traits.test.ts

@@ -419,6 +419,96 @@ describe("U8 onEnter hook degradation (card stays, marker cleared, no wedge)", (
   });
 });
 
+describe("Residual C: plugin onEnter/onExit are INVOKED on the post-commit path", () => {
+  let rootDir = "";
+  let store: TaskStore;
+  const enterTrait = pluginTraitRegistryId("notify-plugin", "enter");
+  const exitTrait = pluginTraitRegistryId("notify-plugin", "exit");
+  let enterCalls = 0;
+  let exitCalls = 0;
+
+  beforeEach(async () => {
+    freshRegistry();
+    enterCalls = 0;
+    exitCalls = 0;
+    const registry = getTraitRegistry();
+    registry.register({ id: enterTrait, name: "Enter", flags: { notify: true }, hooks: { onEnter: true }, builtin: false });
+    registry.register({ id: exitTrait, name: "Exit", flags: { notify: true }, hooks: { onExit: true }, builtin: false });
+    registry.registerTraitHookImpl(enterTrait, "onEnter", () => { enterCalls += 1; });
+    registry.registerTraitHookImpl(exitTrait, "onExit", () => { exitCalls += 1; });
+
+    rootDir = mkdtempSync(join(tmpdir(), "u8-cohooks-"));
+    git(rootDir, "init -b main");
+    git(rootDir, "config user.name 'Fusion'");
+    git(rootDir, "config user.email 'hi@runfusion.ai'");
+    writeFileSync(join(rootDir, "README.md"), "root\n");
+    git(rootDir, "add README.md");
+    git(rootDir, "commit -m init");
+    store = new TaskStore(rootDir, undefined, { inMemoryDb: false });
+    await store.init();
+    await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } });
+  });
+
+  afterEach(() => {
+    try { store?.close(); } catch { /* ignore */ }
+    if (rootDir) rmSync(rootDir, { recursive: true, force: true });
+    __resetTraitRegistryForTests();
+  });
+
+  it("onEnter fires for the to-column's plugin trait; onExit fires for the from-column's", async () => {
+    // Workflow: intake-col(exit trait) → gate-col(enter trait) → done-col.
+    const ir = {
+      version: "v2",
+      name: "CoHooks",
+      columns: [
+        { id: "intake-col", name: "Intake", traits: [{ trait: "intake" }, { trait: exitTrait }] },
+        { id: "gate-col", name: "Gate", traits: [{ trait: enterTrait }] },
+        { id: "done-col", name: "Done", traits: [{ trait: "complete" }] },
+      ],
+      nodes: [
+        { id: "start", kind: "start", column: "intake-col" },
+        { id: "end", kind: "end", column: "done-col" },
+      ],
+      edges: [{ from: "start", to: "end" }],
+    } as WorkflowIr;
+    const def = await store.createWorkflowDefinition({ name: "CoHooks", ir });
+    const task = await store.createTask({ description: "card" });
+    setSelection(store, task.id, def.id);
+    setColumn(store, task.id, "intake-col");
+
+    const moved = await store.moveTask(task.id, "gate-col", { moveSource: "user" });
+    expect(moved.column).toBe("gate-col");
+    expect(enterCalls).toBe(1); // gate-col onEnter
+    expect(exitCalls).toBe(1); // intake-col onExit
+    // Marker cleared (no strand).
+    expect(readTransitionPending(store, task.id)).toBeNull();
+  });
+
+  it("engine-sourced (bypassGuards) moves skip plugin hooks (KTD-9)", async () => {
+    const ir = {
+      version: "v2",
+      name: "CoHooks2",
+      columns: [
+        { id: "intake-col", name: "Intake", traits: [{ trait: "intake" }] },
+        { id: "gate-col", name: "Gate", traits: [{ trait: enterTrait }] },
+        { id: "done-col", name: "Done", traits: [{ trait: "complete" }] },
+      ],
+      nodes: [
+        { id: "start", kind: "start", column: "intake-col" },
+        { id: "end", kind: "end", column: "done-col" },
+      ],
+      edges: [{ from: "start", to: "end" }],
+    } as WorkflowIr;
+    const def = await store.createWorkflowDefinition({ name: "CoHooks2", ir });
+    const task = await store.createTask({ description: "card" });
+    setSelection(store, task.id, def.id);
+    setColumn(store, task.id, "intake-col");
+
+    await store.moveTask(task.id, "gate-col", { moveSource: "engine", bypassGuards: true });
+    expect(enterCalls).toBe(0); // engine move bypasses trait effects
+  });
+});
+
 describe("U8 plugin loader aggregation + disable/force-disable (KTD-7)", () => {
   let rootDir = "";
   let pluginStore: PluginStore;