diff --git a/.changeset/workflow-custom-columns-traits.md b/.changeset/workflow-custom-columns-traits.md new file mode 100644 index 0000000000..c267ef940b --- /dev/null +++ b/.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. diff --git a/CONCEPTS.md b/CONCEPTS.md index 8f4df9fce6..a91b860eef 100644 --- a/CONCEPTS.md +++ b/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. diff --git a/docs/architecture.md b/docs/architecture.md index 98469f2b47..b23c6e640a 100644 --- a/docs/architecture.md +++ b/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: diff --git a/docs/workflow-steps.md b/docs/workflow-steps.md index 33f5614942..4818e04dd3 100644 --- a/docs/workflow-steps.md +++ b/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. diff --git a/packages/core/src/__tests__/migration-workflow-columns.test.ts b/packages/core/src/__tests__/migration-workflow-columns.test.ts new file mode 100644 index 0000000000..1035705577 --- /dev/null +++ b/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; + + 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 { + 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; + + 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; + + 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; + + 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[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); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index d2f17d4950..5d540b150b 100644 --- a/packages/core/src/index.ts +++ b/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"; diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index f0d0a8da05..cdb73291bf 100644 --- a/packages/core/src/store.ts +++ b/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 { 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 { 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> { + const result = new Map>(); + 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(); + 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(); + 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 { // 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 - // synchronously here because the default workflow has no async post-commit - // hook bodies in U4 (merge enqueue is in-txn via the handoff path); - // plugin/async post-commit hooks land in U7/U8 and will defer the clear. + // default workflow's already-committed field effects). + // + // Residual C (U8): AFTER the built-in effects, invoke registered PLUGIN + // 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 { 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 { + 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 { }; } + /** + * 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 diff --git a/packages/core/src/trait-registry.ts b/packages/core/src/trait-registry.ts index b946ee151d..382339b058 100644 --- a/packages/core/src/trait-registry.ts +++ b/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, diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index e40b0be208..ec1ba67da2 100644 --- a/packages/core/src/types.ts +++ b/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 = { 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 = { // FN-4892: intake-side heuristics may cold-archive tasks before execution starts. triage: ["todo", "archived"], diff --git a/packages/core/src/workflow-parity.ts b/packages/core/src/workflow-parity.ts index 56bc4a92f7..28e1b89379 100644 --- a/packages/core/src/workflow-parity.ts +++ b/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(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; +} + +/** + * 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 = {}; + 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, + }; +} diff --git a/packages/dashboard/src/__tests__/workflow-routes.test.ts b/packages/dashboard/src/__tests__/workflow-routes.test.ts index 26dd8b9365..b10d7cc769 100644 --- a/packages/dashboard/src/__tests__/workflow-routes.test.ts +++ b/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"); diff --git a/packages/dashboard/src/routes/register-task-workflow-routes.ts b/packages/dashboard/src/routes/register-task-workflow-routes.ts index d88db69e49..0bb8c13823 100644 --- a/packages/dashboard/src/routes/register-task-workflow-routes.ts +++ b/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) { diff --git a/packages/dashboard/src/routes/register-workflow-routes.ts b/packages/dashboard/src/routes/register-workflow-routes.ts index 2a0a5c1cdf..29a3dbce67 100644 --- a/packages/dashboard/src/routes/register-workflow-routes.ts +++ b/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); } diff --git a/packages/engine/src/__tests__/plugin-traits.test.ts b/packages/engine/src/__tests__/plugin-traits.test.ts index aa520f593d..c7362f0295 100644 --- a/packages/engine/src/__tests__/plugin-traits.test.ts +++ b/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;