From 05c54fbe29d2502fab81e3d5c848d5016950785c Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 29 Jun 2026 04:12:13 -0700 Subject: [PATCH] fix(FN-7228): complete workflow recovery after reviews pass Fusion-Task-Id: FN-7228 --- .../fn-7228-workflow-review-visibility.md | 7 + .../src/__tests__/executor-pause.test.ts | 43 ++++++ .../src/__tests__/mock-provider.test.ts | 62 ++++++++ .../src/__tests__/restart.integration.test.ts | 74 +++++++++ packages/engine/src/executor.ts | 144 +++++++++++++----- .../engine/src/providers/mock-provider.ts | 58 ++++++- 6 files changed, 345 insertions(+), 43 deletions(-) create mode 100644 .changeset/fn-7228-workflow-review-visibility.md diff --git a/.changeset/fn-7228-workflow-review-visibility.md b/.changeset/fn-7228-workflow-review-visibility.md new file mode 100644 index 0000000000..afc7a823bf --- /dev/null +++ b/.changeset/fn-7228-workflow-review-visibility.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Keep Plan Review status visible when execution resumes after stale merge cleanup. +category: fix +dev: Reconstructs passed Plan Review rows from task logs and makes mock test-mode sessions emit workflow-step parser events. diff --git a/packages/engine/src/__tests__/executor-pause.test.ts b/packages/engine/src/__tests__/executor-pause.test.ts index 7539d0586b..bbd39db19b 100644 --- a/packages/engine/src/__tests__/executor-pause.test.ts +++ b/packages/engine/src/__tests__/executor-pause.test.ts @@ -1681,6 +1681,49 @@ describe("FN-2883 fast-path guards", () => { ); }); + it("stale merge cleanup reconstructs passed plan review status from logs when the row is missing", async () => { + const store = createMockStore(); + const executor = new TaskExecutor(store, "/tmp/test"); + const task = { + id: "FN-7228", + title: "stale merge", + description: "desc", + column: "in-progress" as const, + dependencies: [], + steps: [{ name: "Step 0", status: "in-progress" }], + currentStep: 0, + log: [ + { + timestamp: "2026-06-29T10:34:21.917Z", + action: "[pre-merge] Workflow step completed: Plan Review", + outcome: "Plan Review approved the revised specification.", + }, + ], + mergeDetails: { strategy: "manual" }, + workflowStepResults: undefined, + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }; + store.getTask.mockResolvedValue({ ...task, mergeDetails: null }); + + await (executor as any).cleanupMergeStateForReverification(task, "cleanup stale merge state"); + + expect(store.updateTask).toHaveBeenCalledWith( + "FN-7228", + expect.objectContaining({ + workflowStepResults: [ + expect.objectContaining({ + workflowStepId: "plan-review", + workflowStepName: "Plan Review", + status: "passed", + verdict: "APPROVE", + output: "Plan Review approved the revised specification.", + }), + ], + }), + ); + }); + it("resumeOrphaned does not fast-path completed tasks that still have mergeDetails", async () => { const store = createMockStore(); const executor = new TaskExecutor(store, "/tmp/test"); diff --git a/packages/engine/src/__tests__/mock-provider.test.ts b/packages/engine/src/__tests__/mock-provider.test.ts index 6593d6b639..9849f87b3f 100644 --- a/packages/engine/src/__tests__/mock-provider.test.ts +++ b/packages/engine/src/__tests__/mock-provider.test.ts @@ -152,6 +152,68 @@ describe("MockAgentRuntime", () => { expect(updateExecute).toHaveBeenCalledWith(expect.any(String), { step: 1, status: "done" }, undefined, undefined, expect.anything()); }); + it("treats graph-owned executor step sessions as successful without lifecycle tools", async () => { + const runtime = new MockAgentRuntime(); + const { cwd, taskId } = await createWorkspace("FN-7228"); + const onText = vi.fn(); + const taskShowExecute = vi.fn(); + + const { session } = await runtime.createSession({ + cwd, + systemPrompt: "system", + runtimeContext: { sessionPurpose: "executor" }, + customTools: [createTool("fn_task_show", taskShowExecute)], + onText, + taskId, + }); + + await expect(runtime.promptWithFallback(session, "run graph-owned step")).resolves.toBeUndefined(); + expect(taskShowExecute).not.toHaveBeenCalled(); + expect(onText).toHaveBeenCalledWith(expect.stringContaining("graph-owned step session")); + }); + + it("emits mock text through session subscription events for workflow-step parsers", async () => { + const runtime = new MockAgentRuntime(); + const { cwd, taskId } = await createWorkspace("FN-7228-SUBSCRIBE"); + const deltas: string[] = []; + + const { session } = await runtime.createSession({ + cwd, + systemPrompt: "system", + runtimeContext: { sessionPurpose: "reviewer" }, + taskId, + }); + (session as any).subscribe((event: any) => { + if (event.type === "message_update" && event.assistantMessageEvent?.type === "text_delta") { + deltas.push(event.assistantMessageEvent.delta); + } + }); + + await runtime.promptWithFallback(session, "review"); + expect(deltas.join("")).toContain("Verdict: APPROVE"); + }); + + it("emits an approval verdict for executor-backed workflow steps without lifecycle tools", async () => { + const runtime = new MockAgentRuntime(); + const { cwd, taskId } = await createWorkspace("FN-7228-WORKFLOW-STEP"); + const deltas: string[] = []; + + const { session } = await runtime.createSession({ + cwd, + systemPrompt: "system", + runtimeContext: { sessionPurpose: "executor" }, + taskId, + }); + (session as any).subscribe((event: any) => { + if (event.type === "message_update" && event.assistantMessageEvent?.type === "text_delta") { + deltas.push(event.assistantMessageEvent.delta); + } + }); + + await runtime.promptWithFallback(session, "Execute the workflow step \"Code Review\" for task FN-7228-WORKFLOW-STEP."); + expect(deltas.join("")).toContain("\"verdict\":\"APPROVE\""); + }); + it("accumulates synthetic token usage once per session baseline", async () => { const runtime = new MockAgentRuntime(); const { cwd, taskId } = await createWorkspace(); diff --git a/packages/engine/src/__tests__/restart.integration.test.ts b/packages/engine/src/__tests__/restart.integration.test.ts index 3c315e75b7..5519a8c917 100644 --- a/packages/engine/src/__tests__/restart.integration.test.ts +++ b/packages/engine/src/__tests__/restart.integration.test.ts @@ -830,6 +830,80 @@ describe("In-progress task resume after restart", () => { expect(store.moveTask).not.toHaveBeenCalledWith("FN-963", "in-review"); }); + it("recoverCompletedTask() skips graph re-entry when enabled pre-merge gates already passed", async () => { + const store = createMockStore(); + const task = makeTask("FN-7228", "in-progress", { + worktree: "/tmp/wt/FN-7228", + steps: makeSteps("done"), + enabledWorkflowSteps: ["plan-review", "code-review"], + workflowStepResults: [ + { + workflowStepId: "plan-review", + workflowStepName: "Plan Review", + phase: "pre-merge", + status: "passed", + }, + { + workflowStepId: "code-review", + workflowStepName: "Code Review", + phase: "pre-merge", + status: "passed", + }, + ], + }); + store.getTask.mockResolvedValue(makeTaskDetail("FN-7228", "in-progress", { + worktree: "/tmp/wt/FN-7228", + steps: makeSteps("done"), + enabledWorkflowSteps: ["plan-review", "code-review"], + workflowStepResults: task.workflowStepResults, + })); + + const executor = new TaskExecutor(store, "/tmp/test"); + vi.spyOn(executor as any, "captureModifiedFiles").mockResolvedValue([]); + const graphEntry = vi + .spyOn(executor as any, "maybeExecuteWorkflowGraph") + .mockResolvedValue(true); + + const recovered = await executor.recoverCompletedTask(task); + + expect(recovered).toBe(true); + expect(graphEntry).not.toHaveBeenCalled(); + expect(store.handoffToReview).toHaveBeenCalledWith("FN-7228", expect.objectContaining({ + evidence: expect.objectContaining({ reason: "completed-task-recovered" }), + })); + }); + + it("recoverCompletedTask() treats passed default review rows as satisfied when enabled steps are absent", async () => { + const store = createMockStore(); + const task = makeTask("FN-7228-DEFAULT", "in-progress", { + worktree: "/tmp/wt/FN-7228-DEFAULT", + steps: makeSteps("done"), + workflowStepResults: [ + { workflowStepId: "plan-review", workflowStepName: "Plan Review", phase: "pre-merge", status: "passed" }, + { workflowStepId: "code-review", workflowStepName: "Code Review", phase: "pre-merge", status: "passed" }, + ], + }); + store.getTask.mockResolvedValue(makeTaskDetail("FN-7228-DEFAULT", "in-progress", { + worktree: "/tmp/wt/FN-7228-DEFAULT", + steps: makeSteps("done"), + workflowStepResults: task.workflowStepResults, + })); + + const executor = new TaskExecutor(store, "/tmp/test"); + vi.spyOn(executor as any, "captureModifiedFiles").mockResolvedValue([]); + const graphEntry = vi + .spyOn(executor as any, "maybeExecuteWorkflowGraph") + .mockResolvedValue(true); + + const recovered = await executor.recoverCompletedTask(task); + + expect(recovered).toBe(true); + expect(graphEntry).not.toHaveBeenCalled(); + expect(store.handoffToReview).toHaveBeenCalledWith("FN-7228-DEFAULT", expect.objectContaining({ + evidence: expect.objectContaining({ reason: "completed-task-recovered" }), + })); + }); + it("recoverCompletedTask() fails closed (KTD-5) when the store lacks getTaskWorkflowSelection and the task has enabled workflow steps", async () => { // createMockStore does NOT expose getTaskWorkflowSelection, so the workflow // graph cannot resolve a selection — and the legacy runWorkflowSteps path was diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 3bb49d61c7..306f4bf4c1 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -3130,7 +3130,7 @@ export class TaskExecutor { logMessage: string, options?: { preserveVerificationFailureCount?: boolean }, ): Promise { - const preservedWorkflowStepResults = preservePreExecutionWorkflowStepResults(task.workflowStepResults); + const preservedWorkflowStepResults = preservePreExecutionWorkflowStepResults(task); await this.store.updateTask(task.id, { mergeDetails: null, mergeRetries: 0, @@ -3836,42 +3836,54 @@ export class TaskExecutor { // Run workflow steps before transitioning — skip in fast mode if (task.executionMode !== "fast") { - if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow-graph re-entry during completed-task recovery")) { - return false; + if (areEnabledPreMergeWorkflowStepsSatisfied(liveForCompletenessCheck)) { + /* + FNXC:WorkflowLifecycle 2026-06-29-04:37: + Completed graph-owned tasks can be observed briefly as in-progress after + the main graph already recorded every enabled pre-merge gate. Recovery + must not restart the graph from parse in that state; foreach pins from + the completed run make parse fail with pin-mismatch. Hand off to review + instead, which is the same terminal seam the completed graph reached. + */ + executorLog.log(`${task.id}: completed recovery found satisfied workflow gates — skipping graph re-entry`); + } else { + if (await this.shouldDeferCompletionForGlobalPause(task.id, "before workflow-graph re-entry during completed-task recovery")) { + return false; + } + /* + FNXC:WorkflowExecution 2026-06-25-00:00: + U4 (KTD-2) watchdog re-entry. The legacy `runWorkflowSteps` recovery path + was deleted; the workflow graph is the sole executor. A stranded completed + task is recovered by RE-ENTERING the graph via `maybeExecuteWorkflowGraph` + (the same entry execute() uses), which: (1) re-runs any pending + optional-group / gate nodes, (2) records their outcomes into + `task.workflowStepResults` (U2) and emits the `[pre-merge]` logs, and + (3) OWNS the in-review vs back-for-fix transition. The graph's execute seam + registers the normal completion interceptor, so a task whose implementation + already completed resumes at the post-implementation nodes (it does not + re-run the agent from scratch). RECOVERY POLICY mapping (per plan U4): the + old "any failure including REVISE is hard" recovery rule now maps onto the + graph's gate semantics — a GATE node REVISE/failure routes the task back for + fix, while an ADVISORY REVISE is non-blocking and proceeds to review. KTD-5: + for a store lacking `getTaskWorkflowSelection` that has enabled steps, + `maybeExecuteWorkflowGraph` itself fails closed (parks) rather than letting + recovery silently skip the gates. + */ + const graphOwned = await this.maybeExecuteWorkflowGraph(task); + if (graphOwned) { + this.clearCompletedTaskWatchdog(task.id); + await this.store.logEntry( + task.id, + `Auto-recovered: stranded completed task re-dispatched through the workflow graph — the graph re-ran pending workflow steps (recording results) and owns the in-review / back-for-fix transition`, + ).catch(() => undefined); + executorLog.log(`✓ ${task.id} auto-recovered completed task via workflow-graph re-entry`); + return true; + } + // Graph declined (minimal store WITHOUT the workflow-selection API and no + // enabled gates to run — a store WITH enabled steps would have been parked + // fail-closed above): there is nothing to gate, so fall through to the + // legacy in-review handoff below. } - /* - FNXC:WorkflowExecution 2026-06-25-00:00: - U4 (KTD-2) watchdog re-entry. The legacy `runWorkflowSteps` recovery path - was deleted; the workflow graph is the sole executor. A stranded completed - task is recovered by RE-ENTERING the graph via `maybeExecuteWorkflowGraph` - (the same entry execute() uses), which: (1) re-runs any pending - optional-group / gate nodes, (2) records their outcomes into - `task.workflowStepResults` (U2) and emits the `[pre-merge]` logs, and - (3) OWNS the in-review vs back-for-fix transition. The graph's execute seam - registers the normal completion interceptor, so a task whose implementation - already completed resumes at the post-implementation nodes (it does not - re-run the agent from scratch). RECOVERY POLICY mapping (per plan U4): the - old "any failure including REVISE is hard" recovery rule now maps onto the - graph's gate semantics — a GATE node REVISE/failure routes the task back for - fix, while an ADVISORY REVISE is non-blocking and proceeds to review. KTD-5: - for a store lacking `getTaskWorkflowSelection` that has enabled steps, - `maybeExecuteWorkflowGraph` itself fails closed (parks) rather than letting - recovery silently skip the gates. - */ - const graphOwned = await this.maybeExecuteWorkflowGraph(task); - if (graphOwned) { - this.clearCompletedTaskWatchdog(task.id); - await this.store.logEntry( - task.id, - `Auto-recovered: stranded completed task re-dispatched through the workflow graph — the graph re-ran pending workflow steps (recording results) and owns the in-review / back-for-fix transition`, - ).catch(() => undefined); - executorLog.log(`✓ ${task.id} auto-recovered completed task via workflow-graph re-entry`); - return true; - } - // Graph declined (minimal store WITHOUT the workflow-selection API and no - // enabled gates to run — a store WITH enabled steps would have been parked - // fail-closed above): there is nothing to gate, so fall through to the - // legacy in-review handoff below. } else { executorLog.log(`${task.id}: fast mode — skipping workflow steps on auto-recovery`); } @@ -16707,7 +16719,34 @@ function hasNonTerminalWorkflowSteps(task: Pick): boolean { return task.steps.length > 0 && task.steps.some((step) => step.status !== "done" && step.status !== "skipped"); } -function preservePreExecutionWorkflowStepResults(results: Task["workflowStepResults"]): CoreWorkflowStepResult[] { +function areEnabledPreMergeWorkflowStepsSatisfied( + task: Pick | undefined, +): boolean { + const preMergeGateIds = new Set(["plan-review", "browser-verification", "code-review"]); + const enabled = task?.enabledWorkflowSteps; + /* + * FNXC:WorkflowLifecycle 2026-06-29-04:46: + * Older/default coding tasks may not persist an explicit enabledWorkflowSteps + * list even though default-on Plan Review and Code Review have already run. + * Treat those two passed rows as satisfied defaults; keep explicit arrays + * strict so custom/unknown enabled gates still re-enter the graph. + */ + const enabledPreMerge = Array.isArray(enabled) && enabled.length > 0 + ? enabled.filter((id) => preMergeGateIds.has(id)) + : ["plan-review", "code-review"]; + if (enabledPreMerge.length === 0) return false; + if (Array.isArray(enabled) && enabledPreMerge.length !== enabled.length) return false; + const results = task?.workflowStepResults ?? []; + return enabledPreMerge.every((id) => + results.some((result) => + result.workflowStepId === id + && result.phase === "pre-merge" + && result.status === "passed", + ), + ); +} + +function preservePreExecutionWorkflowStepResults(task: Pick): CoreWorkflowStepResult[] { /* * FNXC:WorkflowLifecycle 2026-06-29-03:50: * Reverification cleanup must clear post-implementation verification residue @@ -16715,8 +16754,37 @@ function preservePreExecutionWorkflowStepResults(results: Task["workflowStepResu * Review, then stale merge-state cleanup reset `workflowStepResults` to `[]`; * the dashboard showed Plan Review with no status while execution continued and * the graph no longer had durable proof to skip duplicate plan review. + * + * FNXC:WorkflowLifecycle 2026-06-29-04:19: + * The durable Plan Review row may already be missing when stale merge cleanup + * runs, while the task log still has the authoritative terminal Plan Review + * entry. Reconstruct the passed row from that log so execution can continue + * with a visible pre-execution review status instead of showing an active task + * card with Plan Review blank. */ - return (results ?? []).filter((result) => result.workflowStepId === "plan-review"); + const preserved = (task.workflowStepResults ?? []).filter((result) => result.workflowStepId === "plan-review"); + if (preserved.length > 0) return preserved; + + let latest: { timestamp?: string; outcome?: string; status: "passed" | "failed" } | undefined; + for (const entry of task.log ?? []) { + if (entry.action === "[pre-merge] Workflow step completed: Plan Review") { + latest = { timestamp: entry.timestamp, outcome: entry.outcome, status: "passed" }; + } else if (entry.action === "[pre-merge] Workflow step failed: Plan Review") { + latest = { timestamp: entry.timestamp, outcome: entry.outcome, status: "failed" }; + } + } + if (latest?.status !== "passed") return []; + return [ + { + workflowStepId: "plan-review", + workflowStepName: "Plan Review", + phase: "pre-merge", + status: "passed", + verdict: "APPROVE", + ...(latest.outcome ? { output: latest.outcome, notes: latest.outcome } : {}), + ...(latest.timestamp ? { startedAt: latest.timestamp, completedAt: latest.timestamp } : {}), + }, + ]; } /** diff --git a/packages/engine/src/providers/mock-provider.ts b/packages/engine/src/providers/mock-provider.ts index 0a770a1da4..46d9185b33 100644 --- a/packages/engine/src/providers/mock-provider.ts +++ b/packages/engine/src/providers/mock-provider.ts @@ -99,6 +99,7 @@ interface MockToolCallResult { export class MockAgentSession { readonly __mock: MockAgentSessionState; readonly state: { errorMessage?: string; error?: string } = {}; + private readonly listeners = new Set<(event: unknown) => void>(); constructor(options: AgentRuntimeOptions, sessionPurpose: MockSessionPurpose) { this.__mock = { options, sessionPurpose }; @@ -106,6 +107,30 @@ export class MockAgentSession { dispose(): void {} + subscribe(listener: (event: unknown) => void): { unsubscribe?: () => void } { + this.listeners.add(listener); + return { + unsubscribe: () => { + this.listeners.delete(listener); + }, + }; + } + + emitText(delta: string): void { + const event = { + type: "message_update", + assistantMessageEvent: { + type: "text_delta", + partial: delta, + contentIndex: 0, + delta, + }, + }; + for (const listener of this.listeners) { + listener(event); + } + } + getSessionStats(): { tokens: typeof MOCK_SYNTHETIC_TOKEN_USAGE } { return { tokens: MOCK_SYNTHETIC_TOKEN_USAGE }; } @@ -179,6 +204,18 @@ async function loadTaskSteps(options: AgentRuntimeOptions): Promise = { executor: { async run(ctx) { + /* + FNXC:WorkflowTestMode 2026-06-29-03:58: + Step-session execution is graph-owned and intentionally withholds fn_task_update/fn_task_done so agents cannot mutate task lifecycle directly. Test mode must mirror that contract: use fn_task_update only for legacy full-task executor sessions, and otherwise return success so the graph projection marks the scoped step. + */ + if (!ctx.tools.some((tool) => tool.name === "fn_task_update")) { + if (/Execute the workflow step/i.test(ctx.prompt)) { + ctx.options.onText?.("{\"verdict\":\"APPROVE\",\"notes\":\"Mock workflow step approved scripted output.\"}"); + return; + } + ctx.options.onText?.("Mock executor completed graph-owned step session without lifecycle tool calls."); + return; + } let steps: Array<{ status?: string }> = []; if (ctx.taskId && ctx.tools.some((tool) => tool.name === "fn_task_show")) { const taskDetails = await ctx.invokeTool("fn_task_show", { id: ctx.taskId }); @@ -249,19 +286,30 @@ export class MockAgentRuntime implements AgentRuntime { async promptWithFallback(session: AgentSession, prompt: string, _promptOptions?: unknown): Promise { const mockSession = session as unknown as MockAgentSession; const { options, sessionPurpose } = mockSession.__mock; + /* + FNXC:WorkflowTestMode 2026-06-29-04:02: + Workflow-step gates collect verdict text from AgentSession subscription events, while reviewer lanes also use onText callbacks. Test mode must emit both surfaces so Plan Review and final Code Review exercise the same parsing contract as real runtimes. + */ + const effectiveOptions: AgentRuntimeOptions = { + ...options, + onText: (text) => { + options.onText?.(text); + mockSession.emitText(text); + }, + }; const tools = options.customTools ?? []; const script = mockScriptRegistry.resolveMockScript({ sessionPurpose, - taskId: options.taskId, + taskId: effectiveOptions.taskId, }); await script.run({ sessionPurpose, prompt, - options, + options: effectiveOptions, tools, - taskId: options.taskId, - taskTitle: options.taskTitle, - invokeTool: (name, args) => executeTool(tools, options, name, args), + taskId: effectiveOptions.taskId, + taskTitle: effectiveOptions.taskTitle, + invokeTool: (name, args) => executeTool(tools, effectiveOptions, name, args), }); }