From 38d7f1c436534b18f39d1b81a940986712c9aec9 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 09:00:46 -0700 Subject: [PATCH] FN-6087: skip completed foreach steps on resume Prevent shared-isolation foreach resumes from re-running terminal task steps. - skip foreach step instances whose persisted status is already done or skipped before launching step-execute - log resume skips so restart behavior is auditable during shared-isolation foreach runs - add regression coverage for all-done resumes, mixed done/skipped resumes, and in-progress step replays Files changed: .../src/__tests__/workflow-graph-foreach.test.ts | 85 ++++++++++++++++++++++ packages/engine/src/workflow-graph-foreach.ts | 13 ++++ 2 files changed, 98 insertions(+) Fusion-Task-Id: FN-6087 Fusion-Task-Lineage: 56805a5c-a1ea-41bc-a758-1afed83c6bb7 --- .../__tests__/workflow-graph-foreach.test.ts | 85 +++++++++++++++++++ packages/engine/src/workflow-graph-foreach.ts | 13 +++ 2 files changed, 98 insertions(+) diff --git a/packages/engine/src/__tests__/workflow-graph-foreach.test.ts b/packages/engine/src/__tests__/workflow-graph-foreach.test.ts index 86595ec1d4..bbbaf6cd84 100644 --- a/packages/engine/src/__tests__/workflow-graph-foreach.test.ts +++ b/packages/engine/src/__tests__/workflow-graph-foreach.test.ts @@ -20,6 +20,15 @@ function taskWithSteps(n: number): TaskDetail { return { id: "FN-FOREACH", steps } as unknown as TaskDetail; } +/** Build a TaskDetail with explicit persisted step statuses. */ +function taskWithStepStatuses(statuses: TaskStep["status"][]): TaskDetail { + const steps: TaskStep[] = statuses.map((status, i) => ({ + name: `Step ${i + 1}`, + status, + })); + return { id: "FN-FOREACH", steps } as unknown as TaskDetail; +} + /** * Build a graph: start → foreach → end. The foreach template is provided inline. * Extra edges from the foreach node (e.g. outcome:rework-exhausted) are appended. @@ -98,6 +107,82 @@ describe("WorkflowGraphExecutor foreach (U3)", () => { expect(result.visitedNodeIds.some((id) => id.startsWith("fe#"))).toBe(false); }); + it("resume skips all instances when every step is already done", async () => { + const exec = vi.fn(async () => ({ outcome: "success" as const, value: "step-done" })); + const seams = baseSeams({ stepExecute: exec }); + const executor = new WorkflowGraphExecutor({ seams }); + const result = await executor.run( + taskWithStepStatuses(["done", "done", "done"]), + settingsOn(), + foreachIr(singleExecuteTemplate()), + ); + + expect(result.outcome).toBe("success"); + expect(exec).not.toHaveBeenCalled(); + expect(result.visitedNodeIds).toContain("fe"); + expect(result.visitedNodeIds.some((id) => id.startsWith("fe#"))).toBe(false); + }); + + it("resume skips done instances and runs the first pending step", async () => { + const executedStepIndexes: number[] = []; + const seams = baseSeams({ + stepExecute: async (_t, ctx) => { + const active = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext; + executedStepIndexes.push(active.stepIndex); + return { outcome: "success", value: "step-done" }; + }, + }); + const executor = new WorkflowGraphExecutor({ seams }); + const result = await executor.run( + taskWithStepStatuses(["done", "done", "done", "pending"]), + settingsOn(), + foreachIr(singleExecuteTemplate()), + ); + + expect(result.outcome).toBe("success"); + expect(executedStepIndexes).toEqual([3]); + }); + + it("resume skips mixed done and skipped instances and runs pending steps only", async () => { + const executedStepIndexes: number[] = []; + const seams = baseSeams({ + stepExecute: async (_t, ctx) => { + const active = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext; + executedStepIndexes.push(active.stepIndex); + return { outcome: "success", value: "step-done" }; + }, + }); + const executor = new WorkflowGraphExecutor({ seams }); + const result = await executor.run( + taskWithStepStatuses(["done", "skipped", "done", "pending"]), + settingsOn(), + foreachIr(singleExecuteTemplate()), + ); + + expect(result.outcome).toBe("success"); + expect(executedStepIndexes).toEqual([3]); + }); + + it("resume re-runs in-progress steps instead of treating them as terminal", async () => { + const executedStepIndexes: number[] = []; + const seams = baseSeams({ + stepExecute: async (_t, ctx) => { + const active = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext; + executedStepIndexes.push(active.stepIndex); + return { outcome: "success", value: "step-done" }; + }, + }); + const executor = new WorkflowGraphExecutor({ seams }); + const result = await executor.run( + taskWithStepStatuses(["done", "done", "in-progress", "pending"]), + settingsOn(), + foreachIr(singleExecuteTemplate()), + ); + + expect(result.outcome).toBe("success"); + expect(executedStepIndexes).toEqual([2, 3]); + }); + it("revise-style rework loops twice then completes (custom node routes a rework edge)", async () => { // Template: exec → review. review routes a rework edge back to exec for the // first 2 passes, then approves (success edge → exit). diff --git a/packages/engine/src/workflow-graph-foreach.ts b/packages/engine/src/workflow-graph-foreach.ts index 25a93e91ba..a66387f09a 100644 --- a/packages/engine/src/workflow-graph-foreach.ts +++ b/packages/engine/src/workflow-graph-foreach.ts @@ -413,6 +413,19 @@ export async function runForeach( return { outcome: "failure", value: "aborted", visitedNodeIds }; } + /** + * Engine restart resume semantics: shared-isolation foreach replays the + * graph from the foreach node, but task steps already persisted as terminal + * must not re-run their step-execute instance handlers. + */ + const stepStatus = env.steps[stepIndex]?.status; + if (stepStatus === "done" || stepStatus === "skipped") { + schedulerLog.log( + `foreach ${foreachNode.id} for task ${env.task.id}: skipping step ${stepIndex} — already ${stepStatus}`, + ); + continue; + } + const instanceResult = await runInstance( foreachNode, stepIndex,