diff --git a/packages/core/src/__tests__/db-migrate.test.ts b/packages/core/src/__tests__/db-migrate.test.ts index 9c1eb7dd74..8a989400d8 100644 --- a/packages/core/src/__tests__/db-migrate.test.ts +++ b/packages/core/src/__tests__/db-migrate.test.ts @@ -715,7 +715,7 @@ describe("schema migration", () => { const row = db.prepare("SELECT deletedAt FROM tasks WHERE id = 'FN-legacy'").get() as { deletedAt: string | null }; expect(row.deletedAt).toBeNull(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); db.close(); }); @@ -748,7 +748,7 @@ describe("schema migration", () => { { id: "WS-001", mode: "prompt", gateMode: "advisory" }, { id: "WS-002", mode: "script", gateMode: "advisory" }, ]); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); db.close(); }); @@ -798,7 +798,7 @@ describe("schema migration", () => { reviewerContextRetryCount: 0, reviewerFallbackRetryCount: 0, }); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); db.close(); }); @@ -827,7 +827,7 @@ describe("schema migration", () => { const columns = db.prepare("PRAGMA table_info(milestones)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("acceptanceCriteria"); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); db.close(); }); @@ -868,7 +868,7 @@ describe("schema migration", () => { const missionColumns = db.prepare("PRAGMA table_info(missions)").all() as Array<{ name: string }>; expect(missionColumns.map((column) => column.name)).toContain("autoMerge"); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); db.close(); }); @@ -902,7 +902,7 @@ describe("schema migration", () => { { id: "WS-002", mode: "script", enabled: 1, gateMode: "advisory" }, { id: "WS-003", mode: "prompt", enabled: 0, gateMode: "advisory" }, ]); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); db.close(); }); @@ -939,7 +939,7 @@ describe("schema migration", () => { const indexes = db.prepare("PRAGMA index_list(mission_goals)").all() as Array<{ name: string }>; expect(indexes.some((index) => index.name === "idxMissionGoalsGoalId")).toBe(true); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); db.close(); }); diff --git a/packages/core/src/__tests__/db.test.ts b/packages/core/src/__tests__/db.test.ts index 344d4c1940..72785c7738 100644 --- a/packages/core/src/__tests__/db.test.ts +++ b/packages/core/src/__tests__/db.test.ts @@ -334,7 +334,7 @@ describe("Database", () => { }); it("seeds schema version", () => { - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); }); it("includes tokenUsageCacheWriteTokens on freshly initialized tasks table", () => { @@ -393,7 +393,7 @@ describe("Database", () => { it("is idempotent - calling init() twice does not fail", () => { expect(() => db.init()).not.toThrow(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); }); it("does not overwrite existing config on re-init", () => { // Update the config @@ -1463,7 +1463,7 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 29 (includes v1→v2 through v26→v29) - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1488,11 +1488,11 @@ describe("schema migrations", () => { const db = new Database(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); // Re-init should not fail db.init(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); db.close(); }); @@ -1527,7 +1527,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; expect(cols.map((col) => col.name)).toContain("priority"); @@ -1568,7 +1568,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const colNames = cols.map((col) => col.name); @@ -1640,7 +1640,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const colNames = cols.map((col) => col.name); @@ -1880,7 +1880,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); const cols = db.prepare("PRAGMA table_info(chat_messages)").all() as Array<{ name: string }>; expect(cols.map((col) => col.name)).toContain("attachments"); @@ -1954,7 +1954,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'agentRatings'").all() as Array<{ name: string }>; expect(tables).toEqual([{ name: "agentRatings" }]); @@ -1978,7 +1978,7 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'mission_events'").all() as Array<{ name: string }>; expect(tables).toEqual([{ name: "mission_events" }]); @@ -2082,7 +2082,7 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 29 - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -2301,7 +2301,7 @@ describe("schema migrations", () => { localDb.init(); - expect(localDb.getSchemaVersion()).toBe(106); + expect(localDb.getSchemaVersion()).toBe(107); const columns = localDb.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("tokenUsageCacheWriteTokens"); @@ -2612,7 +2612,7 @@ describe("createDatabase factory", () => { const db = createDatabase(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(106); + expect(db.getSchemaVersion()).toBe(107); expect(db.getLastModified()).toBeGreaterThan(0); db.close(); @@ -2766,7 +2766,7 @@ describe("migration v77 task token budget columns", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(106); + expect(migrated.getSchemaVersion()).toBe(107); const rows = migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const names = new Set(rows.map((row) => row.name)); expect(names.has("tokenBudgetSoftAlertedAt")).toBe(true); @@ -2812,7 +2812,7 @@ describe("migration v67 drops orphan project auth tables", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(106); + expect(migrated.getSchemaVersion()).toBe(107); const tables = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; @@ -2839,7 +2839,7 @@ describe("migration v67 drops orphan project auth tables", () => { try { fresh.init(); - expect(fresh.getSchemaVersion()).toBe(106); + expect(fresh.getSchemaVersion()).toBe(107); const tables = fresh .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index e23e1f543b..0b9d5a0dba 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -149,7 +149,7 @@ export function probeFts5(db: DatabaseSync): boolean { // ── Schema Definition ──────────────────────────────────────────────── -const SCHEMA_VERSION = 106; +const SCHEMA_VERSION = 107; export { SCHEMA_VERSION }; @@ -575,6 +575,20 @@ CREATE TABLE IF NOT EXISTS completion_handoff_markers ( ); CREATE INDEX IF NOT EXISTS idx_completion_handoff_markers_acceptedAt ON completion_handoff_markers(acceptedAt); +-- Per-branch run state for concurrent workflow fan-out/join (U13, KTD-11/R21). +-- Reconstructible per ADR-0001: a crashed parallel run resumes each branch from +-- its persisted node; completed branches are not re-run. Additive-only. +CREATE TABLE IF NOT EXISTS workflow_run_branches ( + taskId TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + runId TEXT NOT NULL, + branchId TEXT NOT NULL, + currentNodeId TEXT NOT NULL, + status TEXT NOT NULL, + updatedAt TEXT NOT NULL, + PRIMARY KEY (taskId, runId, branchId) +); +CREATE INDEX IF NOT EXISTS idx_workflow_run_branches_task_run ON workflow_run_branches(taskId, runId); + -- Task documents (key-value store per task with revision tracking) CREATE TABLE IF NOT EXISTS task_documents ( id TEXT PRIMARY KEY, @@ -4193,6 +4207,28 @@ export class Database { }); } + // Migration 107: Per-branch run state for concurrent workflow fan-out/join + // (workflow-columns U13, KTD-11/R21). Stores {taskId, runId, branchId, + // currentNodeId, status} so a crashed parallel run resumes each branch from + // its persisted node without re-running completed branches. Additive-only, + // idempotent (table-exists guard); no backfill. + if (version < 107) { + this.applyMigration(107, () => { + this.db.exec(` + CREATE TABLE IF NOT EXISTS workflow_run_branches ( + taskId TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + runId TEXT NOT NULL, + branchId TEXT NOT NULL, + currentNodeId TEXT NOT NULL, + status TEXT NOT NULL, + updatedAt TEXT NOT NULL, + PRIMARY KEY (taskId, runId, branchId) + ); + CREATE INDEX IF NOT EXISTS idx_workflow_run_branches_task_run ON workflow_run_branches(taskId, runId); + `); + }); + } + } /** diff --git a/packages/engine/src/__tests__/workflow-graph-fanout.test.ts b/packages/engine/src/__tests__/workflow-graph-fanout.test.ts new file mode 100644 index 0000000000..7cc468b16d --- /dev/null +++ b/packages/engine/src/__tests__/workflow-graph-fanout.test.ts @@ -0,0 +1,360 @@ +import { describe, expect, it, vi } from "vitest"; +import type { TaskDetail, WorkflowIr, WorkflowIrNode } from "@fusion/core"; + +import { WorkflowGraphExecutor, type WorkflowNodeHandler } from "../workflow-graph-executor.js"; +import type { + WorkflowBranchPersistence, + WorkflowBranchProgress, + WorkflowBranchRunState, +} from "../workflow-graph-branches.js"; + +const task = { id: "FN-FANOUT" } as TaskDetail; +const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } }); + +/** A controllable deferred so branches can complete in any order under test control. */ +function deferred() { + let resolve!: (v: T) => void; + let reject!: (e: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +/** start → split → (branchA → branchB) → join → tail → end */ +function twoBranchIr(joinConfig: Record): WorkflowIr { + return { + version: "v2", + name: "two-branch", + columns: [{ id: "work", name: "Work", traits: [] }], + nodes: [ + { id: "start", kind: "start" }, + { id: "split", kind: "split", column: "work" }, + { id: "branchA", kind: "prompt", column: "work", config: { prompt: "a" } }, + { id: "branchB", kind: "prompt", column: "work", config: { prompt: "b" } }, + { id: "join", kind: "join", column: "work", config: joinConfig }, + { id: "tail", kind: "prompt", column: "work", config: { prompt: "tail" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "split" }, + { from: "split", to: "branchA" }, + { from: "split", to: "branchB" }, + { from: "branchA", to: "join", condition: "success" }, + { from: "branchB", to: "join", condition: "success" }, + { from: "join", to: "tail", condition: "success" }, + { from: "join", to: "end", condition: "failure" }, + { from: "tail", to: "end", condition: "success" }, + ], + }; +} + +describe("WorkflowGraphExecutor fan-out/join (U13)", () => { + it("mode:all — both branches complete in any order, join fires once, advances to tail", async () => { + const a = deferred(); + const b = deferred(); + const tail = vi.fn(async () => ({ outcome: "success" as const })); + const prompt: WorkflowNodeHandler = async (node) => { + if (node.id === "branchA") await a.promise; + if (node.id === "branchB") await b.promise; + if (node.id === "tail") return tail(); + return { outcome: "success" as const }; + }; + const executor = new WorkflowGraphExecutor({ handlers: { prompt } }); + const run = executor.run(task, settingsOn(), twoBranchIr({ mode: "all" })); + + // Complete in reverse order to prove order-independence. + b.resolve(); + await Promise.resolve(); + expect(tail).not.toHaveBeenCalled(); + a.resolve(); + + const result = await run; + expect(result.outcome).toBe("success"); + expect(result.visitedNodeIds).toContain("tail"); + expect(tail).toHaveBeenCalledTimes(1); + }); + + it("mode:any with collect — first completion fires join; slower branch finishes without re-firing", async () => { + const slow = deferred(); + const tail = vi.fn(async () => ({ outcome: "success" as const })); + let slowFinished = false; + const prompt: WorkflowNodeHandler = async (node) => { + if (node.id === "branchB") { + await slow.promise; + slowFinished = true; + } + if (node.id === "tail") return tail(); + return { outcome: "success" as const }; + }; + const executor = new WorkflowGraphExecutor({ handlers: { prompt } }); + const run = executor.run(task, settingsOn(), twoBranchIr({ mode: "any", onBranchFailure: "collect" })); + + // branchA resolves immediately → join fires. tail must run exactly once. + await Promise.resolve(); + slow.resolve(); + const result = await run; + + expect(result.outcome).toBe("success"); + expect(tail).toHaveBeenCalledTimes(1); + expect(slowFinished).toBe(true); + }); + + it("mode:any with fail-fast — slower branch is aborted via signal", async () => { + let aborted = false; + const slow = deferred(); + const prompt: WorkflowNodeHandler = async (node, ctx) => { + if (node.id === "branchB") { + ctx.signal?.addEventListener("abort", () => { + aborted = true; + slow.resolve(); + }); + await slow.promise; + if (ctx.signal?.aborted) return { outcome: "failure" as const, value: "aborted" }; + } + return { outcome: "success" as const }; + }; + const executor = new WorkflowGraphExecutor({ handlers: { prompt } }); + const result = await executor.run(task, settingsOn(), twoBranchIr({ mode: "any", onBranchFailure: "fail-fast" })); + + expect(result.outcome).toBe("success"); + expect(aborted).toBe(true); + }); + + it("quorum(2) of 3 — join fires on the second completion", async () => { + const ir: WorkflowIr = { + version: "v2", + name: "quorum", + columns: [{ id: "w", name: "W", traits: [] }], + nodes: [ + { id: "start", kind: "start" }, + { id: "split", kind: "split" }, + { id: "b1", kind: "prompt", config: {} }, + { id: "b2", kind: "prompt", config: {} }, + { id: "b3", kind: "prompt", config: {} }, + { id: "join", kind: "join", config: { mode: { quorum: 2 }, onBranchFailure: "collect" } }, + { id: "tail", kind: "prompt", config: {} }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "split" }, + { from: "split", to: "b1" }, + { from: "split", to: "b2" }, + { from: "split", to: "b3" }, + { from: "b1", to: "join", condition: "success" }, + { from: "b2", to: "join", condition: "success" }, + { from: "b3", to: "join", condition: "success" }, + { from: "join", to: "tail", condition: "success" }, + { from: "join", to: "end", condition: "failure" }, + { from: "tail", to: "end" }, + ], + }; + const d3 = deferred(); + const tail = vi.fn(async () => ({ outcome: "success" as const })); + const prompt: WorkflowNodeHandler = async (node) => { + if (node.id === "b3") await d3.promise; + if (node.id === "tail") return tail(); + return { outcome: "success" as const }; + }; + const executor = new WorkflowGraphExecutor({ handlers: { prompt } }); + const run = executor.run(task, settingsOn(), ir); + // b1 + b2 resolve immediately → quorum(2) satisfied without b3. + await Promise.resolve(); + d3.resolve(); + const result = await run; + expect(result.outcome).toBe("success"); + expect(tail).toHaveBeenCalledTimes(1); + }); + + it("branch failure fail-fast — siblings aborted, join routes the failure edge", async () => { + let siblingAborted = false; + const slow = deferred(); + const prompt: WorkflowNodeHandler = async (node, ctx) => { + if (node.id === "branchA") return { outcome: "failure" as const, value: "boom" }; + if (node.id === "branchB") { + ctx.signal?.addEventListener("abort", () => { + siblingAborted = true; + slow.resolve(); + }); + await slow.promise; + return { outcome: "success" as const }; + } + return { outcome: "success" as const }; + }; + const executor = new WorkflowGraphExecutor({ handlers: { prompt } }); + const result = await executor.run(task, settingsOn(), twoBranchIr({ mode: "all", onBranchFailure: "fail-fast" })); + + expect(result.outcome).toBe("failure"); + expect(result.visitedNodeIds).not.toContain("tail"); + expect(siblingAborted).toBe(true); + }); + + it("branch failure collect — all branches finish; join evaluates combined outcomes", async () => { + const calls: string[] = []; + const prompt: WorkflowNodeHandler = async (node) => { + calls.push(node.id); + if (node.id === "branchA") return { outcome: "failure" as const, value: "boom" }; + return { outcome: "success" as const }; + }; + const executor = new WorkflowGraphExecutor({ handlers: { prompt } }); + const result = await executor.run(task, settingsOn(), twoBranchIr({ mode: "all", onBranchFailure: "collect" })); + + // mode:all unmet (one failed) → join outcome failure; both branches ran. + expect(result.outcome).toBe("failure"); + expect(calls).toContain("branchA"); + expect(calls).toContain("branchB"); + const branchOutcomes = result.context["node:join:branchOutcomes"] as { outcome: string }[]; + expect(branchOutcomes.some((b) => b.outcome === "failure")).toBe(true); + expect(branchOutcomes.some((b) => b.outcome === "success")).toBe(true); + }); + + it("crash mid-branch resume — completed branches' nodes are NOT re-run", async () => { + const calls: string[] = []; + const store: WorkflowBranchRunState[] = []; + const persistence: WorkflowBranchPersistence = { + saveBranchState: (s) => { + const idx = store.findIndex((e) => e.branchId === s.branchId); + if (idx >= 0) store[idx] = s; + else store.push({ ...s }); + }, + loadBranchStates: () => store.map((s) => ({ ...s })), + }; + + // First run: branchA completes, branchB hangs (simulated crash before join). + const hang = deferred(); + const aPersisted = deferred(); + const persistenceA: WorkflowBranchPersistence = { + saveBranchState: (s) => { + persistence.saveBranchState!(s); + if (s.branchId === "branchA" && s.status === "completed") aPersisted.resolve(); + }, + loadBranchStates: persistence.loadBranchStates, + }; + const prompt1: WorkflowNodeHandler = async (node) => { + calls.push(`run1:${node.id}`); + if (node.id === "branchB") await hang.promise; // never resolves this run + return { outcome: "success" as const }; + }; + const exec1 = new WorkflowGraphExecutor({ handlers: { prompt: prompt1 }, branchPersistence: persistenceA }); + const run1 = exec1.run(task, settingsOn(), twoBranchIr({ mode: "all" })); + await aPersisted.promise; + // Don't await run1 (branchB stuck) — simulate process death by starting fresh. + + expect(store.find((s) => s.branchId === "branchA")?.status).toBe("completed"); + + // Resume: a brand-new executor reconstructed from persisted rows. + const prompt2: WorkflowNodeHandler = async (node) => { + calls.push(`run2:${node.id}`); + return { outcome: "success" as const }; + }; + const exec2 = new WorkflowGraphExecutor({ handlers: { prompt: prompt2 }, branchPersistence: persistence }); + const result = await exec2.run(task, settingsOn(), twoBranchIr({ mode: "all" })); + + expect(result.outcome).toBe("success"); + // branchA already completed → not re-run on resume. + expect(calls).not.toContain("run2:branchA"); + // branchB re-runs (it never completed). + expect(calls).toContain("run2:branchB"); + hang.resolve(); + await run1.catch(() => {}); + }); + + it("card-position invariant — no column move occurs during the parallel window", async () => { + // The executor never touches task.column; assert the handler context exposes + // the split's column to all branch nodes and the task object is untouched. + const columnsSeen = new Set(); + const taskColumnBefore = (task as { column?: string }).column; + const prompt: WorkflowNodeHandler = async (node) => { + if (node.id.startsWith("branch")) columnsSeen.add(node.column); + return { outcome: "success" as const }; + }; + const executor = new WorkflowGraphExecutor({ handlers: { prompt } }); + await executor.run(task, settingsOn(), twoBranchIr({ mode: "all" })); + // Branch nodes live in the split's column; task position never forked. + expect(columnsSeen).toEqual(new Set(["work"])); + expect((task as { column?: string }).column).toBe(taskColumnBefore); + }); + + it("semaphore bound — branches queue, never exceeding the limit (fake semaphore)", async () => { + let active = 0; + let peak = 0; + const limit = 1; + const queue: (() => void)[] = []; + const fakeSemaphore = { + async run(fn: () => Promise): Promise { + if (active >= limit) await new Promise((res) => queue.push(res)); + active += 1; + peak = Math.max(peak, active); + try { + return await fn(); + } finally { + active -= 1; + queue.shift()?.(); + } + }, + }; + const executor = new WorkflowGraphExecutor({ + handlers: { prompt: async () => ({ outcome: "success" as const }) }, + branchSemaphore: fakeSemaphore, + }); + const result = await executor.run(task, settingsOn(), twoBranchIr({ mode: "all" })); + expect(result.outcome).toBe("success"); + expect(peak).toBeLessThanOrEqual(limit); + }); + + it("nested split resolves recursively", async () => { + // start → split(outer) → [ branchX | split(inner) → [i1 | i2] → joinInner ] → joinOuter → end + const ir: WorkflowIr = { + version: "v2", + name: "nested", + columns: [{ id: "w", name: "W", traits: [] }], + nodes: [ + { id: "start", kind: "start" }, + { id: "outer", kind: "split" }, + { id: "branchX", kind: "prompt", config: {} }, + { id: "inner", kind: "split" }, + { id: "i1", kind: "prompt", config: {} }, + { id: "i2", kind: "prompt", config: {} }, + { id: "joinInner", kind: "join", config: { mode: "all" } }, + { id: "joinOuter", kind: "join", config: { mode: "all" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "outer" }, + { from: "outer", to: "branchX" }, + { from: "outer", to: "inner" }, + { from: "branchX", to: "joinOuter", condition: "success" }, + { from: "inner", to: "i1" }, + { from: "inner", to: "i2" }, + { from: "i1", to: "joinInner", condition: "success" }, + { from: "i2", to: "joinInner", condition: "success" }, + { from: "joinInner", to: "joinOuter", condition: "success" }, + { from: "joinOuter", to: "end", condition: "success" }, + ], + }; + const calls: string[] = []; + const executor = new WorkflowGraphExecutor({ + handlers: { + prompt: async (node) => { + calls.push(node.id); + return { outcome: "success" as const }; + }, + }, + }); + const result = await executor.run(task, settingsOn(), ir); + expect(result.outcome).toBe("success"); + expect(calls).toEqual(expect.arrayContaining(["branchX", "i1", "i2"])); + }); + + it("reports live per-branch progress for the dashboard", async () => { + const progress: WorkflowBranchProgress[] = []; + const executor = new WorkflowGraphExecutor({ + handlers: { prompt: async () => ({ outcome: "success" as const }) }, + onBranchProgress: (p) => progress.push(p), + }); + await executor.run(task, settingsOn(), twoBranchIr({ mode: "all" })); + expect(progress.some((p) => p.branchId === "branchA" && p.status === "completed")).toBe(true); + expect(progress.some((p) => p.branchId === "branchB" && p.status === "completed")).toBe(true); + }); +}); diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index d864128c60..0f5fadcf31 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -25,6 +25,13 @@ export { type WorkflowGraphExecutorDeps, type WorkflowGraphExecutorResult, } from "./workflow-graph-executor.js"; +export { + runSplitJoin, + type WorkflowBranchPersistence, + type WorkflowBranchProgress, + type WorkflowBranchRunState, + type WorkflowBranchSemaphore, +} from "./workflow-graph-branches.js"; export { createDefaultNodeHandlers, createNoopLegacySeams, @@ -63,6 +70,13 @@ export { getConflictedFiles, type AutostashHandle, } from "./merger.js"; +export { + registerMergeTraitHooks, + resolveMergePolicy, + type ResolvedMergePolicy, + type MergeFileScopeMode, + type MergeTraitStrategy, +} from "./merge-trait.js"; export { resolveIntegrationBranch, resolveIntegrationBranchSync, diff --git a/packages/engine/src/workflow-graph-branches.ts b/packages/engine/src/workflow-graph-branches.ts new file mode 100644 index 0000000000..e6b314cad1 --- /dev/null +++ b/packages/engine/src/workflow-graph-branches.ts @@ -0,0 +1,337 @@ +import type { Settings, TaskDetail, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core"; +import { WorkflowIrError } from "@fusion/core"; + +import type { WorkflowNodeOutcome, WorkflowNodeResult } from "./workflow-graph-executor.js"; + +/** + * Concurrent fan-out/join branch execution (U13, KTD-11, R21). + * + * When the sequential walker reaches a `split` node, every outgoing edge becomes + * a branch that walks concurrently up to the matching `join`. The join then + * synchronizes per its config (`all | any | quorum(n)`) and either fails fast + * (cancelling siblings via an AbortSignal) or collects all branch outcomes. + * + * This module owns ONLY the parallel window: it is handed a `runBranchNode` + * callback that reuses the executor's per-node retry logic, and a small set of + * graph lookups. The card's board position never forks — that invariant is + * upheld by the executor (no handler-driven column moves happen here). + */ + +/** Per-branch persisted run state (ADR-0001 reconstructible). */ +export interface WorkflowBranchRunState { + taskId: string; + runId: string; + branchId: string; + /** Node the branch is currently at / last completed. */ + currentNodeId: string; + status: "running" | "completed" | "failed" | "aborted"; +} + +/** + * Persistence callback surface. Kept as an injected interface so the executor + * stays DI-pure and fake-friendly; the SQLite-backed implementation is wired + * separately (see workflow_run_branches table). All methods are optional so a + * fully in-memory run (tests, flag-off) needs no persistence at all. + */ +export interface WorkflowBranchPersistence { + /** Idempotent upsert of a branch's progress, keyed by (taskId, runId, branchId). */ + saveBranchState?(state: WorkflowBranchRunState): void | Promise; + /** Load any persisted branch states for a run (used on resume). */ + loadBranchStates?(taskId: string, runId: string): WorkflowBranchRunState[] | Promise; +} + +/** Minimal semaphore shape — structurally compatible with AgentSemaphore. */ +export interface WorkflowBranchSemaphore { + run(fn: () => Promise): Promise; +} + +/** Snapshot of a single branch's progress, surfaced for dashboard badges (U9). */ +export interface WorkflowBranchProgress { + branchId: string; + nodeId: string; + status: WorkflowBranchRunState["status"]; +} + +export interface BranchEnvironment { + task: TaskDetail; + settings: Pick | undefined; + runId: string; + nodeMap: Map; + outgoingMap: Map; + /** Reuses the executor's executeNodeWithRetries (+ context bookkeeping). */ + runBranchNode: ( + node: WorkflowIrNode, + signal: AbortSignal, + ) => Promise; + shouldTraverseEdge: (edge: WorkflowIrEdge, source: WorkflowNodeResult) => boolean; + persistence?: WorkflowBranchPersistence; + semaphore?: WorkflowBranchSemaphore; + /** Reports live per-branch progress for the card (no column move). */ + onBranchProgress?: (progress: WorkflowBranchProgress) => void; + /** Node IDs already completed in a prior (crashed) run — skipped on resume. */ + completedNodeIds?: Set; +} + +export interface SplitJoinResult { + /** The join node where branches converged. */ + joinNodeId: string; + /** Join outcome: success if the mode was satisfied, else failure. */ + outcome: WorkflowNodeOutcome; + /** Per-branch outcomes, exposed so the join's outgoing edge conditions can read them. */ + branchOutcomes: { branchId: string; outcome: WorkflowNodeOutcome; nodeId: string }[]; + /** Node IDs visited across all branches (for the executor's visited list). */ + visitedNodeIds: string[]; +} + +interface ResolvedJoinConfig { + mode: "all" | "any" | { quorum: number }; + onBranchFailure: "fail-fast" | "collect"; +} + +function resolveJoinConfig(join: WorkflowIrNode): ResolvedJoinConfig { + const rawMode = join.config?.mode; + let mode: ResolvedJoinConfig["mode"] = "all"; + if (rawMode === "all" || rawMode === "any") mode = rawMode; + else if (rawMode && typeof rawMode === "object" && "quorum" in rawMode) { + const n = (rawMode as { quorum: unknown }).quorum; + if (typeof n === "number" && Number.isInteger(n) && n > 0) mode = { quorum: n }; + else throw new WorkflowIrError(`join '${join.id}' quorum must be a positive integer`); + } + const rawFail = join.config?.onBranchFailure; + const onBranchFailure: ResolvedJoinConfig["onBranchFailure"] = + rawFail === "collect" ? "collect" : "fail-fast"; + return { mode, onBranchFailure }; +} + +/** How many successful completions satisfy this join mode for `branchCount` branches. */ +function requiredCompletions(mode: ResolvedJoinConfig["mode"], branchCount: number): number { + if (mode === "all") return branchCount; + if (mode === "any") return 1; + return Math.min(mode.quorum, branchCount); +} + +/** + * Execute a split's branches concurrently and synchronize at the join. + * + * Returns once the join is satisfied (or definitively cannot be). Sibling + * branches are aborted on fail-fast via the shared AbortSignal; on collect they + * are awaited. Nested splits recurse: a branch walk that itself hits a `split` + * calls back into this function for its inner window. + */ +export async function runSplitJoin( + split: WorkflowIrNode, + env: BranchEnvironment, +): Promise { + const branchEdges = (env.outgoingMap.get(split.id) ?? []).filter( + (e) => env.shouldTraverseEdge(e, { outcome: "success" }), + ); + if (branchEdges.length === 0) { + throw new WorkflowIrError(`split '${split.id}' has no traversable branches`); + } + + const join = findMatchingJoin(branchEdges[0].to, env); + if (!join) throw new WorkflowIrError(`split '${split.id}' has no reachable matching join`); + const joinConfig = resolveJoinConfig(env.nodeMap.get(join)!); + + const controller = new AbortController(); + const branchCount = branchEdges.length; + const required = requiredCompletions(joinConfig.mode, branchCount); + + const visitedNodeIds: string[] = []; + const branchOutcomes: SplitJoinResult["branchOutcomes"] = []; + let succeeded = 0; + let failed = 0; + let settled = false; + let resolveJoin!: (outcome: WorkflowNodeOutcome) => void; + const joinReached = new Promise((res) => { + resolveJoin = res; + }); + + const settle = (outcome: WorkflowNodeOutcome): void => { + if (settled) return; + settled = true; + resolveJoin(outcome); + }; + + // Re-evaluate the join after each branch settles. `lastWasFailure` only + // affects fail-fast (one failure cancels siblings immediately). + const evaluateJoin = (lastWasFailure: boolean): void => { + if (settled) return; + if (joinConfig.onBranchFailure === "fail-fast" && lastWasFailure) { + controller.abort(); + settle("failure"); + return; + } + if (succeeded >= required) { + // Mode satisfied. Fail-fast cancels any laggards; collect lets them finish. + if (joinConfig.onBranchFailure === "fail-fast") controller.abort(); + settle("success"); + return; + } + if (succeeded + failed >= branchCount) { + // All branches settled but the mode is unmet. + settle("failure"); + } + }; + + const branchPromises = branchEdges.map((edge) => { + const branchId = edge.to; + return walkBranch(branchId, join, env, controller.signal, visitedNodeIds) + .then((result) => { + branchOutcomes.push({ branchId, outcome: result.outcome, nodeId: result.lastNodeId }); + if (result.outcome === "success") succeeded += 1; + else failed += 1; + evaluateJoin(result.outcome === "failure"); + }) + .catch((err) => { + // An aborted branch settles silently; any other throw fails the join. + if (controller.signal.aborted) { + branchOutcomes.push({ branchId, outcome: "failure", nodeId: branchId }); + return; + } + failed += 1; + branchOutcomes.push({ branchId, outcome: "failure", nodeId: branchId }); + evaluateJoin(true); + void err; + }); + }); + + // Wait for the join to resolve, then let in-flight branches settle so collect + // semantics (and persistence writes) complete before we return. + const outcome = await joinReached; + await Promise.allSettled(branchPromises); + + return { joinNodeId: join, outcome, branchOutcomes, visitedNodeIds }; +} + +interface BranchWalkResult { + outcome: WorkflowNodeOutcome; + lastNodeId: string; +} + +/** + * Walk a single branch from `startNodeId` up to (but not including) the join. + * Reuses the injected per-node runner; supports nested splits by recursing into + * runSplitJoin. Honors the AbortSignal (fail-fast cancellation) and skips nodes + * already completed in a prior run (crash resume idempotency). + */ +async function walkBranch( + startNodeId: string, + joinId: string, + env: BranchEnvironment, + signal: AbortSignal, + visitedNodeIds: string[], +): Promise { + let currentId = startNodeId; + let lastResult: WorkflowNodeResult = { outcome: "success" }; + + for (;;) { + if (signal.aborted) return { outcome: "failure", lastNodeId: currentId }; + if (currentId === joinId) return { outcome: lastResult.outcome, lastNodeId: currentId }; + + const node = env.nodeMap.get(currentId); + if (!node) throw new WorkflowIrError(`Unknown workflow node: ${currentId}`); + + if (node.kind === "split") { + // Nested split: resolve its inner window, then continue from the inner join. + const inner = await runSplitJoin(node, env); + visitedNodeIds.push(...inner.visitedNodeIds); + lastResult = { outcome: inner.outcome }; + const next = nextEdge(inner.joinNodeId, env, lastResult); + if (!next) return { outcome: inner.outcome, lastNodeId: inner.joinNodeId }; + currentId = next; + continue; + } + + visitedNodeIds.push(currentId); + + const alreadyDone = env.completedNodeIds?.has(currentId) ?? false; + if (alreadyDone) { + lastResult = { outcome: "success" }; + } else { + const exec = async (): Promise => env.runBranchNode(node, signal); + lastResult = env.semaphore ? await env.semaphore.run(exec) : await exec(); + env.persistence?.saveBranchState?.({ + taskId: env.task.id, + runId: env.runId, + branchId: startNodeId, + currentNodeId: currentId, + status: lastResult.outcome === "success" ? "running" : "failed", + }); + env.onBranchProgress?.({ + branchId: startNodeId, + nodeId: currentId, + status: lastResult.outcome === "success" ? "running" : "failed", + }); + } + + if (lastResult.outcome === "failure") { + env.persistence?.saveBranchState?.({ + taskId: env.task.id, + runId: env.runId, + branchId: startNodeId, + currentNodeId: currentId, + status: "failed", + }); + return { outcome: "failure", lastNodeId: currentId }; + } + + const next = nextEdge(currentId, env, lastResult); + if (!next) { + // Dead-end before the join — treat as branch completion. + return { outcome: lastResult.outcome, lastNodeId: currentId }; + } + if (next === joinId) { + env.persistence?.saveBranchState?.({ + taskId: env.task.id, + runId: env.runId, + branchId: startNodeId, + currentNodeId: currentId, + status: "completed", + }); + env.onBranchProgress?.({ branchId: startNodeId, nodeId: currentId, status: "completed" }); + return { outcome: lastResult.outcome, lastNodeId: currentId }; + } + currentId = next; + } +} + +/** The next node along a matching outgoing edge, or undefined if none matches. */ +function nextEdge( + nodeId: string, + env: BranchEnvironment, + source: WorkflowNodeResult, +): string | undefined { + const edges = (env.outgoingMap.get(nodeId) ?? []) + .filter((e) => env.shouldTraverseEdge(e, source)) + .sort((a, b) => a.to.localeCompare(b.to)); + return edges[0]?.to; +} + +/** + * Find the join node a branch starting at `startNodeId` converges on. Walks + * forward through the (non-failure) edges; recurses one level for nested splits + * so balanced nesting resolves to the correct outer join. + */ +function findMatchingJoin(startNodeId: string, env: BranchEnvironment): string | undefined { + const seen = new Set(); + let currentId: string | undefined = startNodeId; + while (currentId && !seen.has(currentId)) { + seen.add(currentId); + const node = env.nodeMap.get(currentId); + if (!node) return undefined; + if (node.kind === "join") return currentId; + if (node.kind === "split") { + const innerJoin = findMatchingJoin( + (env.outgoingMap.get(currentId) ?? [])[0]?.to ?? "", + env, + ); + if (!innerJoin) return undefined; + currentId = (env.outgoingMap.get(innerJoin) ?? []).find((e) => e.condition !== "failure")?.to; + continue; + } + const out = env.outgoingMap.get(currentId) ?? []; + currentId = out.find((e) => e.condition !== "failure")?.to ?? out[0]?.to; + } + return undefined; +} diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 16e2c6da35..91c40b473c 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -7,6 +7,14 @@ import { type WorkflowCustomNodeRunner, type WorkflowLegacySeams, } from "./workflow-node-handlers.js"; +import { + runSplitJoin, + type BranchEnvironment, + type WorkflowBranchPersistence, + type WorkflowBranchProgress, + type WorkflowBranchRunState, + type WorkflowBranchSemaphore, +} from "./workflow-graph-branches.js"; export type WorkflowNodeOutcome = "success" | "failure"; @@ -20,6 +28,9 @@ export interface WorkflowNodeExecutionContext { task: TaskDetail; settings: Pick | undefined; context: Record; + /** Set during concurrent branch execution; fail-fast aborts via this signal. + * Undefined on the sequential path (zero behavior change for linear graphs). */ + signal?: AbortSignal; } export type WorkflowNodeHandler = (node: WorkflowIrNode, context: WorkflowNodeExecutionContext) => Promise; @@ -30,6 +41,15 @@ export interface WorkflowGraphExecutorDeps { /** Executes custom (non-seam) prompt/script/gate nodes. */ runCustomNode?: WorkflowCustomNodeRunner; maxRetriesPerNode?: number; + /** Per-branch run-state persistence (U13). Optional — fully in-memory without it. */ + branchPersistence?: WorkflowBranchPersistence; + /** Bounds concurrent branch-node execution. Omit when the semaphore is + * enforced beneath runCustomNode (the session layer) to avoid double-acquire. */ + branchSemaphore?: WorkflowBranchSemaphore; + /** Live per-branch progress (dashboard badges). */ + onBranchProgress?: (progress: WorkflowBranchProgress) => void; + /** Stable identifier for this run, used to key persisted branch state. */ + runId?: string; } export interface WorkflowGraphExecutorResult { @@ -85,6 +105,34 @@ export class WorkflowGraphExecutor { const context: Record = {}; const visitedNodeIds: string[] = []; const inStack = new Set(); + const runId = this.deps.runId ?? `${task.id}:run`; + + // On resume, completed branch nodes (from a prior crashed run) are skipped + // so their handlers do not re-fire (idempotency). + let completedNodeIds: Set | undefined; + const persisted = await this.deps.branchPersistence?.loadBranchStates?.(task.id, runId); + if (persisted && persisted.length > 0) { + completedNodeIds = new Set( + persisted + .filter((s: WorkflowBranchRunState) => s.status === "completed") + .map((s) => s.currentNodeId), + ); + } + + // Shared branch environment: built lazily so the sequential path pays nothing. + const branchEnv = (): BranchEnvironment => ({ + task, + settings, + runId, + nodeMap, + outgoingMap, + runBranchNode: (node, signal) => this.executeNodeWithRetries(node, task, settings, context, signal), + shouldTraverseEdge: (edge, source) => this.shouldTraverseEdge(edge, source), + persistence: this.deps.branchPersistence, + semaphore: this.deps.branchSemaphore, + onBranchProgress: this.deps.onBranchProgress, + completedNodeIds, + }); const walk = async (nodeId: string): Promise => { const node = nodeMap.get(nodeId); @@ -101,6 +149,23 @@ export class WorkflowGraphExecutor { return { outcome: "success" }; } + if (node.kind === "split") { + // Concurrent fan-out: branches run in parallel up to their join, which + // synchronizes per its config. The card stays in the split's column for + // the whole window (no handler-driven move happens in here). Execution + // then continues sequentially from the join node. + const splitResult = await runSplitJoin(node, branchEnv()); + visitedNodeIds.push(...splitResult.visitedNodeIds); + context[`node:${node.id}:outcome`] = splitResult.outcome; + context[`node:${splitResult.joinNodeId}:outcome`] = splitResult.outcome; + context[`node:${splitResult.joinNodeId}:branchOutcomes`] = splitResult.branchOutcomes; + if (!inStack.has(splitResult.joinNodeId)) visitedNodeIds.push(splitResult.joinNodeId); + return await traverseChildren( + nodeMap.get(splitResult.joinNodeId)!, + { outcome: splitResult.outcome }, + ); + } + const result = await this.executeNodeWithRetries(node, task, settings, context); if (result.contextPatch) Object.assign(context, result.contextPatch); context[`node:${node.id}:outcome`] = result.outcome; @@ -164,6 +229,7 @@ export class WorkflowGraphExecutor { task: TaskDetail, settings: Pick | undefined, context: Record, + signal?: AbortSignal, ): Promise { const handler = this.handlers[node.kind]; if (!handler) { @@ -178,8 +244,10 @@ export class WorkflowGraphExecutor { let lastError: unknown; for (let attempt = 0; attempt < maxAttempts; attempt++) { + // Fail-fast cancellation: a branch aborted mid-retry stops re-trying. + if (signal?.aborted) return { outcome: "failure", value: "aborted" }; try { - return await handler(node, { task, settings, context }); + return await handler(node, { task, settings, context, signal }); } catch (error) { lastError = error; } diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts index ecfb88fac4..f1b892cdbc 100644 --- a/packages/engine/src/workflow-graph-task-runner.ts +++ b/packages/engine/src/workflow-graph-task-runner.ts @@ -3,6 +3,11 @@ import { isExperimentalFeatureEnabled } from "@fusion/core"; import { WorkflowGraphExecutor, type WorkflowNodeOutcome } from "./workflow-graph-executor.js"; import type { WorkflowCustomNodeRunner, WorkflowLegacySeams } from "./workflow-node-handlers.js"; +import type { + WorkflowBranchPersistence, + WorkflowBranchProgress, + WorkflowBranchSemaphore, +} from "./workflow-graph-branches.js"; // (Both types are also used as values in the side-effect tracking wrappers below.) /** @@ -37,6 +42,13 @@ export interface WorkflowGraphTaskRunnerDeps { maxRetriesPerNode?: number; /** Optional diagnostics hook (audit/log emission). Never throws into the run. */ onEvent?: (event: { type: "start" | "terminal" | "fallback"; taskId: string; detail: string }) => void; + /** Per-branch run-state persistence + resume (U13). Additive; in-memory without it. */ + branchPersistence?: WorkflowBranchPersistence; + /** Bounds concurrent branch-node execution (U13); omit when the semaphore is + * enforced beneath runCustomNode at the session layer. */ + branchSemaphore?: WorkflowBranchSemaphore; + /** Live per-branch progress for dashboard badges (U9/U13). */ + onBranchProgress?: (progress: WorkflowBranchProgress) => void; } /** @@ -47,8 +59,18 @@ export interface WorkflowGraphTaskRunnerDeps { * run the legacy pipeline; a task is never stranded by interpreter bugs. */ export class WorkflowGraphTaskRunner { + /** Latest per-branch progress, keyed by branchId. Store/dashboard-readable + * (U9 badges). Reset at the start of each run; the card never moves during a + * parallel window (KTD-11) so this is purely presentational state. */ + private readonly branchProgress = new Map(); + public constructor(private readonly deps: WorkflowGraphTaskRunnerDeps) {} + /** Snapshot of current per-branch progress (branchId, nodeId, status). */ + public getBranchProgress(): WorkflowBranchProgress[] { + return [...this.branchProgress.values()]; + } + private emit(type: "start" | "terminal" | "fallback", taskId: string, detail: string): void { try { this.deps.onEvent?.({ type, taskId, detail }); @@ -91,6 +113,7 @@ export class WorkflowGraphTaskRunner { } this.emit("start", task.id, definition.id); + this.branchProgress.clear(); // Track whether any node side effects ran. A pre-run interpreter error // (bad IR structure, wiring) can safely fall back to the legacy pipeline; @@ -117,6 +140,17 @@ export class WorkflowGraphTaskRunner { seams: wrappedSeams, runCustomNode: wrappedRunCustomNode, maxRetriesPerNode: this.deps.maxRetriesPerNode, + branchPersistence: this.deps.branchPersistence, + branchSemaphore: this.deps.branchSemaphore, + runId: `${task.id}:${definition.id}`, + onBranchProgress: (progress) => { + this.branchProgress.set(progress.branchId, progress); + try { + this.deps.onBranchProgress?.(progress); + } catch { + // Progress reporting must never affect the run. + } + }, }); const result = await executor.run(task, settings, definition.ir); if (!result.executed) {