feat(FN-0000): add workflow loop nodes

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

@@ -0,0 +1,150 @@
+import { describe, expect, it, vi } from "vitest";
+import type { TaskDetail, WorkflowIr, WorkflowIrNode } from "@fusion/core";
+
+import { WorkflowGraphExecutor, type WorkflowNodeHandler } from "../workflow-graph-executor.js";
+
+const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } });
+const task = { id: "FN-LOOP" } as TaskDetail;
+
+function loopIr(config: Record<string, unknown>, extraEdges: WorkflowIr["edges"] = []): WorkflowIr {
+  return {
+    version: "v2",
+    name: "loop-test",
+    columns: [{ id: "work", name: "Work", traits: [] }],
+    nodes: [
+      { id: "start", kind: "start" },
+      {
+        id: "loop",
+        kind: "loop",
+        config: {
+          template: {
+            nodes: [
+              { id: "ask", kind: "prompt", config: { prompt: "try" } },
+              { id: "check", kind: "gate", config: { prompt: "done?" } },
+            ],
+            edges: [{ from: "ask", to: "check" }],
+          },
+          ...config,
+        },
+      },
+      { id: "exhausted", kind: "hold", config: { release: "manual" } },
+      { id: "end", kind: "end" },
+    ],
+    edges: [
+      { from: "start", to: "loop" },
+      { from: "loop", to: "end", condition: "success" },
+      ...extraEdges,
+    ],
+  };
+}
+
+describe("WorkflowGraphExecutor loop", () => {
+  it("exits successfully when the template output matches immediately", async () => {
+    const calls: string[] = [];
+    const prompt: WorkflowNodeHandler = async (node) => {
+      calls.push(node.id);
+      return { outcome: "success", value: node.id === "check" ? "DONE" : "working" };
+    };
+    const executor = new WorkflowGraphExecutor({ handlers: { prompt, gate: prompt } });
+
+    const result = await executor.run(
+      task,
+      settingsOn(),
+      loopIr({ maxIterations: 3, exitWhen: { type: "output-contains", value: "DONE" } }),
+    );
+
+    expect(result.outcome).toBe("success");
+    expect(calls).toEqual(["ask", "check"]);
+    expect(result.visitedNodeIds).toEqual(expect.arrayContaining(["loop", "loop#1:ask", "loop#1:check"]));
+    expect(result.context["node:loop:loop"]).toMatchObject({ iterations: 1, exitReason: "matched" });
+    expect(result.context["loop:active"]).toBeUndefined();
+  });
+
+  it("keeps iterating until the configured output string appears", async () => {
+    let checks = 0;
+    const handler: WorkflowNodeHandler = async (node) => {
+      if (node.id !== "check") return { outcome: "success", value: "working" };
+      checks += 1;
+      return { outcome: "success", value: checks === 3 ? "DONE" : "KEEP_GOING" };
+    };
+    const executor = new WorkflowGraphExecutor({
+      handlers: { prompt: handler, gate: handler, hold: async () => ({ outcome: "success" }) },
+    });
+
+    const result = await executor.run(
+      task,
+      settingsOn(),
+      loopIr({ maxIterations: 4, exitWhen: { type: "output-contains", value: "DONE" } }),
+    );
+
+    expect(result.outcome).toBe("success");
+    expect(checks).toBe(3);
+    expect(result.context["node:loop:loop"]).toMatchObject({ iterations: 3, exitReason: "matched" });
+    expect(result.context["node:check:value"]).toBe("DONE");
+  });
+
+  it("routes iteration exhaustion as a failure outcome value", async () => {
+    const handler = vi.fn(async () => ({ outcome: "success" as const, value: "not yet" }));
+    const executor = new WorkflowGraphExecutor({
+      handlers: { prompt: handler, gate: handler, hold: async () => ({ outcome: "success" }) },
+    });
+
+    const result = await executor.run(
+      task,
+      settingsOn(),
+      loopIr(
+        { maxIterations: 2, exitWhen: { type: "output-contains", value: "DONE" } },
+        [{ from: "loop", to: "exhausted", condition: "outcome:loop-iteration-exhausted" }],
+      ),
+    );
+
+    expect(result.outcome).toBe("success");
+    expect(handler).toHaveBeenCalledTimes(4);
+    expect(result.context["node:loop:outcome"]).toBe("failure");
+    expect(result.context["node:loop:value"]).toBe("loop-iteration-exhausted");
+  });
+
+  it("routes timeout as a failure outcome value", async () => {
+    let now = 0;
+    const handler: WorkflowNodeHandler = async () => {
+      now += 10;
+      return { outcome: "success", value: "not yet" };
+    };
+    const executor = new WorkflowGraphExecutor({
+      handlers: { prompt: handler, gate: handler, hold: async () => ({ outcome: "success" }) },
+      runLoopNowForTests: () => now,
+    });
+
+    const result = await executor.run(
+      task,
+      settingsOn(),
+      loopIr(
+        { maxIterations: 10, timeoutMs: 15, exitWhen: { type: "output-contains", value: "DONE" } },
+        [{ from: "loop", to: "exhausted", condition: "outcome:loop-timeout" }],
+      ),
+    );
+
+    expect(result.outcome).toBe("success");
+    expect(result.context["node:loop:value"]).toBe("loop-timeout");
+  });
+
+  it("can match a regex against a selected template node value", async () => {
+    const handler: WorkflowNodeHandler = async (node: WorkflowIrNode) => ({
+      outcome: "success",
+      value: node.id === "ask" ? "ticket READY-42" : "ignored",
+    });
+    const executor = new WorkflowGraphExecutor({ handlers: { prompt: handler, gate: handler } });
+
+    const result = await executor.run(
+      task,
+      settingsOn(),
+      loopIr({
+        maxIterations: 2,
+        exitWhen: { type: "output-matches", nodeId: "ask", pattern: "READY-\\d+" },
+      }),
+    );
+
+    expect(result.outcome).toBe("success");
+    expect(result.context["node:loop:loop"]).toMatchObject({ exitReason: "matched" });
+  });
+});

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

