diff --git a/packages/core/src/__tests__/workflow-compiler.test.ts b/packages/core/src/__tests__/workflow-compiler.test.ts new file mode 100644 index 0000000000..83fef4360f --- /dev/null +++ b/packages/core/src/__tests__/workflow-compiler.test.ts @@ -0,0 +1,162 @@ +import { describe, it, expect } from "vitest"; + +import { + compileWorkflowToSteps, + validateLinearity, + WorkflowCompileError, +} from "../workflow-compiler.js"; +import { serializeWorkflowIr, parseWorkflowIr } from "../workflow-ir.js"; +import type { WorkflowIr } from "../workflow-ir-types.js"; + +/** Linear graph: start → (user nodes) → execute → review → merge → (user nodes) → end. */ +function graph( + preMerge: WorkflowIr["nodes"], + postMerge: WorkflowIr["nodes"] = [], + { withSeams = true }: { withSeams?: boolean } = {}, +): WorkflowIr { + const seamNodes: WorkflowIr["nodes"] = withSeams + ? [ + { id: "execute", kind: "prompt", config: { seam: "execute" } }, + { id: "review", kind: "prompt", config: { seam: "review" } }, + { id: "merge", kind: "prompt", config: { seam: "merge" } }, + ] + : []; + + const ordered = [ + { id: "start", kind: "start" as const }, + ...preMerge, + ...seamNodes, + ...postMerge, + { id: "end", kind: "end" as const }, + ]; + + const edges: WorkflowIr["edges"] = []; + for (let i = 0; i < ordered.length - 1; i += 1) { + edges.push({ from: ordered[i].id, to: ordered[i + 1].id, condition: "success" }); + } + // Canonical seam failure edges to end. + if (withSeams) { + for (const seam of ["execute", "review", "merge"]) { + edges.push({ from: seam, to: "end", condition: "failure" }); + } + } + + return { version: "v1", name: "test", nodes: ordered, edges }; +} + +describe("compileWorkflowToSteps (U2)", () => { + it("compiles a linear pre-merge gate + prompt in authored order", () => { + const ir = graph([ + { id: "lint", kind: "gate", config: { name: "Lint", scriptName: "lint" } }, + { id: "spec", kind: "prompt", config: { name: "Spec check", prompt: "Check the spec" } }, + ]); + const steps = compileWorkflowToSteps(ir); + expect(steps).toHaveLength(2); + expect(steps[0].name).toBe("Lint"); + expect(steps[0].phase).toBe("pre-merge"); + expect(steps[0].mode).toBe("script"); + expect(steps[0].gateMode).toBe("gate"); + expect(steps[1].name).toBe("Spec check"); + expect(steps[1].mode).toBe("prompt"); + expect(steps[1].gateMode).toBe("advisory"); + }); + + it("partitions nodes after the merge seam into post-merge", () => { + const ir = graph( + [{ id: "pre", kind: "prompt", config: { prompt: "before" } }], + [{ id: "post", kind: "script", config: { scriptName: "notify" } }], + ); + const steps = compileWorkflowToSteps(ir); + expect(steps.map((s) => s.phase)).toEqual(["pre-merge", "post-merge"]); + expect(steps[1].mode).toBe("script"); + expect(steps[1].scriptName).toBe("notify"); + }); + + it("does not emit the execute/review/merge seams as steps", () => { + const ir = graph([{ id: "only", kind: "prompt", config: { prompt: "x" } }]); + const steps = compileWorkflowToSteps(ir); + expect(steps).toHaveLength(1); + expect(steps.every((s) => s.name !== "execute" && s.name !== "review" && s.name !== "merge")).toBe(true); + }); + + it("treats a gate node as gateMode=gate regardless of mode", () => { + const ir = graph([{ id: "g", kind: "gate", config: { prompt: "block?" } }]); + const steps = compileWorkflowToSteps(ir); + expect(steps[0].gateMode).toBe("gate"); + expect(steps[0].mode).toBe("prompt"); + }); + + it("carries prompt-node model overrides into the step", () => { + const ir = graph([ + { + id: "p", + kind: "prompt", + config: { prompt: "x", modelProvider: "anthropic", modelId: "claude-sonnet-4-5" }, + }, + ]); + const [step] = compileWorkflowToSteps(ir); + expect(step.modelProvider).toBe("anthropic"); + expect(step.modelId).toBe("claude-sonnet-4-5"); + }); + + it("rejects a graph with branching (fan-out beyond success/failure)", () => { + const ir: WorkflowIr = { + version: "v1", + name: "branchy", + nodes: [ + { id: "start", kind: "start" }, + { id: "a", kind: "prompt", config: { prompt: "a" } }, + { id: "b", kind: "prompt", config: { prompt: "b" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "a", condition: "success" }, + { from: "a", to: "b", condition: "success" }, + { from: "a", to: "end", condition: "success" }, // illegal second success branch + { from: "b", to: "end", condition: "success" }, + ], + }; + expect(() => compileWorkflowToSteps(ir)).toThrow(WorkflowCompileError); + expect(() => compileWorkflowToSteps(ir)).toThrow(/interpreter \(deferred\)/i); + }); + + it("rejects a graph missing the start/end nodes via parse", () => { + const ir = { version: "v1", name: "x", nodes: [{ id: "p", kind: "prompt" }], edges: [] } as WorkflowIr; + expect(() => compileWorkflowToSteps(ir)).toThrow(); + }); + + it("rejects a disconnected node not on the main path", () => { + const ir: WorkflowIr = { + version: "v1", + name: "orphan", + nodes: [ + { id: "start", kind: "start" }, + { id: "a", kind: "prompt", config: { prompt: "a" } }, + { id: "orphan", kind: "prompt", config: { prompt: "o" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "a", condition: "success" }, + { from: "a", to: "end", condition: "success" }, + { from: "orphan", to: "end", condition: "success" }, + ], + }; + const err = validateLinearity(ir); + expect(err).toBeInstanceOf(WorkflowCompileError); + }); + + it("returns an empty step set for a graph with only start/seams/end", () => { + const ir = graph([]); + expect(compileWorkflowToSteps(ir)).toEqual([]); + }); + + it("is deterministic across a serialize/parse round-trip", () => { + const ir = graph([ + { id: "lint", kind: "gate", config: { name: "Lint", scriptName: "lint" } }, + { id: "spec", kind: "prompt", config: { name: "Spec", prompt: "x" } }, + ]); + const first = compileWorkflowToSteps(ir); + const second = compileWorkflowToSteps(parseWorkflowIr(serializeWorkflowIr(ir))); + expect(second).toEqual(first); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index ccc1ee812e..d94b28b0fb 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -60,6 +60,11 @@ export type { WorkflowDefinitionUpdate, WorkflowNodeLayout, } from "./workflow-definition-types.js"; +export { + compileWorkflowToSteps, + validateLinearity, + WorkflowCompileError, +} from "./workflow-compiler.js"; // ── Engine wiring (set by @fusion/engine at module load) ──────────── export { diff --git a/packages/core/src/workflow-compiler.ts b/packages/core/src/workflow-compiler.ts new file mode 100644 index 0000000000..6df0f4fd4e --- /dev/null +++ b/packages/core/src/workflow-compiler.ts @@ -0,0 +1,201 @@ +import type { WorkflowIr, WorkflowIrNode, WorkflowIrEdge } from "./workflow-ir-types.js"; +import { parseWorkflowIr } from "./workflow-ir.js"; +import type { WorkflowStepInput, WorkflowStepGateMode } from "./types.js"; + +/** + * Raised when a WorkflowIr graph cannot be compiled onto the executable + * WorkflowStep engine — typically because it branches beyond the canonical + * seam success/failure chain and therefore requires the (deferred) graph + * interpreter rather than the linear pre/post-merge step runner. + */ +export class WorkflowCompileError extends Error { + constructor(message: string) { + super(message); + this.name = "WorkflowCompileError"; + } +} + +/** Seam anchor kinds, encoded on IR nodes as `config.seam`. These map to the + * fixed execute → review → merge pipeline and are not emitted as steps. */ +const SEAM_NAMES = new Set(["execute", "review", "merge"]); + +function seamOf(node: WorkflowIrNode): string | undefined { + const seam = node.config?.seam; + return typeof seam === "string" && SEAM_NAMES.has(seam) ? seam : undefined; +} + +function configString(node: WorkflowIrNode, key: string): string | undefined { + const value = node.config?.[key]; + return typeof value === "string" && value.trim() ? value : undefined; +} + +function buildOutgoing(ir: WorkflowIr): Map { + const outgoing = new Map(); + for (const edge of ir.edges) { + const list = outgoing.get(edge.from); + if (list) list.push(edge); + else outgoing.set(edge.from, [edge]); + } + return outgoing; +} + +function mainEdge(edges: WorkflowIrEdge[]): WorkflowIrEdge | undefined { + return edges.find((edge) => edge.condition !== "failure"); +} + +/** + * Validate that a workflow graph reduces to a linear pre-merge → seams → + * post-merge chain the WorkflowStep engine can run. Returns a + * WorkflowCompileError describing the first problem, or null when compilable. + * + * Allowed shape: a single path from start to end. Seam nodes may carry an extra + * `failure` edge to the end node; every other non-terminal node has exactly one + * outgoing edge. Anything else (true branching) requires the deferred interpreter. + */ +export function validateLinearity(ir: WorkflowIr): WorkflowCompileError | null { + const nodesById = new Map(ir.nodes.map((node) => [node.id, node])); + for (const edge of ir.edges) { + if (!nodesById.has(edge.from)) return new WorkflowCompileError(`edge references unknown node '${edge.from}'`); + if (!nodesById.has(edge.to)) return new WorkflowCompileError(`edge references unknown node '${edge.to}'`); + } + + const endNode = ir.nodes.find((node) => node.kind === "end"); + const startNode = ir.nodes.find((node) => node.kind === "start"); + if (!startNode || !endNode) { + return new WorkflowCompileError("workflow must contain exactly one start and one end node"); + } + + const outgoing = buildOutgoing(ir); + + for (const node of ir.nodes) { + const outs = outgoing.get(node.id) ?? []; + if (node.kind === "end") { + if (outs.length > 0) return new WorkflowCompileError("end node must have no outgoing edges"); + continue; + } + + const seam = seamOf(node); + if (seam) { + const failureEdges = outs.filter((edge) => edge.condition === "failure"); + const mainEdges = outs.filter((edge) => edge.condition !== "failure"); + if (mainEdges.length !== 1) { + return new WorkflowCompileError(`seam '${node.id}' must have exactly one success path`); + } + if (failureEdges.length > 1) { + return new WorkflowCompileError(`seam '${node.id}' has multiple failure edges`); + } + if (failureEdges[0] && failureEdges[0].to !== endNode.id) { + return new WorkflowCompileError(`seam '${node.id}' failure edge must target the end node`); + } + continue; + } + + // start or a user node (prompt/script/gate): exactly one outgoing edge. + if (outs.length === 0) { + return new WorkflowCompileError(`node '${node.id}' has no outgoing edge`); + } + if (outs.length > 1) { + return new WorkflowCompileError( + `node '${node.id}' branches into ${outs.length} edges — graphs with branches require the workflow interpreter (deferred)`, + ); + } + } + + // Reachability: the single main path must reach end and cover every node. + const visited = new Set(); + let cursor: string | undefined = startNode.id; + while (cursor && !visited.has(cursor)) { + visited.add(cursor); + if (cursor === endNode.id) break; + cursor = mainEdge(outgoing.get(cursor) ?? [])?.to; + } + if (!visited.has(endNode.id)) { + return new WorkflowCompileError("workflow main path does not reach the end node"); + } + const unreached = ir.nodes.filter((node) => !visited.has(node.id)); + if (unreached.length > 0) { + return new WorkflowCompileError( + `node '${unreached[0].id}' is not on the main path — disconnected nodes require the workflow interpreter (deferred)`, + ); + } + + return null; +} + +function defaultGateMode(node: WorkflowIrNode, mode: "prompt" | "script"): WorkflowStepGateMode { + if (node.kind === "gate") return "gate"; + const explicit = node.config?.gateMode; + if (explicit === "gate" || explicit === "advisory") return explicit; + return mode === "script" ? "gate" : "advisory"; +} + +function nodeToStepInput(node: WorkflowIrNode, phase: "pre-merge" | "post-merge"): WorkflowStepInput { + const scriptName = configString(node, "scriptName"); + const mode: "prompt" | "script" = node.kind === "script" || (node.kind === "gate" && scriptName) ? "script" : "prompt"; + const gateMode = defaultGateMode(node, mode); + + const input: WorkflowStepInput = { + name: configString(node, "name") ?? node.id, + description: configString(node, "description") ?? "", + mode, + phase, + gateMode, + }; + + if (mode === "script") { + input.scriptName = scriptName; + } else { + input.prompt = configString(node, "prompt") ?? ""; + input.toolMode = node.config?.toolMode === "coding" ? "coding" : "readonly"; + const provider = configString(node, "modelProvider"); + const modelId = configString(node, "modelId"); + if (provider && modelId) { + input.modelProvider = provider; + input.modelId = modelId; + } + } + + return input; +} + +/** + * Compile a workflow graph into an ordered list of WorkflowStep inputs ready to + * persist and run on the existing engine. User prompt/script/gate nodes become + * steps; execute/review seams are skipped; the merge seam is the pre-/post-merge + * boundary. Throws WorkflowCompileError for non-linear graphs. + * + * The returned array order is the execution order (it maps directly onto a + * task's `enabledWorkflowSteps`). + */ +export function compileWorkflowToSteps(ir: WorkflowIr): WorkflowStepInput[] { + const parsed = parseWorkflowIr(ir); + const error = validateLinearity(parsed); + if (error) throw error; + + const nodesById = new Map(parsed.nodes.map((node) => [node.id, node])); + const outgoing = buildOutgoing(parsed); + const startNode = parsed.nodes.find((node) => node.kind === "start")!; + + const steps: WorkflowStepInput[] = []; + let phase: "pre-merge" | "post-merge" = "pre-merge"; + const visited = new Set(); + let cursor: string | undefined = startNode.id; + + while (cursor && !visited.has(cursor)) { + visited.add(cursor); + const node = nodesById.get(cursor); + if (!node) break; + + const seam = seamOf(node); + if (seam === "merge") { + phase = "post-merge"; + } else if (!seam && node.kind !== "start" && node.kind !== "end") { + steps.push(nodeToStepInput(node, phase)); + } + + if (node.kind === "end") break; + cursor = mainEdge(outgoing.get(cursor) ?? [])?.to; + } + + return steps; +}