root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(FN-6035): model pre-merge workflow steps in builtin coding

Browse Source
This commit is contained in:
gsxdsm
2026-06-08 22:02:26 -07:00
parent b6800e8622
commit 8d3b869ec1
17 changed files with 283 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts
View File

@@ -1,7 +1,8 @@
// ─────────────────────────────────────────────────────────────────────────────
// PARITY SUBJECT (test-file ownership, U7 / KTD-9):
// This suite owns DEFAULT-WORKFLOW BYTE-IDENTITY parity — it proves the graph
// executor reproduces the workflow-native planning → execute → review → merge seam
// executor reproduces the workflow-native planning → execute → workflow-step
// → review → merge seam
// sequence exactly (the parity ORACLE per KTD-1). It deliberately does NOT
// cover per-step / updateStep-trajectory parity.
//
@@ -22,6 +23,14 @@ import { WorkflowGraphExecutor } from "../workflow-graph-executor.js";
import type { WorkflowLegacySeams } from "../workflow-node-handlers.js";
const task = { id: "FN-5767" } as TaskDetail;
type BaseSeam = "planning" | "execute" | "workflow-step" | "review" | "merge" | "schedule";
function runBaseSeam(seams: WorkflowLegacySeams, seam: BaseSeam, task: TaskDetail, context: Record<string, unknown>) {
if (seam === "workflow-step") {
return seams.workflowStep?.(task, context) ?? Promise.resolve({ outcome: "success" as const });
}
return seams[seam](task, context);
}
function runLegacy(seams: WorkflowLegacySeams) {
return async () => {
@@ -32,6 +41,9 @@ function runLegacy(seams: WorkflowLegacySeams) {
const execute = await seams.execute(task, {});
events.push(`execute:${execute.outcome}`);
if (execute.outcome !== "success") return events;
const workflowStep = await seams.workflowStep?.(task, {}) ?? { outcome: "success" as const };
events.push(`workflow-step:${workflowStep.outcome}`);
if (workflowStep.outcome !== "success") return events;
const review = await seams.review(task, {});
events.push(`review:${review.outcome}`);
if (review.outcome !== "success") return events;
@@ -55,15 +67,15 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {
const seams: WorkflowLegacySeams = {
planning: async () => ({ outcome: "success" }),
execute: async () => ({ outcome: "success" }),
workflowStep: async () => ({ outcome: "success" }),
review: async () => ({ outcome: "success" }),
merge: async () => ({ outcome: "success" }),
schedule: async () => ({ outcome: "success" }),
};
const legacyEvents = await runLegacy(seams)();
type BaseSeam = "planning" | "execute" | "review" | "merge" | "schedule";
const executor = new WorkflowGraphExecutor({ seams, handlers: { prompt: async (node, ctx) => {
const seam = String(node.config?.seam);
const result = await seams[seam as BaseSeam](ctx.task, ctx.context);
const seam = String(node.config?.seam) as BaseSeam;
const result = await runBaseSeam(seams, seam, ctx.task, ctx.context);
events.push(`${seam}:${result.outcome}`);
return result;
} } });
@@ -77,6 +89,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {
const seams: WorkflowLegacySeams = {
planning: async () => ({ outcome: "success" }),
execute: async () => ({ outcome: "success" }),
workflowStep: async () => ({ outcome: "success" }),
review: async () => ({ outcome: "success" }),
merge: async () => ({ outcome: "failure", value: "FileScopeViolationError" }),
schedule: async () => ({ outcome: "success" }),
@@ -85,7 +98,7 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {
const executor = new WorkflowGraphExecutor({ seams });
const result = await executor.run(task, { experimentalFeatures: { workflowGraphExecutor: true } });
expect(result.outcome).toBe("failure");
expect(legacyEvents).toEqual(["planning:success", "execute:success", "review:success", "merge:failure"]);
expect(legacyEvents).toEqual(["planning:success", "execute:success", "workflow-step:success", "review:success", "merge:failure"]);
});
it("preserves autoMerge:false terminal in-review semantics via review failure", async () => {
@@ -162,18 +175,18 @@ describe("column-agent feature is invisible when unbound (U7 / R9)", () => {
const seams: WorkflowLegacySeams = {
planning: async () => ({ outcome: "success" }),
execute: async () => ({ outcome: "success" }),
workflowStep: async () => ({ outcome: "success" }),
review: async () => ({ outcome: "success" }),
merge: async () => ({ outcome: "success" }),
schedule: async () => ({ outcome: "success" }),
};
type BaseSeam = "planning" | "execute" | "review" | "merge" | "schedule";
const executor = new WorkflowGraphExecutor({
seams,
handlers: {
prompt: async (node, ctx) => {
const seam = String(node.config?.seam) as BaseSeam;
stages.push(seam);
return seams[seam](ctx.task, ctx.context);
return runBaseSeam(seams, seam, ctx.task, ctx.context);
},
},
});
@@ -185,11 +198,11 @@ describe("column-agent feature is invisible when unbound (U7 / R9)", () => {
// Bind the invariant to actual executor behavior (PR #1432 review): the
// observation below derives from the run-captured seam sequence, so seam
// drift fails here instead of being masked by a hard-coded literal.
expect(stages).toEqual(["planning", "execute", "review", "merge"]);
expect(stages).toEqual(["planning", "execute", "workflow-step", "review", "merge"]);
// Legacy authoritative observation: a clean run that lands in `done`/merged.
const legacyObs = buildWorkflowObservation({
stageTransitions: ["triage", "planning", "execute", "review", "merge"],
stageTransitions: ["triage", "planning", "execute", "workflow-step", "review", "merge"],
terminalColumn: "done",
terminalStatus: "done",
reviewVerdict: "approve",

3
packages/engine/src/__tests__/workflow-graph-task-runner.test.ts
View File

@@ -69,6 +69,7 @@ function recordingSeams(calls: string[], overrides: Partial<Record<string, Workf
return {
planning: seam("planning"),
execute: seam("execute"),
workflowStep: seam("workflow-step"),
review: seam("review"),
merge: seam("merge"),
schedule: seam("schedule"),
@@ -236,7 +237,7 @@ describe("WorkflowGraphTaskRunner (CU-U2)", () => {
const result = await runner.run(task, flagOn);
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["execute", "review", "merge"]);
expect(calls).toEqual(["planning", "execute", "workflow-step", "review", "merge"]);
expect(result.reason).toBeUndefined();
expect(getWorkflowDefinition).not.toHaveBeenCalled();
});

40
packages/engine/src/__tests__/workflow-task-runtime.test.ts
View File

@@ -29,7 +29,7 @@ function selectedIr(): WorkflowIr {
function recordingPrimitives(
calls: string[],
overrides: Partial<Record<"prepare" | "execute", WorkflowNodeResult>> = {},
overrides: Partial<Record<"prepare" | "execute" | "workflowStep", WorkflowNodeResult>> = {},
observed: { prepared?: PreparedWorktree } = {},
): WorkflowRuntimePrimitives {
const prepared: PreparedWorktree = { worktreePath: "/tmp/fusion-worktree" };
@@ -71,7 +71,16 @@ function recordingPrimitives(
};
},
runVerification: async () => ({ outcome: "success", data: { verdict: "skipped" } }),
runWorkflowStep: async () => ({ outcome: "success", data: { allPassed: true } }),
runWorkflowStep: async () => {
calls.push("workflow-step");
const override = overrides.workflowStep;
return {
outcome: override?.outcome ?? "success",
value: override?.value ?? "workflow-steps-passed",
contextPatch: override?.contextPatch,
data: { allPassed: override?.value !== "remediation-scheduled" },
};
},
updateSteps: async (_ctx, _task, steps) => ({ outcome: "success", data: { count: steps.length } }),
transitionTask: async () => {
calls.push("schedule");
@@ -153,8 +162,31 @@ describe("WorkflowTaskRuntime", () => {
const result = await runtime.run(task, flagOff);
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["planning", "prepare-worktree", "execute", "review", "merge"]);
expect(result.visitedNodeIds).toEqual(["start", "planning", "execute", "review", "merge"]);
expect(calls).toEqual(["planning", "prepare-worktree", "execute", "workflow-step", "review", "merge"]);
expect(result.visitedNodeIds).toEqual(["start", "planning", "execute", "workflow-step", "review", "merge"]);
});
it("stops the built-in workflow before review when workflow-step remediation is scheduled", async () => {
const calls: string[] = [];
const runtime = new WorkflowTaskRuntime({
store: {
getTaskWorkflowSelection: () => undefined,
getWorkflowDefinition: async () => undefined,
},
primitives: recordingPrimitives(calls, {
workflowStep: { outcome: "success", value: "remediation-scheduled" },
}),
runCustomNode: async (node) => {
calls.push(`custom:${node.id}`);
return { outcome: "success" };
},
});
const result = await runtime.run(task, flagOff);
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["planning", "prepare-worktree", "execute", "workflow-step"]);
expect(result.visitedNodeIds).toEqual(["start", "planning", "execute", "workflow-step"]);
});
it("fails selected workflow lookup misses instead of running the built-in workflow", async () => {

118
packages/engine/src/executor.ts
View File

@@ -4775,9 +4775,69 @@ export class TaskExecutor {
runVerification: async () => ({ outcome: "success", value: "verification-skipped", data: {
verdict: "skipped",
} }),
runWorkflowStep: async () => ({ outcome: "success", value: "workflow-step-skipped", data: {
allPassed: true,
} }),
runWorkflowStep: async (_ctx, task, input) => {
if (input.phase !== "pre-merge") {
return { outcome: "success", value: "workflow-step-skipped", data: { allPassed: true } };
}
const live = await this.store.getTask(task.id);
if (live.executionMode === "fast") {
executorLog.log(`${task.id}: fast mode — skipping pre-merge workflow steps`);
await this.store.logEntry(task.id, "Fast mode — pre-merge workflow steps skipped", undefined, this.getRunContextFor(task.id));
return { outcome: "success", value: "workflow-step-skipped", data: { allPassed: true } };
}
if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow steps after task completion")) {
return { outcome: "success", value: "deferred-paused", data: { allPassed: false } };
}
const worktreePath = input.worktreePath || live.worktree || this.rootDir;
const workflowResult = await this.runWorkflowSteps(live, worktreePath, settings, undefined);
if (workflowResult === "deferred-paused") {
if (await this.parkTaskAfterWorkflowStepPause(task.id)) {
this.pausedAborted.delete(task.id);
} else if (this.pausedAborted.has(task.id)) {
this.pausedAborted.delete(task.id);
}
return { outcome: "success", value: "deferred-paused", data: { allPassed: false } };
}
if (!workflowResult.allPassed) {
const feedback = workflowResult.feedback || "Workflow step failed";
const stepName = workflowResult.stepName || "Unknown";
if (workflowResult.revisionRequested) {
const rerunScheduled = await this.handleWorkflowRevisionRequest(
live,
worktreePath,
feedback,
stepName,
settings,
);
if (!rerunScheduled) {
return {
outcome: "failure",
value: "workflow-step-revision-unhandled",
data: workflowResult,
};
}
} else {
const retried = await this.handleWorkflowStepFailure(
live,
worktreePath,
feedback,
stepName,
);
if (!retried) {
await this.sendTaskBackForFix(
live,
worktreePath,
feedback,
stepName,
"Workflow step failed",
);
}
}
return { outcome: "success", value: "remediation-scheduled", data: workflowResult };
}
await this.store.updateTask(task.id, { workflowStepRetries: undefined, taskDoneRetryCount: null });
return { outcome: "success", value: "workflow-steps-passed", data: workflowResult };
},
updateSteps: async (_ctx, task, steps) => {
await this.store.updateTask(task.id, { steps });
return { outcome: "success", value: "steps-updated", data: { count: steps.length } };
@@ -4887,6 +4947,58 @@ export class TaskExecutor {
value: paused ? "implementation-paused" : "implementation-incomplete",
};
},
workflowStep: async (seamTask) => {
const live = await this.store.getTask(seamTask.id);
if (live.executionMode === "fast") {
executorLog.log(`${seamTask.id}: fast mode — skipping pre-merge workflow steps`);
await this.store.logEntry(seamTask.id, "Fast mode — pre-merge workflow steps skipped", undefined, this.getRunContextFor(seamTask.id));
return { outcome: "success", value: "workflow-step-skipped" };
}
const worktreePath = live.worktree || this.rootDir;
const settings = await this.store.getSettings();
const workflowResult = await this.runWorkflowSteps(live, worktreePath, settings, undefined);
if (workflowResult === "deferred-paused") {
if (await this.parkTaskAfterWorkflowStepPause(seamTask.id)) {
this.pausedAborted.delete(seamTask.id);
} else if (this.pausedAborted.has(seamTask.id)) {
this.pausedAborted.delete(seamTask.id);
}
return { outcome: "success", value: "deferred-paused" };
}
if (!workflowResult.allPassed) {
const feedback = workflowResult.feedback || "Workflow step failed";
const stepName = workflowResult.stepName || "Unknown";
if (workflowResult.revisionRequested) {
const rerunScheduled = await this.handleWorkflowRevisionRequest(
live,
worktreePath,
feedback,
stepName,
settings,
);
if (!rerunScheduled) return { outcome: "failure", value: "workflow-step-revision-unhandled" };
} else {
const retried = await this.handleWorkflowStepFailure(
live,
worktreePath,
feedback,
stepName,
);
if (!retried) {
await this.sendTaskBackForFix(
live,
worktreePath,
feedback,
stepName,
"Workflow step failed",
);
}
}
return { outcome: "success", value: "remediation-scheduled" };
}
await this.store.updateTask(seamTask.id, { workflowStepRetries: undefined, taskDoneRetryCount: null });
return { outcome: "success", value: "workflow-steps-passed" };
},
review: async (seamTask) => {
// The legacy "review" stage is the in-review handoff: per-step AI review
// already ran during implementation (fn_review_step), and the in-review

10
packages/engine/src/workflow-authoritative-driver.ts
View File

@@ -101,7 +101,15 @@ function primitivesFromLegacySeams(seams: WorkflowLegacySeams): WorkflowRuntimeP
return { ...result, data: { verdict: result.outcome === "success" ? "APPROVE" : "REVISE" } };
},
runVerification: async () => ({ outcome: "success", data: { verdict: "skipped" } }),
runWorkflowStep: async () => ({ outcome: "success", data: { allPassed: true } }),
runWorkflowStep: async (ctx, task) => {
const result = await seams.workflowStep?.(task, ctx.node.context ?? {});
return {
outcome: result?.outcome ?? "success",
value: result?.value ?? "workflow-step-skipped",
contextPatch: result?.contextPatch,
data: { allPassed: result?.outcome !== "failure" },
};
},
updateSteps: async (_ctx, _task, steps) => ({ outcome: "success", data: { count: steps.length } }),
transitionTask: async (ctx, task) => seams.schedule(task, ctx.node.context ?? {}),
requestMerge: async (ctx, task) => {

7
packages/engine/src/workflow-graph-executor.ts
View File

@@ -430,7 +430,12 @@ export class WorkflowGraphExecutor {
return sourceResult;
}
const matching = edges.filter((edge) => this.shouldTraverseEdge(edge, sourceResult));
const outcomeMatching = edges.filter((edge) =>
edge.condition?.startsWith("outcome:") && this.shouldTraverseEdge(edge, sourceResult)
);
const matching = outcomeMatching.length > 0
? outcomeMatching
: edges.filter((edge) => this.shouldTraverseEdge(edge, sourceResult));
if (matching.length === 0) {
return sourceResult;
}

5
packages/engine/src/workflow-graph-task-runner.ts
View File

@@ -171,6 +171,11 @@ export class WorkflowGraphTaskRunner {
const wrappedSeams: WorkflowLegacySeams = {
planning: (t, c) => ((sideEffectsRan = true), invoked.push("planning"), seams.planning(t, c)),
execute: (t, c) => ((sideEffectsRan = true), invoked.push("execute"), seams.execute(t, c)),
workflowStep: (t, c) => {
sideEffectsRan = true;
invoked.push("workflow-step");
return seams.workflowStep?.(t, c) ?? Promise.resolve({ outcome: "success", value: "workflow-step-skipped" });
},
review: (t, c) => ((sideEffectsRan = true), invoked.push("review"), seams.review(t, c)),
merge: (t, c) => ((sideEffectsRan = true), invoked.push("merge"), seams.merge(t, c)),
schedule: (t, c) => ((sideEffectsRan = true), invoked.push("schedule"), seams.schedule(t, c)),

36
packages/engine/src/workflow-node-handlers.ts
View File

@@ -9,7 +9,14 @@ import {
type WorkflowRuntimePrimitives,
} from "./runtime-primitives.js";
export type WorkflowSeamName = "planning" | "execute" | "review" | "merge" | "schedule" | "step-execute";
export type WorkflowSeamName =
| "planning"
| "execute"
| "workflow-step"
| "review"
| "merge"
| "schedule"
| "step-execute";
export interface WorkflowLegacySeams {
/** Planning/spec stage. Built-in triage runs upstream of the interpreter
@@ -17,6 +24,7 @@ export interface WorkflowLegacySeams {
* custom planning behavior is expressed as a custom prompt node. */
planning: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
execute: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
workflowStep?: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
review: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
merge: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
schedule: (task: TaskDetail, context: Record<string, unknown>) => Promise<WorkflowNodeResult>;
@@ -195,6 +203,7 @@ export function resolveSeamName(node: { config?: Record<string, unknown> }): Wor
if (
seam === "planning" ||
seam === "execute" ||
seam === "workflow-step" ||
seam === "review" ||
seam === "merge" ||
seam === "schedule" ||
@@ -250,6 +259,11 @@ export function createPromptLikeHandler(
// IS the seam node, so its declared column drives the binding. (Other seams
// — planning/review/merge/schedule — stamp it too; only execute reads it.)
context.context[SEAM_GOVERNING_NODE_CONTEXT_KEY] = node.id;
if (seam === "workflow-step") {
return seams.workflowStep
? seams.workflowStep(context.task, context.context)
: { outcome: "success", value: "workflow-step-skipped" };
}
return seams[seam]!(context.task, context.context);
}
if (!runCustomNode) {
@@ -317,7 +331,24 @@ export function createPrimitivePromptLikeHandler(
...(result.contextPatch ?? {}),
}
: undefined;
return { outcome: result.outcome, value: result.value, contextPatch };
return {
outcome: result.outcome,
value: result.value,
contextPatch: {
...(contextPatch ?? {}),
"workflow:worktree-path": prepared.data.worktreePath,
},
};
}
if (seam === "workflow-step") {
const worktreePath = typeof context.context["workflow:worktree-path"] === "string"
? context.context["workflow:worktree-path"]
: undefined;
const result = await primitives.runWorkflowStep(primitiveCtx, context.task, {
phase: "pre-merge",
worktreePath,
});
return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch };
}
if (seam === "review") {
const result = await primitives.runReview(primitiveCtx, context.task, { type: "code" });
@@ -798,6 +829,7 @@ export function createNoopLegacySeams(): WorkflowLegacySeams {
return {
planning: success,
execute: success,
workflowStep: success,
review: success,
merge: success,
schedule: success,