fix(FN-7224): make workflow graph replay idempotent

Move completed-step replay and proven-merge finalization into workflow-owned execution paths so graph restarts continue from live task projection instead of failing stale step nodes.

Fusion-Task-Id: FN-7224

.changeset/fn-7228-parse-resume.md

@@ -0,0 +1,7 @@
+---
+"@runfusion/fusion": patch
+---
+
+summary: Recover workflow retries that restart after step execution has already begun.
+category: fix
+dev: Treat persisted foreach step pins as a parse resume signal instead of a graph failure.

packages/engine/src/__tests__/ce-workflow-step-executor.test.ts

@@ -204,6 +204,97 @@ describe("CE workflow-step executor integration", () => {
       expect(captured.step.prompt).toContain("Plan the work.");
     });
 
+    it("acquires a task worktree when the first CE coding-mode node runs before execute", async () => {
+      const store = createMockStore();
+      let live = baseStepTask({
+        worktree: undefined,
+        branch: undefined,
+        steps: [{ name: "Preflight", status: "pending" }],
+      });
+      store.getTask.mockImplementation(async () => live as any);
+      store.updateTask.mockImplementation(async (_id: string, patch: Record<string, unknown>) => {
+        live = { ...live, ...patch };
+        return live as any;
+      });
+      const { executor } = makeExecutor(store);
+      vi.spyOn(executor as any, "createWorktree").mockResolvedValue({
+        path: "/tmp/test/.worktrees/swift-falcon",
+        branch: "fusion/fn-ce-1",
+      });
+      vi.spyOn(executor as any, "captureBaseCommitSha").mockResolvedValue(undefined);
+
+      const captured: { step?: any; worktreePath?: string } = {};
+      vi.spyOn(executor as any, "executeWorkflowStep").mockImplementation(async (...args: any[]) => {
+        captured.step = args[1];
+        captured.worktreePath = args[2];
+        return { success: true, output: "ok" };
+      });
+
+      const node = {
+        id: "plan",
+        kind: "prompt",
+        column: "in-progress",
+        config: {
+          executor: "skill",
+          skillName: "compound-engineering:ce-plan",
+          toolMode: "coding",
+          prompt: "Run /ce-plan.",
+        },
+      };
+
+      const result = await (executor as any).runGraphCustomNode(node, { id: "FN-CE-1" }, await store.getSettings(), undefined);
+
+      expect(result.outcome).toBe("success");
+      expect((executor as any).createWorktree).toHaveBeenCalled();
+      expect(captured.worktreePath).toBe("/tmp/test/.worktrees/swift-falcon");
+      expect(captured.step.toolMode).toBe("coding");
+      expect(live.worktree).toBe("/tmp/test/.worktrees/swift-falcon");
+    });
+
+    it("finalizes a merge-confirmed workflow graph task that is stranded before done", async () => {
+      const store = createMockStore();
+      let live = baseStepTask({
+        column: "in-progress",
+        status: null,
+        error: null,
+        mergeDetails: { mergeConfirmed: true, commitSha: "abc123" },
+        steps: [{ name: "Preflight", status: "done" }],
+      });
+      store.getTask.mockImplementation(async () => live as any);
+      store.updateTask.mockImplementation(async (_id: string, patch: Record<string, unknown>) => {
+        live = { ...live, ...patch };
+        return live as any;
+      });
+      store.moveTask.mockImplementation(async (_id: string, column: string) => {
+        live = { ...live, column };
+        return live as any;
+      });
+      const { executor } = makeExecutor(store);
+
+      const handled = await (executor as any).finalizeMergeConfirmedWorkflowGraphTask("FN-CE-1", "test");
+
+      expect(handled).toBe(true);
+      expect(store.moveTask).toHaveBeenCalledWith("FN-CE-1", "done", expect.objectContaining({
+        recoveryRehome: true,
+        preserveProgress: true,
+      }));
+      expect(live.column).toBe("done");
+      expect(live.mergeDetails?.mergeConfirmed).toBe(true);
+    });
+
+    it("treats terminal graph step projection as success when the legacy pass rejects", async () => {
+      const store = createMockStore();
+      store.getTask.mockResolvedValue(baseStepTask({
+        steps: [{ name: "Preflight", status: "done" }],
+      }) as any);
+      const { executor } = makeExecutor(store);
+      vi.spyOn(executor as any, "runImplementationPhase").mockRejectedValue(new Error("Agent finished without calling fn_task_done"));
+
+      const result = await (executor as any).runGraphTaskStep(baseStepTask(), 0, "steps#0", "steps#0:step-execute");
+
+      expect(result).toEqual({ success: true });
+    });
+
     it("a non-skill (model) node synthesizes NO skillName and NO preamble", async () => {
       const store = createMockStore();
       store.getTask.mockResolvedValue(baseStepTask() as any);

packages/engine/src/__tests__/workflow-graph-foreach.test.ts

@@ -143,6 +143,26 @@ describe("WorkflowGraphExecutor foreach (U3)", () => {
     expect(executedStepIndexes).toEqual([3]);
   });
 
+  it("resume consults live steps before running a stale in-progress instance", async () => {
+    const exec = vi.fn(async () => ({ outcome: "failure" as const, value: "should-not-run" }));
+    const seams = baseSeams({ stepExecute: exec });
+    const executor = new WorkflowGraphExecutor({
+      seams,
+      getTaskSteps: async () => [
+        { name: "Step 1", status: "done" },
+        { name: "Step 2", status: "done" },
+      ] as TaskStep[],
+    });
+    const result = await executor.run(
+      taskWithStepStatuses(["done", "in-progress"]),
+      settingsOn(),
+      foreachIr(singleExecuteTemplate()),
+    );
+
+    expect(result.outcome).toBe("success");
+    expect(exec).not.toHaveBeenCalled();
+  });
+
   it("resume skips mixed done and skipped instances and runs pending steps only", async () => {
     const executedStepIndexes: number[] = [];
     const seams = baseSeams({

packages/engine/src/__tests__/workflow-parse-steps.test.ts

@@ -145,17 +145,16 @@ describe("parse-steps node handler (U12, KTD-12)", () => {
     expect(written).toEqual([[]]);
   });
 
-  it("pin protection: parse after a foreach expanded → pin-mismatch failure, no write", async () => {
+  it("pin protection: parse after a foreach expanded resumes without rewriting steps", async () => {
     const { deps, written, audits } = makeDeps({
       hasExpandedForeach: async () => true,
     });
-    const ir = parseIr("step-headings", undefined, [
-      { from: "parse", to: "end", condition: "outcome:pin-mismatch" },
-    ]);
+    const ir = parseIr("step-headings");
     const result = await runParse(ir, deps);
     expect(written).toHaveLength(0);
-    expect(result.context["node:parse:value"]).toBe("pin-mismatch");
-    expect(audits.some((a) => a.reason === "pin-mismatch")).toBe(true);
+    expect(result.outcome).toBe("success");
+    expect(result.context["node:parse:value"]).toBe("already-expanded");
+    expect(audits.some((a) => a.reason === "pin-resume")).toBe(true);
   });
 
   it("default workflow parity: registry step-headings == direct parseStepHeadings call", async () => {

packages/engine/src/auto-merge-finalization.ts

@@ -20,7 +20,7 @@ export interface FinalizeProvenAutoMergeTaskOptions {
   audit?: RunAuditor;
   auditAgentId?: string;
   auditPhase?: string;
-  source: "direct-ai-merge" | "merge-confirmed-fast-path" | "self-healing";
+  source: "direct-ai-merge" | "merge-confirmed-fast-path" | "self-healing" | "workflow-graph-merge-finalize";
   log?: (message: string) => void | Promise<void>;
 }
 
@@ -123,7 +123,11 @@ export async function finalizeProvenAutoMergeTask({
 
   const hardBlocker = getTaskHardMergeBlocker({
     ...latest,
-    column: latest.column === "todo" ? "in-review" : latest.column,
+    /*
+    FNXC:WorkflowMerge 2026-06-29-09:15:
+    Proven merge finalization is a recovery path: durable `mergeConfirmed` means the branch already landed, even if a workflow graph crash left the card in `in-progress` or `todo`. Evaluate hard blockers as review-eligible so the column mismatch itself does not block the recovery rehome to `done`; real blockers such as paused/error/incomplete steps still apply.
+    */
+    column: "in-review",
     paused: false,
     status: latest.status === "merging" || latest.status === "merging-pr" || latest.status === "queued" ? undefined : latest.status,
     error: undefined,

packages/engine/src/executor.ts

@@ -12,6 +12,7 @@ import { readFile, rm, writeFile } from "node:fs/promises";
 import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode, WorkflowIrNodeKind, WorkflowStepResult as CoreWorkflowStepResult } from "@fusion/core";
 import { getUnmetSchedulingDependencies } from "./scheduler.js";
 import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId, getWorkflowExtensionRegistry, getBuiltinWorkflow, parseNoOpCompletionMarker, allowsAutoMergeProcessing, isSharedBranchGroupMemberIntegration, resolveMaxAutoMergeRetries, resolveOptionalStepRevisionBudget } from "@fusion/core";
+import { finalizeProvenAutoMergeTask } from "./auto-merge-finalization.js";
 import { mergeEffectiveSettings } from "./effective-settings.js";
 import type { TaskStep, WorkflowIr, WorkflowFieldDefinition, WorkflowColumnAgent, EffectiveAgentInput, WorkflowWorkEngineDispatchResult } from "@fusion/core";
 import {
@@ -4558,6 +4559,7 @@ export class TaskExecutor {
           getWorkflowDefinition: async (id: string) =>
             (await this.store.getWorkflowDefinition?.(id))
               ?? (id === "builtin:coding" ? getBuiltinWorkflow("builtin:coding") : undefined),
+          getTask: (taskId: string) => this.store.getTask(taskId),
         },
         runId: resolvedRunId,
         primitives: this.createAuthoritativeWorkflowPrimitives(settings),
@@ -4683,6 +4685,9 @@ export class TaskExecutor {
         await this.handleGraphFailure(task, result);
       } else if (result.disposition === "completed") {
         const live = await this.store.getTask(task.id).catch(() => task);
+        if ((live as TaskDetail).mergeDetails?.mergeConfirmed === true && (live as TaskDetail).column !== "done") {
+          await this.finalizeMergeConfirmedWorkflowGraphTask(task.id, "graph-completed");
+        }
         if ((live.graphResumeRetryCount ?? 0) !== 0) {
           await this.store.updateTask(task.id, { graphResumeRetryCount: 0 }, this.getRunContextFor(task.id));
         }
@@ -5452,6 +5457,22 @@ export class TaskExecutor {
       if (this.graphStepRunOnce.get(task.id) === phase) {
         this.graphStepRunOnce.delete(task.id);
       }
+      /*
+      FNXC:WorkflowExecution 2026-06-29-09:01:
+      Stepwise graph execution is projection-driven: a shared implementation pass can complete every task step and pass deterministic verification without using the legacy monolithic `task_done` sentinel. If the target step is already terminal in Task.steps[], the workflow node succeeds and the graph continues to its review/merge nodes instead of converting stale legacy completion failure into `steps#N:step-execute`.
+      */
+      try {
+        const live = await this.store.getTask(task.id);
+        const status = live.steps[stepIndex]?.status;
+        if (status === "done" || status === "skipped") {
+          executorLog.warn(
+            `${task.id}: graph step ${stepIndex} completed in projection despite implementation-pass error; continuing workflow (${err instanceof Error ? err.message : String(err)})`,
+          );
+          return { success: true };
+        }
+      } catch {
+        // Fall through to the original failure value below.
+      }
       return { success: false, error: err instanceof Error ? err.message : String(err) };
     }
 
@@ -5591,6 +5612,18 @@ export class TaskExecutor {
           return { outcome: "failure" };
         }
         const live = await this.store.getTask(task.id);
+        /*
+        FNXC:WorkflowResume 2026-06-29-08:53:
+        `step-execute` is a workflow node and must be idempotent on replay. If the live projection already says this foreach instance is terminal, return success before invoking the step runner so retries/restarts cannot fail a fully completed task on a stale step snapshot.
+        */
+        const liveStatus = live.steps[stepIndex]?.status;
+        if (liveStatus === "done" || liveStatus === "skipped") {
+          return {
+            outcome: "success",
+            value: "step-already-terminal",
+            data: { status: liveStatus },
+          };
+        }
         const worktreePath = active.worktreePath || live.worktree || this.rootDir;
         this.graphStepActiveContext.set(this.graphActiveContextKey(task.id, active.instanceId), active);
         const stepGoverningNodeId = context[SEAM_GOVERNING_NODE_CONTEXT_KEY];
@@ -5721,6 +5754,33 @@ export class TaskExecutor {
             return { outcome: "failure", value: "merge-timeout", data: { status: "timeout" } };
           }
           if (result.merged || result.noOp) {
+            /*
+            FNXC:WorkflowMerge 2026-06-29-09:24:
+            The workflow merge primitive owns the normal lifecycle transition after a graph merge node succeeds. Finalize the proven landed task here so `mergeConfirmed` cannot strand a card in `in-progress`; executor preflight recovery is only a fallback for rows already stranded by older runs.
+            */
+            const finalization = await finalizeProvenAutoMergeTask({
+              store: this.store,
+              taskId: task.id,
+              result,
+              audit: createRunAuditor(this.store, {
+                runId: ctx.run.runId,
+                agentId: "executor",
+                taskId: task.id,
+                taskLineageId: task.lineageId,
+                phase: "workflow-merge",
+              }),
+              auditAgentId: "executor",
+              auditPhase: "workflow-merge",
+              source: "workflow-graph-merge-finalize",
+              log: (message) => executorLog.warn(message),
+            });
+            if (finalization.outcome === "blocked" || finalization.outcome === "missing") {
+              return {
+                outcome: "failure",
+                value: `merge-finalize-${finalization.outcome}`,
+                data: { status: "failed", reason: finalization.reason ?? finalization.outcome },
+              };
+            }
             return {
               outcome: "success",
               value: result.noOp ? "merge-noop" : "merged",
@@ -6409,6 +6469,138 @@ export class TaskExecutor {
     };
   }
 
+  private async ensureGraphCustomNodeWorktree(
+    task: TaskDetail,
+    settings: Settings,
+    nodeId: string,
+  ): Promise<TaskDetail> {
+    /*
+    FNXC:WorkflowExecution 2026-06-29-08:21:
+    Custom graph nodes can be the first executable node in a workflow. If such a node is coding/script-capable, acquire the same task worktree the legacy executor would have acquired instead of failing with `no-worktree-for-write-node`; the node remains isolated from main and CE `plan` can run first.
+    */
+    if (this.workspaceConfig === undefined) {
+      this.workspaceConfig = await loadWorkspaceConfig(this.rootDir);
+    }
+    if (this.workspaceConfig && (this.workspaceConfig.repos.length ?? 0) > 0) {
+      return task;
+    }
+
+    const syntheticRunId = generateSyntheticRunId("workflow-node-worktree", task.id);
+    const audit = createRunAuditor(this.store, {
+      runId: syntheticRunId,
+      agentId: task.assignedAgentId ?? "executor",
+      taskId: task.id,
+      phase: "execute",
+    });
+    const commandAbortController = new AbortController();
+    this.registerConfiguredCommandController(task.id, commandAbortController);
+    try {
+      await this.store.logEntry(
+        task.id,
+        `Workflow node '${nodeId}' requires a task worktree — acquiring worktree before node execution`,
+        undefined,
+        this.getRunContextFor(task.id),
+      );
+      const acquisition = await acquireTaskWorktree({
+        task,
+        rootDir: this.rootDir,
+        store: this.store,
+        settings,
+        pool: this.options.pool,
+        logger: executorLog,
+        audit,
+        runContext: this.getRunContextFor(task.id),
+        runInitCommand: true,
+        createWorktree: this.createWorktree.bind(this),
+        runConfiguredCommand: (command, cwd, timeoutMs, env) =>
+          runConfiguredCommand(
+            command,
+            cwd,
+            timeoutMs,
+            env,
+            audit,
+            commandAbortController.signal,
+          ).then((result) => {
+            if (commandAbortController.signal.aborted) {
+              throw this.createConfiguredCommandAbortError(task.id, command);
+            }
+            return result;
+          }),
+        taskEnv: process.env,
+        secretsStore: this.options.secretsStore,
+      });
+      this.addActiveWorktree(task.id, acquisition.worktreePath);
+      if (!acquisition.isResume) {
+        await this.captureBaseCommitSha(task, acquisition.worktreePath, audit, { isResume: false });
+      }
+      this.options.onStart?.(task, acquisition.worktreePath);
+      executorLog.log(`${task.id}: workflow node '${nodeId}' acquired worktree at ${acquisition.worktreePath}`);
+      return await this.store.getTask(task.id);
+    } catch (error) {
+      const message = error instanceof Error ? error.message : String(error);
+      await this.store.logEntry(
+        task.id,
+        `Workflow node '${nodeId}' failed to acquire task worktree: ${message}`,
+        undefined,
+        this.getRunContextFor(task.id),
+      );
+      throw error;
+    } finally {
+      this.unregisterConfiguredCommandController(task.id, commandAbortController);
+    }
+  }
+
+  private async finalizeMergeConfirmedWorkflowGraphTask(taskId: string, reason: string): Promise<boolean> {
+    const live = await this.store.getTask(taskId).catch(() => null);
+    if (!live || live.mergeDetails?.mergeConfirmed !== true || live.column === "done") return false;
+    /*
+    FNXC:WorkflowMerge 2026-06-29-08:32:
+    A workflow graph merge node can await a successful ProjectEngine merge request and return before the row reaches `done`. Merge confirmation is durable proof of landing; the executor must finalize that row from any non-terminal column instead of re-running parse or clearing mergeDetails.
+    */
+    await this.store.logEntry(
+      taskId,
+      `Workflow graph observed confirmed merge while task was '${live.column}' — finalizing to done (${reason})`,
+      undefined,
+      this.getRunContextFor(taskId),
+    );
+    const finalization = await finalizeProvenAutoMergeTask({
+      store: this.store,
+      taskId,
+      result: {
+        task: live,
+        ok: true,
+        merged: true,
+        commitSha: live.mergeDetails?.commitSha,
+        noOp: live.mergeDetails?.noOpMerge === true,
+        reason: live.mergeDetails?.noOpReason,
+        mergeConfirmed: true,
+      } as MergeResult,
+      audit: createRunAuditor(this.store, {
+        runId: generateSyntheticRunId("workflow-graph-merge-finalize", taskId),
+        agentId: "executor",
+        taskId,
+        taskLineageId: live.lineageId,
+        phase: "workflow-graph-merge-finalize",
+      }),
+      auditAgentId: "executor",
+      auditPhase: "workflow-graph-merge-finalize",
+      source: "workflow-graph-merge-finalize",
+      log: (message) => executorLog.warn(message),
+    });
+    if (finalization.outcome === "blocked") {
+      executorLog.warn(`${taskId}: workflow graph merge-confirmed finalization blocked — ${finalization.reason ?? "unknown"}`);
+      await this.store.logEntry(
+        taskId,
+        `Workflow graph merge-confirmed finalization blocked — ${finalization.reason ?? "unknown"}`,
+        undefined,
+        this.getRunContextFor(taskId),
+      );
+      return true;
+    }
+    executorLog.log(`${taskId}: workflow graph merge-confirmed task finalized (${finalization.outcome})`);
+    return true;
+  }
+
   /** Run a custom (non-seam) graph node on the proven WorkflowStep machinery.
    *
    *  `columnBinding` (plan U3) is the agent binding governing this node's
@@ -6504,17 +6696,18 @@ export class TaskExecutor {
     // main checkout and cross-contaminate other tasks. Reject such nodes until a
     // worktree exists. Read-only nodes (default toolMode) are safe against root.
     const writeCapable = cfg.toolMode === "coding" || node.kind === "script" || Boolean(scriptName) || Boolean(rawCliCommand);
-    if (writeCapable && !live.worktree) {
-      await this.store.logEntry(
-        live.id,
-        `Workflow node '${node.id}' is write-capable but no task worktree exists yet — place it after the execute seam`,
-        undefined,
-        this.getRunContextFor(live.id),
-      );
+    /*
+    FNXC:CompoundEngineering 2026-06-29-08:18:
+    Compound engineering starts with a coding-mode `ce-plan` skill node so it can load CE spawn tools before implementation. The graph custom-node path must therefore bootstrap the task worktree itself; requiring an earlier execute seam makes the built-in CE workflow fail at node `plan` before it can start.
+    */
+    const executionTarget = writeCapable && !live.worktree
+      ? await this.ensureGraphCustomNodeWorktree(live, settings, node.id)
+      : live;
+    if (writeCapable && !executionTarget.worktree && !this.workspaceConfig) {
       return { outcome: "failure", value: "no-worktree-for-write-node" };
     }
 
-    const worktreePath = live.worktree || this.rootDir;
+    const worktreePath = executionTarget.worktree || this.rootDir;
     let prompt = typeof cfg.prompt === "string" ? cfg.prompt : "";
     let modelProvider = typeof cfg.modelProvider === "string" && cfg.modelProvider.trim() ? cfg.modelProvider : undefined;
     let modelId = typeof cfg.modelId === "string" && cfg.modelId.trim() ? cfg.modelId : undefined;
@@ -6685,7 +6878,7 @@ export class TaskExecutor {
     if (executorKind === "cli" && prompt) {
       nodeEnv = { ...process.env, FUSION_NODE_PROMPT: prompt };
     } else if (mode === "prompt") {
-      const injected = await this.buildInjectedRuntimeEnv(live.id, worktreePath, live.branch ?? undefined);
+      const injected = await this.buildInjectedRuntimeEnv(live.id, worktreePath, executionTarget.branch ?? undefined);
       nodeEnv = injected.env;
       executorLog.log(
         `${live.id}: graph node '${node.id}' runtime env injected (${injected.pathEntryCount} PATH entries, ${injected.injectedKeyCount} env keys)`,
@@ -7377,6 +7570,12 @@ export class TaskExecutor {
         return;
       }
       const live = loadedLive;
+      if (live.mergeDetails?.mergeConfirmed === true && live.column !== "done") {
+        if (await this.finalizeMergeConfirmedWorkflowGraphTask(live.id, "graph-failure")) {
+          await this.persistTokenUsage(task.id);
+          return;
+        }
+      }
       // A paused/aborted implementation is not a graph failure while the task
       // is still in-progress — leave the pause machinery in charge instead of
       // parking the task in review.
@@ -8135,6 +8334,14 @@ export class TaskExecutor {
     // executor can still recover by falling through to the fresh-worktree
     // path below, but we emit a loud audit record so these states stop being
     // silent.
+    if (task.column === "in-progress" && task.mergeDetails?.mergeConfirmed === true) {
+      if (await this.finalizeMergeConfirmedWorkflowGraphTask(task.id, "execute-preflight")) {
+        this.executing.delete(task.id);
+        executingTaskLock.release(task.id);
+        return;
+      }
+    }
+
     if (task.column === "in-progress" && task.mergeDetails) {
       executorLog.warn(`${task.id}: stale mergeDetails found while executing in-progress task — resetting merge state before continuing`);
       task = await this.cleanupMergeStateForReverification(

packages/engine/src/workflow-graph-executor.ts

@@ -498,6 +498,7 @@ export class WorkflowGraphExecutor {
             task,
             runId,
             steps,
+            getLiveSteps: () => this.resolveTaskSteps(task),
             context,
             runTemplateNode: (tNode, sig, contextOverride) =>
               this.executeNodeWithRetries(tNode, task, settings, contextOverride ?? context, ir, sig),

packages/engine/src/workflow-graph-foreach.ts

@@ -131,6 +131,8 @@ export interface ForeachEnvironment {
   runId: string;
   /** Fresh step list (KTD-3: read at expansion, count pinned). */
   steps: TaskStep[];
+  /** Optional live step projection reader used during restart/replay checks. */
+  getLiveSteps?: () => Promise<TaskStep[]> | TaskStep[];
   /** The shared walk context; the active-instance key is threaded in/out of it. */
   context: Record<string, unknown>;
   /**
@@ -413,12 +415,12 @@ 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;
+    /*
+    FNXC:WorkflowResume 2026-06-29-08:49:
+    The workflow graph owns step replay after engine restarts. A shared-isolation foreach pins the step count at expansion, but must read the live projection before each instance so a completed task does not re-run a stale step snapshot and fail on an already-finished `step-execute` node.
+    */
+    const liveSteps = await Promise.resolve(env.getLiveSteps?.() ?? env.steps).catch(() => env.steps);
+    const stepStatus = liveSteps[stepIndex]?.status ?? env.steps[stepIndex]?.status;
     if (stepStatus === "done" || stepStatus === "skipped") {
       schedulerLog.log(
         `foreach ${foreachNode.id} for task ${env.task.id}: skipping step ${stepIndex} — already ${stepStatus}`,

packages/engine/src/workflow-graph-task-runner.ts

@@ -66,6 +66,7 @@ export interface WorkflowGraphTaskRunResult {
 export interface WorkflowGraphRunnerStore {
   getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined;
   getWorkflowDefinition(id: string): Promise<WorkflowDefinition | undefined>;
+  getTask?(taskId: string): Promise<TaskDetail>;
 }
 
 export interface WorkflowGraphTaskRunnerDeps {
@@ -274,6 +275,11 @@ export class WorkflowGraphTaskRunner {
         runCode: this.deps.runCode,
         notifyDispatch: this.deps.notifyDispatch,
         prNodes: this.deps.prNodes,
+        /*
+        FNXC:WorkflowResume 2026-06-29-08:49:
+        Production graph runs must fetch live task steps during foreach replay. The runner is the workflow boundary that has store access, so it supplies the fresh projection seam instead of making executor self-healing guess after a stale step node fails.
+        */
+        getTaskSteps: async (stepTask) => (await this.deps.store.getTask?.(stepTask.id))?.steps ?? stepTask.steps ?? [],
         // Step-inversion (KTD-11, U10): worktree isolation + parallel scheduling.
         allocateInstanceWorktree: this.deps.allocateInstanceWorktree,
         resolveIntegrationBase: this.deps.resolveIntegrationBase,

packages/engine/src/workflow-node-handlers.ts

@@ -585,13 +585,13 @@ export interface ParseStepsHandlerDeps {
    * expanded for this task+run — either persisted instance rows exist OR a
    * foreach expanded earlier in this walk. Re-parsing after expansion is illegal
    * (it would silently desynchronize the pinned instance set), so the handler
-   * fails with an audited `pin-mismatch` outcome. Optional — absent means no
+   * resumes without rewriting the step projection. Optional — absent means no
    * pin established (always safe to parse).
    */
   hasExpandedForeach?: (task: TaskDetail) => Promise<boolean> | boolean;
   /** Optional audit sink: called with a stable reason code on every routable
-   *  failure outcome (`parse-error`, `pin-mismatch`) so the run audit records it.
-   *  Never throws into the handler. */
+   *  parse outcome (`parse-error`, `pin-mismatch`, `pin-resume`) so the run audit
+   *  records it. Never throws into the handler. */
   audit?: (reason: string, detail: string) => void;
 }
 
@@ -605,7 +605,7 @@ export interface ParseStepsHandlerDeps {
  *   - missing artifact         → `outcome:failure value:"parse-error"` (audited)
  *   - parser throws            → `outcome:failure value:"parse-error"` (audited, never crashes)
  *   - clean empty parse        → `outcome:success value:"no-steps"` (routable; defaults to success)
- *   - foreach already expanded → `outcome:failure value:"pin-mismatch"` (audited, KTD-3)
+ *   - foreach already expanded → `outcome:success value:"already-expanded"` (audited, KTD-3)
  *   - steps parsed             → `outcome:success` (steps written through projection)
  */
 export function createParseStepsHandler(deps: ParseStepsHandlerDeps): WorkflowNodeHandler {
@@ -628,11 +628,15 @@ export function createParseStepsHandler(deps: ParseStepsHandlerDeps): WorkflowNo
     // Pin protection (KTD-3): re-parsing after a foreach has expanded is illegal.
     try {
       if (deps.hasExpandedForeach && (await deps.hasExpandedForeach(ctx.task))) {
+        /*
+        FNXC:WorkflowResume 2026-06-29-08:02:
+        Engine restart/retry re-enters the workflow from the start node, so `parse-steps` can be reached after a foreach already has persisted instance pins. That is a resume boundary, not a fatal graph error: preserve the pinned step list by not rewriting steps, then let foreach continue from its persisted instances. Probe exceptions still fail closed below because the engine cannot prove pins are valid.
+        */
         audit(
-          "pin-mismatch",
-          `parse-steps node '${node.id}' reached after a foreach already expanded for task ${ctx.task.id}`,
+          "pin-resume",
+          `parse-steps node '${node.id}' reached after a foreach already expanded for task ${ctx.task.id}; preserving pinned steps`,
         );
-        return { outcome: "failure", value: "pin-mismatch" };
+        return { outcome: "success", value: "already-expanded" };
       }
     } catch (err) {
       // A pin-probe failure must fail closed (never silently re-parse).