From ffce42ff9c4c976a30faa6c8600ca32616c08c28 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 09:22:35 -0700 Subject: [PATCH 01/45] feat(core): persist named workflow definitions (U1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a workflows table (migration 103) storing WorkflowIr graphs plus editor layout, with CRUD on TaskStore (create/list/get/update/delete) that validates the IR via parseWorkflowIr on write. IDs (WF-001…) use a monotonic __meta counter that never reuses across deletes. --- .../workflow-definition-store.test.ts | 127 ++++++++++++ packages/core/src/db.ts | 32 +++- packages/core/src/index.ts | 6 + packages/core/src/store.ts | 180 ++++++++++++++++++ .../core/src/workflow-definition-types.ts | 43 +++++ 5 files changed, 387 insertions(+), 1 deletion(-) create mode 100644 packages/core/src/__tests__/workflow-definition-store.test.ts create mode 100644 packages/core/src/workflow-definition-types.ts diff --git a/packages/core/src/__tests__/workflow-definition-store.test.ts b/packages/core/src/__tests__/workflow-definition-store.test.ts new file mode 100644 index 0000000000..7f8cf3c807 --- /dev/null +++ b/packages/core/src/__tests__/workflow-definition-store.test.ts @@ -0,0 +1,127 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; + +import { WorkflowIrError } from "../workflow-ir.js"; +import type { WorkflowIr } from "../workflow-ir-types.js"; +import { createTaskStoreTestHarness } from "./store-test-helpers.js"; + +function makeIr(overrides: Partial = {}): WorkflowIr { + return { + version: "v1", + name: "test-workflow", + nodes: [ + { id: "start", kind: "start" }, + { id: "lint", kind: "gate", config: { scriptName: "lint" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "lint" }, + { from: "lint", to: "end" }, + ], + ...overrides, + }; +} + +describe("TaskStore workflow definitions (U1)", () => { + const harness = createTaskStoreTestHarness(); + let store: ReturnType; + + beforeEach(async () => { + await harness.beforeEach(); + store = harness.store(); + }); + + afterEach(async () => { + await harness.afterEach(); + }); + + it("creates and round-trips a workflow with IR and layout intact", async () => { + const created = await store.createWorkflowDefinition({ + name: "Quality Gate", + description: "Runs lint before merge", + ir: makeIr(), + layout: { start: { x: 0, y: 0 }, lint: { x: 120, y: 0 }, end: { x: 240, y: 0 } }, + }); + + expect(created.id).toBe("WF-001"); + const list = await store.listWorkflowDefinitions(); + expect(list).toHaveLength(1); + expect(list[0].name).toBe("Quality Gate"); + expect(list[0].ir.nodes).toHaveLength(3); + expect(list[0].layout.lint).toEqual({ x: 120, y: 0 }); + }); + + it("rejects a workflow whose IR is missing start/end", async () => { + const bad = makeIr({ nodes: [{ id: "only", kind: "prompt" }], edges: [] }); + await expect( + store.createWorkflowDefinition({ name: "Broken", ir: bad }), + ).rejects.toBeInstanceOf(WorkflowIrError); + expect(await store.listWorkflowDefinitions()).toHaveLength(0); + }); + + it("requires a non-empty name", async () => { + await expect( + store.createWorkflowDefinition({ name: " ", ir: makeIr() }), + ).rejects.toThrow(/name is required/i); + }); + + it("updates name, description, IR, and layout and advances updatedAt", async () => { + const created = await store.createWorkflowDefinition({ name: "V1", ir: makeIr() }); + await new Promise((r) => setTimeout(r, 2)); + const updated = await store.updateWorkflowDefinition(created.id, { + name: "V2", + description: "now with a prompt step", + ir: makeIr({ + nodes: [ + { id: "start", kind: "start" }, + { id: "review", kind: "prompt", config: { prompt: "Review the change" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "review" }, + { from: "review", to: "end" }, + ], + }), + layout: { start: { x: 5, y: 5 } }, + }); + + expect(updated.name).toBe("V2"); + expect(updated.description).toBe("now with a prompt step"); + expect(updated.ir.nodes.some((n) => n.id === "review")).toBe(true); + expect(updated.layout.start).toEqual({ x: 5, y: 5 }); + expect(new Date(updated.updatedAt).getTime()).toBeGreaterThanOrEqual( + new Date(created.updatedAt).getTime(), + ); + }); + + it("update rejects an invalid IR without mutating the stored row", async () => { + const created = await store.createWorkflowDefinition({ name: "Keep", ir: makeIr() }); + await expect( + store.updateWorkflowDefinition(created.id, { + ir: { version: "v1", name: "x", nodes: [], edges: [] } as WorkflowIr, + }), + ).rejects.toBeInstanceOf(WorkflowIrError); + const reread = await store.getWorkflowDefinition(created.id); + expect(reread?.ir.nodes).toHaveLength(3); + }); + + it("deletes a workflow and reflects absence", async () => { + const created = await store.createWorkflowDefinition({ name: "Temp", ir: makeIr() }); + await store.deleteWorkflowDefinition(created.id); + expect(await store.getWorkflowDefinition(created.id)).toBeUndefined(); + expect(await store.listWorkflowDefinitions()).toHaveLength(0); + }); + + it("throws when deleting a non-existent workflow", async () => { + await expect(store.deleteWorkflowDefinition("WF-999")).rejects.toThrow(/not found/i); + }); + + it("allocates monotonic ids without reusing across deletes", async () => { + const a = await store.createWorkflowDefinition({ name: "A", ir: makeIr() }); + const b = await store.createWorkflowDefinition({ name: "B", ir: makeIr() }); + expect(a.id).toBe("WF-001"); + expect(b.id).toBe("WF-002"); + await store.deleteWorkflowDefinition(b.id); + const c = await store.createWorkflowDefinition({ name: "C", ir: makeIr() }); + expect(c.id).toBe("WF-003"); + }); +}); diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 8c7385f131..1a25d47924 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 = 102; +const SCHEMA_VERSION = 103; export { SCHEMA_VERSION }; @@ -387,6 +387,19 @@ CREATE TABLE IF NOT EXISTS workflow_steps ( updatedAt TEXT NOT NULL ); +-- Named workflow definitions authored as WorkflowIr graphs (+ editor layout). +-- The ir and layout columns are JSON-encoded TEXT; ir is validated via +-- parseWorkflowIr before persistence at the store layer. +CREATE TABLE IF NOT EXISTS workflows ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + ir TEXT NOT NULL, + layout TEXT NOT NULL DEFAULT '{}', + createdAt TEXT NOT NULL, + updatedAt TEXT NOT NULL +); + -- Activity log with indexed columns for efficient queries CREATE TABLE IF NOT EXISTS activityLog ( id TEXT PRIMARY KEY, @@ -4037,6 +4050,23 @@ export class Database { } } + // Migration 103: Named workflow definitions (WorkflowIr graphs + layout). + if (version < 103) { + this.applyMigration(103, () => { + this.db.exec(` + CREATE TABLE IF NOT EXISTS workflows ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + ir TEXT NOT NULL, + layout TEXT NOT NULL DEFAULT '{}', + createdAt TEXT NOT NULL, + updatedAt TEXT NOT NULL + ) + `); + }); + } + } /** diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index a00a01622f..ccc1ee812e 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -54,6 +54,12 @@ export type { WorkflowIrNodeKind, } from "./workflow-ir-types.js"; export { BUILTIN_CODING_WORKFLOW_IR } from "./builtin-coding-workflow-ir.js"; +export type { + WorkflowDefinition, + WorkflowDefinitionInput, + WorkflowDefinitionUpdate, + WorkflowNodeLayout, +} from "./workflow-definition-types.js"; // ── Engine wiring (set by @fusion/engine at module load) ──────────── export { diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index d5ff59dcb9..663d2d26db 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -7,6 +7,7 @@ import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, import { createActivityLogSnapshot, createRunAuditSnapshot, createTaskMetadataSnapshot, toTaskMetadataRecord, validateSnapshotEnvelope, type ActivityLogSnapshot, type RunAuditSnapshot, type TaskMetadataSnapshot } from "./shared-mesh-state.js"; import { VALID_TRANSITIONS, DEFAULT_SETTINGS, isGlobalOnlySettingsKey, WORKFLOW_STEP_TEMPLATES, validateDocumentKey } from "./types.js"; import { DEFAULT_PROJECT_SETTINGS } from "./settings-schema.js"; +import { parseWorkflowIr, serializeWorkflowIr } from "./workflow-ir.js"; import { resolveWorktrunkSettings, validateWorktrunkSettings } from "./worktrunk-settings.js"; import { normalizeTaskPriority } from "./task-priority.js"; import { canAgentTakeImplementationTaskForExplicitRouting } from "./agent-role-policy.js"; @@ -1133,6 +1134,7 @@ export class TaskStore extends EventEmitter { private lastTaskIdIntegrityLogSignature: string | null = null; /** Cached workflow steps — invalidated on create/update/delete */ private workflowStepsCache: import("./types.js").WorkflowStep[] | null = null; + private workflowDefinitionsCache: import("./workflow-definition-types.js").WorkflowDefinition[] | null = null; /** Plugin-contributed workflow step templates injected by engine runtime. */ private _pluginWorkflowStepTemplates: Array<{ pluginId: string; template: WorkflowStepTemplate }> = []; /** Global settings store (`~/.fusion/settings.json`) */ @@ -10843,6 +10845,184 @@ ${stepsSection}`; } } + // ── Workflow definitions (named WorkflowIr graphs) ───────────────────── + + /** Allocate the next workflow-definition id (WF-001, WF-002, …) using a + * monotonic counter persisted in __meta. Never reuses ids across deletes. */ + private nextWorkflowDefinitionId(): string { + const row = this.db.prepare("SELECT value FROM __meta WHERE key = 'nextWorkflowDefinitionId'").get() as + | { value: string } + | undefined; + const next = row ? parseInt(row.value, 10) || 1 : 1; + this.db + .prepare( + "INSERT INTO __meta (key, value) VALUES ('nextWorkflowDefinitionId', ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value", + ) + .run(String(next + 1)); + return `WF-${String(next).padStart(3, "0")}`; + } + + private toWorkflowDefinition(row: { + id: string; + name: string; + description: string; + ir: string; + layout: string; + createdAt: string; + updatedAt: string; + }): import("./workflow-definition-types.js").WorkflowDefinition { + return { + id: row.id, + name: row.name, + description: row.description, + ir: parseWorkflowIr(row.ir), + layout: this.parseWorkflowLayout(row.layout), + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }; + } + + private parseWorkflowLayout( + raw: string, + ): Record { + try { + const parsed = JSON.parse(raw) as unknown; + if (parsed && typeof parsed === "object") { + return parsed as Record; + } + } catch { + // Corrupt layout JSON falls back to empty (auto-layout) rather than failing the read. + } + return {}; + } + + /** Create a named workflow definition. The IR is validated via parseWorkflowIr. */ + async createWorkflowDefinition( + input: import("./workflow-definition-types.js").WorkflowDefinitionInput, + ): Promise { + return this.withConfigLock(async () => { + const name = input.name?.trim(); + 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); + const layout = input.layout ?? {}; + const now = new Date().toISOString(); + const id = this.nextWorkflowDefinitionId(); + const definition: import("./workflow-definition-types.js").WorkflowDefinition = { + id, + name, + description: input.description ?? "", + ir, + layout, + createdAt: now, + updatedAt: now, + }; + + this.db + .prepare( + `INSERT INTO workflows (id, name, description, ir, layout, createdAt, updatedAt) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + definition.id, + definition.name, + definition.description, + serializeWorkflowIr(definition.ir), + JSON.stringify(definition.layout), + definition.createdAt, + definition.updatedAt, + ); + + this.workflowDefinitionsCache = null; + this.db.bumpLastModified(); + return definition; + }); + } + + /** List all workflow definitions, oldest first. Cached until a mutation. */ + async listWorkflowDefinitions(): Promise { + if (this.workflowDefinitionsCache) return this.workflowDefinitionsCache; + const rows = this.db.prepare("SELECT * FROM workflows ORDER BY createdAt ASC").all() as Array<{ + id: string; + name: string; + description: string; + ir: string; + layout: string; + createdAt: string; + updatedAt: string; + }>; + this.workflowDefinitionsCache = rows.map((row) => this.toWorkflowDefinition(row)); + return this.workflowDefinitionsCache; + } + + /** Get a single workflow definition by id, or undefined when absent. */ + async getWorkflowDefinition( + id: string, + ): Promise { + const row = this.db.prepare("SELECT * FROM workflows WHERE id = ?").get(id) as + | { + id: string; + name: string; + description: string; + ir: string; + layout: string; + createdAt: string; + updatedAt: string; + } + | undefined; + return row ? this.toWorkflowDefinition(row) : undefined; + } + + /** Update a workflow definition. The IR (when supplied) is re-validated. */ + async updateWorkflowDefinition( + id: string, + updates: import("./workflow-definition-types.js").WorkflowDefinitionUpdate, + ): Promise { + return this.withConfigLock(async () => { + const existing = await this.getWorkflowDefinition(id); + if (!existing) throw new Error(`Workflow '${id}' not found`); + + 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; + const next: import("./workflow-definition-types.js").WorkflowDefinition = { + ...existing, + name, + description: updates.description !== undefined ? updates.description : existing.description, + ir, + layout: updates.layout !== undefined ? updates.layout : existing.layout, + updatedAt: new Date().toISOString(), + }; + + this.db + .prepare( + `UPDATE workflows SET name = ?, description = ?, ir = ?, layout = ?, updatedAt = ? WHERE id = ?`, + ) + .run( + next.name, + next.description, + serializeWorkflowIr(next.ir), + JSON.stringify(next.layout), + next.updatedAt, + id, + ); + + this.workflowDefinitionsCache = null; + this.db.bumpLastModified(); + return next; + }); + } + + /** Delete a workflow definition. Throws when the id does not exist. */ + async deleteWorkflowDefinition(id: string): Promise { + const deleted = this.db.prepare("DELETE FROM workflows WHERE id = ?").run(id) as { changes?: number }; + if ((deleted.changes || 0) === 0) { + throw new Error(`Workflow '${id}' not found`); + } + this.workflowDefinitionsCache = null; + this.db.bumpLastModified(); + } + /** * Close the database connection and clean up resources. * Call this when the store is no longer needed (e.g., short-lived per-request stores). diff --git a/packages/core/src/workflow-definition-types.ts b/packages/core/src/workflow-definition-types.ts new file mode 100644 index 0000000000..3383815a9d --- /dev/null +++ b/packages/core/src/workflow-definition-types.ts @@ -0,0 +1,43 @@ +import type { WorkflowIr } from "./workflow-ir-types.js"; + +/** Editor layout position for a single workflow IR node. Persisted separately + * from the IR because the v1 IR contract deliberately excludes node geometry. */ +export interface WorkflowNodeLayout { + x: number; + y: number; +} + +/** A named, persisted workflow authored as a WorkflowIr graph plus editor layout. */ +export interface WorkflowDefinition { + /** Unique identifier (e.g., "WF-001"). */ + id: string; + /** Display name. */ + name: string; + /** Short description for UI display. */ + description: string; + /** The validated workflow graph (v1 IR contract). */ + ir: WorkflowIr; + /** Editor node positions keyed by IR node id. May be empty (auto-layout). */ + layout: Record; + /** ISO-8601 timestamp of creation. */ + createdAt: string; + /** ISO-8601 timestamp of last update. */ + updatedAt: string; +} + +/** Input for creating a workflow definition. */ +export interface WorkflowDefinitionInput { + name: string; + description?: string; + /** Workflow graph; validated via parseWorkflowIr on write. */ + ir: WorkflowIr; + layout?: Record; +} + +/** Partial update for an existing workflow definition. */ +export interface WorkflowDefinitionUpdate { + name?: string; + description?: string; + ir?: WorkflowIr; + layout?: Record; +} From c9250df33c7f0cb22dd3097b761a6ee738f103c6 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 09:24:40 -0700 Subject: [PATCH 02/45] feat(core): compile workflow IR to executable WorkflowSteps (U2) Add compileWorkflowToSteps + validateLinearity: walk the linear main path of a WorkflowIr, emit prompt/script/gate user nodes as ordered WorkflowStep inputs, skip the execute/review seams, and use the merge seam as the pre-/post-merge boundary. Non-linear graphs (branching beyond canonical seam success/failure) throw WorkflowCompileError so they route to the deferred interpreter instead of mis-executing. --- .../src/__tests__/workflow-compiler.test.ts | 162 ++++++++++++++ packages/core/src/index.ts | 5 + packages/core/src/workflow-compiler.ts | 201 ++++++++++++++++++ 3 files changed, 368 insertions(+) create mode 100644 packages/core/src/__tests__/workflow-compiler.test.ts create mode 100644 packages/core/src/workflow-compiler.ts 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; +} From 47a42994edba6323e1bfc192382db939a0a30a6d Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 09:30:48 -0700 Subject: [PATCH 03/45] feat(core): per-task workflow selection + project default (U3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Selecting a workflow compiles it, materializes WorkflowStep rows (tagged and hidden from the step manager), and writes their ids into the task's existing enabledWorkflowSteps — the executor's read path is untouched. Re-selection replaces prior steps with no orphans; non-linear graphs abort before any write. New tasks inherit a project default workflow (settings.defaultWorkflowId) ahead of legacy default-on steps. Adds task_workflow_selection table (migration 104). No scheduler/executor/merger changes. --- .../workflow-selection-store.test.ts | 154 ++++++++++++++ packages/core/src/db.ts | 25 ++- packages/core/src/settings-schema.ts | 1 + packages/core/src/store.ts | 188 +++++++++++++++++- packages/core/src/types.ts | 3 + 5 files changed, 359 insertions(+), 12 deletions(-) create mode 100644 packages/core/src/__tests__/workflow-selection-store.test.ts diff --git a/packages/core/src/__tests__/workflow-selection-store.test.ts b/packages/core/src/__tests__/workflow-selection-store.test.ts new file mode 100644 index 0000000000..4d596a3e1d --- /dev/null +++ b/packages/core/src/__tests__/workflow-selection-store.test.ts @@ -0,0 +1,154 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; + +import { WorkflowCompileError } from "../workflow-compiler.js"; +import type { WorkflowIr } from "../workflow-ir-types.js"; +import { createTaskStoreTestHarness } from "./store-test-helpers.js"; + +/** Linear workflow with two pre-merge steps. */ +function linearIr(): WorkflowIr { + return { + version: "v1", + name: "wf", + nodes: [ + { id: "start", kind: "start" }, + { id: "lint", kind: "gate", config: { name: "Lint", scriptName: "lint" } }, + { id: "spec", kind: "prompt", config: { name: "Spec", prompt: "check" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "lint", condition: "success" }, + { from: "lint", to: "spec", condition: "success" }, + { from: "spec", to: "end", condition: "success" }, + ], + }; +} + +function branchingIr(): WorkflowIr { + return { + 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" }, + { from: "b", to: "end", condition: "success" }, + ], + }; +} + +describe("TaskStore workflow selection (U3)", () => { + const harness = createTaskStoreTestHarness(); + let store: ReturnType; + + beforeEach(async () => { + await harness.beforeEach(); + store = harness.store(); + }); + + afterEach(async () => { + await harness.afterEach(); + }); + + it("selecting a workflow populates enabledWorkflowSteps and records selection", async () => { + const wf = await store.createWorkflowDefinition({ name: "QA", ir: linearIr() }); + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + + await store.selectTaskWorkflow(task.id, wf.id); + + const detail = await store.getTask(task.id); + expect(detail.enabledWorkflowSteps).toHaveLength(2); + const selection = store.getTaskWorkflowSelection(task.id); + expect(selection?.workflowId).toBe(wf.id); + expect(selection?.stepIds).toEqual(detail.enabledWorkflowSteps); + }); + + it("compiled steps are hidden from the user-facing step manager listing", async () => { + const wf = await store.createWorkflowDefinition({ name: "QA", ir: linearIr() }); + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + await store.selectTaskWorkflow(task.id, wf.id); + + expect(await store.listWorkflowSteps()).toHaveLength(0); + // …but the executor can still resolve them directly. + const selection = store.getTaskWorkflowSelection(task.id)!; + expect(await store.getWorkflowStep(selection.stepIds[0])).toBeDefined(); + }); + + it("re-selecting replaces prior compiled steps without accumulating orphans", async () => { + const wfA = await store.createWorkflowDefinition({ name: "A", ir: linearIr() }); + const wfB = await store.createWorkflowDefinition({ name: "B", ir: linearIr() }); + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + + await store.selectTaskWorkflow(task.id, wfA.id); + const firstIds = store.getTaskWorkflowSelection(task.id)!.stepIds; + await store.selectTaskWorkflow(task.id, wfB.id); + const secondIds = store.getTaskWorkflowSelection(task.id)!.stepIds; + + // Old steps are gone, only the new selection's steps remain. + for (const id of firstIds) { + expect(await store.getWorkflowStep(id)).toBeUndefined(); + } + const detail = await store.getTask(task.id); + expect(detail.enabledWorkflowSteps).toEqual(secondIds); + }); + + it("clearing selection empties enabledWorkflowSteps", async () => { + const wf = await store.createWorkflowDefinition({ name: "QA", ir: linearIr() }); + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + await store.selectTaskWorkflow(task.id, wf.id); + + await store.clearTaskWorkflowSelection(task.id); + const detail = await store.getTask(task.id); + expect(detail.enabledWorkflowSteps ?? []).toHaveLength(0); + expect(store.getTaskWorkflowSelection(task.id)).toBeUndefined(); + }); + + it("rejects selecting a non-linear workflow without writing partial state", async () => { + const wf = await store.createWorkflowDefinition({ name: "Branchy", ir: branchingIr() }); + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + + await expect(store.selectTaskWorkflow(task.id, wf.id)).rejects.toBeInstanceOf(WorkflowCompileError); + expect(store.getTaskWorkflowSelection(task.id)).toBeUndefined(); + const detail = await store.getTask(task.id); + expect(detail.enabledWorkflowSteps ?? []).toHaveLength(0); + }); + + it("throws when selecting an unknown workflow", async () => { + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + await expect(store.selectTaskWorkflow(task.id, "WF-404")).rejects.toThrow(/not found/i); + }); + + it("new tasks inherit the project default workflow", async () => { + const wf = await store.createWorkflowDefinition({ name: "Default", ir: linearIr() }); + await store.setDefaultWorkflowId(wf.id); + + const task = await store.createTask({ description: "inherits" }); + const detail = await store.getTask(task.id); + expect(detail.enabledWorkflowSteps).toHaveLength(2); + expect(store.getTaskWorkflowSelection(task.id)?.workflowId).toBe(wf.id); + }); + + it("explicit enabledWorkflowSteps overrides the project default", async () => { + const wf = await store.createWorkflowDefinition({ name: "Default", ir: linearIr() }); + await store.setDefaultWorkflowId(wf.id); + + const task = await store.createTask({ description: "override", enabledWorkflowSteps: [] }); + const detail = await store.getTask(task.id); + expect(detail.enabledWorkflowSteps ?? []).toHaveLength(0); + expect(store.getTaskWorkflowSelection(task.id)).toBeUndefined(); + }); + + it("setDefaultWorkflowId rejects an unknown workflow and clears with null", async () => { + await expect(store.setDefaultWorkflowId("WF-404")).rejects.toThrow(/not found/i); + const wf = await store.createWorkflowDefinition({ name: "D", ir: linearIr() }); + await store.setDefaultWorkflowId(wf.id); + expect(await store.getDefaultWorkflowId()).toBe(wf.id); + await store.setDefaultWorkflowId(null); + expect(await store.getDefaultWorkflowId()).toBeUndefined(); + }); +}); diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 1a25d47924..239ec74878 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 = 103; +const SCHEMA_VERSION = 104; export { SCHEMA_VERSION }; @@ -400,6 +400,15 @@ CREATE TABLE IF NOT EXISTS workflows ( updatedAt TEXT NOT NULL ); +-- Per-task selected workflow. stepIds holds the WorkflowStep ids materialized +-- by compiling the workflow, so re-selection can clean them up (no orphans). +CREATE TABLE IF NOT EXISTS task_workflow_selection ( + taskId TEXT PRIMARY KEY, + workflowId TEXT NOT NULL, + stepIds TEXT NOT NULL DEFAULT '[]', + updatedAt TEXT NOT NULL +); + -- Activity log with indexed columns for efficient queries CREATE TABLE IF NOT EXISTS activityLog ( id TEXT PRIMARY KEY, @@ -4067,6 +4076,20 @@ export class Database { }); } + // Migration 104: Per-task selected workflow (resolves to enabledWorkflowSteps). + if (version < 104) { + this.applyMigration(104, () => { + this.db.exec(` + CREATE TABLE IF NOT EXISTS task_workflow_selection ( + taskId TEXT PRIMARY KEY, + workflowId TEXT NOT NULL, + stepIds TEXT NOT NULL DEFAULT '[]', + updatedAt TEXT NOT NULL + ) + `); + }); + } + } /** diff --git a/packages/core/src/settings-schema.ts b/packages/core/src/settings-schema.ts index c6db71808f..eddab6e7f0 100644 --- a/packages/core/src/settings-schema.ts +++ b/packages/core/src/settings-schema.ts @@ -185,6 +185,7 @@ export const DEFAULT_GLOBAL_SETTINGS = { export const DEFAULT_PROJECT_SETTINGS = { globalPause: false, globalPauseReason: undefined, + defaultWorkflowId: undefined, enginePaused: false, maxConcurrent: 2, maxTriageConcurrent: 2, diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 663d2d26db..0e4ceb1f63 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -8,6 +8,12 @@ import { createActivityLogSnapshot, createRunAuditSnapshot, createTaskMetadataSn import { VALID_TRANSITIONS, DEFAULT_SETTINGS, isGlobalOnlySettingsKey, WORKFLOW_STEP_TEMPLATES, validateDocumentKey } from "./types.js"; import { DEFAULT_PROJECT_SETTINGS } from "./settings-schema.js"; import { parseWorkflowIr, serializeWorkflowIr } from "./workflow-ir.js"; +import { compileWorkflowToSteps } from "./workflow-compiler.js"; +import type { WorkflowIr } from "./workflow-ir-types.js"; + +/** Tags WorkflowStep rows materialized by compiling a workflow so they can be + * filtered out of the user-facing step manager and cleaned up on re-selection. */ +const WORKFLOW_COMPILED_STEP_TEMPLATE_PREFIX = "workflow:"; import { resolveWorktrunkSettings, validateWorktrunkSettings } from "./worktrunk-settings.js"; import { normalizeTaskPriority } from "./task-priority.js"; import { canAgentTakeImplementationTaskForExplicitRouting } from "./agent-role-policy.js"; @@ -3693,23 +3699,41 @@ export class TaskStore extends EventEmitter { ? await this.resolveEnabledWorkflowSteps(input.enabledWorkflowSteps) : undefined; + // When a project default workflow is configured, new tasks inherit it + // (compiled to steps) ahead of the legacy default-on step behavior. + let pendingWorkflowSelection: { workflowId: string; stepIds: string[] } | undefined; if (input.enabledWorkflowSteps === undefined) { try { - const allSteps = await this.listWorkflowSteps(); - const defaultOnSteps = allSteps - .filter((ws) => ws.enabled && ws.defaultOn) - .map((ws) => ws.id); - if (defaultOnSteps.length > 0) { - resolvedWorkflowSteps = defaultOnSteps; + const inherited = await this.materializeDefaultWorkflowSteps(); + if (inherited) { + resolvedWorkflowSteps = inherited.stepIds; + pendingWorkflowSelection = inherited; } } catch (err) { - storeLog.warn("Failed to auto-apply default workflow steps during task creation; auto-defaulting skipped", { - phase: "createTask:workflow-auto-default", - skippedAutoDefaulting: true, + storeLog.warn("Failed to apply default workflow during task creation; falling back to default-on steps", { + phase: "createTask:default-workflow", error: err instanceof Error ? err.message : String(err), - descriptionLength: input.description.length, }); } + + if (resolvedWorkflowSteps === undefined) { + try { + const allSteps = await this.listWorkflowSteps(); + const defaultOnSteps = allSteps + .filter((ws) => ws.enabled && ws.defaultOn) + .map((ws) => ws.id); + if (defaultOnSteps.length > 0) { + resolvedWorkflowSteps = defaultOnSteps; + } + } catch (err) { + storeLog.warn("Failed to auto-apply default workflow steps during task creation; auto-defaulting skipped", { + phase: "createTask:workflow-auto-default", + skippedAutoDefaulting: true, + error: err instanceof Error ? err.message : String(err), + descriptionLength: input.description.length, + }); + } + } } else if (input.enabledWorkflowSteps.length === 0) { resolvedWorkflowSteps = undefined; } @@ -3727,6 +3751,18 @@ export class TaskStore extends EventEmitter { }, }); + // Record the inherited workflow selection now that the task row exists. + if (pendingWorkflowSelection) { + try { + this.writeTaskWorkflowSelection(task.id, pendingWorkflowSelection.workflowId, pendingWorkflowSelection.stepIds); + } catch (err) { + storeLog.warn("Failed to record inherited workflow selection", { + taskId: task.id, + error: err instanceof Error ? err.message : String(err), + }); + } + } + if (hasPendingSummarization && shouldInvokeTaskCreatedHook) { const id = task.id; Promise.resolve().then(async () => { @@ -10620,7 +10656,12 @@ ${stepsSection}`; createdAt: string; updatedAt: string; }>; - const storedSteps = rows.map((row) => this.applyLegacyWorkflowStepOverrides(this.toStoredWorkflowStep(row))); + const storedSteps = rows + .map((row) => this.applyLegacyWorkflowStepOverrides(this.toStoredWorkflowStep(row))) + // Steps materialized by compiling a workflow are an execution detail; keep + // them out of the user-facing step manager listing. The executor resolves + // them directly via getWorkflowStep, which is unaffected by this filter. + .filter((step) => !step.templateId?.startsWith(WORKFLOW_COMPILED_STEP_TEMPLATE_PREFIX)); const pluginSteps = this._pluginWorkflowStepTemplates .map(({ template }) => this.resolvePluginWorkflowStep(template.id)) .filter((step): step is import("./types.js").WorkflowStep => Boolean(step)); @@ -11023,6 +11064,131 @@ ${stepsSection}`; this.db.bumpLastModified(); } + // ── Workflow selection (resolves a workflow to enabledWorkflowSteps) ──── + // + // Selection never touches the engine's scheduler/executor/merger. It compiles + // a workflow into WorkflowStep rows and writes their ids into the task's + // existing `enabledWorkflowSteps`, which the executor already consumes. + + /** The configured project-default workflow id, or undefined when unset. */ + async getDefaultWorkflowId(): Promise { + const settings = await this.getSettings(); + const id = (settings as { defaultWorkflowId?: string }).defaultWorkflowId; + return id && id.trim() ? id : undefined; + } + + /** Set (or clear, with null) the project-default workflow. */ + async setDefaultWorkflowId(workflowId: string | null): Promise { + if (workflowId) { + const exists = await this.getWorkflowDefinition(workflowId); + if (!exists) throw new Error(`Workflow '${workflowId}' not found`); + } + await this.updateSettings({ defaultWorkflowId: workflowId ?? undefined } as Partial); + } + + /** Read the workflow currently selected for a task, if any. */ + getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined { + const row = this.db + .prepare("SELECT workflowId, stepIds FROM task_workflow_selection WHERE taskId = ?") + .get(taskId) as { workflowId: string; stepIds: string } | undefined; + if (!row) return undefined; + let stepIds: string[] = []; + try { + const parsed = JSON.parse(row.stepIds) as unknown; + if (Array.isArray(parsed)) stepIds = parsed.filter((s): s is string => typeof s === "string"); + } catch { + // Corrupt list falls back to empty. + } + return { workflowId: row.workflowId, stepIds }; + } + + private writeTaskWorkflowSelection(taskId: string, workflowId: string, stepIds: string[]): void { + this.db + .prepare( + `INSERT INTO task_workflow_selection (taskId, workflowId, stepIds, updatedAt) + VALUES (?, ?, ?, ?) + ON CONFLICT(taskId) DO UPDATE SET + workflowId = excluded.workflowId, + stepIds = excluded.stepIds, + updatedAt = excluded.updatedAt`, + ) + .run(taskId, workflowId, JSON.stringify(stepIds), new Date().toISOString()); + } + + /** Delete the WorkflowStep rows previously materialized for a task's selection + * and remove the selection record. Best-effort; safe to call when unset. */ + private removeMaterializedSelection(taskId: string): void { + const existing = this.getTaskWorkflowSelection(taskId); + if (existing) { + for (const stepId of existing.stepIds) { + this.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(stepId); + } + this.workflowStepsCache = null; + } + this.db.prepare("DELETE FROM task_workflow_selection WHERE taskId = ?").run(taskId); + } + + /** Compile a workflow into fresh WorkflowStep rows and return their ids in + * execution order. Steps are tagged so they stay out of the step manager. */ + private async materializeWorkflowSteps(workflowId: string, ir: WorkflowIr): Promise { + const inputs = compileWorkflowToSteps(ir); + const ids: string[] = []; + for (const input of inputs) { + const step = await this.createWorkflowStep({ + ...input, + templateId: `${WORKFLOW_COMPILED_STEP_TEMPLATE_PREFIX}${workflowId}`, + enabled: true, + }); + ids.push(step.id); + } + return ids; + } + + /** Resolve the project-default workflow into materialized step ids, or null + * when no default is set / it is missing / it does not compile. */ + private async materializeDefaultWorkflowSteps(): Promise<{ workflowId: string; stepIds: string[] } | undefined> { + const workflowId = await this.getDefaultWorkflowId(); + if (!workflowId) return undefined; + const def = await this.getWorkflowDefinition(workflowId); + if (!def) return undefined; + const stepIds = await this.materializeWorkflowSteps(workflowId, def.ir); + return { workflowId, stepIds }; + } + + /** + * Select a workflow for a task: compile it, materialize its steps, and write + * their ids into the task's enabledWorkflowSteps. Replaces any prior selection + * (no orphaned steps). Throws WorkflowCompileError for non-linear graphs + * before any state is written. + */ + async selectTaskWorkflow(taskId: string, workflowId: string): Promise { + const def = await this.getWorkflowDefinition(workflowId); + if (!def) throw new Error(`Workflow '${workflowId}' not found`); + // Compile first so a non-linear graph aborts before we mutate anything. + const inputs = compileWorkflowToSteps(def.ir); + + this.removeMaterializedSelection(taskId); + + const ids: string[] = []; + for (const input of inputs) { + const step = await this.createWorkflowStep({ + ...input, + templateId: `${WORKFLOW_COMPILED_STEP_TEMPLATE_PREFIX}${workflowId}`, + enabled: true, + }); + ids.push(step.id); + } + + await this.updateTask(taskId, { enabledWorkflowSteps: ids }); + this.writeTaskWorkflowSelection(taskId, workflowId, ids); + } + + /** Clear a task's workflow selection and its enabled steps. */ + async clearTaskWorkflowSelection(taskId: string): Promise { + this.removeMaterializedSelection(taskId); + await this.updateTask(taskId, { enabledWorkflowSteps: [] }); + } + /** * Close the database connection and clean up resources. * Call this when the store is no longer needed (e.g., short-lived per-request stores). diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index a0ffed9bb0..827eccf04d 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -2889,6 +2889,9 @@ export interface ProjectSettings { /** Tracks why globalPause was activated. "rate-limit" for automatic pauses, * "manual" for user-initiated. Cleared on unpause. */ globalPauseReason?: string; + /** Default custom workflow (WF-…) applied to newly created tasks when the + * caller does not specify enabledWorkflowSteps. Overridable per task. */ + defaultWorkflowId?: string; /** Engine pause (soft pause): when true, the scheduler and triage * processor stop dispatching **new** work (scheduling, triage * specification, and auto-merge), but currently running agent sessions From 72f37a842fe26d403af0274f6a6d06fb78243536 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 09:34:27 -0700 Subject: [PATCH 04/45] feat(dashboard): workflow CRUD, compile-preview, and selection API (U4) Add register-workflow-routes: GET/POST/PATCH/DELETE /workflows, POST /workflows/:id/compile (200 steps or 422 for non-linear graphs), PUT/GET /tasks/:taskId/workflow selection, and GET/PUT /project/default-workflow. Handlers are thin pass-throughs to the TaskStore; malformed IR -> 400, unknown id -> 404, non-compilable graph -> 422. --- .../src/__tests__/workflow-routes.test.ts | 150 ++++++++++++++ packages/dashboard/src/routes.ts | 2 + .../src/routes/register-workflow-routes.ts | 196 ++++++++++++++++++ 3 files changed, 348 insertions(+) create mode 100644 packages/dashboard/src/__tests__/workflow-routes.test.ts create mode 100644 packages/dashboard/src/routes/register-workflow-routes.ts diff --git a/packages/dashboard/src/__tests__/workflow-routes.test.ts b/packages/dashboard/src/__tests__/workflow-routes.test.ts new file mode 100644 index 0000000000..e893d9b6df --- /dev/null +++ b/packages/dashboard/src/__tests__/workflow-routes.test.ts @@ -0,0 +1,150 @@ +// @vitest-environment node + +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import express from "express"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { TaskStore } from "@fusion/core"; +import type { WorkflowIr } from "@fusion/core"; +import { registerWorkflowRoutes } from "../routes/register-workflow-routes.js"; +import { ApiError, sendErrorResponse } from "../api-error.js"; +import { request } from "../test-request.js"; + +function linearIr(): WorkflowIr { + return { + version: "v1", + name: "wf", + nodes: [ + { id: "start", kind: "start" }, + { id: "lint", kind: "gate", config: { name: "Lint", scriptName: "lint" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "lint", condition: "success" }, + { from: "lint", to: "end", condition: "success" }, + ], + }; +} + +function branchingIr(): WorkflowIr { + return { + 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" }, + { from: "b", to: "end", condition: "success" }, + ], + }; +} + +describe("workflow routes (U4)", () => { + let store: TaskStore; + let rootDir: string; + let globalDir: string; + let app: express.Express; + + beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "wf-routes-root-")); + globalDir = mkdtempSync(join(tmpdir(), "wf-routes-global-")); + store = new TaskStore(rootDir, globalDir, { inMemoryDb: true }); + await store.init(); + + app = express(); + app.use(express.json()); + const router = express.Router(); + registerWorkflowRoutes({ + router, + getProjectContext: async () => ({ store, engine: undefined, projectId: undefined }), + rethrowAsApiError: (err: unknown) => { + throw err instanceof ApiError ? err : new ApiError(500, err instanceof Error ? err.message : String(err)); + }, + } as unknown as Parameters[0]); + app.use("/api", router); + app.use((err: unknown, _req: express.Request, res: express.Response, _next: express.NextFunction) => { + if (err instanceof ApiError) sendErrorResponse(res, err.statusCode, err.message, { details: err.details }); + else sendErrorResponse(res, 500, err instanceof Error ? err.message : String(err)); + }); + }); + + afterEach(() => { + store.close(); + rmSync(rootDir, { recursive: true, force: true }); + rmSync(globalDir, { recursive: true, force: true }); + }); + + const post = (path: string, body: unknown) => + request(app, "POST", path, JSON.stringify(body), { "content-type": "application/json" }); + const put = (path: string, body: unknown) => + request(app, "PUT", path, JSON.stringify(body), { "content-type": "application/json" }); + const get = (path: string) => request(app, "GET", path); + + it("POST /workflows creates with valid IR and rejects malformed IR", async () => { + const ok = await post("/api/workflows", { name: "QA", ir: linearIr() }); + expect(ok.status).toBe(201); + expect((ok.body as { id: string }).id).toBe("WF-001"); + + const bad = await post("/api/workflows", { name: "Bad", ir: { version: "v1", name: "x", nodes: [], edges: [] } }); + expect(bad.status).toBe(400); + }); + + it("GET /workflows lists created workflows", async () => { + await post("/api/workflows", { name: "A", ir: linearIr() }); + const res = await get("/api/workflows"); + expect(res.status).toBe(200); + expect((res.body as unknown[]).length).toBe(1); + }); + + it("POST /workflows/:id/compile returns steps for linear and 422 for branching", async () => { + const linear = await post("/api/workflows", { name: "L", ir: linearIr() }); + const linearId = (linear.body as { id: string }).id; + const okCompile = await post(`/api/workflows/${linearId}/compile`, {}); + expect(okCompile.status).toBe(200); + expect((okCompile.body as { steps: unknown[] }).steps).toHaveLength(1); + + const branchy = await post("/api/workflows", { name: "B", ir: branchingIr() }); + const branchyId = (branchy.body as { id: string }).id; + const badCompile = await post(`/api/workflows/${branchyId}/compile`, {}); + expect(badCompile.status).toBe(422); + expect((badCompile.body as { error: string }).error).toMatch(/interpreter \(deferred\)/i); + }); + + it("PUT /tasks/:taskId/workflow selects and reflects on the task", async () => { + const wf = await post("/api/workflows", { name: "QA", ir: linearIr() }); + const wfId = (wf.body as { id: string }).id; + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + + const sel = await put(`/api/tasks/${task.id}/workflow`, { workflowId: wfId }); + expect(sel.status).toBe(200); + const detail = await store.getTask(task.id); + expect(detail.enabledWorkflowSteps).toHaveLength(1); + + const read = await get(`/api/tasks/${task.id}/workflow`); + expect((read.body as { workflowId: string }).workflowId).toBe(wfId); + }); + + it("PUT /project/default-workflow then create task inherits the default", async () => { + const wf = await post("/api/workflows", { name: "Def", ir: linearIr() }); + const wfId = (wf.body as { id: string }).id; + const set = await put("/api/project/default-workflow", { workflowId: wfId }); + expect(set.status).toBe(200); + + const task = await store.createTask({ description: "inherits" }); + const detail = await store.getTask(task.id); + expect(detail.enabledWorkflowSteps).toHaveLength(1); + }); + + it("selecting an unknown workflow returns 404", async () => { + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + const res = await put(`/api/tasks/${task.id}/workflow`, { workflowId: "WF-404" }); + expect(res.status).toBe(404); + }); +}); diff --git a/packages/dashboard/src/routes.ts b/packages/dashboard/src/routes.ts index e1bcde6aaa..20d250ae53 100644 --- a/packages/dashboard/src/routes.ts +++ b/packages/dashboard/src/routes.ts @@ -134,6 +134,7 @@ function resolveBundledPluginDirInDashboard(pluginId: string): string | null { import { createSessionDiagnostics } from "./ai-session-diagnostics.js"; import { createApiRoutesContext } from "./routes/context.js"; import { registerTaskWorkflowRoutes } from "./routes/register-task-workflow-routes.js"; +import { registerWorkflowRoutes } from "./routes/register-workflow-routes.js"; import { registerPlanningSubtaskRoutes } from "./routes/register-planning-subtask-routes.js"; import { registerChatRoutes } from "./routes/register-chat-routes.js"; import { registerChatRoomRoutes } from "./routes/register-chat-room-routes.js"; @@ -1053,6 +1054,7 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout triggerCommentWakeForAssignedAgent: (...args) => triggerCommentWakeForAssignedAgent(...args), resolveSelfHealingManager: (...args) => resolveSelfHealingManager(...args), }); + registerWorkflowRoutes(routeContext); registerPlanningSubtaskRoutes(routeContext, { store, aiSessionStore, diff --git a/packages/dashboard/src/routes/register-workflow-routes.ts b/packages/dashboard/src/routes/register-workflow-routes.ts new file mode 100644 index 0000000000..7c2048b404 --- /dev/null +++ b/packages/dashboard/src/routes/register-workflow-routes.ts @@ -0,0 +1,196 @@ +import type { WorkflowIr } from "@fusion/core"; +import { WorkflowCompileError, WorkflowIrError, compileWorkflowToSteps } from "@fusion/core"; +import { ApiError, badRequest, notFound } from "../api-error.js"; +import type { ApiRoutesContext } from "./types.js"; + +/** + * Routes for named workflow definitions, IR compilation preview, per-task + * workflow selection, and the project default workflow. All state changes flow + * through @fusion/core's TaskStore; none touch the engine's scheduler/executor. + */ +export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { + const { router, getProjectContext, rethrowAsApiError } = ctx; + + function requireIr(body: unknown): WorkflowIr { + const ir = (body as { ir?: unknown })?.ir; + if (!ir || typeof ir !== "object") { + throw badRequest("ir is required and must be a workflow graph object"); + } + return ir as WorkflowIr; + } + + // GET /api/workflows — list all workflow definitions for the project. + router.get("/workflows", async (req, res) => { + try { + const { store } = await getProjectContext(req); + res.json(await store.listWorkflowDefinitions()); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); + + // POST /api/workflows — create a workflow. Body: { name, description?, ir, layout? } + router.post("/workflows", async (req, res) => { + try { + const { store } = await getProjectContext(req); + const { name, description, layout } = req.body ?? {}; + if (!name || typeof name !== "string" || !name.trim()) { + throw badRequest("name is required"); + } + const ir = requireIr(req.body); + const created = await store.createWorkflowDefinition({ name, description, ir, layout }); + res.status(201).json(created); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + if (err instanceof WorkflowIrError) throw badRequest(err.message); + rethrowAsApiError(err); + } + }); + + // GET /api/workflows/:id + router.get("/workflows/:id", async (req, res) => { + try { + const { store } = await getProjectContext(req); + const def = await store.getWorkflowDefinition(req.params.id); + if (!def) throw notFound(`Workflow '${req.params.id}' not found`); + res.json(def); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); + + // PATCH /api/workflows/:id — partial update. Body: { name?, description?, ir?, layout? } + router.patch("/workflows/:id", async (req, res) => { + try { + const { store } = await getProjectContext(req); + const { name, description, ir, layout } = req.body ?? {}; + if (name !== undefined && (typeof name !== "string" || !name.trim())) { + throw badRequest("name must be a non-empty string"); + } + if (ir !== undefined && (typeof ir !== "object" || ir === null)) { + throw badRequest("ir must be a workflow graph object"); + } + const updated = await store.updateWorkflowDefinition(req.params.id, { name, description, ir, layout }); + res.json(updated); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + if (err instanceof WorkflowIrError) throw badRequest(err.message); + if (err instanceof Error && /not found/i.test(err.message)) throw notFound(err.message); + rethrowAsApiError(err); + } + }); + + // DELETE /api/workflows/:id + router.delete("/workflows/:id", async (req, res) => { + try { + const { store } = await getProjectContext(req); + await store.deleteWorkflowDefinition(req.params.id); + res.status(204).send(); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + if (err instanceof Error && /not found/i.test(err.message)) throw notFound(err.message); + rethrowAsApiError(err); + } + }); + + // POST /api/workflows/:id/compile — preview the compiled WorkflowSteps. + // 200 with the step set, or 422 when the graph requires the deferred interpreter. + router.post("/workflows/:id/compile", async (req, res) => { + try { + const { store } = await getProjectContext(req); + const def = await store.getWorkflowDefinition(req.params.id); + if (!def) throw notFound(`Workflow '${req.params.id}' not found`); + try { + res.json({ steps: compileWorkflowToSteps(def.ir) }); + } catch (compileErr: unknown) { + if (compileErr instanceof WorkflowCompileError || compileErr instanceof WorkflowIrError) { + throw new ApiError(422, compileErr.message); + } + throw compileErr; + } + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); + + // GET /api/tasks/:taskId/workflow — current selection for a task. + router.get("/tasks/:taskId/workflow", async (req, res) => { + try { + const { store } = await getProjectContext(req); + const selection = store.getTaskWorkflowSelection(req.params.taskId); + res.json({ workflowId: selection?.workflowId ?? null }); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); + + // PUT /api/tasks/:taskId/workflow — select (or clear) a workflow for a task. + // Body: { workflowId: string | null } + router.put("/tasks/:taskId/workflow", async (req, res) => { + try { + const { store } = await getProjectContext(req); + const workflowId = (req.body ?? {}).workflowId; + if (workflowId === null || workflowId === undefined) { + await store.clearTaskWorkflowSelection(req.params.taskId); + res.json({ workflowId: null }); + return; + } + if (typeof workflowId !== "string") { + throw badRequest("workflowId must be a string or null"); + } + try { + await store.selectTaskWorkflow(req.params.taskId, workflowId); + } catch (selectErr: unknown) { + if (selectErr instanceof WorkflowCompileError || selectErr instanceof WorkflowIrError) { + throw new ApiError(422, selectErr.message); + } + if (selectErr instanceof Error && /not found/i.test(selectErr.message)) { + throw notFound(selectErr.message); + } + throw selectErr; + } + res.json({ workflowId }); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); + + // GET /api/project/default-workflow + router.get("/project/default-workflow", async (req, res) => { + try { + const { store } = await getProjectContext(req); + res.json({ workflowId: (await store.getDefaultWorkflowId()) ?? null }); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); + + // PUT /api/project/default-workflow — Body: { workflowId: string | null } + router.put("/project/default-workflow", async (req, res) => { + try { + const { store } = await getProjectContext(req); + const workflowId = (req.body ?? {}).workflowId; + if (workflowId !== null && typeof workflowId !== "string") { + throw badRequest("workflowId must be a string or null"); + } + try { + await store.setDefaultWorkflowId(workflowId); + } catch (setErr: unknown) { + if (setErr instanceof Error && /not found/i.test(setErr.message)) { + throw notFound(setErr.message); + } + throw setErr; + } + res.json({ workflowId: workflowId ?? null }); + } catch (err: unknown) { + if (err instanceof ApiError) throw err; + rethrowAsApiError(err); + } + }); +} From 80470155eb64cc34a8dbdc9de408ae21587ef414 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 09:35:24 -0700 Subject: [PATCH 05/45] feat(dashboard): workflow API client functions (U5) Add fetch/create/update/delete/compile workflow client wrappers plus select/fetch task workflow and get/set project default, mirroring the existing WorkflowStep client conventions (api/withProjectId/dedupe). --- packages/dashboard/app/api/legacy.ts | 90 ++++++++++++++++++++++++++++ 1 file changed, 90 insertions(+) diff --git a/packages/dashboard/app/api/legacy.ts b/packages/dashboard/app/api/legacy.ts index 0a987a4900..d2350e96e9 100644 --- a/packages/dashboard/app/api/legacy.ts +++ b/packages/dashboard/app/api/legacy.ts @@ -4897,6 +4897,96 @@ export function fetchWorkflowResults(taskId: string, projectId?: string): Promis return api(withProjectId(`/tasks/${encodeURIComponent(taskId)}/workflow-results`, projectId)); } +// ── Workflow definitions (graph-authored custom workflows) ─────────────── + +export type { + WorkflowDefinition, + WorkflowDefinitionInput, + WorkflowDefinitionUpdate, + WorkflowIr, +} from "@fusion/core"; + +/** List all workflow definitions for the project. */ +export function fetchWorkflows(projectId?: string): Promise { + const path = withProjectId("/workflows", projectId); + return dedupe(path, () => api(path)); +} + +/** Fetch a single workflow definition. */ +export function fetchWorkflow(id: string, projectId?: string): Promise { + return api(withProjectId(`/workflows/${encodeURIComponent(id)}`, projectId)); +} + +/** Create a workflow definition. */ +export function createWorkflow( + input: import("@fusion/core").WorkflowDefinitionInput, + projectId?: string, +): Promise { + return api(withProjectId("/workflows", projectId), { + method: "POST", + body: JSON.stringify(input), + }); +} + +/** Update a workflow definition (partial). */ +export function updateWorkflow( + id: string, + updates: import("@fusion/core").WorkflowDefinitionUpdate, + projectId?: string, +): Promise { + return api(withProjectId(`/workflows/${encodeURIComponent(id)}`, projectId), { + method: "PATCH", + body: JSON.stringify(updates), + }); +} + +/** Delete a workflow definition. */ +export function deleteWorkflow(id: string, projectId?: string): Promise { + return api(withProjectId(`/workflows/${encodeURIComponent(id)}`, projectId), { method: "DELETE" }); +} + +/** Preview the compiled steps for a workflow. Rejects (422) for non-linear graphs. */ +export function compileWorkflow(id: string, projectId?: string): Promise<{ steps: WorkflowStepInput[] }> { + return api<{ steps: WorkflowStepInput[] }>(withProjectId(`/workflows/${encodeURIComponent(id)}/compile`, projectId), { + method: "POST", + }); +} + +/** Read the workflow currently selected for a task. */ +export function fetchTaskWorkflow(taskId: string, projectId?: string): Promise<{ workflowId: string | null }> { + return api<{ workflowId: string | null }>( + withProjectId(`/tasks/${encodeURIComponent(taskId)}/workflow`, projectId), + ); +} + +/** Select (or clear, with null) a workflow for a task. */ +export function selectTaskWorkflow( + taskId: string, + workflowId: string | null, + projectId?: string, +): Promise<{ workflowId: string | null }> { + return api<{ workflowId: string | null }>(withProjectId(`/tasks/${encodeURIComponent(taskId)}/workflow`, projectId), { + method: "PUT", + body: JSON.stringify({ workflowId }), + }); +} + +/** Read the project default workflow. */ +export function fetchProjectDefaultWorkflow(projectId?: string): Promise<{ workflowId: string | null }> { + return api<{ workflowId: string | null }>(withProjectId("/project/default-workflow", projectId)); +} + +/** Set (or clear, with null) the project default workflow. */ +export function setProjectDefaultWorkflow( + workflowId: string | null, + projectId?: string, +): Promise<{ workflowId: string | null }> { + return api<{ workflowId: string | null }>(withProjectId("/project/default-workflow", projectId), { + method: "PUT", + body: JSON.stringify({ workflowId }), + }); +} + // ── Workflow Step Templates ────────────────────────────────────────────── /** Re-export WorkflowStepTemplate type from core */ From d071aecda6df6823a1ad48bb5e08f17730132ef0 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 09:45:26 -0700 Subject: [PATCH 06/45] feat(dashboard): visual graph node editor for workflows (U6, U7) Add a React Flow (@xyflow/react) based WorkflowNodeEditor: a lazy-loaded modal with a workflow list, a node palette (prompt/script/gate/merge-boundary), drag-to-connect edges, a per-node inspector, and save with compile-validation that surfaces non-linear graphs as a banner. Pure irToFlow/flowToIr mapping round-trips the v1 IR plus editor layout. Reachable via a Graph editor button in the Workflow Steps manager; mounted in AppModals behind a new modal-manager flag. Adds a vendor-reactflow Vite chunk and a feature changeset. --- .changeset/graph-custom-workflows.md | 5 + .../dashboard/app/components/AppModals.tsx | 18 + .../app/components/WorkflowNodeEditor.css | 297 +++++++++++++++ .../app/components/WorkflowNodeEditor.tsx | 359 ++++++++++++++++++ .../app/components/WorkflowStepManager.css | 22 ++ .../app/components/WorkflowStepManager.tsx | 17 +- .../__tests__/WorkflowNodeEditor.test.tsx | 91 +++++ .../components/nodes/WorkflowNodeTypes.tsx | 48 +++ .../app/components/workflow-flow-mapping.ts | 94 +++++ .../dashboard/app/hooks/useModalManager.ts | 10 + packages/dashboard/package.json | 11 +- packages/dashboard/vite.config.ts | 4 + packages/dashboard/vitest.config.ts | 1 + pnpm-lock.yaml | 239 ++++++++++-- 14 files changed, 1184 insertions(+), 32 deletions(-) create mode 100644 .changeset/graph-custom-workflows.md create mode 100644 packages/dashboard/app/components/WorkflowNodeEditor.css create mode 100644 packages/dashboard/app/components/WorkflowNodeEditor.tsx create mode 100644 packages/dashboard/app/components/__tests__/WorkflowNodeEditor.test.tsx create mode 100644 packages/dashboard/app/components/nodes/WorkflowNodeTypes.tsx create mode 100644 packages/dashboard/app/components/workflow-flow-mapping.ts diff --git a/.changeset/graph-custom-workflows.md b/.changeset/graph-custom-workflows.md new file mode 100644 index 0000000000..e19e1e1efc --- /dev/null +++ b/.changeset/graph-custom-workflows.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": minor +--- + +Add executable custom workflows with a visual graph node editor. Author a workflow as a graph (start → prompt/script/gate steps → end) in a new React Flow–based editor, then select it per task or set a project default. Selected workflows compile to the existing WorkflowStep engine and run at the pre/post-merge boundaries — no changes to the scheduler/executor/merger. Non-linear graphs are rejected with a clear message and reserved for the (deferred) graph interpreter. diff --git a/packages/dashboard/app/components/AppModals.tsx b/packages/dashboard/app/components/AppModals.tsx index ee8e5f0bed..9b543557c1 100644 --- a/packages/dashboard/app/components/AppModals.tsx +++ b/packages/dashboard/app/components/AppModals.tsx @@ -29,6 +29,7 @@ import { useNavigationHistoryContext } from "../hooks/useNavigationHistory"; const SetupWizardModal = lazy(() => import("./SetupWizardModal").then((m) => ({ default: m.SetupWizardModal }))); const SettingsModal = lazy(() => import("./SettingsModal").then((m) => ({ default: m.SettingsModal }))); +const WorkflowNodeEditor = lazy(() => import("./WorkflowNodeEditor").then((m) => ({ default: m.WorkflowNodeEditor }))); function prefetchSettingsModal() { const idle: (cb: () => void, opts?: { timeout?: number }) => number = @@ -378,9 +379,26 @@ export function AppModals({ onClose={modalManager.closeWorkflowSteps} addToast={addToast} projectId={projectId} + onOpenGraphEditor={() => { + modalManager.closeWorkflowSteps(); + modalManager.openWorkflowEditor(); + }} /> + {modalManager.workflowEditorOpen && ( + + + + + + )} + void; + addToast: (message: string, type?: ToastType) => void; + projectId?: string; +} + +let nodeSeq = 0; +function newNodeId(): string { + nodeSeq += 1; + return `n-${Date.now().toString(36)}-${nodeSeq}`; +} + +const PALETTE: Array<{ kind: WorkflowEditorNodeKind; label: string; icon: typeof MessageSquare }> = [ + { kind: "prompt", label: "Prompt", icon: MessageSquare }, + { kind: "script", label: "Script", icon: Terminal }, + { kind: "gate", label: "Gate", icon: Shield }, + { kind: "merge", label: "Merge boundary", icon: GitMerge }, +]; + +function InnerEditor({ + onClose, + addToast, + projectId, + modalRef, +}: Omit & { modalRef: React.RefObject }) { + const [workflows, setWorkflows] = useState([]); + const [activeId, setActiveId] = useState(null); + const [loading, setLoading] = useState(false); + const [saving, setSaving] = useState(false); + const [validationError, setValidationError] = useState(null); + const [nodes, setNodes, onNodesChange] = useNodesState>([]); + const [edges, setEdges, onEdgesChange] = useEdgesState([]); + const [selectedNodeId, setSelectedNodeId] = useState(null); + + const activeWorkflow = useMemo(() => workflows.find((w) => w.id === activeId), [workflows, activeId]); + + const loadWorkflows = useCallback(async () => { + setLoading(true); + try { + const data = await fetchWorkflows(projectId); + setWorkflows(data); + setActiveId((prev) => prev ?? data[0]?.id ?? null); + } catch (err) { + addToast(getErrorMessage(err) || "Failed to load workflows", "error"); + } finally { + setLoading(false); + } + }, [projectId, addToast]); + + useEffect(() => { + void loadWorkflows(); + }, [loadWorkflows]); + + // Load the active workflow graph into the canvas. + useEffect(() => { + if (!activeWorkflow) { + setNodes([]); + setEdges([]); + return; + } + const flow = irToFlow(activeWorkflow); + setNodes(flow.nodes); + setEdges(flow.edges); + setSelectedNodeId(null); + setValidationError(null); + }, [activeWorkflow, setNodes, setEdges]); + + const onConnect = useCallback( + (connection: Connection) => { + setEdges((eds) => + addEdge({ ...connection, label: "success", data: { condition: "success" } }, eds), + ); + }, + [setEdges], + ); + + const addNode = useCallback( + (kind: WorkflowEditorNodeKind) => { + const id = newNodeId(); + const label = kind === "merge" ? "Merge boundary" : kind.charAt(0).toUpperCase() + kind.slice(1); + setNodes((ns) => [ + ...ns, + { + id, + type: kind, + position: { x: 200 + ns.length * 40, y: 240 + (ns.length % 3) * 70 }, + data: { kind, label, config: kind === "gate" ? { gateMode: "gate" } : {} }, + deletable: true, + }, + ]); + setSelectedNodeId(id); + }, + [setNodes], + ); + + const updateSelectedData = useCallback( + (patch: Partial | { config: Record }) => { + if (!selectedNodeId) return; + setNodes((ns) => + ns.map((n) => + n.id === selectedNodeId + ? { + ...n, + data: { + ...n.data, + ...("config" in patch ? { config: { ...n.data.config, ...patch.config } } : patch), + }, + } + : n, + ), + ); + }, + [selectedNodeId, setNodes], + ); + + const handleCreateWorkflow = useCallback(async () => { + const name = window.prompt("New workflow name"); + if (!name?.trim()) return; + try { + const created = await createWorkflow( + { name: name.trim(), ir: emptyWorkflowIr(name.trim()), layout: emptyWorkflowLayout() }, + projectId, + ); + setWorkflows((ws) => [...ws, created]); + setActiveId(created.id); + addToast(`Created workflow "${created.name}"`, "success"); + } catch (err) { + addToast(getErrorMessage(err) || "Failed to create workflow", "error"); + } + }, [projectId, addToast]); + + const handleDeleteWorkflow = useCallback(async () => { + if (!activeWorkflow) return; + if (!window.confirm(`Delete workflow "${activeWorkflow.name}"?`)) return; + try { + await deleteWorkflow(activeWorkflow.id, projectId); + setWorkflows((ws) => ws.filter((w) => w.id !== activeWorkflow.id)); + setActiveId(null); + addToast("Workflow deleted", "success"); + } catch (err) { + addToast(getErrorMessage(err) || "Failed to delete workflow", "error"); + } + }, [activeWorkflow, projectId, addToast]); + + const handleSave = useCallback(async () => { + if (!activeWorkflow) return; + setSaving(true); + setValidationError(null); + try { + const { ir, layout } = flowToIr(activeWorkflow.name, nodes, edges); + const updated = await updateWorkflow(activeWorkflow.id, { ir, layout }, projectId); + setWorkflows((ws) => ws.map((w) => (w.id === updated.id ? updated : w))); + // Validate by compiling — surfaces non-linear graphs as a banner. + try { + await compileWorkflow(updated.id, projectId); + addToast("Workflow saved", "success"); + } catch (compileErr) { + setValidationError(getErrorMessage(compileErr) || "Workflow saved but cannot be compiled"); + } + } catch (err) { + const message = getErrorMessage(err) || "Failed to save workflow"; + setValidationError(message); + addToast(message, "error"); + } finally { + setSaving(false); + } + }, [activeWorkflow, nodes, edges, projectId, addToast]); + + const selectedNode = nodes.find((n) => n.id === selectedNodeId) ?? null; + const overlayProps = useOverlayDismiss(onClose); + + return ( +
+
e.stopPropagation()}> +
+

Workflows

+ +
+ +
+ + +
+ {activeWorkflow ? ( + <> +
+
+ {PALETTE.map(({ kind, label, icon: Icon }) => ( + + ))} +
+
+ + +
+
+ + {validationError && ( +
+ {validationError} +
+ )} + +
+ setSelectedNodeId(node.id)} + onPaneClick={() => setSelectedNodeId(null)} + fitView + > + + + + +
+ + ) : ( +
+ Select or create a workflow to start editing. +
+ )} +
+ + {selectedNode && selectedNode.data.kind !== "start" && selectedNode.data.kind !== "end" && ( +