fix: address PR review feedback (#1424)

- restore customFields on unarchive; reconcile all occupants on field-schema edits (store.ts)
- serialize per-field saves + controlled inputs in TaskFieldsSection (race fixes)
- fn_workflow_get includes layout; Array.isArray guards in validateCodeNodeSources
- per-instance graphStepActiveContext keying; rebase in instance worktree; clear run-once memo on RETHINK
- GET /api/step-parsers + registry-backed parser select (plugin parsers reachable from editor)
- translate new workflowNodes/workflowFields strings across all 5 non-en locales

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/engine/src/__tests__/agent-tools.test.ts

@@ -433,6 +433,40 @@ describe("createWorkflowGetTool", () => {
     expect(result.details).toMatchObject({ builtin: true });
   });
 
+  it("includes layout when the definition carries editor positions", async () => {
+    const layout = { n1: { x: 10, y: 20 } };
+    const store = {
+      getWorkflowDefinition: vi.fn().mockResolvedValue({
+        id: "WF-005",
+        name: "Laid out",
+        description: "",
+        ir: { version: "v2", name: "Laid out", columns: [], nodes: [{ id: "n1", kind: "step-execute" }], edges: [] },
+        layout,
+      }),
+    };
+    const tool = createWorkflowGetTool(store as any);
+    const result = await tool.execute("call-1", { workflow_id: "WF-005" } as any, undefined, undefined, {} as any);
+    const text = result.content[0]?.type === "text" ? result.content[0].text : "";
+    expect(JSON.parse(text).layout).toEqual(layout);
+    expect(result.details).toMatchObject({ workflowId: "WF-005", layout });
+  });
+
+  it("omits layout when the definition has none", async () => {
+    const store = {
+      getWorkflowDefinition: vi.fn().mockResolvedValue({
+        id: "WF-006",
+        name: "No layout",
+        description: "",
+        ir: { version: "v2", name: "No layout", columns: [], nodes: [], edges: [] },
+      }),
+    };
+    const tool = createWorkflowGetTool(store as any);
+    const result = await tool.execute("call-1", { workflow_id: "WF-006" } as any, undefined, undefined, {} as any);
+    const text = result.content[0]?.type === "text" ? result.content[0].text : "";
+    expect(JSON.parse(text)).not.toHaveProperty("layout");
+    expect(result.details).not.toHaveProperty("layout");
+  });
+
   it("returns an error result for an unknown id", async () => {
     const store = { getWorkflowDefinition: vi.fn().mockResolvedValue(undefined) };
     const tool = createWorkflowGetTool(store as any);

packages/engine/src/__tests__/code-node.test.ts

@@ -113,6 +113,24 @@ describe("validateCodeNodeSources (U14, save-time helper)", () => {
     const ir = { nodes: [codeNode("export default async () => ({ outcome: 'ok' });")] };
     expect(await validateCodeNodeSources(ir)).toEqual([]);
   });
+
+  it("does not crash on a malformed foreach template.nodes (non-array)", async () => {
+    const ir = {
+      nodes: [
+        // `nodes` is an object, not an array — a truthy-but-malformed config.
+        { id: "fe", kind: "foreach", config: { template: { nodes: { bogus: true } } } } as unknown as WorkflowIrNode,
+      ],
+    };
+    const failures = await validateCodeNodeSources(ir);
+    expect(failures).toEqual([{ nodeId: "fe", error: "foreach template.nodes must be an array" }]);
+  });
+
+  it("ignores a foreach with no template nodes", async () => {
+    const ir = {
+      nodes: [{ id: "fe", kind: "foreach", config: { template: {} } } as unknown as WorkflowIrNode],
+    };
+    expect(await validateCodeNodeSources(ir)).toEqual([]);
+  });
 });
 
 describe("createCodeNodeRunner result mapping (U14, seam-injected)", () => {

packages/engine/src/__tests__/workflow-graph-step-rerun.test.ts

@@ -27,8 +27,14 @@ describe("runGraphTaskStep (FIX 3)", () => {
       steps: stepStatus ? [{ name: "S1", status: stepStatus }] : [{ name: "S1", status: "pending" }],
     });
     const executor: any = new TaskExecutor(store, "/tmp/test", {});
-    // Stamp the active foreach context the seam would normally stamp.
-    if (active) executor.graphStepActiveContext.set("FN-001", { stepIndex: 0, ...active });
+    // Stamp the active foreach context the seam would normally stamp, keyed by
+    // the composite per-instance key (T7).
+    if (active) {
+      executor.graphStepActiveContext.set(
+        executor.graphActiveContextKey("FN-001", "inst-0"),
+        { stepIndex: 0, instanceId: "inst-0", ...active },
+      );
+    }
     return { executor, store };
   }
 
@@ -93,4 +99,55 @@ describe("runGraphTaskStep (FIX 3)", () => {
     const result = await executor.runGraphTaskStep(task, 0);
     expect(result.success).toBe(true);
   });
+
+  // T9: a RETHINK after a SUCCESSFUL pass must clear the memoized implementation
+  // so the rework re-runs implementation rather than re-awaiting the resolved memo.
+  it("clears the memo on rethink reset so implementation re-runs after a successful pass", async () => {
+    const { executor, store } = makeExecutor("done", { deferDoneToReview: true });
+    // No-op the git/step reset machinery — only the memo-clearing path matters here.
+    store.getTask = vi.fn().mockResolvedValue({ id: "FN-001", steps: [{ name: "S1", status: "done" }] });
+    let calls = 0;
+    executor.runImplementationPhase = vi.fn().mockImplementation(async () => {
+      calls += 1;
+      return { taskDone: true, modifiedFiles: [] };
+    });
+
+    // First pass: succeeds and the memo is now resolved.
+    const first = await executor.runGraphTaskStep(task, 0, "inst-0");
+    expect(first.success).toBe(true);
+    expect(calls).toBe(1);
+    expect(executor.graphStepRunOnce.has("FN-001")).toBe(true);
+
+    // RETHINK reset clears the SETTLED memo (guarded against in-flight clobber).
+    await executor.applyGraphRethinkReset("FN-001", { stepIndex: 0, instanceId: "inst-0" });
+    expect(executor.graphStepRunOnce.has("FN-001")).toBe(false);
+
+    // Rework re-run: implementation is invoked AGAIN (the bug re-awaited the memo).
+    const second = await executor.runGraphTaskStep(task, 0, "inst-0");
+    expect(second.success).toBe(true);
+    expect(calls).toBe(2);
+  });
+
+  // T7: parallel instances of the same task keep independent active contexts.
+  it("keys active context per-instance so parallel foreach instances do not clobber", async () => {
+    const store = createMockStore();
+    store.getTask = vi.fn().mockResolvedValue({ id: "FN-001", steps: [{ name: "S1", status: "in-progress" }] });
+    const executor: any = new TaskExecutor(store, "/tmp/test", {});
+    // Instance A defers done to review (non-terminal → success); instance B does not
+    // (non-terminal → failure). A per-task key would let one overwrite the other.
+    executor.graphStepActiveContext.set(
+      executor.graphActiveContextKey("FN-001", "inst-A"),
+      { stepIndex: 0, instanceId: "inst-A", deferDoneToReview: true },
+    );
+    executor.graphStepActiveContext.set(
+      executor.graphActiveContextKey("FN-001", "inst-B"),
+      { stepIndex: 0, instanceId: "inst-B", deferDoneToReview: false },
+    );
+    executor.runImplementationPhase = vi.fn().mockResolvedValue({ taskDone: false, modifiedFiles: [] });
+
+    const a = await executor.runGraphTaskStep(task, 0, "inst-A");
+    const b = await executor.runGraphTaskStep(task, 0, "inst-B");
+    expect(a.success).toBe(true); // review authors done
+    expect(b.success).toBe(false); // implementation left it incomplete
+  });
 });

