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

fix(FN-6035): address workflow primitive review feedback

Browse Source
This commit is contained in:
gsxdsm
2026-06-08 21:16:32 -07:00
parent 7a259de3c4
commit b6800e8622
8 changed files with 176 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -29,13 +29,19 @@ function selectedIr(): WorkflowIr {
function recordingPrimitives(
calls: string[],
overrides: Partial<Record<"execute", WorkflowNodeResult>> = {},
overrides: Partial<Record<"prepare" | "execute", WorkflowNodeResult>> = {},
observed: { prepared?: PreparedWorktree } = {},
): WorkflowRuntimePrimitives {
const prepared: PreparedWorktree = { worktreePath: "/tmp/fusion-worktree" };
return {
prepareWorktree: async () => {
calls.push("prepare-worktree");
return { outcome: "success", data: prepared };
return {
outcome: overrides.prepare?.outcome ?? "success",
value: overrides.prepare?.value,
contextPatch: overrides.prepare?.contextPatch,
data: overrides.prepare?.outcome === "failure" ? undefined : prepared,
};
},
readArtifact: async () => undefined,
writeArtifact: async (_ctx, _task, key) => ({ outcome: "success", data: { key } }),
@@ -43,8 +49,9 @@ function recordingPrimitives(
calls.push("planning");
return { outcome: "success", data: { approved: true, artifactKeys: [] } };
},
runCodingSession: async () => {
runCodingSession: async (_ctx, _task, preparedWorktree) => {
calls.push("execute");
observed.prepared = preparedWorktree;
const override = overrides.execute;
return {
outcome: override?.outcome ?? "success",
@@ -94,6 +101,7 @@ describe("WorkflowTaskRuntime", () => {
it("runs a selected workflow through the graph engine", async () => {
const calls: string[] = [];
const observed: { prepared?: PreparedWorktree } = {};
let workflowSelectionReads = 0;
const runtime = new WorkflowTaskRuntime({
store: {
@@ -103,7 +111,14 @@ describe("WorkflowTaskRuntime", () => {
},
getWorkflowDefinition: async () => ({ ir: selectedIr() }),
},
primitives: recordingPrimitives(calls),
primitives: recordingPrimitives(
calls,
{
prepare: { outcome: "success", contextPatch: { preparedKey: "from-prepare" } },
execute: { outcome: "success", contextPatch: { executeKey: "from-execute" } },
},
observed,
),
runCustomNode: async (node) => {
calls.push(`custom:${node.id}`);
return { outcome: "success" };
@@ -115,6 +130,9 @@ describe("WorkflowTaskRuntime", () => {
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["custom:prepare", "prepare-worktree", "execute"]);
expect(result.visitedNodeIds).toEqual(["start", "prepare", "execute"]);
expect(observed.prepared).toEqual({ worktreePath: "/tmp/fusion-worktree" });
expect(result.context.preparedKey).toBe("from-prepare");
expect(result.context.executeKey).toBe("from-execute");
expect(workflowSelectionReads).toBe(1);
});
@@ -139,9 +157,8 @@ describe("WorkflowTaskRuntime", () => {
expect(result.visitedNodeIds).toEqual(["start", "planning", "execute", "review", "merge"]);
});
it("turns selected workflow lookup failures into the built-in workflow target", async () => {
it("fails selected workflow lookup misses instead of running the built-in workflow", async () => {
const calls: string[] = [];
const observedRunIds: string[] = [];
const runtime = new WorkflowTaskRuntime({
store: {
getTaskWorkflowSelection: () => ({ workflowId: "WF-MISSING", stepIds: [] }),
@@ -149,25 +166,17 @@ describe("WorkflowTaskRuntime", () => {
},
primitives: recordingPrimitives(calls),
runCustomNode: async () => ({ outcome: "success" }),
branchPersistence: {
loadBranchStates: (_taskId, runId) => {
observedRunIds.push(runId);
return [];
},
},
});
const result = await runtime.run(task, flagOff);
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["planning", "prepare-worktree", "execute", "review", "merge"]);
expect(observedRunIds).toContain("FN-9002:builtin:coding");
expect(observedRunIds).not.toContain("FN-9002:WF-MISSING");
expect(result.disposition).toBe("failed");
expect(result.reason).toContain("workflow-resolution-error: workflow-missing: WF-MISSING");
expect(calls).toEqual([]);
});
it("turns corrupt selected workflow definitions into the built-in workflow target", async () => {
it("fails corrupt selected workflow definitions instead of running the built-in workflow", async () => {
const calls: string[] = [];
const observedRunIds: string[] = [];
const runtime = new WorkflowTaskRuntime({
store: {
getTaskWorkflowSelection: () => ({ workflowId: "WF-CORRUPT", stepIds: [] }),
@@ -175,20 +184,13 @@ describe("WorkflowTaskRuntime", () => {
},
primitives: recordingPrimitives(calls),
runCustomNode: async () => ({ outcome: "success" }),
branchPersistence: {
loadBranchStates: (_taskId, runId) => {
observedRunIds.push(runId);
return [];
},
},
});
const result = await runtime.run(task, flagOff);
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["planning", "prepare-worktree", "execute", "review", "merge"]);
expect(observedRunIds).toContain("FN-9002:builtin:coding");
expect(observedRunIds).not.toContain("FN-9002:WF-CORRUPT");
expect(result.disposition).toBe("failed");
expect(result.reason).toContain("workflow-resolution-error:");
expect(calls).toEqual([]);
});
it("forces only the graph executor flag while preserving other settings", async () => {

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

@@ -3637,9 +3637,9 @@ export class TaskExecutor {
private static processWideGraphRouting = new Set<string>();
/** Wired by the runtime to ProjectEngine.onMerge — resolves with the merge outcome. */
private mergeRequester?: (taskId: string) => Promise<MergeResult>;
private mergeRequester?: (taskId: string, options?: { signal?: AbortSignal }) => Promise<MergeResult>;
setMergeRequester(requestMerge: (taskId: string) => Promise<MergeResult>): void {
setMergeRequester(requestMerge: (taskId: string, options?: { signal?: AbortSignal }) => Promise<MergeResult>): void {
this.mergeRequester = requestMerge;
}
@@ -3678,8 +3678,14 @@ export class TaskExecutor {
let selection: { workflowId: string; stepIds: string[] } | undefined;
try {
selection = this.store.getTaskWorkflowSelection?.(task.id);
} catch {
selection = undefined;
} catch (err) {
await this.handleGraphFailure(task, {
disposition: "failed",
outcome: "failure",
reason: `workflow-selection-failed: ${err instanceof Error ? err.message : String(err)}`,
visitedNodeIds: [],
});
return true;
}
selection ??= { workflowId: "builtin:coding", stepIds: [] };
@@ -3737,7 +3743,8 @@ export class TaskExecutor {
getTaskWorkflowSelection: (taskId: string) =>
this.store.getTaskWorkflowSelection?.(taskId) ?? { workflowId: "builtin:coding", stepIds: [] },
getWorkflowDefinition: async (id: string) =>
(await this.store.getWorkflowDefinition?.(id)) ?? getBuiltinWorkflow("builtin:coding"),
(await this.store.getWorkflowDefinition?.(id))
?? (id === "builtin:coding" ? getBuiltinWorkflow("builtin:coding") : undefined),
},
runId: resolvedRunId,
primitives: this.createAuthoritativeWorkflowPrimitives(settings),
@@ -4454,13 +4461,23 @@ export class TaskExecutor {
* interceptor makes execute() stop at the completion boundary instead of
* running workflow steps and the review handoff.
*/
private async runImplementationPhase(task: Task): Promise<{ taskDone: boolean; modifiedFiles: string[] }> {
private async runImplementationPhase(
task: Task,
prepared?: PreparedWorktree,
): Promise<{ taskDone: boolean; modifiedFiles: string[] }> {
let captured: { taskDone: boolean; modifiedFiles: string[] } = { taskDone: false, modifiedFiles: [] };
this.graphCompletionInterceptors.set(task.id, (info) => {
captured = { taskDone: true, modifiedFiles: info.modifiedFiles };
});
const executionTask = prepared
? {
...task,
worktree: prepared.worktreePath || task.worktree,
branch: prepared.branchName || task.branch,
}
: task;
try {
await this.execute(task);
await this.execute(executionTask);
} finally {
this.graphCompletionInterceptors.delete(task.id);
}
@@ -4633,14 +4650,14 @@ export class TaskExecutor {
approved: true,
artifactKeys: [],
} }),
runCodingSession: async (ctx, task) => {
runCodingSession: async (ctx, task, prepared) => {
const governingNodeId = ctx.node.context?.[SEAM_GOVERNING_NODE_CONTEXT_KEY];
if (typeof governingNodeId === "string") {
this.graphSeamGoverningNodeId.set(task.id, governingNodeId);
}
let result: { taskDone: boolean; modifiedFiles: string[] };
try {
result = await this.runImplementationPhase(task);
result = await this.runImplementationPhase(task, prepared);
} finally {
this.graphSeamGoverningNodeId.delete(task.id);
}
@@ -4779,13 +4796,17 @@ export class TaskExecutor {
return { outcome: "failure", value: "merge-unavailable", data: { status: "failed", reason: "merge-unavailable" } };
}
const GRAPH_MERGE_TIMEOUT_MS = 30 * 60 * 1000;
const controller = new AbortController();
let timeoutHandle: ReturnType<typeof setTimeout> | undefined;
const timeout = new Promise<"timeout">((resolve) => {
timeoutHandle = setTimeout(() => resolve("timeout"), GRAPH_MERGE_TIMEOUT_MS);
timeoutHandle = setTimeout(() => {
controller.abort();
resolve("timeout");
}, GRAPH_MERGE_TIMEOUT_MS);
timeoutHandle.unref?.();
});
try {
const result = await Promise.race([this.mergeRequester(task.id), timeout]);
const result = await Promise.race([this.mergeRequester(task.id, { signal: controller.signal }), timeout]);
if (result === "timeout") {
executorLog.warn(`${task.id}: workflow merge primitive timed out after ${GRAPH_MERGE_TIMEOUT_MS}ms`);
return { outcome: "failure", value: "merge-timeout", data: { status: "timeout" } };

71
packages/engine/src/project-engine.ts
View File

@@ -331,6 +331,14 @@ export class ProjectEngine {
else this.manualMergeResolvers.set(taskId, [r]);
}
private removeMergeResolver(taskId: string, resolver: MergeResolver): void {
const list = this.manualMergeResolvers.get(taskId);
if (!list) return;
const next = list.filter((candidate) => candidate !== resolver);
if (next.length > 0) this.manualMergeResolvers.set(taskId, next);
else this.manualMergeResolvers.delete(taskId);
}
/** Remove and return all waiters for a task (empty array if none). */
private takeMergeResolvers(taskId: string): MergeResolver[] {
const list = this.manualMergeResolvers.get(taskId);
@@ -426,7 +434,7 @@ export class ProjectEngine {
// Workflow-graph interpreter merge seam: routes through the auto-merge
// eligibility gate (requestInterpreterMerge), NOT the human "merge now"
// bypass, so a graph merge node can't override an autoMerge-off project.
this.runtime.setMergeRequester?.((taskId) => this.requestInterpreterMerge(taskId));
this.runtime.setMergeRequester?.((taskId, options) => this.requestInterpreterMerge(taskId, options));
}
getActiveMergeTaskId(): string | null {
@@ -1072,21 +1080,56 @@ export class ProjectEngine {
* Returns the full MergeResult so it can be used as the `onMerge` callback
* in createServer().
*/
async onMerge(taskId: string): Promise<MergeResult> {
// If this task is already queued or actively merging, wait for the
// existing merge to finish rather than starting a second one.
if (this.mergeActive.has(taskId)) {
return new Promise<MergeResult>((resolve, reject) => {
this.addMergeResolver(taskId, { resolve, reject });
// Don't re-enqueue — the task is already in the queue/active
});
async onMerge(taskId: string, options: { signal?: AbortSignal } = {}): Promise<MergeResult> {
const signal = options.signal;
if (signal?.aborted) {
throw new Error(`Merge request for ${taskId} aborted`);
}
return new Promise<MergeResult>((resolve, reject) => {
this.addMergeResolver(taskId, { resolve, reject });
let settled = false;
let abort: () => void = () => undefined;
const cleanup = () => {
signal?.removeEventListener("abort", abort);
};
const resolver: MergeResolver = {
resolve: (result) => {
if (settled) return;
settled = true;
cleanup();
resolve(result);
},
reject: (err) => {
if (settled) return;
settled = true;
cleanup();
reject(err);
},
};
abort = () => {
this.removeMergeResolver(taskId, resolver);
if (this.activeMergeTaskId === taskId) {
this.mergeAbortController?.abort();
this.mergeAbortController = null;
this.activeMergeSession?.dispose();
this.activeMergeSession = null;
} else if (!this.hasMergeResolvers(taskId)) {
this.mergeQueue = this.mergeQueue.filter((queuedTaskId) => queuedTaskId !== taskId);
this.mergeActive.delete(taskId);
}
resolver.reject(new Error(`Merge request for ${taskId} aborted`));
};
signal?.addEventListener("abort", abort, { once: true });
this.addMergeResolver(taskId, resolver);
// If this task is already queued or actively merging, wait for the
// existing merge to finish rather than starting a second one.
if (this.mergeActive.has(taskId)) return;
if (!this.internalEnqueueMerge(taskId)) {
// Drop just-added waiter(s) for this task and fail them.
this.rejectMergeResolvers(taskId, new Error(`Merge enqueue rejected for ${taskId}`));
this.removeMergeResolver(taskId, resolver);
resolver.reject(new Error(`Merge enqueue rejected for ${taskId}`));
}
});
}
@@ -1099,7 +1142,7 @@ export class ProjectEngine {
* it as "manual merge required" and parks the task in review — preserving the
* contract that autoMerge-off leaves in-review terminal until a human merges.
*/
async requestInterpreterMerge(taskId: string): Promise<MergeResult> {
async requestInterpreterMerge(taskId: string, options: { signal?: AbortSignal } = {}): Promise<MergeResult> {
let task: Task | null = null;
let settings: Settings | undefined;
try {
@@ -1135,7 +1178,7 @@ export class ProjectEngine {
} as MergeResult;
}
// Eligible: route through the normal serialized merge path.
return this.onMerge(taskId);
return this.onMerge(taskId, options);
}
private setRestoreDiagnostics(

12
packages/engine/src/runtimes/in-process-runtime.ts
View File

@@ -142,7 +142,10 @@ export class InProcessRuntime
* before `start()` via `setMergeEnqueuer`.
*/
private mergeEnqueuer?: (taskId: string) => boolean;
private mergeRequester?: (taskId: string) => Promise<import("@fusion/core").MergeResult>;
private mergeRequester?: (
taskId: string,
options?: { signal?: AbortSignal },
) => Promise<import("@fusion/core").MergeResult>;
private clearMergeActive?: (taskId: string) => void;
private activeMergeTaskIdProvider?: () => string | null;
/** Tracks whether startup recovery was intentionally deferred due to pause state. */
@@ -1140,7 +1143,12 @@ export class InProcessRuntime
* forwards immediately when the executor already exists, and is re-applied at
* executor construction during start().
*/
setMergeRequester(requestMerge: (taskId: string) => Promise<import("@fusion/core").MergeResult>): void {
setMergeRequester(
requestMerge: (
taskId: string,
options?: { signal?: AbortSignal },
) => Promise<import("@fusion/core").MergeResult>,
): void {
this.mergeRequester = requestMerge;
this.executor?.setMergeRequester(requestMerge);
}

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

@@ -14,7 +14,7 @@ import type { TaskExecutor } from "./executor.js";
import { executorLog } from "./logger.js";
import { WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG } from "./workflow-parity-observer.js";
import { WorkflowGraphTaskRunner, type WorkflowGraphTaskRunResult } from "./workflow-graph-task-runner.js";
import type { WorkflowLegacySeams } from "./workflow-node-handlers.js";
import type { StepReviewSeamResult, WorkflowLegacySeams } from "./workflow-node-handlers.js";
import type { PreparedWorktree, WorkflowRuntimePrimitives } from "./runtime-primitives.js";
const AUTHORITATIVE_WORKFLOW_ID = "workflow-interpreter-authoritative";
@@ -52,6 +52,21 @@ function buildAuthoritativeSettings(settings: Settings): Settings {
}
function primitivesFromLegacySeams(seams: WorkflowLegacySeams): WorkflowRuntimePrimitives {
const mapStepReviewValue = (verdict: StepReviewSeamResult["verdict"] | undefined): string => {
switch (verdict) {
case "APPROVE":
return "approve";
case "REVISE":
return "revise";
case "RETHINK":
return "rethink";
default:
return "unavailable";
}
};
// Legacy seams do not consume PreparedWorktree; filesystem/session state is
// still owned inside the seam implementation they delegate to.
const prepared: PreparedWorktree = { worktreePath: "" };
return {
prepareWorktree: async () => ({ outcome: "success", data: prepared }),
@@ -78,7 +93,7 @@ function primitivesFromLegacySeams(seams: WorkflowLegacySeams): WorkflowRuntimeP
const result = await seams.stepReview?.(task, ctx.node.context ?? {}, { type: input.type });
return {
outcome: "success",
value: result?.verdict === "APPROVE" ? "approve" : result?.verdict === "REVISE" ? "revise" : result?.verdict === "RETHINK" ? "rethink" : "unavailable",
value: mapStepReviewValue(result?.verdict),
data: result ?? { verdict: "UNAVAILABLE" },
};
}

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

@@ -311,7 +311,13 @@ export function createPrimitivePromptLikeHandler(
};
}
const result = await primitives.runCodingSession(primitiveCtx, context.task, prepared.data);
return { outcome: result.outcome, value: result.value, contextPatch: result.contextPatch };
const contextPatch = prepared.contextPatch || result.contextPatch
? {
...(prepared.contextPatch ?? {}),
...(result.contextPatch ?? {}),
}
: undefined;
return { outcome: result.outcome, value: result.value, contextPatch };
}
if (seam === "review") {
const result = await primitives.runReview(primitiveCtx, context.task, { type: "code" });
@@ -467,6 +473,7 @@ export function createPrimitiveStepReviewHandler(primitives: WorkflowRuntimePrim
let result: StepReviewSeamResult = {
verdict: "UNAVAILABLE",
};
let primitivePatch: Record<string, unknown> | undefined;
for (let attempt = 0; attempt <= STEP_REVIEW_UNAVAILABLE_RETRY_CAP; attempt++) {
const primitiveResult = await primitives.runReview(
primitiveContextForNode(node, ctx.task, ctx.context, attempt + 1),
@@ -477,6 +484,14 @@ export function createPrimitiveStepReviewHandler(primitives: WorkflowRuntimePrim
baselineSha: config.type === "code" ? active.baselineSha : undefined,
},
);
if (primitiveResult.outcome !== "success") {
return {
outcome: primitiveResult.outcome,
value: primitiveResult.value,
contextPatch: primitiveResult.contextPatch,
};
}
primitivePatch = primitiveResult.contextPatch;
result = primitiveResult.data ?? { verdict: "UNAVAILABLE" as const };
if (result.verdict !== "UNAVAILABLE") break;
}
@@ -485,6 +500,7 @@ export function createPrimitiveStepReviewHandler(primitives: WorkflowRuntimePrim
active.verdict = result.verdict;
}
const patch: Record<string, unknown> = {
...(primitivePatch ?? {}),
[FOREACH_ACTIVE_CONTEXT_KEY]: active,
[`node:${node.id}:verdict`]: result.verdict,
};

28
packages/engine/src/workflow-task-runtime.ts
View File

@@ -40,11 +40,11 @@ export interface WorkflowTaskRuntimeDeps extends Omit<WorkflowGraphExecutorDeps,
/**
* WorkflowTaskRuntime is the workflow-engine execution facade.
*
* It always resolves a task to a workflow IR: explicit selections resolve to
* their selected workflow, and tasks without a selection resolve to the built-in
* coding workflow. This is intentionally
* different from `WorkflowGraphTaskRunner`, whose current contract still models
* "no selection" as legacy fallback.
* It always resolves a task to a workflow IR: explicit selections resolve only
* to their selected workflow, and tasks without a selection resolve to the
* built-in coding workflow. This is intentionally different from
* `WorkflowGraphTaskRunner`, whose current contract still models "no selection"
* as legacy fallback.
*/
export class WorkflowTaskRuntime {
public constructor(private readonly deps: WorkflowTaskRuntimeDeps) {}
@@ -118,27 +118,23 @@ export class WorkflowTaskRuntime {
let workflowId: string | undefined;
try {
workflowId = this.deps.store.getTaskWorkflowSelection(taskId)?.workflowId;
} catch {
return builtinCodingTarget();
} catch (err) {
throw new Error(`workflow-selection-failed: ${err instanceof Error ? err.message : String(err)}`);
}
if (!workflowId) return builtinCodingTarget();
if (isBuiltinWorkflowId(workflowId)) {
const builtin = getBuiltinWorkflow(workflowId);
if (!builtin) return builtinCodingTarget();
if (!builtin) throw new Error(`workflow-missing: ${workflowId}`);
const ir = typeof builtin.ir === "string" ? parseWorkflowIr(builtin.ir) : builtin.ir;
return { workflowId, ir };
}
try {
const def = await this.deps.store.getWorkflowDefinition(workflowId);
if (!def) return builtinCodingTarget();
const ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir;
return { workflowId, ir };
} catch {
return builtinCodingTarget();
}
const def = await this.deps.store.getWorkflowDefinition(workflowId);
if (!def) throw new Error(`workflow-missing: ${workflowId}`);
const ir = typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir;
return { workflowId, ir };
}
private recordingHandlers(invoked: string[]): Partial<Record<WorkflowIrNode["kind"], WorkflowNodeHandler>> {