From a782e5c04ce67d9c15b885a4a1d008b0b50216d2 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 4 Jun 2026 12:03:29 -0700 Subject: [PATCH] =?UTF-8?q?feat(engine):=20U3=20=E2=80=94=20foreach=20expa?= =?UTF-8?q?nsion,=20iterative=20instance=20sub-walk,=20bounded=20rework=20?= =?UTF-8?q?cycles,=20step-execute=20seam?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 (1M context) --- .../workflow-graph-executor-parity.test.ts | 2 +- .../__tests__/workflow-graph-foreach.test.ts | 497 ++++++++++++++++++ packages/engine/src/executor.ts | 49 +- .../engine/src/workflow-graph-executor.ts | 68 ++- packages/engine/src/workflow-graph-foreach.ts | 448 ++++++++++++++++ packages/engine/src/workflow-node-handlers.ts | 78 ++- 6 files changed, 1134 insertions(+), 8 deletions(-) create mode 100644 packages/engine/src/__tests__/workflow-graph-foreach.test.ts create mode 100644 packages/engine/src/workflow-graph-foreach.ts diff --git a/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts b/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts index eff08dc519..e16d871c34 100644 --- a/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts +++ b/packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts @@ -42,7 +42,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => { const legacyEvents = await runLegacy(seams)(); const executor = new WorkflowGraphExecutor({ seams, handlers: { prompt: async (node, ctx) => { const seam = String(node.config?.seam); - const result = await seams[seam as keyof WorkflowLegacySeams](ctx.task, ctx.context); + const result = await seams[seam as keyof WorkflowLegacySeams]!(ctx.task, ctx.context); events.push(`${seam}:${result.outcome}`); return result; } } }); diff --git a/packages/engine/src/__tests__/workflow-graph-foreach.test.ts b/packages/engine/src/__tests__/workflow-graph-foreach.test.ts new file mode 100644 index 0000000000..86c14448ad --- /dev/null +++ b/packages/engine/src/__tests__/workflow-graph-foreach.test.ts @@ -0,0 +1,497 @@ +import { describe, expect, it, vi } from "vitest"; +import type { TaskDetail, TaskStep, WorkflowIr, WorkflowIrNode } from "@fusion/core"; + +import { WorkflowGraphExecutor, type WorkflowNodeHandler } from "../workflow-graph-executor.js"; +import { + FOREACH_ACTIVE_CONTEXT_KEY, + type ForeachActiveContext, + type WorkflowLegacySeams, +} from "../workflow-node-handlers.js"; +import type { WorkflowStepInstanceState } from "../workflow-graph-foreach.js"; + +const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } }); + +/** Build a TaskDetail with a fixed step list. */ +function taskWithSteps(n: number): TaskDetail { + const steps: TaskStep[] = Array.from({ length: n }, (_, i) => ({ + name: `Step ${i + 1}`, + status: "pending" as const, + })); + 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. + */ +function foreachIr( + template: { nodes: WorkflowIrNode[]; edges: WorkflowIr["edges"] }, + opts: { + config?: Record; + extraNodes?: WorkflowIrNode[]; + foreachEdges?: WorkflowIr["edges"]; + } = {}, +): WorkflowIr { + return { + version: "v2", + name: "foreach-test", + columns: [{ id: "work", name: "Work", traits: [] }], + nodes: [ + { id: "start", kind: "start" }, + { + id: "fe", + kind: "foreach", + config: { source: "task-steps", template, ...(opts.config ?? {}) }, + }, + { id: "end", kind: "end" }, + ...(opts.extraNodes ?? []), + ], + edges: [ + { from: "start", to: "fe" }, + { from: "fe", to: "end", condition: "success" }, + ...(opts.foreachEdges ?? []), + ], + }; +} + +/** A single-node template: one step-execute prompt. */ +function singleExecuteTemplate() { + return { + nodes: [{ id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }], + edges: [], + }; +} + +describe("WorkflowGraphExecutor foreach (U3)", () => { + it("3-step expansion runs instances in step order, all 3 template-node instances", async () => { + const order: string[] = []; + const seams = baseSeams({ + stepExecute: async (_t, ctx) => { + const active = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext; + order.push(`exec#${active.stepIndex}`); + return { outcome: "success", value: "step-done" }; + }, + }); + const executor = new WorkflowGraphExecutor({ seams }); + const result = await executor.run(taskWithSteps(3), settingsOn(), foreachIr(singleExecuteTemplate())); + + expect(result.outcome).toBe("success"); + expect(order).toEqual(["exec#0", "exec#1", "exec#2"]); + // Instance ids are materialized deterministically. + expect(result.visitedNodeIds).toEqual( + expect.arrayContaining(["fe#0:exec", "fe#1:exec", "fe#2:exec"]), + ); + // The foreach itself is visited and routes its success edge to end (end is + // intentionally not pushed to visited — same posture as other tail edges). + expect(result.visitedNodeIds).toContain("fe"); + }); + + it("zero steps → foreach traverses its success edge without running any instance", async () => { + const exec = vi.fn(async () => ({ outcome: "success" as const })); + const seams = baseSeams({ stepExecute: exec }); + const executor = new WorkflowGraphExecutor({ seams }); + const result = await executor.run(taskWithSteps(0), 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("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). + let reviewCalls = 0; + const template = { + nodes: [ + { id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }, + { id: "review", kind: "prompt" as const, config: {} }, + ], + edges: [ + { from: "exec", to: "review", condition: "success" }, + // rework loop back to exec when review says "revise" + { from: "review", to: "exec", condition: "outcome:revise", kind: "rework" as const }, + // success/approve exits (no outgoing edge → template exit) + ], + }; + const reviewHandler: WorkflowNodeHandler = async () => { + reviewCalls += 1; + if (reviewCalls <= 2) return { outcome: "success", value: "revise" }; + return { outcome: "success", value: "approve" }; + }; + const seams = baseSeams({ + stepExecute: async () => ({ outcome: "success", value: "step-done" }), + }); + const executor = new WorkflowGraphExecutor({ + seams, + handlers: { prompt: makePromptRouter(seams, { review: reviewHandler }) }, + }); + const result = await executor.run(taskWithSteps(1), settingsOn(), foreachIr(template)); + + expect(result.outcome).toBe("success"); + expect(reviewCalls).toBe(3); // 2 revises + 1 approve + }); + + it("rework exhaustion routes the outcome:rework-exhausted edge", async () => { + // review always says revise → budget (2) exhausts → foreach emits + // rework-exhausted, routed to a hold node. + const template = { + nodes: [ + { id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }, + { id: "review", kind: "prompt" as const, config: {} }, + ], + edges: [ + { from: "exec", to: "review", condition: "success" }, + { from: "review", to: "exec", condition: "outcome:revise", kind: "rework" as const }, + ], + }; + const reviewHandler: WorkflowNodeHandler = async () => ({ outcome: "success", value: "revise" }); + const holdHandler = vi.fn(async () => ({ outcome: "success" as const })); + const seams = baseSeams({ + stepExecute: async () => ({ outcome: "success", value: "step-done" }), + }); + const executor = new WorkflowGraphExecutor({ + seams, + handlers: { + prompt: makePromptRouter(seams, { review: reviewHandler }), + hold: holdHandler, + }, + }); + const result = await executor.run( + taskWithSteps(1), + settingsOn(), + foreachIr(template, { + config: { maxReworkCycles: 2 }, + extraNodes: [{ id: "exhausted-hold", kind: "hold" }], + foreachEdges: [ + { from: "fe", to: "exhausted-hold", condition: "outcome:rework-exhausted" }, + { from: "exhausted-hold", to: "end", condition: "success" }, + ], + }), + ); + + expect(holdHandler).toHaveBeenCalledTimes(1); + expect(result.visitedNodeIds).toContain("exhausted-hold"); + }); + + it("rework exhaustion with NO routed edge falls back to failure", async () => { + const template = { + nodes: [ + { id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }, + { id: "review", kind: "prompt" as const, config: {} }, + ], + edges: [ + { from: "exec", to: "review", condition: "success" }, + { from: "review", to: "exec", condition: "outcome:revise", kind: "rework" as const }, + ], + }; + const reviewHandler: WorkflowNodeHandler = async () => ({ outcome: "success", value: "revise" }); + const seams = baseSeams({ + stepExecute: async () => ({ outcome: "success", value: "step-done" }), + }); + const executor = new WorkflowGraphExecutor({ + seams, + handlers: { prompt: makePromptRouter(seams, { review: reviewHandler }) }, + }); + const result = await executor.run( + taskWithSteps(1), + settingsOn(), + foreachIr(template, { config: { maxReworkCycles: 1 } }), + ); + + expect(result.outcome).toBe("failure"); + }); + + it("rework budget is per-instance, not shared across instances", async () => { + // 2 steps, budget 1 each. Each instance reworks exactly once then approves. + // If the budget were shared, the second instance would exhaust on its first + // rework. Per-instance, both succeed. + const reviewCallsByStep = new Map(); + const template = { + nodes: [ + { id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }, + { id: "review", kind: "prompt" as const, config: {} }, + ], + edges: [ + { from: "exec", to: "review", condition: "success" }, + { from: "review", to: "exec", condition: "outcome:revise", kind: "rework" as const }, + ], + }; + const reviewHandler: WorkflowNodeHandler = async (_node, ctx) => { + const active = ctx.context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext; + const n = (reviewCallsByStep.get(active.stepIndex) ?? 0) + 1; + reviewCallsByStep.set(active.stepIndex, n); + if (n === 1) return { outcome: "success", value: "revise" }; // 1 rework per step + return { outcome: "success", value: "approve" }; + }; + const seams = baseSeams({ + stepExecute: async () => ({ outcome: "success", value: "step-done" }), + }); + const executor = new WorkflowGraphExecutor({ + seams, + handlers: { prompt: makePromptRouter(seams, { review: reviewHandler }) }, + }); + const result = await executor.run( + taskWithSteps(2), + settingsOn(), + foreachIr(template, { config: { maxReworkCycles: 1 } }), + ); + + expect(result.outcome).toBe("success"); + expect(reviewCallsByStep.get(0)).toBe(2); + expect(reviewCallsByStep.get(1)).toBe(2); + }); + + it("a non-rework cycle outside an active instance still throws (recursive detector untouched)", async () => { + // Top-level graph with a plain cycle (no rework kind) — the recursive walk's + // inStack detector must still throw. + const ir: WorkflowIr = { + version: "v2", + name: "cycle", + columns: [{ id: "w", name: "W", traits: [] }], + nodes: [ + { id: "start", kind: "start" }, + { id: "a", kind: "prompt", config: {} }, + { id: "b", kind: "prompt", config: {} }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "a" }, + { from: "a", to: "b", condition: "success" }, + { from: "b", to: "a", condition: "success" }, // non-rework cycle + ], + }; + const executor = new WorkflowGraphExecutor({ + handlers: { prompt: async () => ({ outcome: "success" as const }) }, + }); + await expect(executor.run(taskWithSteps(0), settingsOn(), ir)).rejects.toThrow(/Cycle detected/); + }); + + it("abort mid-instance stops cleanly (signal honored between nodes)", async () => { + const controller = new AbortController(); + const seen: string[] = []; + // Template: exec → second. exec aborts the controller; `second` must not run + // (abort is checked at the top of the loop before the next node). + const template = { + nodes: [ + { id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }, + { id: "second", kind: "prompt" as const, config: {} }, + ], + edges: [{ from: "exec", to: "second", condition: "success" }], + }; + const secondHandler: WorkflowNodeHandler = async () => { + seen.push("second"); + return { outcome: "success" }; + }; + const seams = baseSeams({ + stepExecute: async () => { + seen.push("exec"); + controller.abort(); + return { outcome: "success", value: "step-done" }; + }, + }); + const executor = new WorkflowGraphExecutor({ + seams, + handlers: { prompt: makePromptRouter(seams, { second: secondHandler }) }, + signal: controller.signal, + }); + const result = await executor.run(taskWithSteps(2), settingsOn(), foreachIr(template)); + + expect(result.outcome).toBe("failure"); + expect(seen).toEqual(["exec"]); // second never ran; instance 1 never started + }); + + it("foreach:active context is visible to template handlers and absent outside instances", async () => { + const insideValues: Array = []; + let outsideAfter: unknown = "unset"; + // Template node records the active stepIndex; a tail node after the foreach + // asserts the key was cleared. + const seams = baseSeams({ + stepExecute: async (_t, ctx) => { + const active = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined; + insideValues.push(active?.stepIndex); + return { outcome: "success", value: "step-done" }; + }, + }); + const tailHandler: WorkflowNodeHandler = async (_node, ctx) => { + outsideAfter = ctx.context[FOREACH_ACTIVE_CONTEXT_KEY]; + return { outcome: "success" }; + }; + const executor = new WorkflowGraphExecutor({ + seams, + handlers: { prompt: makePromptRouter(seams, { tail: tailHandler }) }, + }); + const ir = foreachIr(singleExecuteTemplate(), { + extraNodes: [{ id: "tail", kind: "prompt", config: {} }], + foreachEdges: [ + { from: "fe", to: "tail", condition: "success" }, + { from: "tail", to: "end", condition: "success" }, + ], + }); + // Remove the direct fe→end edge so fe→tail is the only success route. + ir.edges = ir.edges.filter((e) => !(e.from === "fe" && e.to === "end")); + const result = await executor.run(taskWithSteps(2), settingsOn(), ir); + + expect(result.outcome).toBe("success"); + expect(insideValues).toEqual([0, 1]); + expect(outsideAfter).toBeUndefined(); // cleared on instance exit + }); + + it("step-execute seam is invoked with the correct stepIndex and captured baseline flows into context", async () => { + const captured: Array<{ stepIndex: number; baseline?: string }> = []; + // step-execute sets a baseline; a following review node reads it from the + // active context to prove the capture threads forward within the instance. + const template = { + nodes: [ + { id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }, + { id: "review", kind: "prompt" as const, config: {} }, + ], + edges: [{ from: "exec", to: "review", condition: "success" }], + }; + const seams = baseSeams({ + stepExecute: async (_t, ctx) => { + const active = ctx[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext; + active.baselineSha = `sha-for-${active.stepIndex}`; + active.checkpointId = `ckpt-${active.stepIndex}`; + return { + outcome: "success", + value: "step-done", + contextPatch: { [FOREACH_ACTIVE_CONTEXT_KEY]: active }, + }; + }, + }); + const reviewHandler: WorkflowNodeHandler = async (_node, ctx) => { + const active = ctx.context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext; + captured.push({ stepIndex: active.stepIndex, baseline: active.baselineSha }); + return { outcome: "success" }; + }; + const executor = new WorkflowGraphExecutor({ + seams, + handlers: { prompt: makePromptRouter(seams, { review: reviewHandler }) }, + }); + const result = await executor.run(taskWithSteps(2), settingsOn(), foreachIr(template)); + + expect(result.outcome).toBe("success"); + expect(captured).toEqual([ + { stepIndex: 0, baseline: "sha-for-0" }, + { stepIndex: 1, baseline: "sha-for-1" }, + ]); + }); + + it("step-execute with no seam wired fails closed (does not silently succeed)", async () => { + // No stepExecute seam provided → step-execute node fails with a clear value. + const seams = baseSeams({}); + const executor = new WorkflowGraphExecutor({ seams }); + const result = await executor.run(taskWithSteps(1), settingsOn(), foreachIr(singleExecuteTemplate())); + expect(result.outcome).toBe("failure"); + }); + + it("parallel mode is guarded with a clear not-yet-wired failure (U10 replaces it)", async () => { + const seams = baseSeams({ + stepExecute: async () => ({ outcome: "success", value: "step-done" }), + }); + const executor = new WorkflowGraphExecutor({ seams }); + const result = await executor.run( + taskWithSteps(2), + settingsOn(), + foreachIr(singleExecuteTemplate(), { config: { mode: "parallel", concurrency: 2 } }), + ); + expect(result.outcome).toBe("failure"); + expect(result.context["node:fe:value"]).toBe("parallel-not-wired"); + }); + + it("getTaskSteps dep is used to read a fresh count when injected", async () => { + const exec = vi.fn(async () => ({ outcome: "success" as const, value: "step-done" })); + const seams = baseSeams({ stepExecute: exec }); + // task.steps is empty, but the injected accessor returns 2 steps. + const executor = new WorkflowGraphExecutor({ + seams, + getTaskSteps: () => [ + { name: "fresh-1", status: "pending" }, + { name: "fresh-2", status: "pending" }, + ], + }); + const result = await executor.run(taskWithSteps(0), settingsOn(), foreachIr(singleExecuteTemplate())); + expect(result.outcome).toBe("success"); + expect(exec).toHaveBeenCalledTimes(2); + }); + + it("step instance persistence hook is called at start/completion/rework (no-op default safe)", async () => { + const saved: WorkflowStepInstanceState[] = []; + const template = { + nodes: [ + { id: "exec", kind: "prompt" as const, config: { seam: "step-execute" } }, + { id: "review", kind: "prompt" as const, config: {} }, + ], + edges: [ + { from: "exec", to: "review", condition: "success" }, + { from: "review", to: "exec", condition: "outcome:revise", kind: "rework" as const }, + ], + }; + let reviewCalls = 0; + const reviewHandler: WorkflowNodeHandler = async () => { + reviewCalls += 1; + return reviewCalls === 1 + ? { outcome: "success", value: "revise" } + : { outcome: "success", value: "approve" }; + }; + const seams = baseSeams({ + stepExecute: async () => ({ outcome: "success", value: "step-done" }), + }); + const executor = new WorkflowGraphExecutor({ + seams, + handlers: { prompt: makePromptRouter(seams, { review: reviewHandler }) }, + stepInstancePersistence: { + saveInstanceState: (s) => { + saved.push({ ...s }); + }, + }, + }); + const result = await executor.run( + taskWithSteps(1), + settingsOn(), + foreachIr(template, { config: { maxReworkCycles: 2 } }), + ); + + expect(result.outcome).toBe("success"); + // in-progress at start, a rework in-progress bump, and a final completed. + expect(saved.some((s) => s.status === "in-progress" && s.reworkCount === 0)).toBe(true); + expect(saved.some((s) => s.status === "in-progress" && s.reworkCount === 1)).toBe(true); + expect(saved.some((s) => s.status === "completed")).toBe(true); + expect(saved.every((s) => s.pinnedStepCount === 1)).toBe(true); + }); +}); + +// ── helpers ─────────────────────────────────────────────────────────────── + +/** Base no-op seams with an optional override (stepExecute etc.). */ +function baseSeams(overrides: Partial): WorkflowLegacySeams { + const ok = async () => ({ outcome: "success" as const }); + return { + planning: ok, + execute: ok, + review: ok, + merge: ok, + schedule: ok, + ...overrides, + }; +} + +/** + * A prompt handler that dispatches: step-execute seam → seams.stepExecute; + * otherwise to a per-node-id custom handler map (review/tail/second/etc.). + */ +function makePromptRouter( + seams: WorkflowLegacySeams, + byId: Record, +): WorkflowNodeHandler { + return async (node, ctx) => { + if (node.config?.seam === "step-execute") { + if (!seams.stepExecute) return { outcome: "failure", value: "step-execute-unwired" }; + return seams.stepExecute(ctx.task, ctx.context); + } + const handler = byId[node.id]; + if (handler) return handler(node, ctx); + return { outcome: "success" }; + }; +} diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index c94f134d78..b53ead6215 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -19,7 +19,11 @@ import { import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js"; import type { WorkflowBranchPersistence, WorkflowBranchRunState } from "./workflow-graph-branches.js"; import { observeWorkflowParity, WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from "./workflow-parity-observer.js"; -import type { WorkflowLegacySeams } from "./workflow-node-handlers.js"; +import { + FOREACH_ACTIVE_CONTEXT_KEY, + type ForeachActiveContext, + type WorkflowLegacySeams, +} from "./workflow-node-handlers.js"; import type { WorkflowNodeResult } from "./workflow-graph-executor.js"; import { ApprovalRequestStore, @@ -103,7 +107,7 @@ import type { StuckTaskDetector, StuckTaskEvent } from "./stuck-task-detector.js import type { PluginRunner } from "./plugin-runner.js"; import { isContextLimitError } from "./context-limit-detector.js"; import { StepSessionExecutor } from "./step-session-executor.js"; -import { resetStepToBaseline } from "./step-runner.js"; +import { resetStepToBaseline, runTaskStep } from "./step-runner.js"; import { acquireTaskWorktree } from "./worktree-acquisition.js"; import { resolveCapturedBaseCommitSha } from "./base-commit-capture.js"; import { installTaskWorktreeIdentityGuard } from "./worktree-hooks.js"; @@ -3519,6 +3523,47 @@ export class TaskExecutor { } }, schedule: async () => ({ outcome: "success" }), + // Step-inversion (KTD-2/KTD-4, U3): run exactly the foreach-active step. + // The foreach sub-walk has set `foreach:active` with the step index; here + // we drive runTaskStep (step-runner.ts) over the task's worktree, then + // capture the per-step baselineSha/checkpointId back INTO the active + // context object so a later RETHINK (U5) can reset the step. The full + // single-step session physics (a StepSessionExecutor scoped to one step) + // is U5/U7 territory; U3 wires the seam and the context capture, using the + // existing implementation phase as the single-pass step driver. + stepExecute: async (seamTask, context) => { + const active = context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined; + if (!active || typeof active.stepIndex !== "number") { + return { outcome: "failure", value: "no-active-step-instance" }; + } + const live = await this.store.getTask(seamTask.id); + const worktreePath = live.worktree || this.rootDir; + const result = await runTaskStep( + { + store: this.store, + worktreePath, + // Single-pass step driver. The agent authors the step's commit; this + // only observes (KTD-2). Refined to per-step session physics in U5/U7. + runStep: async () => { + const phase = await this.runImplementationPhase(seamTask); + return { success: phase.taskDone }; + }, + }, + { id: seamTask.id, steps: live.steps }, + active.stepIndex, + ); + // Capture baseline/checkpoint back into the reserved active context so the + // foreach sub-walk threads them to later template nodes (step-review/reset). + active.baselineSha = result.baselineSha; + active.checkpointId = result.checkpointId; + return { + outcome: result.outcome, + value: result.outcome === "success" ? "step-done" : "step-failed", + contextPatch: { + [FOREACH_ACTIVE_CONTEXT_KEY]: active, + }, + }; + }, }; } diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index fe64000cbc..32e34e5ce3 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -1,4 +1,4 @@ -import type { Settings, TaskDetail, WorkflowIr, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core"; +import type { Settings, TaskDetail, TaskStep, WorkflowIr, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core"; import { BUILTIN_CODING_WORKFLOW_IR, WorkflowIrError, isExperimentalFeatureEnabled } from "@fusion/core"; import { @@ -15,6 +15,10 @@ import { type WorkflowBranchRunState, type WorkflowBranchSemaphore, } from "./workflow-graph-branches.js"; +import { + runForeach, + type WorkflowStepInstancePersistence, +} from "./workflow-graph-foreach.js"; export type WorkflowNodeOutcome = "success" | "failure"; @@ -50,6 +54,28 @@ export interface WorkflowGraphExecutorDeps { onBranchProgress?: (progress: WorkflowBranchProgress) => void; /** Stable identifier for this run, used to key persisted branch state. */ runId?: string; + /** + * Step-inversion (KTD-3, U3): fresh `Task.steps[]` accessor used by a `foreach` + * node at expansion time. Defaults to reading `task.steps` off the run's task. + * A production caller may inject a fresh store fetch so the count reflects the + * planning seam's latest write; tests inject a fixed list. + */ + getTaskSteps?: (task: TaskDetail) => Promise | TaskStep[]; + /** + * Step-inversion (KTD-6, U3 stub): per-instance run-state persistence for + * foreach instances. Optional with no-op default — the real SQLite adapter is + * U4's executor-half wiring; the sub-walk already calls into this so that + * wiring is purely additive. + */ + stepInstancePersistence?: WorkflowStepInstancePersistence; + /** + * Step-inversion (U3): top-level abort signal honored between foreach instance + * nodes (existing posture, mirrors the branch path's per-branch signal). When a + * run is cancelled (pause/abort), the in-flight instance stops cleanly between + * nodes and the foreach fails with `value: "aborted"`. Undefined on normal + * runs (zero behavior change for non-foreach graphs). + */ + signal?: AbortSignal; } export interface WorkflowGraphExecutorResult { @@ -171,6 +197,34 @@ export class WorkflowGraphExecutor { ); } + if (node.kind === "foreach") { + // Step-inversion (KTD-3/KTD-5, U3): expand the foreach into per-step + // instances run through an iterative region sub-walk. The recursive + // walk's inStack cycle detector is untouched — rework loops are + // expressed inside the sub-walk only. The foreach node's own outcome + // routes its outgoing edges (success / outcome:rework-exhausted / ...). + const steps = await this.resolveTaskSteps(task); + const foreachResult = await runForeach(node, { + task, + runId, + steps, + context, + runTemplateNode: (tNode, sig) => + this.executeNodeWithRetries(tNode, task, settings, context, sig), + shouldTraverseEdge: (edge, src) => this.shouldTraverseEdge(edge, src), + persistence: this.deps.stepInstancePersistence, + signal: this.deps.signal, + }); + visitedNodeIds.push(...foreachResult.visitedNodeIds); + const result: WorkflowNodeResult = { + outcome: foreachResult.outcome, + value: foreachResult.value, + }; + context[`node:${node.id}:outcome`] = result.outcome; + if (result.value !== undefined) context[`node:${node.id}:value`] = result.value; + return await traverseChildren(node, result); + } + const result = await this.executeNodeWithRetries(node, task, settings, context); if (result.contextPatch) Object.assign(context, result.contextPatch); context[`node:${node.id}:outcome`] = result.outcome; @@ -222,6 +276,18 @@ export class WorkflowGraphExecutor { }; } + /** + * Resolve the task's step list for a foreach expansion (KTD-3). Defaults to + * the steps already on the run's task; a caller may inject `getTaskSteps` to + * fetch fresh state (e.g. after the planning seam populated steps). + */ + private async resolveTaskSteps(task: TaskDetail): Promise { + if (this.deps.getTaskSteps) { + return await this.deps.getTaskSteps(task); + } + return task.steps ?? []; + } + /** Best-effort prune of stale-run branch rows; never throws into the run. */ private async pruneStaleBranches(taskId: string, keepRunId: string): Promise { try { diff --git a/packages/engine/src/workflow-graph-foreach.ts b/packages/engine/src/workflow-graph-foreach.ts new file mode 100644 index 0000000000..1b442c12ba --- /dev/null +++ b/packages/engine/src/workflow-graph-foreach.ts @@ -0,0 +1,448 @@ +import type { TaskDetail, TaskStep, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core"; +import { WorkflowIrError } from "@fusion/core"; + +import type { WorkflowNodeOutcome, WorkflowNodeResult } from "./workflow-graph-executor.js"; +import { + FOREACH_ACTIVE_CONTEXT_KEY, + type ForeachActiveContext, +} from "./workflow-node-handlers.js"; +import { schedulerLog } from "./logger.js"; + +/** + * Foreach region expansion + instance sub-walk (step-inversion KTD-3/KTD-5, U3). + * + * When the sequential walker reaches a `foreach` node it does NOT recurse through + * the main `walk` (whose `inStack` cycle detector intentionally throws on any + * back-edge). Instead it hands control here, which: + * + * - reads `Task.steps[]` and pins the count at expansion time; + * - for each step `i` in order, runs the inline template subgraph as an + * **iterative region sub-walk** (a `for(;;)` over `currentId`, modeled on + * `walkBranch` in workflow-graph-branches.ts), from the template entry to its + * exit, materializing deterministic instance node ids + * `#:` purely as walk state (the IR/nodeMap are + * never mutated); + * - permits `kind: "rework"` edges as the only legal cycles — each traversal + * decrements a per-instance budget seeded from `config.maxReworkCycles` + * (default 3, defensively clamped to ≤10); exhaustion emits the + * `outcome:rework-exhausted` outcome from the foreach node; + * - threads the active instance under the reserved `foreach:active` context key + * so template handlers (step-execute now; step-review in U5) know which step + * they operate on, clearing it on instance exit; + * - honors the abort signal between nodes (existing posture). + * + * Only sequential + shared physics are implemented here (concurrency 1). The + * scheduler is intentionally a runnable-set loop running one instance at a time + * so U10 can extend it to parallel/worktree without restructuring. Parallel mode + * is guarded to a clean failure (U10 replaces it). + */ + +/** Default rework budget when the foreach config omits `maxReworkCycles`. */ +const DEFAULT_MAX_REWORK_CYCLES = 3; +/** Defensive cap mirroring core's validation clamp (KTD-5). */ +const MAX_REWORK_CYCLES_CAP = 10; + +/** The foreach node's config shape this module reads (subset of WorkflowForeachConfig). */ +interface ForeachConfig { + source?: unknown; + maxReworkCycles?: number; + mode?: "sequential" | "parallel"; + concurrency?: number; + isolation?: "shared" | "worktree"; + template?: { nodes: WorkflowIrNode[]; edges: WorkflowIrEdge[] }; +} + +/** + * Narrow persistence hook for foreach instance run-state (KTD-6, U3 stub). + * + * The real SQLite-backed adapter lands in U4 (executor half); this interface is + * shaped so that wiring is a pure additive change. All methods are optional and + * default to no-ops — the sub-walk calls them at instance start / completion / + * each rework pass, but a fully in-memory run (tests, flag-off, pre-U4 store) + * needs none of them. Instance identity is deterministic + * (`#`), so a future resume can seed the sub-walk + * position directly from a loaded `currentNodeId` + `reworkCount` (KTD-6) — + * this hook is the seam where that seeding will plug in. + */ +export interface WorkflowStepInstanceState { + taskId: string; + runId: string; + foreachNodeId: string; + stepIndex: number; + pinnedStepCount: number; + /** Template node id (NOT the materialized instance id) the instance is at. */ + currentNodeId: string; + status: "in-progress" | "completed" | "failed"; + baselineSha?: string; + checkpointId?: string; + reworkCount: number; +} + +export interface WorkflowStepInstancePersistence { + /** Idempotent upsert keyed by (taskId, runId, foreachNodeId, stepIndex). */ + saveInstanceState?(state: WorkflowStepInstanceState): void | Promise; + /** Load any persisted instance states for a run (used on resume — U4). */ + loadInstanceStates?( + taskId: string, + runId: string, + ): WorkflowStepInstanceState[] | Promise; + /** Prune stale instance rows for a task, keeping only `keepRunId` (U4). */ + clearStaleInstanceStates?(taskId: string, keepRunId: string): void | Promise; +} + +/** + * Await a persistence call inside a guard so a Promise-returning impl cannot + * escape as an unhandled rejection, and a persistence failure never kills + * instance execution (log-and-continue). Mirrors `persistBranchState`. + */ +async function persistInstanceState( + persistence: WorkflowStepInstancePersistence | undefined, + state: WorkflowStepInstanceState, +): Promise { + try { + await persistence?.saveInstanceState?.(state); + } catch (err) { + const message = err instanceof Error ? err.message : String(err); + schedulerLog.warn( + `saveInstanceState failed for task ${state.taskId} run ${state.runId} foreach ${state.foreachNodeId} step ${state.stepIndex}: ${message}`, + ); + } +} + +export interface ForeachEnvironment { + task: TaskDetail; + runId: string; + /** Fresh step list (KTD-3: read at expansion, count pinned). */ + steps: TaskStep[]; + /** The shared walk context; the active-instance key is threaded in/out of it. */ + context: Record; + /** + * Runs one template node through the executor's executeNodeWithRetries (so + * per-node maxRetries still applies inside the sub-walk). The node passed is + * the ORIGINAL template node; the executor reads/writes the shared context, + * which already carries `foreach:active` for the current instance. + */ + runTemplateNode: ( + node: WorkflowIrNode, + signal?: AbortSignal, + ) => Promise; + shouldTraverseEdge: (edge: WorkflowIrEdge, source: WorkflowNodeResult) => boolean; + persistence?: WorkflowStepInstancePersistence; + /** Honored between nodes (existing posture). */ + signal?: AbortSignal; +} + +export interface ForeachRunResult { + /** Foreach node outcome: success when all instances completed; otherwise the + * routed outcome value (e.g. "rework-exhausted") with a failure outcome unless + * the caller routes it. */ + outcome: WorkflowNodeOutcome; + /** Outcome value for `outcome:` edge routing (e.g. "rework-exhausted"). */ + value?: string; + /** Materialized instance node ids visited, for the executor's visited list. */ + visitedNodeIds: string[]; +} + +/** Materialize a deterministic instance node id (KTD-3) — pure, no IR mutation. */ +export function instanceNodeId(foreachNodeId: string, stepIndex: number, templateNodeId: string): string { + return `${foreachNodeId}#${stepIndex}:${templateNodeId}`; +} + +/** Resolve the foreach config, validating the bits this module relies on. */ +function resolveForeachConfig(node: WorkflowIrNode): { + template: { nodes: WorkflowIrNode[]; edges: WorkflowIrEdge[] }; + maxReworkCycles: number; + mode: "sequential" | "parallel"; +} { + const cfg = (node.config ?? {}) as ForeachConfig; + const template = cfg.template; + if (!template || !Array.isArray(template.nodes) || !Array.isArray(template.edges)) { + throw new WorkflowIrError(`foreach node '${node.id}' has no template subgraph`); + } + const raw = typeof cfg.maxReworkCycles === "number" ? cfg.maxReworkCycles : DEFAULT_MAX_REWORK_CYCLES; + const maxReworkCycles = Math.max(1, Math.min(MAX_REWORK_CYCLES_CAP, Math.floor(raw))); + const mode = cfg.mode === "parallel" ? "parallel" : "sequential"; + return { template, maxReworkCycles, mode }; +} + +/** Find the single template entry node (no non-rework incoming edge). */ +function findTemplateEntry( + nodes: WorkflowIrNode[], + edges: WorkflowIrEdge[], + foreachId: string, +): WorkflowIrNode { + const incoming = new Map(); + for (const edge of edges) { + if (edge.kind === "rework") continue; + incoming.set(edge.to, (incoming.get(edge.to) ?? 0) + 1); + } + const entries = nodes.filter((n) => (incoming.get(n.id) ?? 0) === 0); + if (entries.length !== 1) { + throw new WorkflowIrError( + `foreach node '${foreachId}' template must have exactly one entry node (found ${entries.length})`, + ); + } + return entries[0]; +} + +/** + * Expand a foreach node and run its instances sequentially in step order. + * Returns the foreach node's aggregate outcome (KTD-3). + */ +export async function runForeach( + foreachNode: WorkflowIrNode, + env: ForeachEnvironment, +): Promise { + const { template, maxReworkCycles, mode } = resolveForeachConfig(foreachNode); + + // U3 scope guard: parallel mode is U10. Fail cleanly with a routable outcome + // rather than silently running it as sequential. + if (mode === "parallel") { + return { + outcome: "failure", + value: "parallel-not-wired", + visitedNodeIds: [], + }; + } + + // Pin the count at expansion (KTD-3). Zero steps → success edge (no instances). + const pinnedStepCount = env.steps.length; + const visitedNodeIds: string[] = []; + if (pinnedStepCount === 0) { + return { outcome: "success", visitedNodeIds }; + } + + const templateById = new Map(template.nodes.map((n) => [n.id, n])); + const templateOutgoing = new Map(); + for (const edge of template.edges) { + const list = templateOutgoing.get(edge.from) ?? []; + list.push(edge); + templateOutgoing.set(edge.from, list); + } + const entry = findTemplateEntry(template.nodes, template.edges, foreachNode.id); + + // Sequential + shared: a runnable-set loop with concurrency 1 (U10 extends this + // to parallel/worktree). Instances run strictly in step order. + for (let stepIndex = 0; stepIndex < pinnedStepCount; stepIndex++) { + if (env.signal?.aborted) { + return { outcome: "failure", value: "aborted", visitedNodeIds }; + } + + const instanceResult = await runInstance( + foreachNode, + stepIndex, + pinnedStepCount, + entry, + templateById, + templateOutgoing, + maxReworkCycles, + env, + visitedNodeIds, + ); + + if (instanceResult.outcome === "failure") { + // Rework exhaustion routes a dedicated outcome; other failures propagate. + return { + outcome: "failure", + value: instanceResult.value, + visitedNodeIds, + }; + } + } + + // All instances completed → foreach success edge (KTD-3). + return { outcome: "success", visitedNodeIds }; +} + +interface InstanceResult { + outcome: WorkflowNodeOutcome; + value?: string; +} + +/** + * Run one foreach instance (step `stepIndex`) as an iterative region sub-walk. + * Threads `foreach:active` into the shared context on entry and clears it on + * exit. Rework edges loop `currentId` back, bounded by the per-instance budget. + */ +async function runInstance( + foreachNode: WorkflowIrNode, + stepIndex: number, + pinnedStepCount: number, + entry: WorkflowIrNode, + templateById: Map, + templateOutgoing: Map, + maxReworkCycles: number, + env: ForeachEnvironment, + visitedNodeIds: string[], +): Promise { + // Per-instance rework budget (KTD-5) — NOT shared across instances. + let reworkBudget = maxReworkCycles; + let reworkCount = 0; + + // Active-instance context (KTD-3). baselineSha/checkpointId start undefined and + // are captured by step-execute (U3) into this same object so later template + // nodes (step-review/reset, U5) can read them. + const active: ForeachActiveContext = { + foreachNodeId: foreachNode.id, + stepIndex, + instanceId: `${foreachNode.id}#${stepIndex}`, + }; + env.context[FOREACH_ACTIVE_CONTEXT_KEY] = active; + + await persistInstanceState(env.persistence, { + taskId: env.task.id, + runId: env.runId, + foreachNodeId: foreachNode.id, + stepIndex, + pinnedStepCount, + currentNodeId: entry.id, + status: "in-progress", + baselineSha: active.baselineSha, + checkpointId: active.checkpointId, + reworkCount, + }); + + try { + let currentId = entry.id; + let lastResult: WorkflowNodeResult = { outcome: "success" }; + + for (;;) { + if (env.signal?.aborted) { + await persistInstanceState(env.persistence, { + taskId: env.task.id, + runId: env.runId, + foreachNodeId: foreachNode.id, + stepIndex, + pinnedStepCount, + currentNodeId: currentId, + status: "failed", + baselineSha: active.baselineSha, + checkpointId: active.checkpointId, + reworkCount, + }); + return { outcome: "failure", value: "aborted" }; + } + + const node = templateById.get(currentId); + if (!node) throw new WorkflowIrError(`Unknown foreach template node: ${currentId}`); + + visitedNodeIds.push(instanceNodeId(foreachNode.id, stepIndex, currentId)); + + lastResult = await env.runTemplateNode(node, env.signal); + // step-execute (and U5 nodes) write captured baseline/checkpoint into the + // active context via their contextPatch; mirror them onto `active` so the + // reserved key stays the single source of truth for later nodes. + syncActiveFromContext(env.context, active); + + if (lastResult.outcome === "failure") { + await persistInstanceState(env.persistence, { + taskId: env.task.id, + runId: env.runId, + foreachNodeId: foreachNode.id, + stepIndex, + pinnedStepCount, + currentNodeId: currentId, + status: "failed", + baselineSha: active.baselineSha, + checkpointId: active.checkpointId, + reworkCount, + }); + return { outcome: "failure", value: lastResult.value }; + } + + // Pick the next edge. Rework edges are the only legal back-edges. + const next = chooseNextEdge(currentId, templateOutgoing, lastResult, env.shouldTraverseEdge); + if (!next) { + // No outgoing edge matched → template exit reached. Instance complete. + await persistInstanceState(env.persistence, { + taskId: env.task.id, + runId: env.runId, + foreachNodeId: foreachNode.id, + stepIndex, + pinnedStepCount, + currentNodeId: currentId, + status: "completed", + baselineSha: active.baselineSha, + checkpointId: active.checkpointId, + reworkCount, + }); + return { outcome: "success" }; + } + + if (next.kind === "rework") { + if (reworkBudget <= 0) { + // Budget exhausted (KTD-5): emit rework-exhausted from the foreach node. + await persistInstanceState(env.persistence, { + taskId: env.task.id, + runId: env.runId, + foreachNodeId: foreachNode.id, + stepIndex, + pinnedStepCount, + currentNodeId: currentId, + status: "failed", + baselineSha: active.baselineSha, + checkpointId: active.checkpointId, + reworkCount, + }); + return { outcome: "failure", value: "rework-exhausted" }; + } + reworkBudget -= 1; + reworkCount += 1; + await persistInstanceState(env.persistence, { + taskId: env.task.id, + runId: env.runId, + foreachNodeId: foreachNode.id, + stepIndex, + pinnedStepCount, + currentNodeId: next.to, + status: "in-progress", + baselineSha: active.baselineSha, + checkpointId: active.checkpointId, + reworkCount, + }); + } + + currentId = next.to; + } + } finally { + // Clear the active-instance context on exit (KTD-3): absent outside instances. + delete env.context[FOREACH_ACTIVE_CONTEXT_KEY]; + } +} + +/** Sync baseline/checkpoint a handler wrote into the shared `foreach:active` + * context object back onto our local `active` snapshot. Handlers that patch the + * reserved key (step-execute) update the SAME object reference, but a handler + * could replace it via contextPatch — re-read defensively. */ +function syncActiveFromContext( + context: Record, + active: ForeachActiveContext, +): void { + const fromContext = context[FOREACH_ACTIVE_CONTEXT_KEY] as ForeachActiveContext | undefined; + if (fromContext && fromContext !== active) { + active.baselineSha = fromContext.baselineSha ?? active.baselineSha; + active.checkpointId = fromContext.checkpointId ?? active.checkpointId; + // Keep the canonical object reference stable for later nodes. + context[FOREACH_ACTIVE_CONTEXT_KEY] = active; + } +} + +/** + * Choose the single next edge from `nodeId`. A rework edge wins only when no + * non-rework edge matches the outcome (rework is the explicit loop-back, not a + * primary forward edge); among matching forward edges the lowest `to` id wins + * (deterministic, mirrors walkBranch/traverseChildren ordering). + */ +function chooseNextEdge( + nodeId: string, + templateOutgoing: Map, + source: WorkflowNodeResult, + shouldTraverseEdge: (edge: WorkflowIrEdge, source: WorkflowNodeResult) => boolean, +): WorkflowIrEdge | undefined { + const edges = (templateOutgoing.get(nodeId) ?? []).filter((e) => shouldTraverseEdge(e, source)); + if (edges.length === 0) return undefined; + const forward = edges.filter((e) => e.kind !== "rework").sort((a, b) => a.to.localeCompare(b.to)); + if (forward.length > 0) return forward[0]; + const rework = edges.filter((e) => e.kind === "rework").sort((a, b) => a.to.localeCompare(b.to)); + return rework[0]; +} diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts index 2abb1da2dc..f1223d57c5 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/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) => Promise; merge: (task: TaskDetail, context: Record) => Promise; schedule: (task: TaskDetail, context: Record) => Promise; + /** + * 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) => Promise; +} + +/** 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 }): 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, }; }