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) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
@@ -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";
|
||||
|
||||
|
||||
@@ -31,13 +31,32 @@ export interface WorkflowIrNode {
|
||||
config?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** 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";
|
||||
}
|
||||
|
||||
|
||||
264
packages/engine/src/__tests__/pr-graph-flow.test.ts
Normal file
264
packages/engine/src/__tests__/pr-graph-flow.test.ts
Normal file
@@ -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> = {}): 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>): 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<PrEntity> = {
|
||||
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");
|
||||
});
|
||||
});
|
||||
203
packages/engine/src/__tests__/pr-rework-bound.test.ts
Normal file
203
packages/engine/src/__tests__/pr-rework-bound.test.ts
Normal file
@@ -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)
|
||||
});
|
||||
});
|
||||
@@ -48,6 +48,7 @@ export {
|
||||
} from "./workflow-node-handlers.js";
|
||||
export {
|
||||
createPrNodeHandlers,
|
||||
createAutoMergeGateHandler,
|
||||
buildPrNodeDeps,
|
||||
type PrNodeDeps,
|
||||
type PrNodeGithubOps,
|
||||
|
||||
@@ -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<PrNodeDeps, "getStore" | "audit">): 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" };
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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<string>();
|
||||
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<string>();
|
||||
for (const edge of ir.edges) {
|
||||
if (edge.kind === "rework") reworkHeads.add(edge.to);
|
||||
}
|
||||
const reworkBudget = new Map<string, number>();
|
||||
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<string> | undefined;
|
||||
@@ -220,14 +257,11 @@ export class WorkflowGraphExecutor {
|
||||
completedNodeIds,
|
||||
});
|
||||
|
||||
const walk = async (nodeId: string): Promise<WorkflowNodeResult> => {
|
||||
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<WorkflowNodeResult | ReworkSignal> => {
|
||||
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<WorkflowNodeResult | ReworkSignal> => {
|
||||
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<WorkflowNodeResult> => {
|
||||
const traverseChildren = async (
|
||||
node: WorkflowIrNode,
|
||||
sourceResult: WorkflowNodeResult,
|
||||
): Promise<WorkflowNodeResult | ReworkSignal> => {
|
||||
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);
|
||||
|
||||
@@ -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.)
|
||||
|
||||
@@ -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),
|
||||
|
||||
Reference in New Issue
Block a user