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

fix(FN-7228): complete workflow recovery after reviews pass

Browse Source 
Fusion-Task-Id: FN-7228
This commit is contained in:
gsxdsm
2026-06-29 04:12:13 -07:00
parent c088f8a412
commit 05c54fbe29
6 changed files with 345 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fn-7228-workflow-review-visibility.md Normal file
View File

@@ -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.

43
packages/engine/src/__tests__/executor-pause.test.ts
View File

@@ -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");

62
packages/engine/src/__tests__/mock-provider.test.ts
View File

@@ -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();

74
packages/engine/src/__tests__/restart.integration.test.ts
View File

@@ -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

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

@@ -3130,7 +3130,7 @@ export class TaskExecutor {
logMessage: string,
options?: { preserveVerificationFailureCount?: boolean },
): Promise<Task> {
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<TaskDetail, "steps">): 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<Task, "enabledWorkflowSteps" | "workflowStepResults"> | 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<Task, "workflowStepResults" | "log">): 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 } : {}),
},
];
}
/**

58
packages/engine/src/providers/mock-provider.ts
View File

@@ -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<Array<{ stat
const DEFAULT_SCRIPTS: Record<MockSessionPurpose, MockScript> = {
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<void> {
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),
});
}