@@ -25,6 +25,7 @@ import {
   type ForeachEnvironment,
   type WorkflowStepInstancePersistence,
 } from "./workflow-graph-foreach.js";
+import { runLoop } from "./workflow-graph-loop.js";
 
 export type WorkflowNodeOutcome = "success" | "failure";
 
@@ -70,6 +71,8 @@ export interface WorkflowGraphExecutorDeps {
   onBranchProgress?: (progress: WorkflowBranchProgress) => void;
   /** Stable identifier for this run, used to key persisted branch state. */
   runId?: string;
+  /** Test seam for bounded loop timeout checks. Defaults to Date.now. */
+  runLoopNowForTests?: () => number;
   /**
    * 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.
@@ -338,6 +341,25 @@ export class WorkflowGraphExecutor {
           return await traverseChildren(node, result);
         }
 
+        if (node.kind === "loop") {
+          const loopResult = await runLoop(node, {
+            context,
+            runTemplateNode: (tNode, sig, contextOverride) =>
+              this.executeNodeWithRetries(tNode, task, settings, contextOverride ?? context, ir, sig),
+            shouldTraverseEdge: (edge, src) => this.shouldTraverseEdge(edge, src),
+            signal: this.deps.signal,
+            now: this.deps.runLoopNowForTests,
+          });
+          visitedNodeIds.push(...loopResult.visitedNodeIds);
+          const result: WorkflowNodeResult = {
+            outcome: loopResult.outcome,
+            value: loopResult.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, ir);
         if (result.contextPatch) Object.assign(context, result.contextPatch);
         context[`node:${node.id}:outcome`] = result.outcome;

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

@@ -0,0 +1,205 @@
+import type { WorkflowIrEdge, WorkflowIrNode, WorkflowLoopConfig } from "@fusion/core";
+import { WorkflowIrError } from "@fusion/core";
+
+import type { WorkflowNodeOutcome, WorkflowNodeResult } from "./workflow-graph-executor.js";
+
+const DEFAULT_MAX_ITERATIONS = 3;
+const MAX_ITERATIONS_CAP = 50;
+const DEFAULT_TIMEOUT_MS = 300_000;
+const MAX_TIMEOUT_MS = 3_600_000;
+
+interface LoopConfig {
+  template: { nodes: WorkflowIrNode[]; edges: WorkflowIrEdge[] };
+  exitWhen: WorkflowLoopConfig["exitWhen"];
+  maxIterations: number;
+  timeoutMs: number;
+}
+
+export interface LoopEnvironment {
+  context: Record<string, unknown>;
+  runTemplateNode: (
+    node: WorkflowIrNode,
+    signal?: AbortSignal,
+    contextOverride?: Record<string, unknown>,
+  ) => Promise<WorkflowNodeResult>;
+  shouldTraverseEdge: (edge: WorkflowIrEdge, source: WorkflowNodeResult) => boolean;
+  signal?: AbortSignal;
+  now?: () => number;
+}
+
+export interface LoopRunResult {
+  outcome: WorkflowNodeOutcome;
+  value?: string;
+  visitedNodeIds: string[];
+}
+
+function resolveLoopConfig(node: WorkflowIrNode): LoopConfig {
+  const cfg = (node.config ?? {}) as Partial<WorkflowLoopConfig>;
+  if (!cfg.template || !Array.isArray(cfg.template.nodes) || !Array.isArray(cfg.template.edges)) {
+    throw new WorkflowIrError(`loop node '${node.id}' has no template subgraph`);
+  }
+  if (!cfg.exitWhen) {
+    throw new WorkflowIrError(`loop node '${node.id}' has no exitWhen condition`);
+  }
+  const maxIterations =
+    typeof cfg.maxIterations === "number" && Number.isFinite(cfg.maxIterations)
+      ? Math.max(1, Math.min(MAX_ITERATIONS_CAP, Math.floor(cfg.maxIterations)))
+      : DEFAULT_MAX_ITERATIONS;
+  const timeoutMs =
+    typeof cfg.timeoutMs === "number" && Number.isFinite(cfg.timeoutMs)
+      ? Math.max(1, Math.min(MAX_TIMEOUT_MS, Math.floor(cfg.timeoutMs)))
+      : DEFAULT_TIMEOUT_MS;
+  return {
+    template: cfg.template,
+    exitWhen: cfg.exitWhen,
+    maxIterations,
+    timeoutMs,
+  };
+}
+
+function buildOutgoing(edges: WorkflowIrEdge[]): Map<string, WorkflowIrEdge[]> {
+  const outgoing = new Map<string, WorkflowIrEdge[]>();
+  for (const edge of edges) {
+    const list = outgoing.get(edge.from) ?? [];
+    list.push(edge);
+    outgoing.set(edge.from, list);
+  }
+  return outgoing;
+}
+
+function findTemplateEntry(nodes: WorkflowIrNode[], edges: WorkflowIrEdge[], loopId: string): WorkflowIrNode {
+  const incoming = new Map<string, number>();
+  for (const edge of edges) 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(`loop node '${loopId}' template must have exactly one entry node`);
+  }
+  return entries[0];
+}
+
+function exitNodeId(nodes: WorkflowIrNode[], edges: WorkflowIrEdge[], loopId: string): string {
+  const outgoing = new Map<string, number>();
+  for (const edge of edges) outgoing.set(edge.from, (outgoing.get(edge.from) ?? 0) + 1);
+  const exits = nodes.filter((n) => (outgoing.get(n.id) ?? 0) === 0);
+  if (exits.length !== 1) {
+    throw new WorkflowIrError(`loop node '${loopId}' template must have exactly one exit node`);
+  }
+  return exits[0].id;
+}
+
+function matchesExit(condition: WorkflowLoopConfig["exitWhen"], value: unknown): boolean {
+  const text = typeof value === "string" ? value : value == null ? "" : String(value);
+  if (condition.type === "output-contains") {
+    return text.includes(condition.value);
+  }
+  return new RegExp(condition.pattern, condition.flags).test(text);
+}
+
+function publishIterationContext(
+  target: Record<string, unknown>,
+  iterationContext: Record<string, unknown>,
+): void {
+  const { ["loop:active"]: _active, ...publicContext } = iterationContext;
+  Object.assign(target, publicContext);
+}
+
+export async function runLoop(
+  loopNode: WorkflowIrNode,
+  env: LoopEnvironment,
+): Promise<LoopRunResult> {
+  const config = resolveLoopConfig(loopNode);
+  const templateById = new Map(config.template.nodes.map((n) => [n.id, n]));
+  const outgoing = buildOutgoing(config.template.edges);
+  const entry = findTemplateEntry(config.template.nodes, config.template.edges, loopNode.id);
+  const defaultExitNodeId = exitNodeId(config.template.nodes, config.template.edges, loopNode.id);
+  const sourceNodeId = config.exitWhen.nodeId ?? defaultExitNodeId;
+  const now = env.now ?? (() => Date.now());
+  const deadline = now() + config.timeoutMs;
+  const visitedNodeIds: string[] = [];
+  const iterationSummaries: Array<{ iteration: number; outcome: string; value?: string }> = [];
+
+  for (let iteration = 1; iteration <= config.maxIterations; iteration++) {
+    if (env.signal?.aborted) {
+      return { outcome: "failure", value: "aborted", visitedNodeIds };
+    }
+    if (now() >= deadline) {
+      env.context[`node:${loopNode.id}:loop`] = {
+        iterations: iteration - 1,
+        exitReason: "timeout",
+        history: iterationSummaries,
+      };
+      return { outcome: "failure", value: "loop-timeout", visitedNodeIds };
+    }
+
+    const iterationContext: Record<string, unknown> = {
+      ...env.context,
+      "loop:active": {
+        loopNodeId: loopNode.id,
+        iteration,
+      },
+    };
+    let current: WorkflowIrNode | undefined = entry;
+    let lastResult: WorkflowNodeResult = { outcome: "success" };
+
+    while (current) {
+      if (env.signal?.aborted) {
+        return { outcome: "failure", value: "aborted", visitedNodeIds };
+      }
+      if (now() >= deadline) {
+        env.context[`node:${loopNode.id}:loop`] = {
+          iterations: iteration - 1,
+          exitReason: "timeout",
+          history: iterationSummaries,
+        };
+        return { outcome: "failure", value: "loop-timeout", visitedNodeIds };
+      }
+
+      const materializedId = `${loopNode.id}#${iteration}:${current.id}`;
+      visitedNodeIds.push(materializedId);
+      lastResult = await env.runTemplateNode(current, env.signal, iterationContext);
+      if (lastResult.contextPatch) Object.assign(iterationContext, lastResult.contextPatch);
+      iterationContext[`node:${current.id}:outcome`] = lastResult.outcome;
+      if (lastResult.value !== undefined) iterationContext[`node:${current.id}:value`] = lastResult.value;
+      if (lastResult.outcome === "failure") {
+        publishIterationContext(env.context, iterationContext);
+        env.context[`node:${loopNode.id}:loop`] = {
+          iterations: iteration,
+          exitReason: "node-failure",
+          history: iterationSummaries,
+        };
+        return { outcome: "failure", value: lastResult.value, visitedNodeIds };
+      }
+
+      const edges: WorkflowIrEdge[] = outgoing.get(current.id) ?? [];
+      const matching: WorkflowIrEdge[] = edges.filter((edge: WorkflowIrEdge) =>
+        env.shouldTraverseEdge(edge, lastResult),
+      );
+      current = matching.length > 0 ? templateById.get(matching[0].to) : undefined;
+    }
+
+    const sourceValue = iterationContext[`node:${sourceNodeId}:value`];
+    const finalValue = sourceValue ?? lastResult.value;
+    iterationSummaries.push({
+      iteration,
+      outcome: lastResult.outcome,
+      ...(finalValue !== undefined ? { value: String(finalValue) } : {}),
+    });
+    publishIterationContext(env.context, iterationContext);
+    if (matchesExit(config.exitWhen, finalValue)) {
+      env.context[`node:${loopNode.id}:loop`] = {
+        iterations: iteration,
+        exitReason: "matched",
+        finalValue,
+        history: iterationSummaries,
+      };
+      return { outcome: "success", visitedNodeIds };
+    }
+  }
+
+  env.context[`node:${loopNode.id}:loop`] = {
+    iterations: config.maxIterations,
+    exitReason: "iteration-exhausted",
+    history: iterationSummaries,
+  };
+  return { outcome: "failure", value: "loop-iteration-exhausted", visitedNodeIds };
+}