From e89a58f2765c5818f9665dd5016121c58e559231 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 26 Jun 2026 14:17:07 -0700 Subject: [PATCH] FN-7066: schedule fixes for failed optional steps Route failed pre-merge optional workflow steps back through executor remediation before review/merge. - Add a graph-executor seam that schedules bounded executor fixes for pre-merge optional REVISE results. - Track optional-step fix attempts with the existing post-review fix budget before falling back to advisory/gate behavior. - Cover optional Code Review / Browser Verification fix scheduling and exhausted-budget behavior with engine tests. - Document the pre-merge remediation semantics and add a release changeset. Files changed: .changeset/fn-7066-optional-step-fix.md | 7 + docs/workflow-steps.md | 11 +- packages/engine/src/__tests__/self-healing.test.ts | 43 +++++ .../workflow-graph-optional-group.test.ts | 206 +++++++++++++++++++++ .../workflow-graph-optional-step-fix.test.ts | 89 +++++++++ packages/engine/src/executor.ts | 50 ++++- packages/engine/src/workflow-graph-executor.ts | 29 +++ packages/engine/src/workflow-graph-task-runner.ts | 5 +- 8 files changed, 436 insertions(+), 4 deletions(-) Fusion-Task-Id: FN-7066 Fusion-Task-Lineage: 6b584293-4f2a-4851-a0d7-3ec9e68fe31a --- .changeset/fn-7066-optional-step-fix.md | 7 + docs/workflow-steps.md | 11 +- .../engine/src/__tests__/self-healing.test.ts | 43 ++++ .../workflow-graph-optional-group.test.ts | 206 ++++++++++++++++++ .../workflow-graph-optional-step-fix.test.ts | 89 ++++++++ packages/engine/src/executor.ts | 50 ++++- .../engine/src/workflow-graph-executor.ts | 29 +++ .../engine/src/workflow-graph-task-runner.ts | 5 +- 8 files changed, 436 insertions(+), 4 deletions(-) create mode 100644 .changeset/fn-7066-optional-step-fix.md create mode 100644 packages/engine/src/__tests__/workflow-graph-optional-step-fix.test.ts diff --git a/.changeset/fn-7066-optional-step-fix.md b/.changeset/fn-7066-optional-step-fix.md new file mode 100644 index 0000000000..7435ed4348 --- /dev/null +++ b/.changeset/fn-7066-optional-step-fix.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": minor +--- + +summary: Failed optional workflow steps now send tasks back for a bounded executor fix pass. +category: feature +dev: New `requestPreMergeOptionalStepFix` graph-executor seam wired to `sendTaskBackForFix`; bounded by `maxPostReviewFixes`/`postReviewFixCount`; falls through to prior advisory/gate behavior once the budget is exhausted. Pre-merge phase only; post-merge optional groups stay non-blocking. diff --git a/docs/workflow-steps.md b/docs/workflow-steps.md index 344a5425b0..1ea4c0c414 100644 --- a/docs/workflow-steps.md +++ b/docs/workflow-steps.md @@ -550,7 +550,16 @@ Authoritative cutover now depends on existing/current parity summary evidence, n If a task is found in `in-review` with failed pre-merge workflow results and no active executor, self-healing can auto-revive it (bounded by `maxPostReviewFixes`) by replaying the same remediation send-back flow. -Advisory failures are intentionally excluded from merge blocking and auto-revive. + + +During a live graph run, an enabled **pre-merge** optional step that returns `REVISE` (including the built-in **Code Review** / `code-review` and **Browser Verification** / `browser-verification` groups) sends the task back to the executor for a bounded fix pass before the graph continues to review or merge. The same `postReviewFixCount` / `maxPostReviewFixes` budget used by self-healing caps this inline remediation. When the budget is exhausted or disabled (`maxPostReviewFixes <= 0`), behavior falls through to the prior semantics: advisory results remain non-blocking and gate failures remain failed/parked. + +Post-merge optional groups never trigger this send-back path because merge has already happened; their failures are recorded/logged as non-blocking post-merge results. + +Advisory failures are intentionally excluded from merge blocking and self-healing auto-revive after a task is already in review; their live-run fix pass is only the bounded pre-review remediation described above. ## Viewing Results diff --git a/packages/engine/src/__tests__/self-healing.test.ts b/packages/engine/src/__tests__/self-healing.test.ts index 293d2f7852..b5f31b1443 100644 --- a/packages/engine/src/__tests__/self-healing.test.ts +++ b/packages/engine/src/__tests__/self-healing.test.ts @@ -5425,6 +5425,49 @@ describe("SelfHealingManager", () => { managerWithRecovery.stop(); }); + it("does not double-fire against an inline-rescheduled in-progress task", async () => { + const recoverFn = vi.fn().mockResolvedValue(true); + const managerWithRecovery = new SelfHealingManager(store, { + rootDir: "/tmp/test-project", + recoverFailedPreMergeStep: recoverFn, + }); + (store.getSettings as ReturnType).mockResolvedValue({ + maxPostReviewFixes: 2, + }); + (store.listTasks as ReturnType).mockResolvedValue([ + { ...baseTask, column: "in-progress", postReviewFixCount: 1 }, + ]); + + const result = await managerWithRecovery.recoverReviewTasksWithFailedPreMergeSteps(); + + expect(result).toBe(0); + expect(recoverFn).not.toHaveBeenCalled(); + expect(store.updateTask).not.toHaveBeenCalled(); + + managerWithRecovery.stop(); + }); + + it("does not revive an already in-review task when auto-merge processing is disabled", async () => { + const recoverFn = vi.fn().mockResolvedValue(true); + const managerWithRecovery = new SelfHealingManager(store, { + rootDir: "/tmp/test-project", + recoverFailedPreMergeStep: recoverFn, + }); + (store.getSettings as ReturnType).mockResolvedValue({ + maxPostReviewFixes: 2, + autoMerge: false, + }); + (store.listTasks as ReturnType).mockResolvedValue([{ ...baseTask }]); + + const result = await managerWithRecovery.recoverReviewTasksWithFailedPreMergeSteps(); + + expect(result).toBe(0); + expect(recoverFn).not.toHaveBeenCalled(); + expect(store.updateTask).not.toHaveBeenCalled(); + + managerWithRecovery.stop(); + }); + it("leaves tasks with non-pre-merge blockers alone (e.g. incomplete steps)", async () => { const recoverFn = vi.fn().mockResolvedValue(true); const managerWithRecovery = new SelfHealingManager(store, { diff --git a/packages/engine/src/__tests__/workflow-graph-optional-group.test.ts b/packages/engine/src/__tests__/workflow-graph-optional-group.test.ts index dbf29e77cb..8817d74a70 100644 --- a/packages/engine/src/__tests__/workflow-graph-optional-group.test.ts +++ b/packages/engine/src/__tests__/workflow-graph-optional-group.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it, vi } from "vitest"; +import { BUILTIN_CODING_WORKFLOW_IR, BUILTIN_STEPWISE_CODING_WORKFLOW_IR } from "@fusion/core"; import type { TaskDetail, WorkflowIr } from "@fusion/core"; import { WorkflowGraphExecutor, type WorkflowNodeHandler } from "../workflow-graph-executor.js"; @@ -87,6 +88,38 @@ function taskWith(enabled: string[] | undefined): TaskDetail { return { id: "FN-OG", enabledWorkflowSteps: enabled } as TaskDetail; } +function reviseGroupIr(options: { phase?: "pre-merge" | "post-merge"; gateMode?: "advisory" | "gate" } = {}): WorkflowIr { + return { + version: "v2", + name: "optional-group-revise-test", + columns: [{ id: "work", name: "Work", traits: [] }], + nodes: [ + { id: "start", kind: "start" }, + { + id: "group", + kind: "optional-group", + config: { + name: options.phase === "post-merge" ? "Post-merge verification" : "Code Review", + defaultOn: true, + phase: options.phase, + template: { + nodes: [{ id: "review", kind: options.gateMode === "gate" ? "gate" : "prompt", config: { prompt: "review" } }], + edges: [], + }, + }, + }, + { id: "after", kind: "prompt", config: { prompt: "after" } }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "group" }, + { from: "group", to: "after", condition: "success" }, + { from: "group", to: "end", condition: "failure" }, + { from: "after", to: "end" }, + ], + }; +} + describe("WorkflowGraphExecutor optional-group", () => { it("two-task divergence: only the task whose enabledWorkflowSteps includes the group id runs the template; the sibling runs none and both reach downstream", async () => { const ir = optionalGroupIr(); @@ -220,4 +253,177 @@ describe("WorkflowGraphExecutor optional-group", () => { expect(calls).toContain("after"); expect(result.outcome).toBe("success"); }); + + it("pre-merge advisory REVISE requests a bounded fix and aborts forward traversal when scheduled", async () => { + const calls: string[] = []; + const records: unknown[] = []; + const requestFix = vi.fn(async () => true); + const executor = new WorkflowGraphExecutor({ + handlers: { + prompt: async (node) => { + calls.push(node.id); + if (node.id === "review") { + return { outcome: "success", value: "REVISE", contextPatch: { output: "Fix the review finding" } }; + } + return { outcome: "success" }; + }, + }, + recordWorkflowStepResult: async (_taskId, result) => { records.push(result); }, + requestPreMergeOptionalStepFix: requestFix, + }); + + const result = await executor.run(taskWith(["group"]), settingsOn(), reviseGroupIr()); + + expect(requestFix).toHaveBeenCalledWith("FN-OG", { + stepName: "Code Review", + feedback: "Fix the review finding", + phase: "pre-merge", + status: "advisory_failure", + verdict: "REVISE", + }); + expect(calls).not.toContain("after"); + expect(result.context["node:group:fixScheduled"]).toBe(true); + expect(records).toEqual(expect.arrayContaining([ + expect.objectContaining({ workflowStepId: "group", status: "advisory_failure", verdict: "REVISE", output: "Fix the review finding" }), + ])); + }); + + it("falls through unchanged when the pre-merge fix seam is absent or declines", async () => { + for (const requestFix of [undefined, vi.fn(async () => false)] as const) { + const calls: string[] = []; + const executor = new WorkflowGraphExecutor({ + handlers: { + prompt: async (node) => { + calls.push(node.id); + if (node.id === "review") return { outcome: "success", value: "REVISE", contextPatch: { output: "still advisory" } }; + return { outcome: "success" }; + }, + }, + ...(requestFix ? { requestPreMergeOptionalStepFix: requestFix } : {}), + }); + + const result = await executor.run(taskWith(["group"]), settingsOn(), reviseGroupIr()); + + expect(calls).toContain("after"); + expect(result.context["node:group:fixScheduled"]).toBeUndefined(); + if (requestFix) expect(requestFix).toHaveBeenCalledOnce(); + } + }); + + it("requests fixes for pre-merge gate REVISE but not post-merge, non-REVISE, or fast-mode skipped outcomes", async () => { + const requestFix = vi.fn(async () => true); + const gateExecutor = new WorkflowGraphExecutor({ + handlers: { + gate: async () => ({ outcome: "failure", value: "REVISE", contextPatch: { output: "gate finding" } }), + prompt: async () => ({ outcome: "success" }), + }, + requestPreMergeOptionalStepFix: requestFix, + }); + await gateExecutor.run(taskWith(["group"]), settingsOn(), reviseGroupIr({ gateMode: "gate" })); + expect(requestFix).toHaveBeenLastCalledWith("FN-OG", expect.objectContaining({ status: "failed", feedback: "gate finding" })); + + requestFix.mockClear(); + const postMergeCalls: string[] = []; + const postMergeExecutor = new WorkflowGraphExecutor({ + handlers: { + prompt: async (node) => { + postMergeCalls.push(node.id); + if (node.id === "review") return { outcome: "success", value: "REVISE", contextPatch: { output: "post merge finding" } }; + return { outcome: "success" }; + }, + }, + requestPreMergeOptionalStepFix: requestFix, + }); + await postMergeExecutor.run(taskWith(["group"]), settingsOn(), reviseGroupIr({ phase: "post-merge" })); + expect(requestFix).not.toHaveBeenCalled(); + expect(postMergeCalls).toContain("after"); + + const approveExecutor = new WorkflowGraphExecutor({ + handlers: { + prompt: async () => ({ outcome: "success", value: "APPROVE_WITH_NOTES", contextPatch: { output: "notes only" } }), + }, + requestPreMergeOptionalStepFix: requestFix, + }); + await approveExecutor.run(taskWith(["group"]), settingsOn(), reviseGroupIr()); + expect(requestFix).not.toHaveBeenCalled(); + + const fastExecutor = new WorkflowGraphExecutor({ + handlers: { + prompt: async (_node, context) => context.task.executionMode === "fast" + ? { outcome: "success", value: "workflow-step-skipped" } + : { outcome: "success", value: "REVISE", contextPatch: { output: "would revise outside fast mode" } }, + }, + requestPreMergeOptionalStepFix: requestFix, + }); + await fastExecutor.run({ ...taskWith(["group"]), executionMode: "fast" } as TaskDetail, settingsOn(), reviseGroupIr()); + expect(requestFix).not.toHaveBeenCalled(); + }); + + it("builtin coding optional Code Review and Browser Verification REVISE abort before review, and stepwise carries the same pre-merge path", async () => { + for (const groupId of ["code-review", "browser-verification"] as const) { + const requestFix = vi.fn(async () => true); + const calls: string[] = []; + const executor = new WorkflowGraphExecutor({ + handlers: { + prompt: async (node) => { + calls.push(node.id); + if ((groupId === "code-review" && node.id === "code-review-step") + || (groupId === "browser-verification" && node.id === "browser-verification-step")) { + return { outcome: "success", value: "REVISE", contextPatch: { output: `${groupId} finding` } }; + } + return { outcome: "success" }; + }, + }, + requestPreMergeOptionalStepFix: requestFix, + }); + + const result = await executor.run( + { ...taskWith(groupId === "code-review" ? ["code-review"] : ["browser-verification", "code-review"]), id: `FN-${groupId}` } as TaskDetail, + settingsOn(), + BUILTIN_CODING_WORKFLOW_IR, + ); + + expect(requestFix).toHaveBeenCalledWith(`FN-${groupId}`, expect.objectContaining({ + stepName: groupId === "code-review" ? "Code Review" : "Browser Verification", + feedback: `${groupId} finding`, + })); + expect(calls).not.toContain("review"); + expect(result.context[`node:${groupId}:fixScheduled`]).toBe(true); + } + + for (const ir of [BUILTIN_CODING_WORKFLOW_IR, BUILTIN_STEPWISE_CODING_WORKFLOW_IR]) { + for (const groupId of ["browser-verification", "code-review"] as const) { + const node = ir.nodes.find((candidate) => candidate.id === groupId); + expect(node).toMatchObject({ kind: "optional-group" }); + expect(node?.config?.phase).toBeUndefined(); + expect(ir.edges).toEqual(expect.arrayContaining([ + expect.objectContaining({ from: groupId, to: groupId === "browser-verification" ? "code-review" : "review", condition: "success" }), + expect.objectContaining({ from: groupId, to: "end", condition: "failure" }), + ])); + } + } + + const stepwiseRequestFix = vi.fn(async () => true); + const stepwiseExecutor = new WorkflowGraphExecutor({ + handlers: { + "parse-steps": async () => ({ outcome: "success", value: "no-steps" }), + prompt: async (node) => node.id === "code-review-step" + ? { outcome: "success", value: "REVISE", contextPatch: { output: "stepwise code-review finding" } } + : { outcome: "success" }, + }, + requestPreMergeOptionalStepFix: stepwiseRequestFix, + }); + + const stepwiseResult = await stepwiseExecutor.run( + { ...taskWith(["code-review"]), id: "FN-stepwise", steps: [] } as TaskDetail, + settingsOn(), + BUILTIN_STEPWISE_CODING_WORKFLOW_IR, + ); + + expect(stepwiseRequestFix).toHaveBeenCalledWith("FN-stepwise", expect.objectContaining({ + stepName: "Code Review", + feedback: "stepwise code-review finding", + })); + expect(stepwiseResult.context["node:code-review:fixScheduled"]).toBe(true); + }); }); diff --git a/packages/engine/src/__tests__/workflow-graph-optional-step-fix.test.ts b/packages/engine/src/__tests__/workflow-graph-optional-step-fix.test.ts new file mode 100644 index 0000000000..82fb5b3662 --- /dev/null +++ b/packages/engine/src/__tests__/workflow-graph-optional-step-fix.test.ts @@ -0,0 +1,89 @@ +import "./executor-test-helpers.js"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { Task } from "@fusion/core"; + +import { TaskExecutor } from "../executor.js"; +import { createMockStore, resetExecutorMocks } from "./executor-test-helpers.js"; + +function task(overrides: Partial = {}): Task { + return { + id: "FN-7066", + title: "Optional step fix", + description: "Fix optional workflow findings", + column: "in-progress", + status: null, + dependencies: [], + steps: [{ name: "Implement", status: "done" }], + currentStep: 0, + log: [], + prompt: "# Task\n## Steps\n### Step 0: Implement\n- [x] done", + worktree: "/tmp/fusion/fn-7066", + postReviewFixCount: 0, + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + ...overrides, + } as Task; +} + +const reviseInfo = { + stepName: "Code Review", + feedback: "packages/engine/src/example.ts:1 needs a guard", + phase: "pre-merge" as const, + status: "advisory_failure" as const, + verdict: "REVISE", +}; + +describe("TaskExecutor pre-merge optional-step fix seam", () => { + beforeEach(() => { + resetExecutorMocks(); + }); + + it("consumes budget before sending the task back for optional-step remediation", async () => { + const store = createMockStore(); + const liveTask = task({ postReviewFixCount: 0 }); + store.getTask.mockResolvedValue(liveTask); + store.getSettings.mockResolvedValue({ maxPostReviewFixes: 2 }); + const executor = new TaskExecutor(store, "/tmp/test"); + const sendBack = vi.spyOn(executor as any, "sendTaskBackForFix").mockResolvedValue(undefined); + + const scheduled = await (executor as any).requestPreMergeOptionalStepFix(liveTask.id, liveTask, reviseInfo); + + expect(scheduled).toBe(true); + expect(store.updateTask).toHaveBeenCalledWith("FN-7066", { postReviewFixCount: 1 }, undefined); + expect(store.logEntry).toHaveBeenCalledWith( + "FN-7066", + expect.stringContaining("attempt 1/2"), + expect.stringContaining("packages/engine/src/example.ts:1 needs a guard"), + undefined, + ); + expect(sendBack).toHaveBeenCalledWith( + liveTask, + "/tmp/fusion/fn-7066", + "packages/engine/src/example.ts:1 needs a guard", + "Code Review", + expect.stringContaining("requested revision"), + ); + expect(store.updateTask.mock.invocationCallOrder[0]).toBeLessThan(sendBack.mock.invocationCallOrder[0]); + }); + + it("declines without sending back when maxPostReviewFixes disables or exhausts the budget", async () => { + for (const { settingsMax, count } of [ + { settingsMax: 0, count: 0 }, + { settingsMax: 1, count: 1 }, + ]) { + const store = createMockStore(); + const liveTask = task({ postReviewFixCount: count }); + store.getTask.mockResolvedValue(liveTask); + store.getSettings.mockResolvedValue({ maxPostReviewFixes: settingsMax }); + const executor = new TaskExecutor(store, "/tmp/test"); + const sendBack = vi.spyOn(executor as any, "sendTaskBackForFix").mockResolvedValue(undefined); + + const scheduled = await (executor as any).requestPreMergeOptionalStepFix(liveTask.id, liveTask, reviseInfo); + + expect(scheduled).toBe(false); + expect(store.updateTask).not.toHaveBeenCalledWith(liveTask.id, expect.objectContaining({ postReviewFixCount: expect.any(Number) }), expect.anything()); + expect(store.updateTask).not.toHaveBeenCalledWith(liveTask.id, expect.objectContaining({ postReviewFixCount: expect.any(Number) }), undefined); + expect(sendBack).not.toHaveBeenCalled(); + } + }); +}); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 58181c8768..6ab5b68e5e 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -8,7 +8,7 @@ const execAsync = promisify(exec); import { delimiter, isAbsolute, join, relative, resolve as resolvePath } from "node:path"; import { existsSync, lstatSync, realpathSync } from "node:fs"; import { readFile, rm, writeFile } from "node:fs/promises"; -import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode, WorkflowIrNodeKind } from "@fusion/core"; +import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode, WorkflowIrNodeKind, WorkflowStepResult as CoreWorkflowStepResult } from "@fusion/core"; import { getUnmetSchedulingDependencies } from "./scheduler.js"; import { RetryStormError, TaskDeletedError, serializeRetryStormError, isExperimentalFeatureEnabled, resolveWorkflowIrForTask, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId, getWorkflowExtensionRegistry, getBuiltinWorkflow, parseNoOpCompletionMarker, allowsAutoMergeProcessing, isSharedBranchGroupMemberIntegration, resolveMaxAutoMergeRetries } from "@fusion/core"; import { mergeEffectiveSettings } from "./effective-settings.js"; @@ -3703,6 +3703,51 @@ export class TaskExecutor { } } + /* + * FNXC:WorkflowOptionalStepFix 2026-06-26-16:35: + * Inline graph optional-step remediation consumes `postReviewFixCount` BEFORE calling `sendTaskBackForFix`, matching self-healing's budget-first ordering. This prevents a persistent Code Review / Browser Verification REVISE from ping-ponging forever: when `postReviewFixCount >= maxPostReviewFixes` (or max <= 0), the seam declines and graph execution falls through to the prior advisory/gate behavior. + */ + private async requestPreMergeOptionalStepFix( + taskId: string, + fallbackTask: Task, + info: { + stepName: string; + feedback: string; + phase: CoreWorkflowStepResult["phase"]; + status: CoreWorkflowStepResult["status"]; + verdict?: string; + }, + ): Promise { + if (info.phase !== "pre-merge") return false; + if (info.verdict !== "REVISE") return false; + if (info.status !== "advisory_failure" && info.status !== "failed") return false; + + const liveTask = await this.store.getTask(taskId).catch(() => fallbackTask); + const settings = await mergeEffectiveSettings(this.store, liveTask, await this.store.getSettings()); + const maxFixes = settings.maxPostReviewFixes ?? 1; + if (!Number.isFinite(maxFixes) || maxFixes <= 0) return false; + + const currentCount = liveTask.postReviewFixCount ?? 0; + if (currentCount >= maxFixes) return false; + + const nextCount = currentCount + 1; + await this.store.updateTask(taskId, { postReviewFixCount: nextCount }, this.getRunContextFor(taskId)); + await this.store.logEntry( + taskId, + `Pre-merge optional workflow step requested executor fixes (attempt ${nextCount}/${maxFixes})`, + `Step: ${info.stepName}\nStatus: ${info.status}\nFeedback:\n${info.feedback}`, + this.getRunContextFor(taskId), + ); + await this.sendTaskBackForFix( + liveTask, + liveTask.worktree ?? "", + info.feedback, + info.stepName, + `Pre-merge optional workflow step "${info.stepName}" requested revision`, + ); + return true; + } + /** * Auto-revive an `in-review` task whose pre-merge workflow step(s) failed, by * replaying the same send-back-for-fix flow the executor uses during a live @@ -4326,7 +4371,7 @@ export class TaskExecutor { no-op when the store lacks updateTask, and swallow read/write errors (the executor wrapper also swallows) so result recording never affects the run. */ - recordWorkflowStepResult: async (taskId: string, result: import("@fusion/core").WorkflowStepResult) => { + recordWorkflowStepResult: async (taskId: string, result: CoreWorkflowStepResult) => { if (typeof this.store.updateTask !== "function") return; try { const live = await this.store.getTask(taskId); @@ -4341,6 +4386,7 @@ export class TaskExecutor { // Result recording is additive visibility — never affect the run. } }, + requestPreMergeOptionalStepFix: (taskId, info) => this.requestPreMergeOptionalStepFix(taskId, task, info), }); let result: WorkflowGraphTaskRunResult; try { diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 808f2b5b6d..f6f8cfbf61 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -166,6 +166,17 @@ export interface WorkflowGraphExecutorDeps { * this seam only forwards the terminal/pending entry. */ recordWorkflowStepResult?: (taskId: string, result: WorkflowStepResult) => void | Promise; + /* + * FNXC:WorkflowOptionalStepFix 2026-06-26-16:20: + * Enabled PRE-merge optional workflow steps that return REVISE must offer the executor one bounded remediation path before normal advisory/gate fall-through. This seam returns true only when the caller already consumed the `maxPostReviewFixes` budget and scheduled `sendTaskBackForFix`; the graph must then stop before review/merge. Absent or false preserves prior byte-inert behavior for in-memory tests and exhausted budgets. + */ + requestPreMergeOptionalStepFix?: (taskId: string, info: { + stepName: string; + feedback: string; + phase: WorkflowStepResult["phase"]; + status: WorkflowStepResult["status"]; + verdict?: string; + }) => Promise | boolean; /** Project node-published task metadata onto the task row for dispatcher/UI. */ publishTaskProjection?: (taskId: string, patch: WorkflowTaskProjection, source: { nodeId: string; nodeKind: WorkflowIrNode["kind"] }) => void | Promise; /** @deprecated use publishTaskProjection. Kept for older callers. */ @@ -649,6 +660,24 @@ export class WorkflowGraphExecutor { }; context[`node:${node.id}:outcome`] = result.outcome; if (result.value !== undefined) context[`node:${node.id}:value`] = result.value; + if ( + stepPhase === "pre-merge" + && verdict === "REVISE" + && (stepStatus === "advisory_failure" || stepStatus === "failed") + ) { + const feedback = stepOutput?.trim() || stepNotes?.trim() || "(no feedback captured)"; + const fixScheduled = await this.deps.requestPreMergeOptionalStepFix?.(task.id, { + stepName: groupName, + feedback, + phase: stepPhase, + status: stepStatus, + verdict, + }); + if (fixScheduled) { + context[`node:${node.id}:fixScheduled`] = true; + return { outcome: "success", value: "pre-merge-optional-step-fix-scheduled" }; + } + } return await traverseChildren(node, result); } diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts index 844021041d..6589a92ae8 100644 --- a/packages/engine/src/workflow-graph-task-runner.ts +++ b/packages/engine/src/workflow-graph-task-runner.ts @@ -1,7 +1,7 @@ import type { Settings, TaskDetail, WorkflowDefinition, WorkflowStepResult } from "@fusion/core"; import { getBuiltinWorkflow, isBuiltinWorkflowId } from "@fusion/core"; -import { WorkflowGraphExecutor, type WorkflowNodeOutcome, type WorkflowTaskProjection } from "./workflow-graph-executor.js"; +import { WorkflowGraphExecutor, type WorkflowGraphExecutorDeps, type WorkflowNodeOutcome, type WorkflowTaskProjection } from "./workflow-graph-executor.js"; import type { CodeNodeRunner, ForeachActiveContext, @@ -92,6 +92,8 @@ export interface WorkflowGraphTaskRunnerDeps { * node's outcome into `task.workflowStepResults` keyed by node id. Additive; * absent → graph records nothing (disabled groups + unwired stores byte-inert). */ recordWorkflowStepResult?: (taskId: string, result: WorkflowStepResult) => void | Promise; + /** Enabled pre-merge optional-step REVISE remediation seam. Additive; absent preserves prior graph traversal. */ + requestPreMergeOptionalStepFix?: WorkflowGraphExecutorDeps["requestPreMergeOptionalStepFix"]; /** Project node-published task metadata onto the task row for dispatcher/UI. */ publishTaskProjection?: (taskId: string, patch: WorkflowTaskProjection, source: { nodeId: string; nodeKind: string }) => void | Promise; /** @deprecated use publishTaskProjection. */ @@ -236,6 +238,7 @@ export class WorkflowGraphTaskRunner { resumeReconcile: this.deps.resumeReconcile, logTaskEntry: this.deps.logTaskEntry, recordWorkflowStepResult: this.deps.recordWorkflowStepResult, + requestPreMergeOptionalStepFix: this.deps.requestPreMergeOptionalStepFix, publishTaskProjection: this.deps.publishTaskProjection, publishTouchedFiles: this.deps.publishTouchedFiles, // Single source of truth (KTD-6): prefer the caller-threaded run id so the