packages/engine/src/__tests__/workflow-step-integration-cwd.test.ts

@@ -0,0 +1,77 @@
+// -nocheck
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+import "./executor-test-helpers.js";
+import { TaskExecutor } from "../executor.js";
+import { createMockStore, resetExecutorMocks, mockedExecSync } from "./executor-test-helpers.js";
+
+/**
+ * T8: the integration rebase (and rebase --abort) must run in the INSTANCE
+ * worktree — the instance branch is checked out there, so running the rebase
+ * from the task's MAIN worktree fails with "branch is already checked out in
+ * another worktree". The final fast-forward merge still runs from the main
+ * worktree (it advances the target branch checked out there).
+ *
+ * The shared executor harness routes the promisified `exec` through the
+ * `execSync` mock (see executor-test-helpers), so we drive behavior + capture
+ * cwds via `mockedExecSync`.
+ */
+describe("buildForeachWorktreeDeps integrate() cwd (T8)", () => {
+  beforeEach(() => {
+    resetExecutorMocks();
+    mockedExecSync.mockReset();
+  });
+  afterEach(() => vi.restoreAllMocks());
+
+  function makeDeps() {
+    const store = createMockStore();
+    store.getTask = vi.fn().mockResolvedValue({
+      id: "FN-PAR",
+      worktree: "/main/wt",
+      branch: "fusion/FN-PAR",
+    });
+    const executor: any = new TaskExecutor(store, "/root", {});
+    // Stub createWorktree so allocateInstanceWorktree records the instance path
+    // without touching the filesystem.
+    executor.createWorktree = vi.fn(async (branch: string, _path: string) => ({
+      path: `/inst/step-${branch}`,
+      branch,
+    }));
+    const deps = executor.buildForeachWorktreeDeps({ id: "FN-PAR", branch: "fusion/FN-PAR" });
+    return { executor, deps };
+  }
+
+  it("runs rebase in the instance worktree and ff-merge in the main worktree", async () => {
+    const { deps } = makeDeps();
+    const alloc = await deps.allocateInstanceWorktree(2, "base-sha");
+    const calls: Array<{ cmd: string; cwd: string }> = [];
+    mockedExecSync.mockImplementation((cmd: string, opts: any) => {
+      calls.push({ cmd, cwd: String(opts?.cwd ?? "") });
+      return "deadbeef";
+    });
+
+    const result = await deps.integrationGitOps.integrate(alloc.branchName, 2);
+    expect(result.kind).toBe("integrated");
+
+    const rebase = calls.find((c) => c.cmd.startsWith("git rebase ") && !c.cmd.includes("--abort"));
+    const merge = calls.find((c) => c.cmd.startsWith("git merge --ff-only"));
+    expect(rebase?.cwd).toBe(`/inst/step-${alloc.branchName}`); // instance worktree
+    expect(merge?.cwd).toBe("/main/wt"); // main worktree
+  });
+
+  it("falls back to the main worktree cwd when no instance path is recorded", async () => {
+    // Defensive: an integrate() for a stepIndex with no allocated instance path
+    // (e.g. shared isolation) must not pass an undefined cwd to the rebase.
+    const { deps } = makeDeps();
+    const calls: Array<{ cmd: string; cwd: string }> = [];
+    mockedExecSync.mockImplementation((cmd: string, opts: any) => {
+      calls.push({ cmd, cwd: String(opts?.cwd ?? "") });
+      return "deadbeef";
+    });
+
+    const result = await deps.integrationGitOps.integrate("fusion/FN-PAR-step-9", 9);
+    expect(result.kind).toBe("integrated");
+    const rebase = calls.find((c) => c.cmd.startsWith("git rebase ") && !c.cmd.includes("--abort"));
+    expect(rebase?.cwd).toBe("/main/wt"); // fallback to main worktree, never undefined
+  });
+});

