From ce8402e37b31251b14edd6dfffcce9f1bfe68654 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 7 Jun 2026 17:59:24 -0700 Subject: [PATCH] fix(FN-6021): align workflow runtime fallback run ids --- .../__tests__/workflow-task-runtime.test.ts | 41 ++++-- packages/engine/src/workflow-task-runtime.ts | 125 +++++++++--------- 2 files changed, 98 insertions(+), 68 deletions(-) diff --git a/packages/engine/src/__tests__/workflow-task-runtime.test.ts b/packages/engine/src/__tests__/workflow-task-runtime.test.ts index f23e07852a..5c73270f2b 100644 --- a/packages/engine/src/__tests__/workflow-task-runtime.test.ts +++ b/packages/engine/src/__tests__/workflow-task-runtime.test.ts @@ -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 () => { diff --git a/packages/engine/src/workflow-task-runtime.ts b/packages/engine/src/workflow-task-runtime.ts index 5dd95dd77a..8ca46f8aac 100644 --- a/packages/engine/src/workflow-task-runtime.ts +++ b/packages/engine/src/workflow-task-runtime.ts @@ -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 { 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>; 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 { + 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> { + 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> = {}; + 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 & Partial) | undefined, ): Pick & Partial {