FN-7113: validate workflow graph integrity before execution
Reject malformed workflow graphs at save and run boundaries. - Add central IR validation for duplicate top-level node ids, unregistered extension keys, and unreachable required nodes. - Re-validate resolved workflow IR in the graph runner before side effects and fail closed on invalid graphs. - Expand graph integrity tests and document save/run validation behavior with a release changeset. Files changed: .changeset/FN-7113-workflow-graph-integrity.md | 7 + docs/custom-workflow-reliability-acceptance-map.md | 7 +- docs/workflow-steps.md | 20 +++ .../workflow-ir-extension-metadata.test.ts | 49 ++++++- packages/core/src/__tests__/workflow-ir.test.ts | 106 +++++++++++++++- packages/core/src/workflow-ir.ts | 51 +++++++- .../__tests__/workflow-graph-task-runner.test.ts | 141 +++++++++++++++++---- packages/engine/src/workflow-graph-task-runner.ts | 46 +++++-- 8 files changed, 386 insertions(+), 41 deletions(-) Fusion-Task-Id: FN-7113 Fusion-Task-Lineage: f41e0d81-7cdf-497a-96ab-fe42504d0be4 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/FN-7113-workflow-graph-integrity.md
Normal file
7
.changeset/FN-7113-workflow-graph-integrity.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": minor
|
||||
---
|
||||
|
||||
summary: Reject malformed workflow graphs before they can be saved or launched.
|
||||
category: feature
|
||||
dev: Hardens the central parseWorkflowIr/validateV2 gate (duplicate-node-id and required top-level reachability rejection) and fail-closed re-validation at the WorkflowGraphTaskRunner run boundary before any side effects (FN-7113).
|
||||
@@ -9,6 +9,9 @@ This artifact distinguishes MVP/blocking requirements from nice-to-have enhancem
|
||||
|
||||
FNXC:WorkflowRouting 2026-06-22-12:00:
|
||||
Workflow selection acceptance must distinguish operator intent and task creator ownership from executor opportunism. Agents can assign workflows when the user asked or when creating the task; executors cannot reroute the task under execution unless instructed.
|
||||
|
||||
FNXC:WorkflowValidation 2026-06-27-00:00:
|
||||
FN-7113 upgrades graph integrity from an authoring-only expectation to a save-and-run acceptance criterion: malformed workflow DAGs must be rejected before persistence and revalidated before graph execution side effects.
|
||||
-->
|
||||
|
||||
## Purpose
|
||||
@@ -37,7 +40,7 @@ Use this document to write engineering tasks, QA plans, and release checks. It i
|
||||
- **Actor / need:** A workflow author needs to create or copy a workflow that can be reviewed, saved, and selected without corrupting built-in definitions.
|
||||
- **Trigger:** Open the [Workflow Editor](./workflow-editor.md) from the dashboard, duplicate a built-in with **Duplicate to customize**, start from Blank, import a JSON envelope, or use workflow tools such as `fn_workflow_create` / `fn_workflow_update`.
|
||||
- **Expected happy path + lifecycle transitions + feedback:** The editor serializes graph nodes/edges, columns, fields, and setting declarations into Workflow IR, saves the custom definition, and keeps built-ins read-only. The saved workflow appears in the editor picker and `fn_workflow_list`; no task lifecycle transition occurs until a task selects the workflow. The editor reports whether the workflow can run on the linear engine or must run on the graph interpreter.
|
||||
- **Failure / recovery expectation:** Invalid JSON, dangling edges, illegal cycles, unplaced nodes, blocking column-trait violations, invalid setting/field declarations, and attempts to mutate built-ins are rejected before partial persistence. Import errors and server validation errors render in a persistent inline error region; built-ins show read-only hints and disable mutation controls.
|
||||
- **Failure / recovery expectation:** Invalid JSON, duplicate top-level node ids, missing/multiple start or end nodes, dangling edges, illegal cycles, invalid step/template references, unknown plugin workflow-extension keys, unplaced nodes, blocking column-trait violations, invalid setting/field declarations, and attempts to mutate built-ins are rejected before partial persistence. Import errors and server validation errors render in a persistent inline error region; built-ins show read-only hints and disable mutation controls.
|
||||
- **Measurable success signal:** A stable workflow ID is returned/listed by `fn_workflow_list`; `fn_workflow_get` or the editor reload shows the saved IR; invalid saves return a typed validation failure without changing the prior persisted definition.
|
||||
- **Priority:** MVP/blocking for save/validation/discovery; enhancement for AI-assisted design quality and richer telemetry around definition registration.
|
||||
|
||||
@@ -64,7 +67,7 @@ Use this document to write engineering tasks, QA plans, and release checks. It i
|
||||
- **Actor / need:** The scheduler/executor needs to run the selected workflow deterministically while preserving Fusion's observable task lifecycle.
|
||||
- **Trigger:** A schedulable task with a selected or default workflow is picked up for execution.
|
||||
- **Expected happy path + lifecycle transitions + feedback:** `TaskExecutor.execute()` resolves the workflow, pins graph execution for the run, and `WorkflowGraphExecutor` traverses nodes through workflow runtime primitives such as planning, execute, workflow-step, review, merge, schedule, and step-execute. Standard coding work continues to show `todo → in-progress → in-review → done` (or equivalent workflow-defined columns/holds where enabled), workflow checks appear on task cards/list/detail, and task documents/artifacts are persisted as produced.
|
||||
- **Failure / recovery expectation:** Unsupported edge conditions throw `WorkflowIrError`; explicit custom workflow resolution failures fail closed; interpreter failures park as workflow failures rather than re-running a legacy imperative path. File-scope guards (`FileScopeViolationError`), squash overlap enforcement, `autoMerge:false` terminal-until-human behavior, and `moveTask(in-progress → todo)` hard-cancel semantics remain non-bypassable.
|
||||
- **Failure / recovery expectation:** Unsupported edge conditions throw `WorkflowIrError`; explicit custom workflow resolution failures fail closed; resolved IR is revalidated before any graph side effects and malformed graphs fail closed with `invalid-ir: <message>` rather than partially running or falling back into the wrong legacy workflow; interpreter failures after side effects park as workflow failures rather than re-running a legacy imperative path. File-scope guards (`FileScopeViolationError`), squash overlap enforcement, `autoMerge:false` terminal-until-human behavior, and `moveTask(in-progress → todo)` hard-cancel semantics remain non-bypassable.
|
||||
- **Measurable success signal:** Workflow results are visible in task card/list/detail surfaces; node outcomes route according to `success`, `failure`, or `outcome:<value>` edges; relevant run-audit records exist for lifecycle/git/database mutations; parity instrumentation emits `workflow:parity-observed` or `workflow:parity-drift` when dual-observe is enabled.
|
||||
- **Priority:** MVP/blocking.
|
||||
|
||||
|
||||
@@ -55,6 +55,26 @@ Decision-only or investigation tasks can also declare `noCommitsExpected` / `**N
|
||||
|
||||
Use the dashboard [Workflow Editor](./workflow-editor.md) to inspect built-ins, tune built-in prompts, duplicate workflows, or author custom workflows. Custom workflows can declare graph nodes and edges, columns/traits, task fields, typed workflow settings, model lanes, optional workflow-step templates, and author-time validation. Use this page for runtime semantics; use the editor guide for the visual authoring surface.
|
||||
|
||||
### Workflow graph integrity validation
|
||||
|
||||
<!--
|
||||
FNXC:WorkflowValidation 2026-06-27-00:00:
|
||||
FN-7113 makes graph integrity a save-and-run invariant. The central `parseWorkflowIr`/`validateV2` gate rejects malformed author/plugin graphs before persistence, and `WorkflowGraphTaskRunner` re-validates the resolved IR before side-effecting seams so stale or plugin-supplied invalid graphs fail closed instead of partially running or falling back into the wrong legacy workflow.
|
||||
-->
|
||||
|
||||
Workflow definitions are validated through the same central IR gate before they can be saved, imported, AI-designed, selected/materialized for a task, or launched by the graph interpreter. Dashboard routes and workflow tools surface `WorkflowIrError` / `WorkflowCompileError` messages as author-facing validation failures instead of persisting partial definitions.
|
||||
|
||||
The enforced integrity classes include:
|
||||
|
||||
- exactly one `start` node and exactly one `end` node;
|
||||
- unique top-level node ids and unique column/field/setting ids;
|
||||
- every top-level edge endpoint references a declared top-level node;
|
||||
- no illegal non-rework cycles in DAG-required regions;
|
||||
- required reachability/dominance rules: every required top-level node must be reachable from `start`, and `parse-steps` must dominate `foreach(source:"task-steps")`; interpreter-owned recovery entry primitives remain valid even when they are re-entered from persisted runtime state instead of the author-facing start path;
|
||||
- valid node-specific references, including `parse-steps` artifacts, `loop.exitWhen.nodeId`, foreach/loop/optional-group template entry/exit references, and registered plugin workflow-extension keys.
|
||||
|
||||
At run time, `WorkflowGraphTaskRunner` resolves the selected built-in or custom workflow, re-runs this integrity validation before any seam, primitive, or custom-node side effect, and fails closed with an `invalid-ir: <message>` reason when the resolved IR is malformed. Once a node side effect has run, runtime failures keep the existing failed-run behavior rather than re-running the legacy pipeline.
|
||||
|
||||
### Overriding built-in workflow prompts
|
||||
|
||||
<!--
|
||||
|
||||
@@ -29,7 +29,22 @@ describe("workflow IR extension metadata", () => {
|
||||
__resetWorkflowExtensionRegistryForTests();
|
||||
});
|
||||
|
||||
it("accepts plugin-namespaced column and node extension metadata", () => {
|
||||
it("accepts registered plugin-namespaced column and node extension metadata", () => {
|
||||
getWorkflowExtensionRegistry().register("workflow-pack", {
|
||||
extensionId: "role",
|
||||
name: "Role",
|
||||
kind: "column-metadata",
|
||||
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
|
||||
fallback: "failClosed",
|
||||
});
|
||||
getWorkflowExtensionRegistry().register("workflow-pack", {
|
||||
extensionId: "node-handler",
|
||||
name: "Node handler",
|
||||
kind: "node-handler",
|
||||
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
|
||||
fallback: "failClosed",
|
||||
});
|
||||
|
||||
const parsed = parseWorkflowIr(ir({
|
||||
columns: [
|
||||
{
|
||||
@@ -60,6 +75,22 @@ describe("workflow IR extension metadata", () => {
|
||||
expect(parsed.nodes[0].extensions?.["plugin:workflow-pack:node-handler"]).toEqual({ handler: "plan" });
|
||||
});
|
||||
|
||||
it("rejects unknown plugin-namespaced extension metadata keys", () => {
|
||||
expect(() =>
|
||||
parseWorkflowIr(ir({
|
||||
nodes: [
|
||||
{
|
||||
id: "start",
|
||||
kind: "start",
|
||||
column: "todo",
|
||||
extensions: { "plugin:workflow-pack:missing": {} },
|
||||
},
|
||||
{ id: "end", kind: "end", column: "todo" },
|
||||
],
|
||||
})),
|
||||
).toThrow(/Workflow node 'start' extension key 'plugin:workflow-pack:missing' is not registered/);
|
||||
});
|
||||
|
||||
it("rejects extension metadata keys outside the plugin namespace", () => {
|
||||
expect(() =>
|
||||
parseWorkflowIr(ir({
|
||||
@@ -76,6 +107,14 @@ describe("workflow IR extension metadata", () => {
|
||||
});
|
||||
|
||||
it("rejects non-object extension metadata values", () => {
|
||||
getWorkflowExtensionRegistry().register("workflow-pack", {
|
||||
extensionId: "node-handler",
|
||||
name: "Node handler",
|
||||
kind: "node-handler",
|
||||
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
|
||||
fallback: "failClosed",
|
||||
});
|
||||
|
||||
expect(() =>
|
||||
parseWorkflowIr(ir({
|
||||
nodes: [
|
||||
@@ -140,6 +179,14 @@ describe("workflow IR extension metadata", () => {
|
||||
});
|
||||
|
||||
it("keeps v2 when otherwise-pure workflows carry extension metadata", () => {
|
||||
getWorkflowExtensionRegistry().register("workflow-pack", {
|
||||
extensionId: "role",
|
||||
name: "Role",
|
||||
kind: "column-metadata",
|
||||
schemaVersion: WORKFLOW_EXTENSION_SCHEMA_VERSION,
|
||||
fallback: "failClosed",
|
||||
});
|
||||
|
||||
const parsed = parseWorkflowIr(ir({
|
||||
columns: [
|
||||
{
|
||||
|
||||
@@ -13,6 +13,7 @@ import type {
|
||||
WorkflowIrNode,
|
||||
WorkflowIrEdge,
|
||||
} from "../workflow-ir-types.js";
|
||||
import { BUILTIN_WORKFLOWS } from "../builtin-workflows.js";
|
||||
|
||||
function v2(
|
||||
columns: WorkflowIrV2["columns"],
|
||||
@@ -153,6 +154,25 @@ describe("parseWorkflowIr — v2 columns & placement", () => {
|
||||
);
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(/duplicate column id 'dup'/);
|
||||
});
|
||||
|
||||
it("rejects duplicate top-level node ids before Map de-duplication can mask them", () => {
|
||||
const ir = v2(
|
||||
[{ id: "only", name: "Only", traits: [] }],
|
||||
[
|
||||
{ id: "start", kind: "start", column: "only" },
|
||||
{ id: "dup", kind: "prompt", column: "only" },
|
||||
{ id: "dup", kind: "script", column: "only" },
|
||||
{ id: "end", kind: "end", column: "only" },
|
||||
],
|
||||
[
|
||||
{ from: "start", to: "dup" },
|
||||
{ from: "dup", to: "end" },
|
||||
],
|
||||
);
|
||||
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(WorkflowIrError);
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(/Workflow IR has duplicate node id 'dup'/);
|
||||
});
|
||||
});
|
||||
|
||||
// FNXC:WorkflowOptionalGroup 2026-06-21-18:00:
|
||||
@@ -659,8 +679,92 @@ describe("parseWorkflowIr — version & shape guards", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects missing start/end nodes", () => {
|
||||
const ir = v2([{ id: "c", name: "C", traits: [] }], [{ id: "start", kind: "start", column: "c" }], []);
|
||||
it("rejects missing start nodes", () => {
|
||||
const ir = v2([{ id: "c", name: "C", traits: [] }], [{ id: "end", kind: "end", column: "c" }], []);
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(WorkflowIrError);
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(/exactly one start and one end/);
|
||||
});
|
||||
|
||||
it("rejects missing end nodes", () => {
|
||||
const ir = v2([{ id: "c", name: "C", traits: [] }], [{ id: "start", kind: "start", column: "c" }], []);
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(WorkflowIrError);
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(/exactly one start and one end/);
|
||||
});
|
||||
|
||||
it("rejects illegal non-rework cycles", () => {
|
||||
const ir = v2(
|
||||
[{ id: "c", name: "C", traits: [] }],
|
||||
[
|
||||
{ id: "start", kind: "start", column: "c" },
|
||||
{ id: "a", kind: "prompt", column: "c" },
|
||||
{ id: "b", kind: "prompt", column: "c" },
|
||||
{ id: "end", kind: "end", column: "c" },
|
||||
],
|
||||
[
|
||||
{ from: "start", to: "a" },
|
||||
{ from: "a", to: "b" },
|
||||
{ from: "b", to: "a" },
|
||||
],
|
||||
);
|
||||
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(WorkflowIrError);
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(/illegal cycle.*edge 'b' -> 'a'/);
|
||||
});
|
||||
|
||||
it("rejects unreachable required top-level nodes with the offending node id", () => {
|
||||
const ir = v2(
|
||||
[{ id: "c", name: "C", traits: [] }],
|
||||
[
|
||||
{ id: "start", kind: "start", column: "c" },
|
||||
{ id: "reachable", kind: "prompt", column: "c" },
|
||||
{ id: "orphan", kind: "prompt", column: "c" },
|
||||
{ id: "end", kind: "end", column: "c" },
|
||||
],
|
||||
[
|
||||
{ from: "start", to: "reachable" },
|
||||
{ from: "reachable", to: "end" },
|
||||
],
|
||||
);
|
||||
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(WorkflowIrError);
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(/Workflow node 'orphan' is not reachable from the start node/);
|
||||
});
|
||||
|
||||
it("rejects invalid parse-steps artifact references with the offending node and artifact", () => {
|
||||
const ir = v2(
|
||||
[{ id: "c", name: "C", traits: [] }],
|
||||
[
|
||||
{ id: "start", kind: "start", column: "c" },
|
||||
{ id: "parse", kind: "parse-steps", column: "c", config: { artifact: "missing.md", parser: "step-headings" } },
|
||||
{ id: "end", kind: "end", column: "c" },
|
||||
],
|
||||
[
|
||||
{ from: "start", to: "parse" },
|
||||
{ from: "parse", to: "end" },
|
||||
],
|
||||
);
|
||||
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(WorkflowIrError);
|
||||
expect(() => parseWorkflowIr(ir)).toThrow(/parse-steps node 'parse' references artifact 'missing.md'/);
|
||||
});
|
||||
|
||||
it("parses valid v2 graphs and every built-in workflow without false rejection", () => {
|
||||
const valid = v2(
|
||||
[{ id: "c", name: "C", traits: [] }],
|
||||
[
|
||||
{ id: "start", kind: "start", column: "c" },
|
||||
{ id: "custom", kind: "prompt", column: "c" },
|
||||
{ id: "end", kind: "end", column: "c" },
|
||||
],
|
||||
[
|
||||
{ from: "start", to: "custom" },
|
||||
{ from: "custom", to: "end" },
|
||||
],
|
||||
);
|
||||
|
||||
expect(() => parseWorkflowIr(valid)).not.toThrow();
|
||||
for (const workflow of BUILTIN_WORKFLOWS) {
|
||||
expect(() => parseWorkflowIr(workflow.ir)).not.toThrow();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -307,6 +307,36 @@ function reachableFrom(
|
||||
return seen;
|
||||
}
|
||||
|
||||
const INTERPRETER_ENTRY_NODE_KINDS: ReadonlySet<WorkflowIrNodeKind> = new Set([
|
||||
"merge-gate",
|
||||
"merge-attempt",
|
||||
"manual-merge-hold",
|
||||
"retry-backoff",
|
||||
"recovery-router",
|
||||
"branch-group-member-integration",
|
||||
"branch-group-promotion",
|
||||
"pr-create",
|
||||
"pr-respond",
|
||||
"pr-merge",
|
||||
]);
|
||||
|
||||
function validateRequiredTopLevelReachability(
|
||||
nodes: WorkflowIrNode[],
|
||||
outgoing: Map<string, WorkflowIrEdge[]>,
|
||||
): void {
|
||||
/*
|
||||
FNXC:WorkflowValidation 2026-06-27-07:40:
|
||||
FN-7113 requires required top-level workflow nodes to be reachable from start at parse time, including interpreter-deferred branch graphs. Engine-owned recovery entry primitives stay exempt because they can be re-entered by persisted runtime state rather than by the author-facing start path.
|
||||
*/
|
||||
const startNode = nodes.find((node) => node.kind === "start");
|
||||
if (!startNode) return;
|
||||
const reachable = reachableFrom(startNode.id, outgoing);
|
||||
for (const node of nodes) {
|
||||
if (reachable.has(node.id) || INTERPRETER_ENTRY_NODE_KINDS.has(node.kind)) continue;
|
||||
throw new WorkflowIrError(`Workflow node '${node.id}' is not reachable from the start node`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate a foreach `template` subgraph recursively (KTD-3):
|
||||
* - non-empty;
|
||||
@@ -1234,7 +1264,14 @@ function validateRegisteredExtensionMetadata(
|
||||
value: Record<string, unknown>,
|
||||
): void {
|
||||
const definition = getWorkflowExtensionRegistry().get(key);
|
||||
const fields = definition?.extension.configSchema?.fields;
|
||||
/*
|
||||
FNXC:WorkflowValidation 2026-06-27-00:00:
|
||||
FN-7113 requires plugin-referencing workflow graphs to validate against the same central gate as built-in/custom graphs. Reject unknown workflow extension keys by name so authoring surfaces cannot persist a graph whose plugin node/column contract is missing at save or launch time.
|
||||
*/
|
||||
if (!definition) {
|
||||
throw new WorkflowIrError(`${owner} extension key '${key}' is not registered`);
|
||||
}
|
||||
const fields = definition.extension.configSchema?.fields;
|
||||
if (!fields || fields.length === 0) return;
|
||||
for (const field of fields) {
|
||||
if (field.required && !(field.key in value)) {
|
||||
@@ -1314,6 +1351,17 @@ function validateV2(ir: WorkflowIrV2): void {
|
||||
validateColumns(ir);
|
||||
|
||||
const columnIds = new Set(ir.columns.map((c) => c.id));
|
||||
const nodeIds = new Set<string>();
|
||||
for (const node of ir.nodes) {
|
||||
/*
|
||||
FNXC:WorkflowValidation 2026-06-27-00:00:
|
||||
FN-7113 requires top-level duplicate node ids to fail before persistence or launch. Keep this check before nodesById is built so Map de-duplication cannot silently mask a malformed author/plugin workflow graph.
|
||||
*/
|
||||
if (nodeIds.has(node.id)) {
|
||||
throw new WorkflowIrError(`Workflow IR has duplicate node id '${node.id}'`);
|
||||
}
|
||||
nodeIds.add(node.id);
|
||||
}
|
||||
const nodesById = new Map(ir.nodes.map((n) => [n.id, n]));
|
||||
|
||||
for (const node of ir.nodes) {
|
||||
@@ -1394,6 +1442,7 @@ function validateV2(ir: WorkflowIrV2): void {
|
||||
}
|
||||
|
||||
validateNoIllegalCycles(ir.nodes, outgoing);
|
||||
validateRequiredTopLevelReachability(ir.nodes, outgoing);
|
||||
validateForeachDominance(ir.nodes, ir.edges, outgoing);
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { EventEmitter } from "node:events";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import type { Settings, TaskDetail, WorkflowDefinition, WorkflowIr } from "@fusion/core";
|
||||
import { TaskStore } from "@fusion/core";
|
||||
|
||||
import { NotificationService } from "../notification/notification-service.js";
|
||||
import { WorkflowGraphTaskRunner, type WorkflowGraphRunnerStore } from "../workflow-graph-task-runner.js";
|
||||
@@ -229,6 +233,56 @@ describe("WorkflowGraphTaskRunner (CU-U2)", () => {
|
||||
expect(result.reason).toMatch(/workflow-missing/);
|
||||
});
|
||||
|
||||
it("persists a valid workflow through the store and launches it through the graph runner", async () => {
|
||||
const rootDir = mkdtempSync(join(tmpdir(), "fn-7113-workflow-run-"));
|
||||
const globalDir = mkdtempSync(join(tmpdir(), "fn-7113-workflow-global-"));
|
||||
const store = new TaskStore(rootDir, globalDir, { inMemoryDb: true });
|
||||
try {
|
||||
const invalidIr: WorkflowIr = {
|
||||
version: "v2",
|
||||
name: "invalid-save",
|
||||
columns: [{ id: "todo", name: "Todo", traits: [] }],
|
||||
nodes: [
|
||||
{ id: "start", kind: "start", column: "todo" },
|
||||
{ id: "dup", kind: "prompt", column: "todo" },
|
||||
{ id: "dup", kind: "script", column: "todo" },
|
||||
{ id: "end", kind: "end", column: "todo" },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "dup" },
|
||||
{ from: "dup", to: "end" },
|
||||
],
|
||||
};
|
||||
await expect(store.createWorkflowDefinition({ name: "Invalid", ir: invalidIr })).rejects.toThrow(
|
||||
/Workflow IR has duplicate node id 'dup'/,
|
||||
);
|
||||
|
||||
const workflow = await store.createWorkflowDefinition({ name: "Valid", ir: fullLifecycleIr() });
|
||||
const persisted = await store.getWorkflowDefinition(workflow.id);
|
||||
expect(persisted?.id).toBe(workflow.id);
|
||||
const savedTask = await store.createTask({ description: "save run", enabledWorkflowSteps: [] });
|
||||
await store.selectTaskWorkflow(savedTask.id, workflow.id);
|
||||
|
||||
const calls: string[] = [];
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store,
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runner.run(savedTask, flagOn);
|
||||
|
||||
expect(result.disposition).toBe("completed");
|
||||
expect(calls).toEqual(["custom:lint", "execute", "review", "merge", "custom:notify"]);
|
||||
} finally {
|
||||
rmSync(rootDir, { recursive: true, force: true });
|
||||
rmSync(globalDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("resolves built-in workflow selections without requiring the store to return a definition", async () => {
|
||||
const calls: string[] = [];
|
||||
const getWorkflowDefinition = vi.fn(async () => undefined);
|
||||
@@ -252,61 +306,92 @@ describe("WorkflowGraphTaskRunner (CU-U2)", () => {
|
||||
expect(getWorkflowDefinition).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("falls back (never strands the task) when the interpreter throws", async () => {
|
||||
// Malformed graph: edge references unknown node → WorkflowIrError inside run().
|
||||
it("fails closed with invalid-ir before any side-effect seam when resolved IR is malformed", async () => {
|
||||
const badIr: WorkflowIr = {
|
||||
version: "v1",
|
||||
version: "v2",
|
||||
name: "bad",
|
||||
columns: [{ id: "c", name: "C", traits: [] }],
|
||||
nodes: [
|
||||
{ id: "start", kind: "start" },
|
||||
{ id: "end", kind: "end" },
|
||||
{ id: "start", kind: "start", column: "c" },
|
||||
{ id: "dup", kind: "prompt", column: "c" },
|
||||
{ id: "dup", kind: "script", column: "c" },
|
||||
{ id: "end", kind: "end", column: "c" },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "dup" },
|
||||
{ from: "dup", to: "end" },
|
||||
],
|
||||
edges: [{ from: "start", to: "ghost" }],
|
||||
};
|
||||
const calls: string[] = [];
|
||||
const events: string[] = [];
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(badIr)),
|
||||
seams: recordingSeams([]),
|
||||
runCustomNode: async () => ({ outcome: "success" }),
|
||||
onEvent: (e) => events.push(e.type),
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
},
|
||||
onEvent: (e) => events.push(`${e.type}:${e.detail}`),
|
||||
});
|
||||
|
||||
const result = await runner.run(task, flagOn);
|
||||
expect(result.disposition).toBe("fell-back");
|
||||
expect(result.reason).toMatch(/interpreter-error/);
|
||||
expect(events).toContain("fallback");
|
||||
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(result.outcome).toBe("failure");
|
||||
expect(result.reason).toMatch(/invalid-ir: Workflow IR has duplicate node id 'dup'/);
|
||||
expect(result.visitedNodeIds).toEqual([]);
|
||||
expect(calls).toEqual([]);
|
||||
expect(events.some((event) => event.includes("terminal:invalid-ir"))).toBe(true);
|
||||
});
|
||||
|
||||
it("an interpreter error AFTER side effects terminates as failed, not fell-back", async () => {
|
||||
// Cycle reached only after custom nodes execute: re-running legacy would
|
||||
// repeat the implementation, so the runner must not signal fallback.
|
||||
const cyclicIr: WorkflowIr = {
|
||||
version: "v1",
|
||||
name: "cyclic",
|
||||
it("fails closed with invalid-ir before any custom node when resolved IR has a dangling edge", async () => {
|
||||
const badIr: WorkflowIr = {
|
||||
version: "v2",
|
||||
name: "bad-edge",
|
||||
columns: [{ id: "c", name: "C", traits: [] }],
|
||||
nodes: [
|
||||
{ id: "start", kind: "start" },
|
||||
{ id: "a", kind: "prompt", config: { prompt: "a" } },
|
||||
{ id: "b", kind: "prompt", config: { prompt: "b" } },
|
||||
{ id: "end", kind: "end" },
|
||||
{ id: "start", kind: "start", column: "c" },
|
||||
{ id: "a", kind: "prompt", column: "c" },
|
||||
{ id: "end", kind: "end", column: "c" },
|
||||
],
|
||||
edges: [
|
||||
{ from: "start", to: "a", condition: "success" },
|
||||
{ from: "a", to: "b", condition: "success" },
|
||||
{ from: "b", to: "a", condition: "success" },
|
||||
{ from: "start", to: "a" },
|
||||
{ from: "a", to: "ghost" },
|
||||
],
|
||||
};
|
||||
const calls: string[] = [];
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(cyclicIr)),
|
||||
store: storeWith(definition(badIr)),
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async (node) => {
|
||||
calls.push(`custom:${node.id}`);
|
||||
return { outcome: "success" };
|
||||
},
|
||||
});
|
||||
|
||||
const result = await runner.run(task, flagOn);
|
||||
expect(calls.length).toBeGreaterThan(0);
|
||||
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(result.reason).toMatch(/interpreter-error/);
|
||||
expect(result.outcome).toBe("failure");
|
||||
expect(result.reason).toMatch(/invalid-ir: Workflow edge 'a' -> 'ghost' references undefined node 'ghost'/);
|
||||
expect(result.visitedNodeIds).toEqual([]);
|
||||
expect(calls).toEqual([]);
|
||||
});
|
||||
|
||||
it("a custom-node failure AFTER side effects terminates as failed, not fell-back", async () => {
|
||||
const calls: string[] = [];
|
||||
const runner = new WorkflowGraphTaskRunner({
|
||||
store: storeWith(definition(fullLifecycleIr())),
|
||||
seams: recordingSeams(calls),
|
||||
runCustomNode: async (node) => {
|
||||
throw new Error(`custom boom: ${node.id}`);
|
||||
},
|
||||
});
|
||||
const result = await runner.run(task, flagOn);
|
||||
expect(calls).toEqual([]);
|
||||
expect(result.visitedNodeIds).toEqual(["start", "lint"]);
|
||||
expect(result.disposition).toBe("failed");
|
||||
expect(result.reason).toBeUndefined();
|
||||
});
|
||||
|
||||
it("exposes node outcomes in the shared context for downstream consumers", async () => {
|
||||
|
||||
@@ -1,5 +1,11 @@
|
||||
import type { Settings, TaskDetail, WorkflowDefinition, WorkflowStepResult } from "@fusion/core";
|
||||
import { getBuiltinWorkflow, isBuiltinWorkflowId } from "@fusion/core";
|
||||
import type { Settings, TaskDetail, WorkflowDefinition, WorkflowIr, WorkflowStepResult } from "@fusion/core";
|
||||
import {
|
||||
compileWorkflowToSteps,
|
||||
getBuiltinWorkflow,
|
||||
isBuiltinWorkflowId,
|
||||
parseWorkflowIr,
|
||||
WorkflowCompileError,
|
||||
} from "@fusion/core";
|
||||
|
||||
import { WorkflowGraphExecutor, type WorkflowGraphExecutorDeps, type WorkflowNodeOutcome, type WorkflowTaskProjection } from "./workflow-graph-executor.js";
|
||||
import type {
|
||||
@@ -29,11 +35,15 @@ import type { WorkflowPrimitiveContext, WorkflowRuntimePrimitives } from "./runt
|
||||
*/
|
||||
export type WorkflowGraphRunDisposition = "completed" | "failed" | "fell-back";
|
||||
|
||||
function isInterpreterDeferredCompileError(error: unknown): boolean {
|
||||
return error instanceof WorkflowCompileError && error.message.includes("require the workflow interpreter (deferred)");
|
||||
}
|
||||
|
||||
export interface WorkflowGraphTaskRunResult {
|
||||
disposition: WorkflowGraphRunDisposition;
|
||||
outcome?: WorkflowNodeOutcome;
|
||||
visitedNodeIds: string[];
|
||||
/** Why the runner fell back (flag-off, no-selection, workflow-missing, interpreter-error). */
|
||||
/** Why the runner fell back or failed before execution (flag-off, no-selection, workflow-missing, invalid-ir, interpreter-error). */
|
||||
reason?: string;
|
||||
/** Shared graph context after the run (node outcomes/values). */
|
||||
context?: Record<string, unknown>;
|
||||
@@ -141,6 +151,11 @@ export class WorkflowGraphTaskRunner {
|
||||
return { disposition: "fell-back", reason, visitedNodeIds: [] };
|
||||
}
|
||||
|
||||
private failBeforeSideEffects(taskId: string, reason: string): WorkflowGraphTaskRunResult {
|
||||
this.emit("terminal", taskId, reason);
|
||||
return { disposition: "failed", outcome: "failure", reason, visitedNodeIds: [] };
|
||||
}
|
||||
|
||||
public async run(
|
||||
task: TaskDetail,
|
||||
settings: Pick<Settings, "experimentalFeatures"> | undefined,
|
||||
@@ -167,13 +182,28 @@ export class WorkflowGraphTaskRunner {
|
||||
return this.fallBack(task.id, `workflow-missing: ${selection.workflowId}`);
|
||||
}
|
||||
|
||||
let validatedIr: WorkflowIr;
|
||||
try {
|
||||
/*
|
||||
FNXC:WorkflowExecution 2026-06-27-07:40:
|
||||
FN-7113 requires the interpreter to re-validate the resolved built-in/custom/plugin workflow IR before any seam, primitive, or custom-node side effects. Invalid persisted or plugin-authored graphs fail closed with an author-facing invalid-ir reason instead of partially running or falling back into the wrong legacy workflow.
|
||||
*/
|
||||
validatedIr = parseWorkflowIr(definition.ir);
|
||||
try {
|
||||
compileWorkflowToSteps(validatedIr);
|
||||
} catch (err) {
|
||||
if (!isInterpreterDeferredCompileError(err)) throw err;
|
||||
}
|
||||
} catch (err) {
|
||||
return this.failBeforeSideEffects(task.id, `invalid-ir: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
|
||||
this.emit("start", task.id, definition.id);
|
||||
this.branchProgress.clear();
|
||||
|
||||
// Track whether any node side effects ran. A pre-run interpreter error
|
||||
// (bad IR structure, wiring) can safely fall back to the legacy pipeline;
|
||||
// a mid-run error cannot — re-running legacy would repeat the implementation
|
||||
// session — so it terminates as "failed" for the caller to park instead.
|
||||
// Track whether any node side effects ran. Invalid IR is rejected above as
|
||||
// failed/terminal before this point; once execution starts, no error can
|
||||
// safely fall back to legacy because that could repeat a partial workflow run.
|
||||
let sideEffectsRan = false;
|
||||
const invoked: string[] = [];
|
||||
const seams = this.deps.seams;
|
||||
@@ -254,7 +284,7 @@ export class WorkflowGraphTaskRunner {
|
||||
}
|
||||
},
|
||||
});
|
||||
const result = await executor.run(task, settings, definition.ir);
|
||||
const result = await executor.run(task, settings, validatedIr);
|
||||
if (!result.executed) {
|
||||
return this.fallBack(task.id, "not-executed");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user