From ffce42ff9c4c976a30faa6c8600ca32616c08c28 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 3 Jun 2026 09:22:35 -0700 Subject: [PATCH] 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; +}