From b6800e8622167dde1fe32553776aa9330bd2b8f1 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 8 Jun 2026 21:16:32 -0700 Subject: [PATCH] fix(FN-6035): address workflow primitive review feedback --- packages/core/src/workflow-compiler.ts | 2 +- .../__tests__/workflow-task-runtime.test.ts | 58 +++++++-------- packages/engine/src/executor.ts | 43 ++++++++--- packages/engine/src/project-engine.ts | 71 +++++++++++++++---- .../engine/src/runtimes/in-process-runtime.ts | 12 +++- .../src/workflow-authoritative-driver.ts | 19 ++++- packages/engine/src/workflow-node-handlers.ts | 18 ++++- packages/engine/src/workflow-task-runtime.ts | 28 ++++---- 8 files changed, 176 insertions(+), 75 deletions(-) diff --git a/packages/core/src/workflow-compiler.ts b/packages/core/src/workflow-compiler.ts index b920782a3c..b240105470 100644 --- a/packages/core/src/workflow-compiler.ts +++ b/packages/core/src/workflow-compiler.ts @@ -131,7 +131,7 @@ export function validateLinearity(ir: WorkflowIr): WorkflowCompileError | null { nextExpectedSeamIndex += 1; } if (expectedSeamOrder[nextExpectedSeamIndex] !== seam) { - return new WorkflowCompileError("seams must follow the execute -> review -> merge order"); + return new WorkflowCompileError("seams must follow the planning -> execute -> review -> merge order"); } seenSeams.add(seam); nextExpectedSeamIndex += 1; diff --git a/packages/engine/src/__tests__/workflow-task-runtime.test.ts b/packages/engine/src/__tests__/workflow-task-runtime.test.ts index 8e3b4ba2d2..3ad467a306 100644 --- a/packages/engine/src/__tests__/workflow-task-runtime.test.ts +++ b/packages/engine/src/__tests__/workflow-task-runtime.test.ts @@ -29,13 +29,19 @@ function selectedIr(): WorkflowIr { function recordingPrimitives( calls: string[], - overrides: Partial> = {}, + overrides: Partial> = {}, + 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 () => { diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 600f705a8e..b0d68b98b8 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -3637,9 +3637,9 @@ export class TaskExecutor { private static processWideGraphRouting = new Set(); /** Wired by the runtime to ProjectEngine.onMerge — resolves with the merge outcome. */ - private mergeRequester?: (taskId: string) => Promise; + private mergeRequester?: (taskId: string, options?: { signal?: AbortSignal }) => Promise; - setMergeRequester(requestMerge: (taskId: string) => Promise): void { + setMergeRequester(requestMerge: (taskId: string, options?: { signal?: AbortSignal }) => Promise): 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 | 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" } }; diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index c9c735d945..578d9d7d00 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -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 { - // 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((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 { + const signal = options.signal; + if (signal?.aborted) { + throw new Error(`Merge request for ${taskId} aborted`); } return new Promise((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 { + async requestInterpreterMerge(taskId: string, options: { signal?: AbortSignal } = {}): Promise { 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( diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index c83aa10650..33c2804279 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -142,7 +142,10 @@ export class InProcessRuntime * before `start()` via `setMergeEnqueuer`. */ private mergeEnqueuer?: (taskId: string) => boolean; - private mergeRequester?: (taskId: string) => Promise; + private mergeRequester?: ( + taskId: string, + options?: { signal?: AbortSignal }, + ) => Promise; 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): void { + setMergeRequester( + requestMerge: ( + taskId: string, + options?: { signal?: AbortSignal }, + ) => Promise, + ): void { this.mergeRequester = requestMerge; this.executor?.setMergeRequester(requestMerge); } diff --git a/packages/engine/src/workflow-authoritative-driver.ts b/packages/engine/src/workflow-authoritative-driver.ts index 6ec2fa4fb1..e156a17125 100644 --- a/packages/engine/src/workflow-authoritative-driver.ts +++ b/packages/engine/src/workflow-authoritative-driver.ts @@ -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" }, }; } diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts index a8d86fecfa..caf75de157 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/packages/engine/src/workflow-node-handlers.ts @@ -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 | 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 = { + ...(primitivePatch ?? {}), [FOREACH_ACTIVE_CONTEXT_KEY]: active, [`node:${node.id}:verdict`]: result.verdict, }; diff --git a/packages/engine/src/workflow-task-runtime.ts b/packages/engine/src/workflow-task-runtime.ts index daf6aa5644..ecf5d15b2d 100644 --- a/packages/engine/src/workflow-task-runtime.ts +++ b/packages/engine/src/workflow-task-runtime.ts @@ -40,11 +40,11 @@ export interface WorkflowTaskRuntimeDeps extends Omit> {