diff --git a/packages/core/src/__tests__/store-persistence.test.ts b/packages/core/src/__tests__/store-persistence.test.ts index c88174d48e..6403a8c59e 100644 --- a/packages/core/src/__tests__/store-persistence.test.ts +++ b/packages/core/src/__tests__/store-persistence.test.ts @@ -428,5 +428,26 @@ describe("TaskStore", () => { const listed = tasks.find((t) => t.id === paused.id); expect(listed?.pausedReason).toBe("worktrunk_operation_failed"); }); + + it("survives a disk-backed store reload", async () => { + // The original bug was "pause state vanished on reload" — an in-memory + // cache could mask a missing persist column, so close and reopen the + // store from disk before asserting. + harness.store().close(); + await harness.reopenDiskBackedStore(); + + const task = await harness.store().createTask({ description: "Pause across reload" }); + await harness.store().updateTask(task.id, { + paused: true, + pausedReason: "workflow-input:ask: What environment should this deploy to?", + }); + + harness.store().close(); + await harness.reopenDiskBackedStore(); + + const reloaded = await harness.store().getTask(task.id); + expect(reloaded.paused).toBe(true); + expect(reloaded.pausedReason).toBe("workflow-input:ask: What environment should this deploy to?"); + }); }); }); diff --git a/packages/core/src/__tests__/workflow-compiler.test.ts b/packages/core/src/__tests__/workflow-compiler.test.ts index 83fef4360f..596ccfb489 100644 --- a/packages/core/src/__tests__/workflow-compiler.test.ts +++ b/packages/core/src/__tests__/workflow-compiler.test.ts @@ -145,6 +145,48 @@ describe("compileWorkflowToSteps (U2)", () => { expect(err).toBeInstanceOf(WorkflowCompileError); }); + it("rejects seams that are out of the execute -> review -> merge order", () => { + const ir: WorkflowIr = { + version: "v1", + name: "misordered-seams", + nodes: [ + { id: "start", kind: "start" }, + { id: "merge", kind: "prompt", config: { seam: "merge" } }, + { id: "review", kind: "prompt", config: { seam: "review" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "merge", condition: "success" }, + { from: "merge", to: "review", condition: "success" }, + { from: "review", to: "end", condition: "success" }, + ], + }; + const err = validateLinearity(ir); + expect(err).toBeInstanceOf(WorkflowCompileError); + expect(err?.message).toMatch(/execute -> review -> merge order/); + }); + + it("rejects a graph with a duplicated seam role", () => { + const ir: WorkflowIr = { + version: "v1", + name: "dup-merge", + nodes: [ + { id: "start", kind: "start" }, + { id: "merge1", kind: "prompt", config: { seam: "merge" } }, + { id: "merge2", kind: "prompt", config: { seam: "merge" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "merge1", condition: "success" }, + { from: "merge1", to: "merge2", condition: "success" }, + { from: "merge2", to: "end", condition: "success" }, + ], + }; + const err = validateLinearity(ir); + expect(err).toBeInstanceOf(WorkflowCompileError); + expect(err?.message).toMatch(/appears more than once/); + }); + it("returns an empty step set for a graph with only start/seams/end", () => { const ir = graph([]); expect(compileWorkflowToSteps(ir)).toEqual([]); diff --git a/packages/core/src/__tests__/workflow-definition-store.test.ts b/packages/core/src/__tests__/workflow-definition-store.test.ts index 7f8cf3c807..27f17e95fa 100644 --- a/packages/core/src/__tests__/workflow-definition-store.test.ts +++ b/packages/core/src/__tests__/workflow-definition-store.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect, beforeEach, afterEach } from "vitest"; import { WorkflowIrError } from "../workflow-ir.js"; +import { isBuiltinWorkflowId } from "../builtin-workflows.js"; import type { WorkflowIr } from "../workflow-ir-types.js"; import { createTaskStoreTestHarness } from "./store-test-helpers.js"; @@ -43,11 +44,12 @@ describe("TaskStore workflow definitions (U1)", () => { }); 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 }); + // The list prepends read-only built-ins; assert on the user workflows only. + const userList = (await store.listWorkflowDefinitions()).filter((w) => !isBuiltinWorkflowId(w.id)); + expect(userList).toHaveLength(1); + expect(userList[0].name).toBe("Quality Gate"); + expect(userList[0].ir.nodes).toHaveLength(3); + expect(userList[0].layout.lint).toEqual({ x: 120, y: 0 }); }); it("rejects a workflow whose IR is missing start/end", async () => { @@ -55,7 +57,7 @@ describe("TaskStore workflow definitions (U1)", () => { await expect( store.createWorkflowDefinition({ name: "Broken", ir: bad }), ).rejects.toBeInstanceOf(WorkflowIrError); - expect(await store.listWorkflowDefinitions()).toHaveLength(0); + expect((await store.listWorkflowDefinitions()).filter((w) => !isBuiltinWorkflowId(w.id))).toHaveLength(0); }); it("requires a non-empty name", async () => { @@ -88,7 +90,7 @@ describe("TaskStore workflow definitions (U1)", () => { 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( + expect(new Date(updated.updatedAt).getTime()).toBeGreaterThan( new Date(created.updatedAt).getTime(), ); }); @@ -108,7 +110,7 @@ describe("TaskStore workflow definitions (U1)", () => { 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); + expect((await store.listWorkflowDefinitions()).filter((w) => !isBuiltinWorkflowId(w.id))).toHaveLength(0); }); it("throws when deleting a non-existent workflow", async () => { diff --git a/packages/core/src/__tests__/workflow-selection-store.test.ts b/packages/core/src/__tests__/workflow-selection-store.test.ts index 4d596a3e1d..cfc4b37157 100644 --- a/packages/core/src/__tests__/workflow-selection-store.test.ts +++ b/packages/core/src/__tests__/workflow-selection-store.test.ts @@ -118,6 +118,27 @@ describe("TaskStore workflow selection (U3)", () => { expect(detail.enabledWorkflowSteps ?? []).toHaveLength(0); }); + it("force-resurrecting over a tombstoned task purges its prior workflow 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 priorIds = store.getTaskWorkflowSelection(task.id)!.stepIds; + expect(priorIds).toHaveLength(2); + + // Soft-delete then physically resurrect the same id; the physical purge of + // the old tasks row must drop the orphaned selection + its compiled steps. + await store.deleteTask(task.id); + await store.createTaskWithReservedId( + { description: "resurrected", enabledWorkflowSteps: [], forceResurrect: true }, + { taskId: task.id, applyDefaultWorkflowSteps: false }, + ); + + expect(store.getTaskWorkflowSelection(task.id)).toBeUndefined(); + for (const id of priorIds) { + expect(await store.getWorkflowStep(id)).toBeUndefined(); + } + }); + 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); diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 07ad49e9c8..2bf09735fd 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 = 104; +const SCHEMA_VERSION = 105; export { SCHEMA_VERSION }; @@ -4092,6 +4092,29 @@ export class Database { }); } + // Migration 105: task_workflow_selection has no FK to tasks(id) (SQLite can't + // add one to an existing table without a rebuild), so physical task deletes + // before this version could leave orphaned selection rows and unreclaimable + // compiled workflow_steps. Drop any already-orphaned rows and their steps. + if (version < 105) { + this.applyMigration(105, () => { + // Delete the compiled steps referenced by orphaned selections first, then + // the orphaned selection rows themselves. json_each expands the stepIds + // JSON array; the WHERE guards against malformed (non-array) stepIds. + this.db.exec(` + DELETE FROM workflow_steps WHERE id IN ( + SELECT je.value + FROM task_workflow_selection sel + JOIN json_each(sel.stepIds) je + WHERE json_valid(sel.stepIds) + AND sel.taskId NOT IN (SELECT id FROM tasks) + ); + DELETE FROM task_workflow_selection + WHERE taskId NOT IN (SELECT id FROM tasks); + `); + }); + } + } /** diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 9e3b21b9c4..a5f33e2865 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -2529,6 +2529,7 @@ export class TaskStore extends EventEmitter { const allowResurrection = existing.allowResurrection === true; if (input.forceResurrect === true || allowResurrection) { + this.purgeTaskWorkflowSelectionRows(id); this.db.prepare("DELETE FROM tasks WHERE id = ?").run(id); this.db.bumpLastModified(); return; @@ -3779,18 +3780,26 @@ export class TaskStore extends EventEmitter { resolvedWorkflowSteps = undefined; } - const task = await this.createTaskWithDistributedReservation(input, { - createTaskWithId: async (taskId) => { - await this.assertNoDependencyCycle(taskId, input.dependencies ?? [], "createTask"); - return this._createTaskInternal( - input, - title, - resolvedWorkflowSteps, - taskId, - { invokeTaskCreatedHook: shouldInvokeTaskCreatedHook && !hasPendingSummarization }, - ); - }, - }); + let task: Task; + try { + task = await this.createTaskWithDistributedReservation(input, { + createTaskWithId: async (taskId) => { + await this.assertNoDependencyCycle(taskId, input.dependencies ?? [], "createTask"); + return this._createTaskInternal( + input, + title, + resolvedWorkflowSteps, + taskId, + { invokeTaskCreatedHook: shouldInvokeTaskCreatedHook && !hasPendingSummarization }, + ); + }, + }); + } catch (err) { + // The task row was never created, so any default-workflow steps we + // materialized above would orphan with no task/selection pointing at them. + this.cleanupOrphanedMaterializedSteps(pendingWorkflowSelection?.stepIds); + throw err; + } // Record the inherited workflow selection now that the task row exists. if (pendingWorkflowSelection) { @@ -3942,12 +3951,20 @@ export class TaskStore extends EventEmitter { resolvedWorkflowSteps = undefined; } - const createdTask = await this._createTaskInternal(input, title, resolvedWorkflowSteps, id, { - createdAt: options.createdAt, - updatedAt: options.updatedAt, - promptOverride: options.prompt, - invokeTaskCreatedHook: options.invokeTaskCreatedHook, - }); + let createdTask: Task; + try { + createdTask = await this._createTaskInternal(input, title, resolvedWorkflowSteps, id, { + createdAt: options.createdAt, + updatedAt: options.updatedAt, + promptOverride: options.prompt, + invokeTaskCreatedHook: options.invokeTaskCreatedHook, + }); + } catch (err) { + // The task row was never created, so any default-workflow steps we + // materialized above would orphan with no task/selection pointing at them. + this.cleanupOrphanedMaterializedSteps(pendingWorkflowSelection?.stepIds); + throw err; + } // Record the inherited workflow selection now that the task row exists. if (pendingWorkflowSelection) { @@ -7827,6 +7844,7 @@ export class TaskStore extends EventEmitter { private deleteTaskById(taskId: string): void { this.clearLinkedAgentTaskIds(taskId); + this.purgeTaskWorkflowSelectionRows(taskId); this.db.prepare('DELETE FROM tasks WHERE id = ?').run(taskId); this.db.bumpLastModified(); } @@ -8454,6 +8472,7 @@ export class TaskStore extends EventEmitter { this.db.transaction(() => { rewrittenLineageChildren = this.rewriteLineageChildrenForRemoval(id, lineageChildIds); this.clearLinkedAgentTaskIds(id, task.updatedAt); + this.purgeTaskWorkflowSelectionRows(id); this.db.prepare('DELETE FROM tasks WHERE id = ?').run(id); this.db.bumpLastModified(); }); @@ -10472,6 +10491,7 @@ export class TaskStore extends EventEmitter { const archivedAt = task.columnMovedAt ?? task.updatedAt ?? new Date().toISOString(); const entry = await this.taskToArchiveEntry(task, archivedAt); this.archiveDb.upsert(entry); + this.purgeTaskWorkflowSelectionRows(task.id); this.db.prepare("DELETE FROM tasks WHERE id = ?").run(task.id); await rm(this.taskDir(task.id), { recursive: true, force: true }); if (this.isWatching) { @@ -10507,6 +10527,7 @@ export class TaskStore extends EventEmitter { this.archiveDb.upsert(entry); // Remove task from tasks table + this.purgeTaskWorkflowSelectionRows(task.id); this.db.prepare('DELETE FROM tasks WHERE id = ?').run(task.id); this.db.bumpLastModified(); @@ -11018,16 +11039,21 @@ ${stepsSection}`; /** 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")}`; + // Serialize the read+increment in one write transaction so two TaskStore + // instances cannot both observe the same counter and allocate the same + // WF-id (which would collide on the workflows primary key). + return this.db.transactionImmediate(() => { + 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: { @@ -11322,6 +11348,49 @@ ${stepsSection}`; this.db.prepare("DELETE FROM task_workflow_selection WHERE taskId = ?").run(taskId); } + /** Purge a task's workflow selection and its materialized WorkflowStep rows + * when the task row itself is being physically removed. `task_workflow_selection` + * has no FK to `tasks(id)` (SQLite can't add one to an existing table without a + * rebuild), so deletion must be mirrored here to avoid orphaned selection rows + * and unreclaimable compiled steps. Best-effort and synchronous: unlike + * clearTaskWorkflowSelection it does not touch enabledWorkflowSteps, since the + * owning task row no longer exists. */ + private purgeTaskWorkflowSelectionRows(taskId: string): void { + const row = this.db + .prepare("SELECT stepIds FROM task_workflow_selection WHERE taskId = ?") + .get(taskId) as { stepIds: string } | undefined; + if (!row) return; + try { + const parsed = JSON.parse(row.stepIds) as unknown; + if (Array.isArray(parsed)) { + for (const stepId of parsed) { + if (typeof stepId === "string") { + this.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(stepId); + } + } + } + } catch { + // Corrupt stepIds list — still remove the selection row below. + } + this.db.prepare("DELETE FROM task_workflow_selection WHERE taskId = ?").run(taskId); + this.workflowStepsCache = null; + } + + /** Delete a set of freshly materialized WorkflowStep rows that were never + * successfully attached to a task/selection (e.g. the owning task create + * failed). Best-effort; tolerates already-removed ids. */ + private cleanupOrphanedMaterializedSteps(stepIds: string[] | undefined): void { + if (!stepIds || stepIds.length === 0) return; + for (const stepId of stepIds) { + try { + this.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(stepId); + } catch { + // Best-effort cleanup. + } + } + this.workflowStepsCache = null; + } + /** Persist pre-compiled workflow steps as fresh WorkflowStep rows and return * their ids in execution order. Steps are tagged so they stay out of the * step manager. Compile via compileWorkflowToSteps before calling. */ @@ -11372,8 +11441,23 @@ ${stepsSection}`; // referencing already-deleted step ids. const priorSelection = this.getTaskWorkflowSelection(taskId); const ids = await this.materializeWorkflowSteps(workflowId, inputs); - await this.updateTask(taskId, { enabledWorkflowSteps: ids }); - this.writeTaskWorkflowSelection(taskId, workflowId, ids); + try { + await this.updateTask(taskId, { enabledWorkflowSteps: ids }); + this.writeTaskWorkflowSelection(taskId, workflowId, ids); + } catch (err) { + // The owner write (updateTask / selection upsert) failed, so the steps we + // just materialized would orphan with no selection row pointing at them. + // Delete them before propagating; the prior selection is left untouched. + for (const stepId of ids) { + try { + this.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(stepId); + } catch { + // Best-effort cleanup; surface the original error below. + } + } + this.workflowStepsCache = null; + throw err; + } if (priorSelection) { for (const stepId of priorSelection.stepIds) { diff --git a/packages/core/src/workflow-compiler.ts b/packages/core/src/workflow-compiler.ts index 6df0f4fd4e..dfbd3f1bcd 100644 --- a/packages/core/src/workflow-compiler.ts +++ b/packages/core/src/workflow-compiler.ts @@ -102,10 +102,36 @@ export function validateLinearity(ir: WorkflowIr): WorkflowCompileError | null { } // Reachability: the single main path must reach end and cover every node. + // While walking, enforce the canonical seam pipeline: each of execute/review/ + // merge may appear at most once and only in that order. The compiler treats + // seams as a fixed execute → review → merge boundary (merge flips pre- to + // post-merge), so out-of-order or duplicate seams would compile inconsistently + // with the runtime contract. + const expectedSeamOrder = ["execute", "review", "merge"] as const; + const seenSeams = new Set(); + let nextExpectedSeamIndex = 0; const visited = new Set(); let cursor: string | undefined = startNode.id; while (cursor && !visited.has(cursor)) { visited.add(cursor); + const node = nodesById.get(cursor); + const seam = node ? seamOf(node) : undefined; + if (seam) { + if (seenSeams.has(seam)) { + return new WorkflowCompileError(`seam '${seam}' appears more than once`); + } + while ( + nextExpectedSeamIndex < expectedSeamOrder.length && + expectedSeamOrder[nextExpectedSeamIndex] !== seam + ) { + nextExpectedSeamIndex += 1; + } + if (expectedSeamOrder[nextExpectedSeamIndex] !== seam) { + return new WorkflowCompileError("seams must follow the execute -> review -> merge order"); + } + seenSeams.add(seam); + nextExpectedSeamIndex += 1; + } if (cursor === endNode.id) break; cursor = mainEdge(outgoing.get(cursor) ?? [])?.to; } diff --git a/packages/dashboard/app/components/InlineCreateCard.tsx b/packages/dashboard/app/components/InlineCreateCard.tsx index 60a7914f64..03403d20b4 100644 --- a/packages/dashboard/app/components/InlineCreateCard.tsx +++ b/packages/dashboard/app/components/InlineCreateCard.tsx @@ -647,6 +647,7 @@ export function InlineCreateCard({ onPlanningMode?.(trimmed); // Clear the input after triggering planning mode setDescription(""); + setSelectedWorkflowId(null); setDependencies([]); setExecutorProvider(undefined); setExecutorModelId(undefined); @@ -672,6 +673,7 @@ export function InlineCreateCard({ onSubtaskBreakdown?.(trimmed); // Clear the input after triggering subtask breakdown setDescription(""); + setSelectedWorkflowId(null); setDependencies([]); setExecutorProvider(undefined); setExecutorModelId(undefined); diff --git a/packages/dashboard/app/components/NewTaskModal.tsx b/packages/dashboard/app/components/NewTaskModal.tsx index 3e7c3158a9..15fc8bf42b 100644 --- a/packages/dashboard/app/components/NewTaskModal.tsx +++ b/packages/dashboard/app/components/NewTaskModal.tsx @@ -157,6 +157,7 @@ export function NewTaskModal({ isOpen, onClose, projectId, tasks, onCreateTask, description.trim() !== "" || dependencies.length > 0 || pendingImages.length > 0 || + selectedWorkflowId !== null || executorModel !== "" || validatorModel !== "" || planningModel !== "" || @@ -173,7 +174,7 @@ export function NewTaskModal({ isOpen, onClose, projectId, tasks, onCreateTask, githubTrackingEnabled || githubRepoOverrideTrimmed !== ""; setHasDirtyState(isDirty); - }, [description, dependencies, pendingImages, executorModel, validatorModel, planningModel, thinkingLevel, selectedWorkflowSteps, selectedAgentId, reviewLevel, autoMerge, priority, nodeId, branchMode, branch, baseBranch, githubTrackingEnabled, githubRepoOverrideTrimmed]); + }, [description, dependencies, pendingImages, selectedWorkflowId, executorModel, validatorModel, planningModel, thinkingLevel, selectedWorkflowSteps, selectedAgentId, reviewLevel, autoMerge, priority, nodeId, branchMode, branch, baseBranch, githubTrackingEnabled, githubRepoOverrideTrimmed]); const handleClose = useCallback(async () => { if (hasDirtyState) { @@ -196,6 +197,7 @@ export function NewTaskModal({ isOpen, onClose, projectId, tasks, onCreateTask, setThinkingLevel(""); setSelectedPresetId(""); setPresetMode("default"); + setSelectedWorkflowId(null); setSelectedWorkflowSteps([]); setWorkflowStepsExplicitlySet(false); setSelectedAgentId(null); diff --git a/packages/dashboard/app/components/TaskCard.tsx b/packages/dashboard/app/components/TaskCard.tsx index f8c8c71b6c..0b8ca3ef7e 100644 --- a/packages/dashboard/app/components/TaskCard.tsx +++ b/packages/dashboard/app/components/TaskCard.tsx @@ -1706,16 +1706,11 @@ function TaskCardComponent({ {pausedByAgent ? "paused by agent" : "paused"} )} - {isAwaitingInput && ( - - Needs input - - )} {!isPaused && visualStatus && visualStatus !== "queued" && ( - {isStuck ? "Stuck" : isAwaitingApproval ? "Awaiting Approval" : getTaskStatusLabel(visualStatus)} + {isStuck ? "Stuck" : isAwaitingApproval ? "Awaiting Approval" : isAwaitingInput ? "Needs input" : getTaskStatusLabel(visualStatus)} )} {hasInReviewStall && stallCopy && ( diff --git a/packages/dashboard/app/components/TaskDetailModal.tsx b/packages/dashboard/app/components/TaskDetailModal.tsx index 2c19d6e799..97007bb736 100644 --- a/packages/dashboard/app/components/TaskDetailModal.tsx +++ b/packages/dashboard/app/components/TaskDetailModal.tsx @@ -2742,7 +2742,14 @@ export function TaskDetailContent({ enabledWorkflowSteps={workflowEnabledSteps} canEdit={canEdit} projectId={projectId} - isTaskInProgress={task.column === "in-progress" && task.status !== "paused"} + isTaskInProgress={ + task.column === "in-progress" + && !task.paused + && !task.userPaused + && task.status !== "paused" + && task.status !== "awaiting-user-input" + && task.status !== "awaiting-cli-approval" + } onWorkflowStepsChange={handleWorkflowStepsChange} taskStatus={task.status} taskPausedReason={task.pausedReason} diff --git a/packages/dashboard/app/components/WorkflowNodeEditor.tsx b/packages/dashboard/app/components/WorkflowNodeEditor.tsx index c9c1cde39a..5e8ed32753 100644 --- a/packages/dashboard/app/components/WorkflowNodeEditor.tsx +++ b/packages/dashboard/app/components/WorkflowNodeEditor.tsx @@ -151,7 +151,15 @@ function InnerEditor({ ); const updateSelectedData = useCallback( - (patch: Partial | { config: Record }) => { + ( + patch: + | Partial + | { + config: + | Record + | ((prev: Record) => Record); + }, + ) => { if (!selectedNodeId) return; setNodes((ns) => ns.map((n) => @@ -160,7 +168,14 @@ function InnerEditor({ ...n, data: { ...n.data, - ...("config" in patch ? { config: { ...n.data.config, ...patch.config } } : patch), + ...("config" in patch + ? { + config: + typeof patch.config === "function" + ? patch.config((n.data.config ?? {}) as Record) + : { ...(n.data.config ?? {}), ...patch.config }, + } + : patch), }, } : n, @@ -269,8 +284,16 @@ function InnerEditor({ addToast(getErrorMessage(err) || "Failed to load skills", "error"); }); } - // eslint-disable-next-line react-hooks/exhaustive-deps - }, [currentExecutor, selectedNode?.id]); + }, [ + currentExecutor, + selectedNode?.id, + selectedNode?.data.kind, + projectId, + addToast, + models.length, + agents.length, + skills.length, + ]); const overlayProps = useOverlayDismiss(onClose); @@ -544,9 +567,13 @@ function InnerEditor({ onChange={(e) => { const val = e.target.value.trim(); if (val === "") { - const patch: Record = { ...selectedNode.data.config }; - delete patch.maxRetries; - updateSelectedData({ config: patch }); + updateSelectedData({ + config: (prev) => { + const next = { ...prev }; + delete next.maxRetries; + return next; + }, + }); } else { const num = parseInt(val, 10); if (!isNaN(num)) updateSelectedData({ config: { maxRetries: num } }); diff --git a/packages/dashboard/app/components/WorkflowResultsTab.css b/packages/dashboard/app/components/WorkflowResultsTab.css index 5089adb4b8..ffa0953148 100644 --- a/packages/dashboard/app/components/WorkflowResultsTab.css +++ b/packages/dashboard/app/components/WorkflowResultsTab.css @@ -90,6 +90,11 @@ opacity: 0.7; } +.workflow-input-error { + font-size: 0.78rem; + color: var(--color-error); +} + .workflow-results-loading { display: flex; align-items: center; diff --git a/packages/dashboard/app/components/WorkflowResultsTab.tsx b/packages/dashboard/app/components/WorkflowResultsTab.tsx index 7b39607ff0..faa8501a8a 100644 --- a/packages/dashboard/app/components/WorkflowResultsTab.tsx +++ b/packages/dashboard/app/components/WorkflowResultsTab.tsx @@ -4,6 +4,7 @@ import { Check, ChevronDown, ChevronUp, Maximize2, Pencil, X } from "lucide-reac import ReactMarkdown from "react-markdown"; import remarkGfm from "remark-gfm"; import type { AgentLogEntry, WorkflowStep, WorkflowStepResult } from "@fusion/core"; +import { getErrorMessage } from "@fusion/core"; import { fetchWorkflowSteps, fetchTaskWorkflow, selectTaskWorkflow, submitTaskWorkflowInput, approveTaskWorkflowCli } from "../api"; import { WorkflowSelector } from "./WorkflowSelector"; import { useAgentLogs } from "../hooks/useAgentLogs"; @@ -233,6 +234,17 @@ export function WorkflowResultsTab({ const [allWorkflowSteps, setAllWorkflowSteps] = useState([]); const [isEditing, setIsEditing] = useState(false); const [selectedWorkflowId, setSelectedWorkflowId] = useState(null); + const [resumeError, setResumeError] = useState(null); + + // Reset the paused-action UI whenever the blocked node/task changes, so a new + // awaiting-user-input / awaiting-cli-approval pause starts with fresh controls + // instead of a stale "Resuming…" banner. + useEffect(() => { + setInputText(""); + setSubmitting(false); + setSubmitted(false); + setResumeError(null); + }, [taskId, taskStatus, taskPausedReason]); // Load the task's current workflow selection (if any). useEffect(() => { @@ -690,10 +702,13 @@ export function WorkflowResultsTab({ const handleSubmitInput = async () => { if (!inputText.trim() || submitting) return; setSubmitting(true); + setResumeError(null); try { await submitTaskWorkflowInput(taskId, inputText, projectId); setInputText(""); setSubmitted(true); + } catch (err) { + setResumeError(getErrorMessage(err) || "Failed to resume task"); } finally { setSubmitting(false); } @@ -702,9 +717,12 @@ export function WorkflowResultsTab({ const handleApproveCli = async () => { if (submitting) return; setSubmitting(true); + setResumeError(null); try { await approveTaskWorkflowCli(taskId, projectId); setSubmitted(true); + } catch (err) { + setResumeError(getErrorMessage(err) || "Failed to approve command"); } finally { setSubmitting(false); } @@ -736,6 +754,9 @@ export function WorkflowResultsTab({ > {submitting ? "Submitting…" : "Submit & resume"} + {resumeError && ( + {resumeError} + )} )} @@ -760,6 +781,9 @@ export function WorkflowResultsTab({ {submitting ? "Approving…" : "Approve & run"} To reject, keep the task paused and do not approve. + {resumeError && ( + {resumeError} + )} )} diff --git a/packages/dashboard/app/components/WorkflowSelector.tsx b/packages/dashboard/app/components/WorkflowSelector.tsx index 3710468830..76baf38b62 100644 --- a/packages/dashboard/app/components/WorkflowSelector.tsx +++ b/packages/dashboard/app/components/WorkflowSelector.tsx @@ -34,12 +34,16 @@ export function WorkflowSelector({ useEffect(() => { let cancelled = false; + setWorkflows([]); setLoading(true); fetchWorkflows(projectId) .then((data) => { if (!cancelled) setWorkflows(data); }) - .catch((err) => addToast?.(getErrorMessage(err) || "Failed to load workflows", "error")) + .catch((err) => { + if (!cancelled) setWorkflows([]); + addToast?.(getErrorMessage(err) || "Failed to load workflows", "error"); + }) .finally(() => { if (!cancelled) setLoading(false); }); @@ -103,11 +107,13 @@ export function ProjectDefaultWorkflowField({ projectId, addToast, onManage }: P useEffect(() => { let cancelled = false; + setValue(null); fetchProjectDefaultWorkflow(projectId) .then((res) => { if (!cancelled) setValue(res.workflowId); }) .catch(() => { + if (!cancelled) setValue(null); /* default is optional; ignore load failures */ }); return () => { diff --git a/packages/dashboard/app/components/__tests__/InlineCreateCard.test.tsx b/packages/dashboard/app/components/__tests__/InlineCreateCard.test.tsx index 3c0d6dbc4b..3fd127d525 100644 --- a/packages/dashboard/app/components/__tests__/InlineCreateCard.test.tsx +++ b/packages/dashboard/app/components/__tests__/InlineCreateCard.test.tsx @@ -25,6 +25,7 @@ vi.mock("lucide-react", () => ({ Server: () => null, Maximize2: () => null, Minimize2: () => null, + Workflow: () => null, })); // Mock ModelSelectionModal (renders via portal, so mock for testability) @@ -117,6 +118,10 @@ vi.mock("../../api", () => ({ uploadAttachment: vi.fn(), updateGlobalSettings: vi.fn(), fetchAgents: vi.fn().mockResolvedValue([]), + selectTaskWorkflow: vi.fn().mockResolvedValue({ workflowId: null, enabledWorkflowSteps: [] }), + fetchWorkflows: vi.fn().mockResolvedValue([]), + fetchProjectDefaultWorkflow: vi.fn().mockResolvedValue({ workflowId: null }), + setProjectDefaultWorkflow: vi.fn().mockResolvedValue({ workflowId: null }), checkDuplicateTasks: vi.fn().mockResolvedValue([]), DuplicateCandidatesError: class DuplicateCandidatesError extends Error { matches: unknown[]; diff --git a/packages/dashboard/app/components/__tests__/workflow-flow-mapping.test.ts b/packages/dashboard/app/components/__tests__/workflow-flow-mapping.test.ts new file mode 100644 index 0000000000..783f7c336d --- /dev/null +++ b/packages/dashboard/app/components/__tests__/workflow-flow-mapping.test.ts @@ -0,0 +1,95 @@ +import { describe, expect, it } from "vitest"; +import type { WorkflowDefinition } from "@fusion/core"; +import { irToFlow, flowToIr } from "../workflow-flow-mapping"; + +function makeDef(ir: WorkflowDefinition["ir"]): WorkflowDefinition { + return { + id: "WF-001", + name: ir.name, + description: "", + ir, + layout: {}, + createdAt: "2024-01-01T00:00:00.000Z", + updatedAt: "2024-01-01T00:00:00.000Z", + }; +} + +describe("workflow-flow-mapping name preservation", () => { + it("does not inject synthetic names for unnamed start/end/merge nodes on round-trip", () => { + const ir: WorkflowDefinition["ir"] = { + version: "v1", + name: "wf", + nodes: [ + { id: "start", kind: "start" }, + { id: "n1", kind: "prompt", config: { prompt: "do work" } }, + { id: "m1", kind: "prompt", config: { seam: "merge" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "n1", condition: "success" }, + { from: "n1", to: "m1", condition: "success" }, + { from: "m1", to: "end", condition: "success" }, + ], + }; + + const { nodes, edges } = irToFlow(makeDef(ir)); + const { ir: out } = flowToIr("wf", nodes, edges); + + const byId = Object.fromEntries(out.nodes.map((n) => [n.id, n])); + // start/end carry no config or a config without an injected name + expect(byId.start.config?.name).toBeUndefined(); + expect(byId.end.config?.name).toBeUndefined(); + // merge boundary keeps its seam but does not gain a synthetic "Merge boundary" name + expect(byId.m1.config?.seam).toBe("merge"); + expect(byId.m1.config?.name).toBeUndefined(); + // an unnamed prompt node keeps no synthetic id-as-name + expect(byId.n1.config?.name).toBeUndefined(); + expect(byId.n1.config?.prompt).toBe("do work"); + }); + + it("preserves an explicit node name across round-trips", () => { + const ir: WorkflowDefinition["ir"] = { + version: "v1", + name: "wf", + nodes: [ + { id: "start", kind: "start" }, + { id: "n1", kind: "prompt", config: { name: "Implement", prompt: "do work" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "n1", condition: "success" }, + { from: "n1", to: "end", condition: "success" }, + ], + }; + + const { nodes, edges } = irToFlow(makeDef(ir)); + const { ir: out } = flowToIr("wf", nodes, edges); + const n1 = out.nodes.find((n) => n.id === "n1"); + expect(n1?.config?.name).toBe("Implement"); + }); + + it("persists a user-entered label as the node name", () => { + const ir: WorkflowDefinition["ir"] = { + version: "v1", + name: "wf", + nodes: [ + { id: "start", kind: "start" }, + { id: "n1", kind: "prompt", config: { prompt: "do work" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "n1", condition: "success" }, + { from: "n1", to: "end", condition: "success" }, + ], + }; + + const { nodes, edges } = irToFlow(makeDef(ir)); + // Simulate the editor renaming the node via its label input. + const renamed = nodes.map((n) => + n.id === "n1" ? { ...n, data: { ...n.data, label: "Build feature" } } : n, + ); + const { ir: out } = flowToIr("wf", renamed, edges); + const n1 = out.nodes.find((n) => n.id === "n1"); + expect(n1?.config?.name).toBe("Build feature"); + }); +}); diff --git a/packages/dashboard/app/components/workflow-flow-mapping.ts b/packages/dashboard/app/components/workflow-flow-mapping.ts index f52c0619d4..cb0a7e975b 100644 --- a/packages/dashboard/app/components/workflow-flow-mapping.ts +++ b/packages/dashboard/app/components/workflow-flow-mapping.ts @@ -56,7 +56,21 @@ export function flowToIr( const irNodes: WorkflowIr["nodes"] = nodes.map((node) => { const data = node.data; const config: Record = { ...(data.config ?? {}) }; - if (data.label) config.name = data.label; + // `irToFlow` synthesizes display labels for unnamed nodes (matching the + // fallback below), so only persist a label that the user actually set — + // otherwise saving an untouched workflow injects synthetic names like + // "start"/"end"/"Merge boundary" and breaks IR round-trips. + const fallbackLabel = data.kind === "merge" ? "Merge boundary" : node.id; + if ( + data.kind !== "start" && + data.kind !== "end" && + data.label && + data.label !== fallbackLabel + ) { + config.name = data.label; + } else { + delete config.name; + } if (data.kind === "merge") { config.seam = "merge"; return { id: node.id, kind: "prompt", config }; diff --git a/packages/dashboard/src/__tests__/workflow-routes.test.ts b/packages/dashboard/src/__tests__/workflow-routes.test.ts index 6a4f508314..25a4fba391 100644 --- a/packages/dashboard/src/__tests__/workflow-routes.test.ts +++ b/packages/dashboard/src/__tests__/workflow-routes.test.ts @@ -135,6 +135,26 @@ describe("workflow routes (U4)", () => { expect((read.body as { workflowId: string }).workflowId).toBe(wfId); }); + it("PUT /tasks/:taskId/workflow rejects an omitted workflowId but clears on explicit null", 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: [] }); + await put(`/api/tasks/${task.id}/workflow`, { workflowId: wfId }); + + // Malformed body ({}) must not silently wipe the selection. + const omitted = await put(`/api/tasks/${task.id}/workflow`, {}); + expect(omitted.status).toBe(400); + const stillSelected = await get(`/api/tasks/${task.id}/workflow`); + expect((stillSelected.body as { workflowId: string }).workflowId).toBe(wfId); + + // Explicit null is the only clear signal. + const cleared = await put(`/api/tasks/${task.id}/workflow`, { workflowId: null }); + expect(cleared.status).toBe(200); + expect((cleared.body as { workflowId: string | null }).workflowId).toBeNull(); + const read = await get(`/api/tasks/${task.id}/workflow`); + expect((read.body as { workflowId: string | null }).workflowId).toBeNull(); + }); + 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; @@ -181,4 +201,33 @@ describe("workflow routes (U4)", () => { }); expect(res.status).toBe(400); }); + + it("approve-cli 400s when a CLI-approval reason lingers but the task is not paused", async () => { + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + // Stale reason string with no active pause must not be approvable. + await store.updateTask(task.id, { + paused: false, + pausedReason: "workflow-cli-approval:build: npm run build", + }); + const res = await post(`/api/tasks/${task.id}/workflow/approve-cli`, {}); + expect(res.status).toBe(400); + expect(await store.isWorkflowCliCommandApproved("npm run build")).toBe(false); + }); + + it("POST /workflow/input resumes without clearing pausedReason", async () => { + const task = await store.createTask({ description: "T", enabledWorkflowSteps: [] }); + await store.updateTask(task.id, { + paused: true, + pausedReason: "workflow-await-input:ask: please confirm", + }); + + const res = await post(`/api/tasks/${task.id}/workflow/input`, { text: "yes" }); + expect(res.status).toBe(200); + + const detail = await store.getTask(task.id); + expect(detail.paused).toBeFalsy(); + // The route deliberately leaves pausedReason intact; the await-input node + // consumes the marker itself on re-run. + expect(detail.pausedReason).toBe("workflow-await-input:ask: please confirm"); + }); }); diff --git a/packages/dashboard/src/routes/register-workflow-routes.ts b/packages/dashboard/src/routes/register-workflow-routes.ts index 841785748d..e93df43dfb 100644 --- a/packages/dashboard/src/routes/register-workflow-routes.ts +++ b/packages/dashboard/src/routes/register-workflow-routes.ts @@ -134,7 +134,13 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { try { const { store } = await getProjectContext(req); const workflowId = (req.body ?? {}).workflowId; - if (workflowId === null || workflowId === undefined) { + // Only an explicit null clears the selection. An omitted field + // (e.g. a malformed `{}` body) must fail validation rather than + // silently wiping the task's workflow. + if (workflowId === undefined) { + throw badRequest("workflowId is required (string to select, null to clear)"); + } + if (workflowId === null) { await store.clearTaskWorkflowSelection(req.params.taskId); res.json({ workflowId: null, enabledWorkflowSteps: [] }); return; @@ -174,7 +180,12 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { // would let any client approve an arbitrary command the task is not // actually paused on, bypassing trust-on-first-use entirely. const command = match ? match[1].trim() : ""; - if (!command) throw badRequest("No pending CLI command to approve for this task"); + // Require an active CLI-approval pause: a non-empty command parsed from + // pausedReason AND the task actually paused. This rejects approvals + // against a stale reason string on an already-resumed task. + if (!task.paused || !command) { + throw badRequest("No pending CLI command to approve for this task"); + } await store.approveWorkflowCliCommand(command); await store.updateTask(req.params.taskId, { status: null, paused: false, pausedReason: null }); res.json({ approved: command }); @@ -192,7 +203,12 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { const text = (req.body?.text as string | undefined)?.trim(); if (!text) throw badRequest("Input text is required"); await store.addSteeringComment(req.params.taskId, text); - await store.updateTask(req.params.taskId, { status: null, paused: false, pausedReason: null }); + // Do NOT clear pausedReason here: runAwaitInputNode checks + // (live.pausedReason ?? "").startsWith(marker) to confirm this specific + // node previously paused the task. Clearing it would make every re-run + // re-pause without ever consuming the answer. The node clears the marker + // itself once it consumes the input. + await store.updateTask(req.params.taskId, { status: null, paused: false }); res.json({ ok: true }); } catch (err: unknown) { if (err instanceof ApiError) throw err; diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 7fcb77e5aa..0e2ce55cc4 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -3243,8 +3243,18 @@ export class TaskExecutor { runCustomNode: (node, nodeTask) => this.runGraphCustomNode(node, nodeTask, settings), onEvent: (event) => executorLog.log(`[workflow-graph] ${event.type} ${event.taskId}: ${event.detail}`), }); - const detail = await this.store.getTask(task.id); - const result = await runner.run(detail, settings); + let result: WorkflowGraphTaskRunResult; + try { + const detail = await this.store.getTask(task.id); + result = await runner.run(detail, settings); + } catch (err) { + // A thrown interpreter error must not strand the task in-progress: fall + // back to the legacy pipeline so the normal executor lock + flow runs. + executorLog.error( + `[workflow-graph] ${task.id} interpreter threw — falling back to legacy pipeline: ${err instanceof Error ? err.message : String(err)}`, + ); + return false; + } if (result.disposition === "fell-back") { executorLog.log(`[workflow-graph] ${task.id} fell back to legacy pipeline: ${result.reason}`); return false; @@ -3362,20 +3372,44 @@ export class TaskExecutor { // pausedReason). A pre-existing steering comment (e.g. one added at task // creation) must never short-circuit the pause on the node's first run — // otherwise the node consumes a stale comment and never asks the user. - const pausedByThisNode = (live.pausedReason ?? "").startsWith(marker); - if (!live.paused && pausedByThisNode && steering.length > 0) { - // Input has arrived (user replied and unpaused): consume the latest comment. - const latest = steering[steering.length - 1] as { text?: string; comment?: string }; - const answer = (latest?.text ?? latest?.comment ?? "").toString(); - await this.store.updateTask(live.id, { status: null }, this.getRunContextFor(live.id)); - await this.store.logEntry(live.id, `Workflow input received for node '${node.id}'`, undefined, this.getRunContextFor(live.id)); - return { outcome: "success", value: "input-received", contextPatch: { [`input:${node.id}`]: answer } }; + const pausedReason = live.pausedReason ?? ""; + const pausedByThisNode = pausedReason.startsWith(marker); + if (!live.paused && pausedByThisNode) { + // Correlate the reply to THIS pause: the marker embeds a watermark + // (`${marker}@${pauseEpochMs}: …`) recorded when the node paused. Only + // count steering comments created at/after that watermark as the answer, + // so an unpause-without-reply can't consume a comment that predates the + // pause. The watermark is epoch milliseconds (colon-free) so it never + // collides with the `:` that separates the marker from the question, nor + // with the dashboard's colon-delimited question parser. + const watermark = (() => { + const m = pausedReason.slice(marker.length).match(/^@(\d+)/); + const t = m ? Number(m[1]) : NaN; + return Number.isFinite(t) ? t : undefined; + })(); + const replies = watermark === undefined + ? steering + : steering.filter((c) => { + const created = Date.parse((c as { createdAt?: string }).createdAt ?? ""); + return Number.isFinite(created) ? created >= watermark : false; + }); + if (replies.length > 0) { + // Input has arrived (user replied and unpaused): consume the latest + // post-pause comment and clear this node's marker so a future fresh + // visit re-asks instead of silently consuming a stale comment. + const latest = replies[replies.length - 1] as { text?: string; comment?: string }; + const answer = (latest?.text ?? latest?.comment ?? "").toString(); + await this.store.updateTask(live.id, { status: null, pausedReason: null }, this.getRunContextFor(live.id)); + await this.store.logEntry(live.id, `Workflow input received for node '${node.id}'`, undefined, this.getRunContextFor(live.id)); + return { outcome: "success", value: "input-received", contextPatch: { [`input:${node.id}`]: answer } }; + } + // Unpaused but no post-pause reply yet — re-park below and keep waiting. } await this.store.logEntry(live.id, `Workflow paused for user input: ${question}`, undefined, this.getRunContextFor(live.id)); await this.store.updateTask( live.id, - { status: "awaiting-user-input", paused: true, pausedReason: `${marker}: ${question}` }, + { status: "awaiting-user-input", paused: true, pausedReason: `${marker}@${Date.now()}: ${question}` }, this.getRunContextFor(live.id), ); // Failure outcome ends the walk; handleGraphFailure leaves paused tasks @@ -3523,6 +3557,15 @@ export class TaskExecutor { if (!skipApproval && !(await this.store.isWorkflowCliCommandApproved(rawCommand))) { return this.pauseForCliApproval(node, live, rawCommand); } + // We are proceeding to execute. If this task was previously paused by + // THIS node's CLI-approval gate, clear that status/pausedReason now — + // otherwise the task keeps the "awaiting-cli-approval" status through + // later graph nodes even though approval already happened (mirrors the + // status reset in runAwaitInputNode). + const approvalMarker = `workflow-cli-approval:${node.id}`; + if ((live.pausedReason ?? "").startsWith(approvalMarker)) { + await this.store.updateTask(live.id, { status: null, pausedReason: null }, this.getRunContextFor(live.id)); + } const env = prompt ? { ...process.env, FUSION_NODE_PROMPT: prompt } : undefined; const out = await this.runRawCliCommand( live, diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts index 6d89959eab..2abb1da2dc 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/packages/engine/src/workflow-node-handlers.ts @@ -77,7 +77,13 @@ export function createGateHandler(runCustomNode?: WorkflowCustomNodeRunner): Wor const hasExecutableConfig = typeof node.config?.prompt === "string" || typeof node.config?.scriptName === "string"; - if (hasExecutableConfig && runCustomNode) { + if (hasExecutableConfig) { + // Fail closed: an executable gate with no runner must NOT auto-pass — that + // would silently bypass the gate and let the workflow continue. Mirror the + // prompt/script handler, which throws in the same situation. + if (!runCustomNode) { + throw new WorkflowIrError(`No custom-node runner registered for node: ${node.id}`); + } return runCustomNode(node, context.task, context.context); }