packages/engine/src/agent-tools.ts

@@ -1080,10 +1080,14 @@ export function createWorkflowGetTool(store: TaskStore): ToolDefinition {
           description: def.description,
           builtin,
           ir: def.ir,
+          // Preserve editor node positions so a read→modify→write cycle does not
+          // strip the layout. May be absent for older/built-in defs; only include
+          // when present to keep the payload tidy.
+          ...(def.layout ? { layout: def.layout } : {}),
         };
         return {
           content: [{ type: "text" as const, text: JSON.stringify(payload, null, 2) }],
-          details: { workflowId: def.id, builtin },
+          details: { workflowId: def.id, builtin, ...(def.layout ? { layout: def.layout } : {}) },
         };
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
       } catch (err: any) {

packages/engine/src/code-node-runner.ts

@@ -401,9 +401,15 @@ export async function validateCodeNodeSources(
   // Recurse into any foreach templates (code nodes are legal inside them, KTD-15).
   for (const node of ir.nodes) {
     if (node.kind !== "foreach") continue;
-    const template = (node.config as { template?: { nodes?: WorkflowIrNode[] } } | undefined)?.template;
-    if (template?.nodes) {
-      failures.push(...(await validateCodeNodeSources({ nodes: template.nodes })));
+    const template = (node.config as { template?: { nodes?: unknown } } | undefined)?.template;
+    // Guard with an explicit array check: a malformed config where `nodes` is a
+    // non-array truthy value would otherwise break the `for...of` inside the
+    // recursive validateCodeNodeSources call and bubble as a 500 rather than a
+    // clean validation failure.
+    if (Array.isArray(template?.nodes)) {
+      failures.push(...(await validateCodeNodeSources({ nodes: template.nodes as WorkflowIrNode[] })));
+    } else if (template?.nodes != null) {
+      failures.push({ nodeId: node.id, error: "foreach template.nodes must be an array" });
     }
   }
   return failures;

packages/engine/src/executor.ts

@@ -3249,9 +3249,16 @@ export class TaskExecutor {
    *  `deferDoneToReview` when deciding whether a non-terminal step is a success
    *  (review will author done) or a failure (implementation left it incomplete).
    *  Stamped by the stepExecute seam around the runTaskStep call; cleared with the
-   *  per-run pins. */
+   *  per-run pins. Keyed by `${task.id}:${instanceId}` so parallel foreach
+   *  instances of the same task cannot clobber each other's active context
+   *  (the read path threads the same instanceId through `runGraphTaskStep`). */
   private graphStepActiveContext = new Map<string, ForeachActiveContext>();
 
+  /** Composite key for {@link graphStepActiveContext}: per-instance, not per-task. */
+  private graphActiveContextKey(taskId: string, instanceId: string): string {
+    return `${taskId}:${instanceId}`;
+  }
+
   /** Tasks currently being orchestrated by the graph runner. Process-wide for
    *  the same reason as executingTaskLock (FN-4811): duplicate execute()
    *  invocations can arrive from different TaskExecutor instances in one
@@ -3378,7 +3385,11 @@ export class TaskExecutor {
       // Clear per-run step-inversion pins (KTD-8: pinned only for the run's life).
       this.graphStepSessionPinned.delete(task.id);
       this.graphStepRunOnce.delete(task.id);
-      this.graphStepActiveContext.delete(task.id);
+      // Per-instance keys: clear every instance slot owned by this task.
+      const ctxPrefix = `${task.id}:`;
+      for (const key of this.graphStepActiveContext.keys()) {
+        if (key.startsWith(ctxPrefix)) this.graphStepActiveContext.delete(key);
+      }
     }
   }
 
@@ -3656,17 +3667,26 @@ export class TaskExecutor {
         integrate: async (branchName, stepIndex): Promise<import("./step-integration.js").IntegrationAttemptResult> => {
           const cwd = await mainWorktree();
           const target = await mainBranch();
+          // The instance branch is checked out in its OWN worktree, so the rebase
+          // (which checks out `branchName`) must run THERE — running it from the
+          // main worktree fails with "branch is already checked out in another
+          // worktree". The final fast-forward merge still runs from the main
+          // worktree (it only advances `target`, which is checked out there).
+          const instanceCwd = instancePaths.get(stepIndex) ?? cwd;
           try {
-            // Rebase the instance branch onto the current main tip, then ff main.
-            await execAsync(`git rebase ${target} ${branchName}`, { cwd });
+            // Rebase the instance branch onto the current main tip (in its own
+            // worktree), then ff main from the main worktree.
+            await execAsync(`git rebase ${target} ${branchName}`, { cwd: instanceCwd });
             await execAsync(`git checkout ${target}`, { cwd });
             await execAsync(`git merge --ff-only ${branchName}`, { cwd });
             return { kind: "integrated", integratedAt: new Date().toISOString() };
           } catch (err) {
             // Conflict (or other rebase failure): classify via merger helper, abort.
-            const conflictedFiles = await getConflictedFiles(cwd);
+            // The rebase ran in the instance worktree, so conflicts live there and
+            // the abort must target that same cwd.
+            const conflictedFiles = await getConflictedFiles(instanceCwd);
             try {
-              await execAsync("git rebase --abort", { cwd });
+              await execAsync("git rebase --abort", { cwd: instanceCwd });
             } catch {
               // best-effort; leave the worktree recoverable.
             }
@@ -3808,6 +3828,28 @@ export class TaskExecutor {
    * the documented KTD-2 semantics; the git reset + step→pending are authoritative.
    */
   private async applyGraphRethinkReset(taskId: string, active: ForeachActiveContext): Promise<void> {
+    // Clear the memoized implementation pass so the next `runGraphTaskStep`
+    // re-executes (T9): the per-run pass is memoized in `graphStepRunOnce` keyed
+    // by task id and is normally only cleared on REJECTION. A RETHINK fires AFTER
+    // a SUCCESSFUL pass (a review verdict resets git/step state via this reset),
+    // so without clearing the memo the rework re-awaits the already-resolved
+    // promise and implementation never re-runs — leaving the instance permanently
+    // pending or falsely successful under `deferDoneToReview`. Mirrors the
+    // rejection-clear guard: only delete the memo when the stored promise is the
+    // SETTLED pass (a fresh in-flight attempt another caller installed is left
+    // untouched). At rethink time the pass under review has already resolved, so
+    // checking settled-ness avoids clobbering a concurrent re-dispatch.
+    const memo = this.graphStepRunOnce.get(taskId);
+    if (memo) {
+      let settled = false;
+      await Promise.race([memo.then(
+        () => { settled = true; },
+        () => { settled = true; },
+      ), Promise.resolve()]);
+      if (settled && this.graphStepRunOnce.get(taskId) === memo) {
+        this.graphStepRunOnce.delete(taskId);
+      }
+    }
     // Worktree isolation (KTD-11): reset the instance's OWN branch/worktree only —
     // sibling instances and the integration base are untouched, so the blast-radius
     // guard is STRUCTURAL (skipped) in this mode. Shared isolation resets the task's
@@ -4015,7 +4057,11 @@ export class TaskExecutor {
    *
    * Returns whether the targeted step ended up `done`/`skipped` in the projection.
    */
-  private async runGraphTaskStep(task: Task, stepIndex: number): Promise<{ success: boolean; error?: string }> {
+  private async runGraphTaskStep(
+    task: Task,
+    stepIndex: number,
+    instanceId?: string,
+  ): Promise<{ success: boolean; error?: string }> {
     // Pin step-session physics for the run before the implementation pass.
     this.graphStepSessionPinned.add(task.id);
 
@@ -4049,7 +4095,7 @@ export class TaskExecutor {
     // completes; a step-review node (when present) decides done-ness instead.
     try {
       const live = await this.store.getTask(task.id);
-      const active = this.foreachActiveForTask(task.id);
+      const active = this.foreachActiveForTask(task.id, instanceId);
       const status = live.steps[stepIndex]?.status;
       if (status === "done" || status === "skipped") return { success: true };
       // Step not terminal after the pass: when a review will author done-ness
@@ -4071,8 +4117,21 @@ export class TaskExecutor {
    *  the step driver can honor `deferDoneToReview`. The active context is threaded
    *  through the foreach sub-walk; we surface it via a per-task slot the
    *  step-execute seam stamps. Returns undefined outside a foreach instance. */
-  private foreachActiveForTask(taskId: string): ForeachActiveContext | undefined {
-    return this.graphStepActiveContext.get(taskId);
+  private foreachActiveForTask(taskId: string, instanceId?: string): ForeachActiveContext | undefined {
+    if (typeof instanceId === "string") {
+      const byInstance = this.graphStepActiveContext.get(this.graphActiveContextKey(taskId, instanceId));
+      if (byInstance) return byInstance;
+    }
+    // Fallback (single-instance / no instanceId threaded): return the sole slot
+    // owned by this task if exactly one exists.
+    const prefix = `${taskId}:`;
+    let only: ForeachActiveContext | undefined;
+    for (const [key, value] of this.graphStepActiveContext) {
+      if (!key.startsWith(prefix)) continue;
+      if (only) return undefined; // ambiguous: more than one instance active
+      only = value;
+    }
+    return only;
   }
 
   /** Seam implementations delegating to the legacy engine (KTD-1: delegate, never reimplement). */
@@ -4160,7 +4219,7 @@ export class TaskExecutor {
         const worktreePath = active.worktreePath || live.worktree || this.rootDir;
         // Stamp the active instance so `runGraphTaskStep` can honor
         // `deferDoneToReview` when judging a non-terminal step (FIX 3).
-        this.graphStepActiveContext.set(seamTask.id, active);
+        this.graphStepActiveContext.set(this.graphActiveContextKey(seamTask.id, active.instanceId), active);
         const result = await runTaskStep(
           {
             store: this.store,
@@ -4168,8 +4227,9 @@ export class TaskExecutor {
             // U6/U8: per-step session physics — graph-owned runs force
             // step-session mode for the run (KTD-2/KTD-8) regardless of the
             // runStepsInNewSessions setting. The agent authors the step's commit;
-            // this driver only observes (KTD-2).
-            runStep: (stepIndex) => this.runGraphTaskStep(seamTask, stepIndex),
+            // this driver only observes (KTD-2). Thread the instanceId so the
+            // active-context read is per-instance (parallel-foreach safe).
+            runStep: (stepIndex) => this.runGraphTaskStep(seamTask, stepIndex, active.instanceId),
           },
           { id: seamTask.id, steps: live.steps },
           active.stepIndex,