root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(FN-6021): align workflow runtime fallback run ids

Browse Source
This commit is contained in:
gsxdsm
2026-06-07 17:59:24 -07:00
parent cee81e66ac
commit ce8402e37b
2 changed files with 98 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
packages/engine/src/__tests__/workflow-task-runtime.test.ts
View File

@@ -56,9 +56,13 @@ describe("WorkflowTaskRuntime", () => {
it("runs a selected workflow through the graph engine", async () => {
const calls: string[] = [];
let workflowSelectionReads = 0;
const runtime = new WorkflowTaskRuntime({
store: {
getTaskWorkflowSelection: () => ({ workflowId: "WF-001", stepIds: [] }),
getTaskWorkflowSelection: () => {
workflowSelectionReads += 1;
return { workflowId: "WF-001", stepIds: [] };
},
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
},
seams: recordingSeams(calls),
@@ -73,6 +77,7 @@ describe("WorkflowTaskRuntime", () => {
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["custom:prepare", "execute"]);
expect(result.visitedNodeIds).toEqual(["start", "prepare", "execute"]);
expect(workflowSelectionReads).toBe(1);
});
it("resolves an unselected task to the built-in coding workflow instead of falling back", async () => {
@@ -96,8 +101,9 @@ describe("WorkflowTaskRuntime", () => {
expect(result.visitedNodeIds).toEqual(["start", "execute", "review", "merge"]);
});
it("turns selected workflow lookup failures into the built-in workflow via the shared resolver", async () => {
it("turns selected workflow lookup failures into the built-in workflow target", async () => {
const calls: string[] = [];
const observedRunIds: string[] = [];
const runtime = new WorkflowTaskRuntime({
store: {
getTaskWorkflowSelection: () => ({ workflowId: "WF-MISSING", stepIds: [] }),
@@ -105,16 +111,25 @@ describe("WorkflowTaskRuntime", () => {
},
seams: recordingSeams(calls),
runCustomNode: async () => ({ outcome: "success" }),
branchPersistence: {
loadBranchStates: (_taskId, runId) => {
observedRunIds.push(runId);
return [];
},
},
});
const result = await runtime.run(task, flagOff);
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["execute", "review", "merge"]);
expect(observedRunIds).toContain("FN-9002:builtin:coding");
expect(observedRunIds).not.toContain("FN-9002:WF-MISSING");
});
it("turns corrupt selected workflow definitions into the built-in workflow via the shared resolver", async () => {
it("turns corrupt selected workflow definitions into the built-in workflow target", async () => {
const calls: string[] = [];
const observedRunIds: string[] = [];
const runtime = new WorkflowTaskRuntime({
store: {
getTaskWorkflowSelection: () => ({ workflowId: "WF-CORRUPT", stepIds: [] }),
@@ -122,12 +137,20 @@ describe("WorkflowTaskRuntime", () => {
},
seams: recordingSeams(calls),
runCustomNode: async () => ({ outcome: "success" }),
branchPersistence: {
loadBranchStates: (_taskId, runId) => {
observedRunIds.push(runId);
return [];
},
},
});
const result = await runtime.run(task, flagOff);
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["execute", "review", "merge"]);
expect(observedRunIds).toContain("FN-9002:builtin:coding");
expect(observedRunIds).not.toContain("FN-9002:WF-CORRUPT");
});
it("forces only the graph executor flag while preserving other settings", async () => {
@@ -249,19 +272,19 @@ describe("WorkflowTaskRuntime", () => {
expect(result.reason).toMatch(/workflow-execution-error/);
});
it("preserves invoked node ids when the graph throws after side effects", async () => {
it("preserves graph node ids when the graph throws after seam and custom side effects", async () => {
const cyclicIr: WorkflowIr = {
version: "v1",
name: "cyclic",
nodes: [
{ id: "start", kind: "start" },
{ id: "prepare", kind: "prompt", config: { prompt: "prepare" } },
{ id: "do-execute", kind: "prompt", config: { seam: "execute" } },
{ id: "loop", kind: "prompt", config: { prompt: "loop" } },
],
edges: [
{ from: "start", to: "prepare", condition: "success" },
{ from: "prepare", to: "loop", condition: "success" },
{ from: "loop", to: "prepare", condition: "success" },
{ from: "start", to: "do-execute", condition: "success" },
{ from: "do-execute", to: "loop", condition: "success" },
{ from: "loop", to: "do-execute", condition: "success" },
],
};
const runtime = new WorkflowTaskRuntime({
@@ -277,7 +300,7 @@ describe("WorkflowTaskRuntime", () => {
expect(result.disposition).toBe("failed");
expect(result.reason).toMatch(/workflow-execution-error/);
expect(result.visitedNodeIds).toEqual(["prepare", "loop"]);
expect(result.visitedNodeIds).toEqual(["do-execute", "loop"]);
});
it("diagnostic event failures do not affect execution", async () => {

125
packages/engine/src/workflow-task-runtime.ts
View File

@@ -1,12 +1,23 @@
import type { Settings, TaskDetail, WorkflowIr } from "@fusion/core";
import { resolveWorkflowIrForTask, type WorkflowIrResolverStore } from "@fusion/core";
import type { Settings, TaskDetail, WorkflowIr, WorkflowIrNode } from "@fusion/core";
import {
BUILTIN_CODING_WORKFLOW_IR,
getBuiltinWorkflow,
isBuiltinWorkflowId,
parseWorkflowIr,
type WorkflowIrResolverStore,
} from "@fusion/core";
import {
WorkflowGraphExecutor,
type WorkflowGraphExecutorDeps,
type WorkflowNodeHandler,
type WorkflowNodeOutcome,
} from "./workflow-graph-executor.js";
import type { WorkflowCustomNodeRunner, WorkflowLegacySeams } from "./workflow-node-handlers.js";
import {
createDefaultNodeHandlers,
type WorkflowCustomNodeRunner,
type WorkflowLegacySeams,
} from "./workflow-node-handlers.js";
export type WorkflowTaskRuntimeDisposition = "completed" | "failed";
@@ -30,7 +41,7 @@ export interface WorkflowTaskRuntimeDeps extends Omit<WorkflowGraphExecutorDeps,
*
* It always resolves a task to a workflow IR: explicit selections resolve to
* their selected workflow, and tasks without a selection resolve to the built-in
* coding workflow through `resolveWorkflowIrForTask`. This is intentionally
* coding workflow. This is intentionally
* different from `WorkflowGraphTaskRunner`, whose current contract still models
* "no selection" as legacy fallback.
*/
@@ -51,10 +62,9 @@ export class WorkflowTaskRuntime {
): Promise<WorkflowTaskRuntimeResult> {
this.emit("start", task.id, "resolve-workflow");
const workflowId = this.resolveWorkflowId(task.id);
let ir: WorkflowIr;
let target: WorkflowRuntimeTarget;
try {
ir = await resolveWorkflowIrForTask(this.deps.store, task.id);
target = await this.resolveRuntimeTarget(task.id);
} catch (err) {
const reason = `workflow-resolution-error: ${err instanceof Error ? err.message : String(err)}`;
this.emit("terminal", task.id, `failed:${reason}`);
@@ -68,25 +78,19 @@ export class WorkflowTaskRuntime {
}
const invoked: string[] = [];
const wrappedSeams = this.wrapSeams(invoked);
const wrappedRunCustomNode: WorkflowCustomNodeRunner = (node, nodeTask, context) => {
invoked.push(node.id);
return this.deps.runCustomNode(node, nodeTask, context);
};
const executor = new WorkflowGraphExecutor({
...this.deps,
seams: wrappedSeams,
runCustomNode: wrappedRunCustomNode,
handlers: this.recordingHandlers(invoked),
// WorkflowTaskRuntime is the execution engine, so internally the graph
// executor is authoritative even before the old feature flag plumbing is
// deleted from legacy entry points.
runId: this.deps.runId ?? `${task.id}:${workflowId}`,
runId: this.deps.runId ?? `${task.id}:${target.workflowId}`,
});
const runtimeSettings = forceWorkflowGraphExecutor(settings);
let result: Awaited<ReturnType<WorkflowGraphExecutor["run"]>>;
try {
result = await executor.run(task, runtimeSettings, ir);
result = await executor.run(task, runtimeSettings, target.ir);
} catch (err) {
const reason = `workflow-execution-error: ${err instanceof Error ? err.message : String(err)}`;
this.emit("terminal", task.id, `failed:${reason}`);
@@ -108,57 +112,60 @@ export class WorkflowTaskRuntime {
};
}
private resolveWorkflowId(taskId: string): string {
private async resolveRuntimeTarget(taskId: string): Promise<WorkflowRuntimeTarget> {
let workflowId: string | undefined;
try {
return this.deps.store.getTaskWorkflowSelection(taskId)?.workflowId ?? "builtin:coding";
workflowId = this.deps.store.getTaskWorkflowSelection(taskId)?.workflowId;
} catch {
return "builtin:coding";
return builtinCodingTarget();
}
if (!workflowId) return builtinCodingTarget();
if (isBuiltinWorkflowId(workflowId)) {
const builtin = getBuiltinWorkflow(workflowId);
if (!builtin) return builtinCodingTarget();
const ir = typeof builtin.ir === "string" ? parseWorkflowIr(builtin.ir) : builtin.ir;
return { workflowId, ir };
}
try {
const def = await this.deps.store.getWorkflowDefinition(workflowId);
if (!def) return builtinCodingTarget();
const ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir;
return { workflowId, ir };
} catch {
return builtinCodingTarget();
}
}
private wrapSeams(invoked: string[]): WorkflowLegacySeams {
const seams = this.deps.seams;
return {
planning: (task, context) => {
invoked.push("planning");
return seams.planning(task, context);
},
execute: (task, context) => {
invoked.push("execute");
return seams.execute(task, context);
},
review: (task, context) => {
invoked.push("review");
return seams.review(task, context);
},
merge: (task, context) => {
invoked.push("merge");
return seams.merge(task, context);
},
schedule: (task, context) => {
invoked.push("schedule");
return seams.schedule(task, context);
},
...(seams.stepExecute
? {
stepExecute: (task, context) => {
invoked.push("step-execute");
return seams.stepExecute!(task, context);
},
}
: {}),
...(seams.stepReview
? {
stepReview: (task, context, config) => {
invoked.push("step-review");
return seams.stepReview!(task, context, config);
},
}
: {}),
};
private recordingHandlers(invoked: string[]): Partial<Record<WorkflowIrNode["kind"], WorkflowNodeHandler>> {
const defaultHandlers = createDefaultNodeHandlers(this.deps.seams, this.deps.runCustomNode, {
parseSteps: this.deps.parseStepsDeps,
runCode: this.deps.runCode,
prNodes: this.deps.prNodes,
});
const handlers = { ...defaultHandlers, ...(this.deps.handlers ?? {}) };
const wrapped: Partial<Record<WorkflowIrNode["kind"], WorkflowNodeHandler>> = {};
for (const [kind, handler] of Object.entries(handlers) as Array<[WorkflowIrNode["kind"], WorkflowNodeHandler]>) {
wrapped[kind] = async (node, context) => {
invoked.push(node.id);
return handler(node, context);
};
}
return wrapped;
}
}
interface WorkflowRuntimeTarget {
workflowId: string;
ir: WorkflowIr;
}
function builtinCodingTarget(): WorkflowRuntimeTarget {
return { workflowId: "builtin:coding", ir: BUILTIN_CODING_WORKFLOW_IR };
}
function forceWorkflowGraphExecutor(
settings: (Pick<Settings, "experimentalFeatures"> & Partial<Settings>) | undefined,
): Pick<Settings, "experimentalFeatures"> & Partial<Settings> {