feat(engine): U3 — foreach expansion, iterative instance sub-walk, bounded rework cycles, step-execute seam

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

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

@@ -3,7 +3,7 @@ import type { TaskDetail, WorkflowIrNode } from "@fusion/core";
 
 import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js";
 
-export type WorkflowSeamName = "planning" | "execute" | "review" | "merge" | "schedule";
+export type WorkflowSeamName = "planning" | "execute" | "review" | "merge" | "schedule" | "step-execute";
 
 export interface WorkflowLegacySeams {
   /** Planning/spec stage. Built-in triage runs upstream of the interpreter
@@ -14,6 +14,32 @@ export interface WorkflowLegacySeams {
   review: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
   merge: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
   schedule: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
+  /**
+   * Step-inversion (KTD-2/KTD-4, U3): run exactly the foreach-active step inside
+   * the task's session/worktree. Only invoked for `step-execute` prompt nodes
+   * inside a foreach template, where `context["foreach:active"]` carries the
+   * active instance's `stepIndex`. Optional — a workflow that never uses a
+   * foreach/step-execute node needs no implementation (the noop seams omit it,
+   * and a step-execute node reached without this wired fails cleanly rather than
+   * silently no-opping). The engine wires this to `runTaskStep` (executor.ts
+   * createGraphSeams); it returns the per-step `baselineSha`/`checkpointId` in
+   * its `contextPatch` so a later RETHINK (U5) can reset the step.
+   */
+  stepExecute?: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
+}
+
+/** The reserved context key carrying the active foreach instance (KTD-3, U3).
+ *  Template node handlers (step-execute now; step-review in U5) read it to learn
+ *  which step they operate on and the per-instance baseline/checkpoint state. */
+export const FOREACH_ACTIVE_CONTEXT_KEY = "foreach:active";
+
+/** Shape of the value stored under {@link FOREACH_ACTIVE_CONTEXT_KEY}. */
+export interface ForeachActiveContext {
+  foreachNodeId: string;
+  stepIndex: number;
+  instanceId: string;
+  baselineSha?: string;
+  checkpointId?: string;
 }
 
 /**
@@ -31,7 +57,14 @@ export type WorkflowCustomNodeRunner = (
 export function resolveSeamName(node: { config?: Record<string, unknown> }): WorkflowSeamName | undefined {
   const seam = node.config?.seam;
   if (seam === undefined) return undefined;
-  if (seam === "planning" || seam === "execute" || seam === "review" || seam === "merge" || seam === "schedule") {
+  if (
+    seam === "planning" ||
+    seam === "execute" ||
+    seam === "review" ||
+    seam === "merge" ||
+    seam === "schedule" ||
+    seam === "step-execute"
+  ) {
     return seam;
   }
   throw new WorkflowIrError(`Unsupported workflow seam: ${String(seam)}`);
@@ -47,8 +80,27 @@ export function createPromptLikeHandler(
 ): WorkflowNodeHandler {
   return async (node, context) => {
     const seam = resolveSeamName(node);
+    if (seam === "step-execute") {
+      // Step-inversion (U3): step-execute resolves the active foreach instance
+      // from the reserved context key and runs exactly that step. The active
+      // context is set by the executor's foreach sub-walk on instance entry.
+      const active = context.context[FOREACH_ACTIVE_CONTEXT_KEY] as
+        | ForeachActiveContext
+        | undefined;
+      if (!active || typeof active.stepIndex !== "number") {
+        throw new WorkflowIrError(
+          `step-execute node '${node.id}' reached without an active foreach instance context`,
+        );
+      }
+      if (!seams.stepExecute) {
+        // Fail closed: a step-execute node with no seam wired must NOT silently
+        // succeed — that would merge a task with no step work done.
+        return { outcome: "failure", value: "step-execute-unwired" };
+      }
+      return seams.stepExecute(context.task, context.context);
+    }
     if (seam) {
-      return seams[seam](context.task, context.context);
+      return seams[seam]!(context.task, context.context);
     }
     if (!runCustomNode) {
       throw new WorkflowIrError(`No custom-node runner registered for node: ${node.id}`);
@@ -91,15 +143,33 @@ export function createGateHandler(runCustomNode?: WorkflowCustomNodeRunner): Wor
   };
 }
 
+/**
+ * Placeholder handler for the `step-review` node kind (KTD-4). The real verdict
+ * logic (delegating to `reviewStep`, mapping APPROVE/REVISE/RETHINK/UNAVAILABLE
+ * to outcome edges, and triggering RETHINK reset on rework traversal) is U5, NOT
+ * U3. Until U5 wires it, a step-review node reached during a foreach instance
+ * fails cleanly with a documented not-implemented value rather than throwing an
+ * unhandled-node-kind error — keeping a foreach with a step-review node from
+ * crashing the walk while making the gap explicit and routable.
+ */
+export const stepReviewNotImplementedHandler: WorkflowNodeHandler = async (node) => ({
+  outcome: "failure",
+  value: "step-review-not-implemented",
+  contextPatch: {
+    [`node:${node.id}:error`]: "step-review handler is not implemented until U5",
+  },
+});
+
 export function createDefaultNodeHandlers(
   seams: WorkflowLegacySeams,
   runCustomNode?: WorkflowCustomNodeRunner,
-): Record<"prompt" | "script" | "gate", WorkflowNodeHandler> {
+): Record<"prompt" | "script" | "gate" | "step-review", WorkflowNodeHandler> {
   const promptLike = createPromptLikeHandler(seams, runCustomNode);
   return {
     prompt: promptLike,
     script: promptLike,
     gate: createGateHandler(runCustomNode),
+    "step-review": stepReviewNotImplementedHandler,
   };
 }