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

This commit is contained in:
gsxdsm
2026-06-04 02:13:48 -07:00
parent 3aa8d2d28b
commit 60605fafa9
14 changed files with 1152 additions and 8 deletions

View File

@@ -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.

View File

@@ -134,6 +134,31 @@ The user's mid-stage feedback channel: free-text guidance attached to an answer,
### Rehydration ### 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. 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 ## 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. - "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.

View File

@@ -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. 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 ## 10) Agent System
Fusion has two complementary agent models: Fusion has two complementary agent models:

View File

@@ -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. 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 ## What They Are
A workflow step is a reusable check (AI prompt or script) that can be enabled on tasks. A workflow step is a reusable check (AI prompt or script) that can be enabled on tasks.

View File

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

View File

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

View File

@@ -15,7 +15,7 @@ import {
findWorkflowColumn, findWorkflowColumn,
resolveColumnPluginGates, resolveColumnPluginGates,
} from "./plugin-gate-verdict.js"; } 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 { resolveColumnCapacity } from "./workflow-capacity.js";
import { import {
OccupiedColumnsError, OccupiedColumnsError,
@@ -38,7 +38,7 @@ import {
} from "./transition-types.js"; } from "./transition-types.js";
import { writeTransitionPending, clearTransitionPending } from "./transition-pending.js"; import { writeTransitionPending, clearTransitionPending } from "./transition-pending.js";
import { BUILTIN_CODING_WORKFLOW_IR } from "./builtin-coding-workflow-ir.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 // 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). // shared trait registry on load (the flag-ON path resolves traits by id).
import "./builtin-traits.js"; import "./builtin-traits.js";
@@ -53,8 +53,11 @@ import { BUILTIN_WORKFLOWS, getBuiltinWorkflow, isBuiltinWorkflowId } from "./bu
import { import {
WORKFLOW_PARITY_OBSERVED_MUTATION, WORKFLOW_PARITY_OBSERVED_MUTATION,
WORKFLOW_PARITY_DRIFT_MUTATION, WORKFLOW_PARITY_DRIFT_MUTATION,
DUAL_ACCEPT_PARITY_MUTATIONS,
computeWorkflowColumnsGraduationReport,
type WorkflowParityDiff, type WorkflowParityDiff,
type WorkflowParitySummary, type WorkflowParitySummary,
type WorkflowColumnsGraduationReport,
} from "./workflow-parity.js"; } from "./workflow-parity.js";
/** Tags WorkflowStep rows materialized by compiling a workflow so they can be /** 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), 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 ──────────────────────────────────────── // ── Row <-> Task Conversion ────────────────────────────────────────
@@ -4982,6 +5001,82 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
return sorted.slice(offset, offset + Math.max(0, limit)); 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 }> { async listTasksForGithubTrackingReconcile(options?: { offset?: number; limit?: number }): Promise<{ tasks: Task[]; hasMore: boolean }> {
const reconcileScanLimit = 200; const reconcileScanLimit = 200;
const offset = Math.max(0, options?.offset ?? 0); 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 // 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 // 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 // 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) { 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 { try {
clearTransitionPending(this.db, id); clearTransitionPending(this.db, id);
} catch { } catch {
@@ -6202,6 +6315,111 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
return task; 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 { private resetAllStepsToPending(task: Task): void {
if (task.steps.length === 0) { if (task.steps.length === 0) {
return; 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 { enqueueMergeQueue(taskId: string, opts: MergeQueueEnqueueOptions = {}): MergeQueueEntry {
let invalidColumn: Column | null = null; let invalidColumn: Column | null = null;
const entry = this.db.transactionImmediate(() => { const entry = this.db.transactionImmediate(() => {
@@ -11556,6 +11805,17 @@ ${stepsSection}`;
return {}; 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. */ /** Create a named workflow definition. The IR is validated via parseWorkflowIr. */
async createWorkflowDefinition( async createWorkflowDefinition(
input: WorkflowDefinitionInput, input: WorkflowDefinitionInput,
@@ -11565,6 +11825,9 @@ ${stepsSection}`;
if (!name) throw new Error("Workflow name is required"); if (!name) throw new Error("Workflow name is required");
// Validate the IR shape up front so we never persist a malformed graph. // Validate the IR shape up front so we never persist a malformed graph.
const ir = parseWorkflowIr(input.ir); 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 layout = input.layout ?? {};
const now = new Date().toISOString(); const now = new Date().toISOString();
const id = this.nextWorkflowDefinitionId(); const id = this.nextWorkflowDefinitionId();
@@ -11681,6 +11944,9 @@ ${stepsSection}`;
const name = updates.name !== undefined ? updates.name.trim() : existing.name; const name = updates.name !== undefined ? updates.name.trim() : existing.name;
if (!name) throw new Error("Workflow name is required"); if (!name) throw new Error("Workflow name is required");
const ir = updates.ir !== undefined ? parseWorkflowIr(updates.ir) : existing.ir; 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 = { const next: WorkflowDefinition = {
...existing, ...existing,
name, 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) ──── // ── Workflow selection (resolves a workflow to enabledWorkflowSteps) ────
// //
// Selection never touches the engine's scheduler/executor/merger. It compiles // Selection never touches the engine's scheduler/executor/merger. It compiles

View File

@@ -71,6 +71,23 @@ export interface TraitViolation {
message: string; 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 /** 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 * 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. */ * 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); 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( export function registerTraitHookImpl(
traitId: string, traitId: string,
hookKind: TraitHookKind, hookKind: TraitHookKind,

View File

@@ -15,7 +15,22 @@ export type { CapacityRiskSignal } from "./capacity.js";
export const THINKING_LEVELS = ["off", "minimal", "low", "medium", "high"] as const; export const THINKING_LEVELS = ["off", "minimal", "low", "medium", "high"] as const;
export type ThinkingLevel = (typeof THINKING_LEVELS)[number]; 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; 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]; export type Column = (typeof COLUMNS)[number];
/** /**
@@ -29,10 +44,22 @@ export type ColumnId = Column | (string & {});
export const DEFAULT_COLUMN: Column = "triage"; 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 { export function isColumn(value: unknown): value is Column {
return typeof value === "string" && (COLUMNS as readonly string[]).includes(value); 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 { export function normalizeColumn(value: unknown, fallback: Column = DEFAULT_COLUMN): Column {
return isColumn(value) ? value : fallback; return isColumn(value) ? value : fallback;
} }
@@ -3959,6 +3986,15 @@ export const COLUMN_DESCRIPTIONS: Record<Column, string> = {
archived: "Completed and archived", 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[]> = { export const VALID_TRANSITIONS: Record<Column, Column[]> = {
// FN-4892: intake-side heuristics may cold-archive tasks before execution starts. // FN-4892: intake-side heuristics may cold-archive tasks before execution starts.
triage: ["todo", "archived"], triage: ["todo", "archived"],

View File

@@ -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_OBSERVED_MUTATION = "workflow:parity-observed" as const;
export const WORKFLOW_PARITY_DRIFT_MUTATION = "workflow:parity-drift" 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 }, 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,
};
}

View File

@@ -96,6 +96,68 @@ describe("workflow routes (U4)", () => {
expect(bad.status).toBe(400); 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 () => { it("GET /workflows lists created workflows (ahead of read-only built-ins)", async () => {
await post("/api/workflows", { name: "A", ir: linearIr() }); await post("/api/workflows", { name: "A", ir: linearIr() });
const res = await get("/api/workflows"); const res = await get("/api/workflows");

View File

@@ -776,6 +776,28 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
const listOptions = { limit, offset, slim: true, includeArchived, ...(column ? { column } : {}) }; const listOptions = { limit, offset, slim: true, includeArchived, ...(column ? { column } : {}) };
tasks = await scopedStore.listTasks(listOptions); 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); res.json(tasks);
} catch (err: unknown) { } catch (err: unknown) {
if (err instanceof ApiError) { if (err instanceof ApiError) {

View File

@@ -1,5 +1,5 @@
import type { WorkflowIr } from "@fusion/core"; 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 { ApiError, badRequest, conflict, notFound } from "../api-error.js";
import type { ApiRoutesContext } from "./types.js"; import type { ApiRoutesContext } from "./types.js";
@@ -70,6 +70,11 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void {
} catch (err: unknown) { } catch (err: unknown) {
if (err instanceof ApiError) throw err; if (err instanceof ApiError) throw err;
if (err instanceof WorkflowIrError) throw badRequest(err.message); 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); rethrowAsApiError(err);
} }
}); });
@@ -118,6 +123,9 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void {
throw conflict(err.message, { workflowId: err.workflowId, occupancies: err.occupancies }); throw conflict(err.message, { workflowId: err.workflowId, occupancies: err.occupancies });
} }
if (err instanceof WorkflowIrError) throw badRequest(err.message); 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); if (err instanceof Error && /not found/i.test(err.message)) throw notFound(err.message);
rethrowAsApiError(err); rethrowAsApiError(err);
} }

View File

@@ -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)", () => { describe("U8 plugin loader aggregation + disable/force-disable (KTD-7)", () => {
let rootDir = ""; let rootDir = "";
let pluginStore: PluginStore; let pluginStore: PluginStore;