From 96f801c019eb9f7ccb22d5ee58735c8d047efe14 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 5 Jun 2026 20:32:24 -0700 Subject: [PATCH] feat(pr): top-level bounded rework + auto-merge gate + legacy-bypass pin (U6) Generalizes the foreach-only bounded-rework mechanism to the top-level graph walk so the PR review loop (await-review -> pr-respond -> rework back) is a legal bounded cycle: a kind:rework back-edge to a stacked node returns a ReworkSignal the loop-head re-runs up to maxReworkCycles, then routes outcome:rework-exhausted. Non-rework cycles still throw 'Cycle detected' (safety preserved); foreach rework unchanged (shared core constants). Adds createAutoMergeGateHandler (live entity re-fetch + isPrEntityAutoMergeReady -> auto-on/auto-off). Pins R14: graph-executed PR tasks merge through pr-merge, never the legacy queue. 122 graph tests green. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/core/src/index.ts | 5 + packages/core/src/workflow-ir-types.ts | 25 +- .../src/__tests__/pr-graph-flow.test.ts | 264 ++++++++++++++++++ .../src/__tests__/pr-rework-bound.test.ts | 203 ++++++++++++++ packages/engine/src/index.ts | 1 + packages/engine/src/pr-nodes.ts | 49 ++++ .../engine/src/workflow-graph-executor.ts | 111 +++++++- packages/engine/src/workflow-graph-foreach.ts | 12 +- packages/engine/src/workflow-node-handlers.ts | 14 +- 9 files changed, 662 insertions(+), 22 deletions(-) create mode 100644 packages/engine/src/__tests__/pr-graph-flow.test.ts create mode 100644 packages/engine/src/__tests__/pr-rework-bound.test.ts diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 65dbc70d59..7fba542ed3 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -71,6 +71,11 @@ export type { WorkflowFieldOption, WorkflowFieldRender, } from "./workflow-ir-types.js"; +export { + DEFAULT_MAX_REWORK_CYCLES, + MAX_REWORK_CYCLES_CAP, + resolveMaxReworkCycles, +} from "./workflow-ir-types.js"; export { BUILTIN_CODING_WORKFLOW_IR } from "./builtin-coding-workflow-ir.js"; export { BUILTIN_STEPWISE_CODING_WORKFLOW_IR } from "./builtin-stepwise-coding-workflow-ir.js"; diff --git a/packages/core/src/workflow-ir-types.ts b/packages/core/src/workflow-ir-types.ts index e96a0eeba6..720c89cf70 100644 --- a/packages/core/src/workflow-ir-types.ts +++ b/packages/core/src/workflow-ir-types.ts @@ -31,13 +31,32 @@ export interface WorkflowIrNode { config?: Record; } +/** Default bounded-rework budget when a rework region omits `maxReworkCycles` + * (KTD-5 foreach default; U6 reuses it for the top-level review loop). */ +export const DEFAULT_MAX_REWORK_CYCLES = 3; +/** Defensive clamp on any rework budget (KTD-5; shared by foreach + U6). */ +export const MAX_REWORK_CYCLES_CAP = 10; + +/** Resolve a bounded-rework budget from a config bag, applying the shared + * default + clamp. Used by the foreach sub-walk and the top-level rework loop so + * the bound semantics cannot drift between the two. */ +export function resolveMaxReworkCycles(raw: unknown): number { + const n = typeof raw === "number" ? raw : DEFAULT_MAX_REWORK_CYCLES; + return Math.max(1, Math.min(MAX_REWORK_CYCLES_CAP, Math.floor(n))); +} + export interface WorkflowIrEdge { from: string; to: string; condition?: string; - /** Step-inversion (KTD-5): `rework` edges are the only legal cycles, scoped to - * one foreach template instance and bounded by the foreach `maxReworkCycles`. - * They are exempt from cycle/parallelism complaints. */ + /** Step-inversion (KTD-5) + PR review loop (U6): `rework` edges are the only + * legal cycles. Originally scoped to one foreach template instance and bounded + * by the foreach `maxReworkCycles`; U6 generalizes the same mechanism to the + * top-level walk so a PR review region (await-review → pr-respond → back to + * await-review) is a legal bounded cycle too. The bound on a top-level rework + * edge is `maxReworkCycles` on this edge's `from` node config (the loop head), + * defaulting to {@link DEFAULT_MAX_REWORK_CYCLES}. Either way, rework edges are + * exempt from "Cycle detected"; every other back-edge still throws. */ kind?: "rework"; } diff --git a/packages/engine/src/__tests__/pr-graph-flow.test.ts b/packages/engine/src/__tests__/pr-graph-flow.test.ts new file mode 100644 index 0000000000..a5c146c865 --- /dev/null +++ b/packages/engine/src/__tests__/pr-graph-flow.test.ts @@ -0,0 +1,264 @@ +/** + * U6 — auto-merge gate routing + legacy-queue bypass pin (R14). + * + * Auto-merge gate (R10): a `gate` node carrying `config.gate === "auto-merge"` + * consults the LIVE PR entity and routes: + * - `outcome:auto-on` when the entity is auto-merge-ready (opted in + approved + * + checks success + mergeable clean + verified) → toward pr-merge; + * - `outcome:auto-off` for every non-ready case (pending checks, UNKNOWN + * mergeable, unverified, not opted in, no entity) → park for manual merge. + * + * R14 pin: a graph-executed PR task merges THROUGH the pr-merge node's injected + * mergePr callback — the merge node IS the merge path — and never falls into the + * legacy merge queue. The executor's graph/legacy routing enforces this; this + * test pins the merge-node behavior so a regression can't silently re-introduce a + * double-merge. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { TaskStore } from "@fusion/core"; +import type { PrEntity, TaskDetail, WorkflowIr, WorkflowIrNode } from "@fusion/core"; + +import { WorkflowGraphExecutor } from "../workflow-graph-executor.js"; +import { createAutoMergeGateHandler } from "../pr-nodes.js"; +import type { PrMergeCallResult, PrNodeDeps, PrSourceDescriptor } from "../pr-nodes.js"; +import type { WorkflowNodeExecutionContext } from "../workflow-graph-executor.js"; + +const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } }); + +const SOURCE: PrSourceDescriptor = { + sourceType: "task", + sourceId: "T-1", + repo: "owner/repo", + headBranch: "fusion/t-1", +}; + +function ctx(taskId = "T-1"): WorkflowNodeExecutionContext { + return { task: { id: taskId } as unknown as TaskDetail, settings: undefined, context: {} }; +} + +const GATE_NODE = { id: "g", kind: "gate", config: { gate: "auto-merge" } } as WorkflowIrNode; + +describe("auto-merge gate (U6, R10)", () => { + let rootDir: string; + let store: TaskStore; + + beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "fusion-pr-graph-flow-")); + store = new TaskStore(rootDir, join(rootDir, ".fusion-global")); + await store.init(); + }); + + afterEach(async () => { + store.close(); + await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + }); + + function deps(overrides: Partial = {}): PrNodeDeps { + return { + getStore: () => store, + resolvePrSource: () => SOURCE, + createPr: async () => ({ prNumber: 1, prUrl: "u" }), + mergePr: async () => ({ status: "merged-requested" }) as PrMergeCallResult, + ...overrides, + }; + } + + /** Seed a live `open` entity and patch it to a chosen readiness state. */ + function seedEntity(patch: Partial): PrEntity { + const entity = store.ensurePrEntityForSource({ ...SOURCE, state: "open" }); + return store.updatePrEntity(entity.id, { + state: "open", + autoMerge: patch.autoMerge, + reviewDecision: patch.reviewDecision, + checksRollup: patch.checksRollup, + mergeable: patch.mergeable, + unverified: patch.unverified, + }); + } + + const READY: Partial = { + autoMerge: true, + reviewDecision: "APPROVED", + checksRollup: "success", + mergeable: "clean", + unverified: false, + }; + + it("ready entity → auto-on", async () => { + seedEntity(READY); + const gate = createAutoMergeGateHandler(deps()); + const result = await gate(GATE_NODE, ctx()); + expect(result).toEqual({ outcome: "success", value: "auto-on" }); + }); + + it("not opted in → auto-off", async () => { + seedEntity({ ...READY, autoMerge: false }); + const gate = createAutoMergeGateHandler(deps()); + expect(await gate(GATE_NODE, ctx())).toEqual({ outcome: "success", value: "auto-off" }); + }); + + it("pending checks → auto-off", async () => { + seedEntity({ ...READY, checksRollup: "pending" }); + const gate = createAutoMergeGateHandler(deps()); + expect(await gate(GATE_NODE, ctx())).toEqual({ outcome: "success", value: "auto-off" }); + }); + + it("unknown mergeability → auto-off", async () => { + seedEntity({ ...READY, mergeable: "unknown" }); + const gate = createAutoMergeGateHandler(deps()); + expect(await gate(GATE_NODE, ctx())).toEqual({ outcome: "success", value: "auto-off" }); + }); + + it("unverified entity → auto-off (R19 hard gate)", async () => { + seedEntity({ ...READY, unverified: true }); + const gate = createAutoMergeGateHandler(deps()); + expect(await gate(GATE_NODE, ctx())).toEqual({ outcome: "success", value: "auto-off" }); + }); + + it("not approved → auto-off", async () => { + seedEntity({ ...READY, reviewDecision: "CHANGES_REQUESTED" }); + const gate = createAutoMergeGateHandler(deps()); + expect(await gate(GATE_NODE, ctx())).toEqual({ outcome: "success", value: "auto-off" }); + }); + + it("no live entity → auto-off (never blocks the run)", async () => { + const gate = createAutoMergeGateHandler(deps()); + expect(await gate(GATE_NODE, ctx())).toEqual({ outcome: "success", value: "auto-off" }); + }); + + it("routes a graph end-to-end: approve → auto-on gate → pr-merge", async () => { + seedEntity(READY); + const mergePr = vi.fn(async () => ({ status: "merged-requested" }) as PrMergeCallResult); + const ir: WorkflowIr = { + version: "v1", + name: "auto-merge-flow", + nodes: [ + { id: "start", kind: "start" }, + { id: "gate", kind: "gate", config: { gate: "auto-merge" } }, + { id: "merge", kind: "pr-merge" }, + { id: "park", kind: "script" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "gate" }, + { from: "gate", to: "merge", condition: "outcome:auto-on" }, + { from: "gate", to: "park", condition: "outcome:auto-off" }, + { from: "merge", to: "end" }, + { from: "park", to: "end" }, + ], + }; + const park = vi.fn(async () => ({ outcome: "success" as const })); + const executor = new WorkflowGraphExecutor({ + prNodes: deps({ mergePr }), + handlers: { script: park }, + }); + + const result = await executor.run({ id: "T-1" } as TaskDetail, settingsOn(), ir); + expect(result.outcome).toBe("success"); + expect(result.visitedNodeIds).toContain("merge"); + expect(mergePr).toHaveBeenCalledTimes(1); + expect(park).not.toHaveBeenCalled(); + }); + + it("auto-off entity parks for manual merge (pr-merge not reached)", async () => { + seedEntity({ ...READY, checksRollup: "pending" }); + const mergePr = vi.fn(async () => ({ status: "merged-requested" }) as PrMergeCallResult); + const ir: WorkflowIr = { + version: "v1", + name: "auto-merge-park", + nodes: [ + { id: "start", kind: "start" }, + { id: "gate", kind: "gate", config: { gate: "auto-merge" } }, + { id: "merge", kind: "pr-merge" }, + { id: "park", kind: "script" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "gate" }, + { from: "gate", to: "merge", condition: "outcome:auto-on" }, + { from: "gate", to: "park", condition: "outcome:auto-off" }, + { from: "merge", to: "end" }, + { from: "park", to: "end" }, + ], + }; + const park = vi.fn(async () => ({ outcome: "success" as const })); + const executor = new WorkflowGraphExecutor({ + prNodes: deps({ mergePr }), + handlers: { script: park }, + }); + + const result = await executor.run({ id: "T-1" } as TaskDetail, settingsOn(), ir); + expect(result.visitedNodeIds).toContain("park"); + expect(result.visitedNodeIds).not.toContain("merge"); + expect(mergePr).not.toHaveBeenCalled(); + }); +}); + +describe("legacy-queue bypass pin (U6, R14)", () => { + let rootDir: string; + let store: TaskStore; + + beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "fusion-pr-r14-")); + store = new TaskStore(rootDir, join(rootDir, ".fusion-global")); + await store.init(); + }); + + afterEach(async () => { + store.close(); + await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + }); + + function deps(mergePr: PrNodeDeps["mergePr"]): PrNodeDeps { + return { + getStore: () => store, + resolvePrSource: () => SOURCE, + createPr: async () => ({ prNumber: 1, prUrl: "u" }), + mergePr, + }; + } + + it("a graph-executed PR task merges through the pr-merge node, not a legacy queue", async () => { + // Seed an actionable entity so pr-merge proceeds (the merge node IS the merge + // path under the graph executor). + const entity = store.ensurePrEntityForSource({ ...SOURCE, state: "open" }); + store.updatePrEntity(entity.id, { state: "open", unverified: false, headOid: "deadbeef" }); + + // A legacy merge-queue sink. If the graph path EVER routed a PR task into the + // legacy merger this spy would be hit — pinning the bypass (R14). + const legacyMergeEnqueue = vi.fn(); + const mergePr = vi.fn(async () => ({ status: "merged-requested" }) as PrMergeCallResult); + + const ir: WorkflowIr = { + version: "v1", + name: "r14-merge-node-only", + nodes: [ + { id: "start", kind: "start" }, + { id: "merge", kind: "pr-merge" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "merge" }, + { from: "merge", to: "end" }, + ], + }; + const executor = new WorkflowGraphExecutor({ prNodes: deps(mergePr) }); + + const result = await executor.run({ id: "T-1" } as TaskDetail, settingsOn(), ir); + + // Merge happened exactly once, via the injected node callback with the + // entity's head OID — the graph node IS the merge, no legacy enqueue. + expect(result.outcome).toBe("success"); + expect(mergePr).toHaveBeenCalledTimes(1); + expect(mergePr).toHaveBeenCalledWith(expect.objectContaining({ expectedHeadOid: "deadbeef" })); + expect(legacyMergeEnqueue).not.toHaveBeenCalled(); + + // The node does NOT write the terminal `merged` state (reconcile corroborates), + // so there is no path for a second/legacy merge to also act on a `merged` row. + expect(store.getActivePrEntityBySource("task", "T-1")?.state).toBe("open"); + }); +}); diff --git a/packages/engine/src/__tests__/pr-rework-bound.test.ts b/packages/engine/src/__tests__/pr-rework-bound.test.ts new file mode 100644 index 0000000000..6d4d108ea0 --- /dev/null +++ b/packages/engine/src/__tests__/pr-rework-bound.test.ts @@ -0,0 +1,203 @@ +/** + * U6 — bounded-rework generalization to the top-level walk. + * + * U1–U5 confined `kind: "rework"` cycles to the foreach sub-walk; the top-level + * recursive `walk` threw "Cycle detected" on ANY back-edge. U6 lifts the same + * bounded-rework mechanism to the top level so the PR review loop (await-review → + * pr-respond → rework → await-review) is a legal, bounded cycle. These tests pin: + * + * - a top-level rework cycle loops up to the cap then routes + * `outcome:rework-exhausted` (finite; never infinite; never "Cycle detected"); + * - a NON-rework top-level back-edge still throws "Cycle detected" (safety); + * - the bound is honored exactly (cap traversals of the rework edge). + */ +import { describe, expect, it, vi } from "vitest"; +import type { TaskDetail, WorkflowIr } from "@fusion/core"; + +import { WorkflowGraphExecutor } from "../workflow-graph-executor.js"; + +const task = { id: "FN-U6" } as TaskDetail; +const settingsOn = () => ({ experimentalFeatures: { workflowGraphExecutor: true } }); + +describe("WorkflowGraphExecutor bounded-rework generalization (U6)", () => { + it("loops a top-level rework cycle up to the cap then routes rework-exhausted (never infinite)", async () => { + // start → A(head) → B → rework back to A; A also has an + // `outcome:rework-exhausted` forward edge to `done`. B always emits + // value:"again" so the rework edge keeps firing until the budget runs out. + const ir: WorkflowIr = { + version: "v1", + name: "toplevel-rework", + nodes: [ + { id: "start", kind: "start" }, + { id: "A", kind: "gate", config: { maxReworkCycles: 2 } }, + { id: "B", kind: "prompt" }, + { id: "done", kind: "script" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "A" }, + { from: "A", to: "B", condition: "success" }, + // exhaustion routes the head forward via the rework-exhausted value. + { from: "A", to: "done", condition: "outcome:rework-exhausted" }, + { from: "B", to: "A", kind: "rework", condition: "outcome:again" }, + { from: "done", to: "end" }, + ], + }; + + const a = vi.fn(async () => ({ outcome: "success" as const })); + const b = vi.fn(async () => ({ outcome: "success" as const, value: "again" })); + const done = vi.fn(async () => ({ outcome: "success" as const })); + const executor = new WorkflowGraphExecutor({ + handlers: { gate: a, prompt: b, script: done }, + }); + + const result = await executor.run(task, settingsOn(), ir); + + expect(result.outcome).toBe("success"); + // cap = 2 → A runs the initial pass + 2 rework re-entries = 3 times; B once + // per A pass = 3; then exhaustion routes `done` exactly once. + expect(a).toHaveBeenCalledTimes(3); + expect(b).toHaveBeenCalledTimes(3); + expect(done).toHaveBeenCalledTimes(1); + expect(result.visitedNodeIds).toContain("done"); + }); + + it("never throws 'Cycle detected' for the legal rework edge", async () => { + const ir: WorkflowIr = { + version: "v1", + name: "rework-no-throw", + nodes: [ + { id: "start", kind: "start" }, + { id: "A", kind: "gate", config: { maxReworkCycles: 1 } }, + { id: "B", kind: "prompt" }, + { id: "done", kind: "script" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "A" }, + { from: "A", to: "B", condition: "success" }, + { from: "A", to: "done", condition: "outcome:rework-exhausted" }, + { from: "B", to: "A", kind: "rework", condition: "outcome:again" }, + { from: "done", to: "end" }, + ], + }; + const executor = new WorkflowGraphExecutor({ + handlers: { + gate: async () => ({ outcome: "success" }), + prompt: async () => ({ outcome: "success", value: "again" }), + script: async () => ({ outcome: "success" }), + }, + }); + + // Must resolve, not reject with "Cycle detected". + await expect(executor.run(task, settingsOn(), ir)).resolves.toMatchObject({ + outcome: "success", + }); + }); + + it("a rework cycle that resolves before the cap takes the forward edge (no exhaustion)", async () => { + // B emits value:"again" on the first pass (rework), then value:"ok" so A's + // forward edge to `done` is taken on the second pass — exhaustion never fires. + const ir: WorkflowIr = { + version: "v1", + name: "rework-resolves", + nodes: [ + { id: "start", kind: "start" }, + { id: "A", kind: "gate", config: { maxReworkCycles: 5 } }, + { id: "B", kind: "prompt" }, + { id: "done", kind: "script" }, + { id: "exhausted", kind: "script" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "A" }, + // A routes forward to B on every pass; B decides rework vs. proceed. + { from: "A", to: "B", condition: "success" }, + { from: "A", to: "exhausted", condition: "outcome:rework-exhausted" }, + { from: "B", to: "done", condition: "outcome:ok" }, + { from: "B", to: "A", kind: "rework", condition: "outcome:again" }, + { from: "done", to: "end" }, + { from: "exhausted", to: "end" }, + ], + }; + let bCalls = 0; + const done = vi.fn(async () => ({ outcome: "success" as const })); + const exhausted = vi.fn(async () => ({ outcome: "success" as const })); + const executor = new WorkflowGraphExecutor({ + handlers: { + gate: async () => ({ outcome: "success" }), + prompt: async () => { + bCalls += 1; + return { outcome: "success" as const, value: bCalls === 1 ? "again" : "ok" }; + }, + script: async (node) => (node.id === "done" ? done() : exhausted()), + }, + }); + + const result = await executor.run(task, settingsOn(), ir); + expect(result.outcome).toBe("success"); + expect(done).toHaveBeenCalledTimes(1); + expect(exhausted).not.toHaveBeenCalled(); + expect(result.visitedNodeIds).toContain("done"); + }); + + it("a NON-rework top-level back-edge still throws 'Cycle detected' (safety preserved)", async () => { + // A → B → A with NO kind:"rework" on the back-edge. This must still be + // rejected as an illegal cycle. + const ir: WorkflowIr = { + version: "v1", + name: "illegal-cycle", + nodes: [ + { id: "start", kind: "start" }, + { id: "A", kind: "prompt" }, + { id: "B", kind: "prompt" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "A" }, + { from: "A", to: "B", condition: "success" }, + // plain back-edge — NOT a rework edge. + { from: "B", to: "A", condition: "success" }, + ], + }; + const executor = new WorkflowGraphExecutor({ + handlers: { prompt: async () => ({ outcome: "success" }) }, + }); + + await expect(executor.run(task, settingsOn(), ir)).rejects.toThrow(/Cycle detected/); + }); + + it("defaults the rework cap when the head omits maxReworkCycles", async () => { + // No config.maxReworkCycles → DEFAULT_MAX_REWORK_CYCLES (3): A runs 1 + 3 = 4. + const ir: WorkflowIr = { + version: "v1", + name: "rework-default-cap", + nodes: [ + { id: "start", kind: "start" }, + { id: "A", kind: "gate" }, + { id: "B", kind: "prompt" }, + { id: "done", kind: "script" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "A" }, + { from: "A", to: "B", condition: "success" }, + { from: "A", to: "done", condition: "outcome:rework-exhausted" }, + { from: "B", to: "A", kind: "rework", condition: "outcome:again" }, + { from: "done", to: "end" }, + ], + }; + const a = vi.fn(async () => ({ outcome: "success" as const })); + const executor = new WorkflowGraphExecutor({ + handlers: { + gate: a, + prompt: async () => ({ outcome: "success", value: "again" }), + script: async () => ({ outcome: "success" }), + }, + }); + + const result = await executor.run(task, settingsOn(), ir); + expect(result.outcome).toBe("success"); + expect(a).toHaveBeenCalledTimes(4); // initial + 3 reworks (default cap) + }); +}); diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 0974757eb3..4b329ace7f 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -48,6 +48,7 @@ export { } from "./workflow-node-handlers.js"; export { createPrNodeHandlers, + createAutoMergeGateHandler, buildPrNodeDeps, type PrNodeDeps, type PrNodeGithubOps, diff --git a/packages/engine/src/pr-nodes.ts b/packages/engine/src/pr-nodes.ts index 12d0f8da4e..c82c735a30 100644 --- a/packages/engine/src/pr-nodes.ts +++ b/packages/engine/src/pr-nodes.ts @@ -17,6 +17,7 @@ import { isPrEntityActionable, + isPrEntityAutoMergeReady, type PrEntity, type PrEntityCreateInput, type PrEntityUpdate, @@ -425,3 +426,51 @@ export function createPrNodeHandlers(deps: PrNodeDeps): Record< "pr-merge": prMerge, }; } + +/** + * Auto-merge gate handler (U6, R10). Placed after the approval step. It + * re-evaluates the LIVE PR entity each time (never trusts a cached/SSE copy) and + * routes: + * + * - `outcome:auto-on` → toward `pr-merge`, when {@link isPrEntityAutoMergeReady} + * (opted in + approved + all checks concluded success + mergeable clean + + * verified). + * - `outcome:auto-off` → park for a manual-release merge, for EVERY non-ready + * case: not opted in, pending/failed checks, UNKNOWN/conflicting mergeability, + * unverified entity, or no live entity at all. The gate never blocks the run. + * + * Reuses the gate-routing contract (`{ outcome: "success", value }` consumed by + * `outcome:` edges in shouldTraverseEdge) rather than forking a parallel routing + * mechanism. The store/entity lookup is injected via {@link PrNodeDeps} so the + * engine stays dashboard-import-free. The `auto-merge ready` predicate lives in + * @fusion/core so the gate, the dashboard, and the reconcile share one + * definition and cannot drift. + */ +export function createAutoMergeGateHandler(deps: Pick): WorkflowNodeHandler { + const audit = (reason: string, detail: string): void => { + try { + deps.audit?.(reason, detail); + } catch { + // Audit must never affect the run. + } + }; + return async (node, ctx) => { + const store = deps.getStore(); + const entity = store.getActivePrEntityBySource("task", ctx.task.id) + ?? store.getActivePrEntityBySource("branch-group", ctx.task.id); + + if (!entity) { + // No live entity → cannot auto-merge; park for manual handling (never block). + audit("auto-merge-gate-no-entity", `auto-merge gate '${node.id}' found no live PR entity for task ${ctx.task.id}`); + return { outcome: "success", value: "auto-off" }; + } + + // Re-fetch authoritative state: the entity row IS the live copy here (store + // read), so pending checks / UNKNOWN mergeable / unverified / not-opted-in all + // fall to auto-off via the shared predicate. + if (isPrEntityAutoMergeReady(entity)) { + return { outcome: "success", value: "auto-on" }; + } + return { outcome: "success", value: "auto-off" }; + }; +} diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 23f29bfc5d..edb3c76a06 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -1,5 +1,5 @@ import type { Settings, TaskDetail, TaskStep, WorkflowIr, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core"; -import { BUILTIN_CODING_WORKFLOW_IR, WorkflowIrError, isExperimentalFeatureEnabled } from "@fusion/core"; +import { BUILTIN_CODING_WORKFLOW_IR, WorkflowIrError, isExperimentalFeatureEnabled, resolveMaxReworkCycles } from "@fusion/core"; import { createDefaultNodeHandlers, @@ -183,6 +183,43 @@ export class WorkflowGraphExecutor { const inStack = new Set(); const runId = this.deps.runId ?? `${task.id}:run`; + // Bounded-rework generalization (U6). A `kind: "rework"` edge is the only + // legal cycle: it loops back to a "rework region head" (the edge's `to` node). + // The same mechanism the foreach sub-walk uses (bounded budget, exhaustion + // routes `outcome:rework-exhausted`) is lifted to the top-level walk so the PR + // review loop (await-review → pr-respond → rework → await-review) is legal and + // bounded. Every NON-rework back-edge still throws "Cycle detected" below. + // + // - reworkHeads: every node that is the target of a rework edge. + // - reworkBudget: per-head remaining traversals, seeded lazily from the head + // node's `config.maxReworkCycles` (shared default + clamp from core). + // - The loop is iterative at the head frame: when a downstream node takes its + // rework edge back to a head currently on the stack, the head's walk frame + // catches a REWORK_SIGNAL sentinel and re-iterates (under budget) instead of + // recursing — so `inStack` never sees the head re-entered as a cycle. + const reworkHeads = new Set(); + for (const edge of ir.edges) { + if (edge.kind === "rework") reworkHeads.add(edge.to); + } + const reworkBudget = new Map(); + const reworkBudgetFor = (headId: string): number => { + const existing = reworkBudget.get(headId); + if (existing !== undefined) return existing; + const head = nodeMap.get(headId); + const seeded = resolveMaxReworkCycles(head?.config?.maxReworkCycles); + reworkBudget.set(headId, seeded); + return seeded; + }; + // Sentinel a downstream rework edge returns up the recursion to its loop head. + interface ReworkSignal { + readonly __rework: true; + readonly headId: string; + /** The source node's result, carried so the head re-runs against fresh state. */ + readonly source: WorkflowNodeResult; + } + const isReworkSignal = (r: WorkflowNodeResult | ReworkSignal): r is ReworkSignal => + (r as ReworkSignal).__rework === true; + // On resume, completed branch nodes (from a prior crashed run) are skipped // so their handlers do not re-fire (idempotency). let completedNodeIds: Set | undefined; @@ -220,14 +257,11 @@ export class WorkflowGraphExecutor { completedNodeIds, }); - const walk = async (nodeId: string): Promise => { - const node = nodeMap.get(nodeId); - if (!node) throw new WorkflowIrError(`Unknown workflow node: ${nodeId}`); - if (inStack.has(nodeId)) throw new WorkflowIrError(`Cycle detected at node: ${nodeId}`); - inStack.add(nodeId); - visitedNodeIds.push(nodeId); - - try { + // Execute one node and traverse its outgoing edges. May return a ReworkSignal + // (a rework back-edge fired); the caller frame propagates or consumes it. + const runNodeAndTraverse = async ( + node: WorkflowIrNode, + ): Promise => { if (node.kind === "start") { return await traverseChildren(node, { outcome: "success" }); } @@ -310,12 +344,55 @@ export class WorkflowGraphExecutor { if (result.value !== undefined) context[`node:${node.id}:value`] = result.value; return await traverseChildren(node, result); + }; + + // Recursive walk into a node. A rework region head (target of a `kind: + // "rework"` edge) is wrapped in an iterative loop: while a downstream rework + // edge fires back to it (returned as a ReworkSignal under budget) the head + // re-runs; budget exhaustion re-routes the head with an + // `outcome:rework-exhausted` source so its forward edge carries the flow out. + // Every NON-rework back-edge still hits the cycle detector and throws. + const walk = async (nodeId: string): Promise => { + const node = nodeMap.get(nodeId); + if (!node) throw new WorkflowIrError(`Unknown workflow node: ${nodeId}`); + if (inStack.has(nodeId)) throw new WorkflowIrError(`Cycle detected at node: ${nodeId}`); + inStack.add(nodeId); + visitedNodeIds.push(nodeId); + + try { + const isReworkHead = reworkHeads.has(nodeId); + for (;;) { + const outcome = await runNodeAndTraverse(node); + if (!isReworkSignal(outcome)) return outcome; + // A rework back-edge fired. It must target THIS head (the deepest + // enclosing rework head); a signal for an outer head propagates up. + if (!isReworkHead || outcome.headId !== nodeId) return outcome; + const remaining = reworkBudgetFor(nodeId); + if (remaining > 0) { + reworkBudget.set(nodeId, remaining - 1); + continue; // re-run the head node fresh (await-review re-evaluates) + } + // Budget exhausted: route the head's `outcome:rework-exhausted` forward + // edge (mirrors the foreach node's `{outcome:"failure", value: + // "rework-exhausted"}`). The `failure` outcome is deliberate so the + // exhausted re-route does NOT also satisfy the head's generic + // `condition:"success"` forward edge (which would re-enter the loop body + // and never terminate). Never loops forever; never throws "Cycle + // detected" for the legal rework edge. + const exhausted = await traverseChildren(node, { outcome: "failure", value: "rework-exhausted" }); + // If no `outcome:rework-exhausted` edge exists the source bubbles back + // (a failure outcome) — a finite, routable terminal, never an infinite loop. + return exhausted; + } } finally { inStack.delete(nodeId); } }; - const traverseChildren = async (node: WorkflowIrNode, sourceResult: WorkflowNodeResult): Promise => { + const traverseChildren = async ( + node: WorkflowIrNode, + sourceResult: WorkflowNodeResult, + ): Promise => { const edges = outgoingMap.get(node.id) ?? []; if (edges.length === 0) { return sourceResult; @@ -327,13 +404,22 @@ export class WorkflowGraphExecutor { } let aggregate: WorkflowNodeResult = sourceResult; + // Forward edges first, deterministic by target id (matches prior ordering); + // a rework edge is a loop-back and is handled distinctly below. for (const edge of matching.sort((a, b) => a.to.localeCompare(b.to))) { + // Rework back-edge: do NOT recurse (the head is on the stack — that would + // be a cycle). Bubble a ReworkSignal up to the head's iterative loop. + if (edge.kind === "rework" && inStack.has(edge.to)) { + return { __rework: true, headId: edge.to, source: sourceResult } satisfies ReworkSignal; + } const target = nodeMap.get(edge.to); if (target?.kind === "end") { aggregate = sourceResult; continue; } const child = await walk(edge.to); + // A ReworkSignal propagated from deeper: bubble it further up unchanged. + if (isReworkSignal(child)) return child; if (child.outcome === "failure") { aggregate = child; break; @@ -344,6 +430,11 @@ export class WorkflowGraphExecutor { }; const terminal = await walk(startNode.id); + if (isReworkSignal(terminal)) { + // A rework edge whose target is not an enclosing head on the stack — i.e. a + // rework edge pointing at a node never entered as a loop head. Malformed IR. + throw new WorkflowIrError(`Rework edge targets a node that is not a region head: ${terminal.headId}`); + } // Prune again on run completion (#1412): keeps only this run's rows so the // table does not accumulate historical runs for a long-lived task. await this.pruneStaleBranches(task.id, runId); diff --git a/packages/engine/src/workflow-graph-foreach.ts b/packages/engine/src/workflow-graph-foreach.ts index 9c0ee826c0..bd776f5a74 100644 --- a/packages/engine/src/workflow-graph-foreach.ts +++ b/packages/engine/src/workflow-graph-foreach.ts @@ -1,5 +1,5 @@ import type { TaskDetail, TaskStep, WorkflowIrEdge, WorkflowIrNode } from "@fusion/core"; -import { WorkflowIrError } from "@fusion/core"; +import { WorkflowIrError, resolveMaxReworkCycles } from "@fusion/core"; import type { WorkflowNodeOutcome, WorkflowNodeResult } from "./workflow-graph-executor.js"; import { @@ -43,10 +43,9 @@ import { schedulerLog } from "./logger.js"; * 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; +// Rework budget default + clamp live in @fusion/core (DEFAULT_MAX_REWORK_CYCLES / +// MAX_REWORK_CYCLES_CAP / resolveMaxReworkCycles) so the foreach sub-walk and the +// top-level PR review loop (U6) share one definition and cannot drift. /** Default parallel concurrency (KTD-3). */ const DEFAULT_CONCURRENCY = 2; /** Hard cap on parallel concurrency (KTD-3). */ @@ -265,8 +264,7 @@ function resolveForeachConfig(node: WorkflowIrNode): { 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 maxReworkCycles = resolveMaxReworkCycles(cfg.maxReworkCycles); const mode = cfg.mode === "parallel" ? "parallel" : "sequential"; // Default isolation: worktree for parallel mode, shared for sequential (KTD-3). // (Core validation rejects parallel+shared; this default mirrors that intent.) diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts index f867a18136..b81202258e 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/packages/engine/src/workflow-node-handlers.ts @@ -2,7 +2,7 @@ import { WorkflowIrError, getStepParser } from "@fusion/core"; import type { TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core"; import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js"; -import { createPrNodeHandlers, type PrNodeDeps } from "./pr-nodes.js"; +import { createPrNodeHandlers, createAutoMergeGateHandler, type PrNodeDeps } from "./pr-nodes.js"; export type WorkflowSeamName = "planning" | "execute" | "review" | "merge" | "schedule" | "step-execute"; @@ -554,10 +554,20 @@ export function createDefaultNodeHandlers( "pr-respond": async () => ({ outcome: "failure", value: "pr-nodes-unwired" }), "pr-merge": async () => ({ outcome: "failure", value: "pr-nodes-unwired" }), }; + // Auto-merge gate (U6): a `gate` node carrying `config.gate === "auto-merge"` + // routes on live PR-entity state (outcome:auto-on/auto-off) instead of the + // generic context/executable gate. Wired only when PR deps are present; absent + // them it falls back to the generic gate (fail-closed, no silent auto-merge). + const genericGate = createGateHandler(runCustomNode); + const autoMergeGate = deps?.prNodes ? createAutoMergeGateHandler(deps.prNodes) : undefined; + const gate: WorkflowNodeHandler = autoMergeGate + ? (node, ctx) => + node.config?.gate === "auto-merge" ? autoMergeGate(node, ctx) : genericGate(node, ctx) + : genericGate; return { prompt: promptLike, script: promptLike, - gate: createGateHandler(runCustomNode), + gate, "step-review": createStepReviewHandler(seams), "parse-steps": parseSteps, code: createCodeNodeHandler(deps?.runCode),