fix(FN-7229): stop parking execution errors in review

This commit is contained in:
gsxdsm
2026-06-29 03:33:18 -07:00
parent f430f5db39
commit 984e36255d
5 changed files with 211 additions and 50 deletions

View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Stop routing failed workflow execution into the review column.
category: fix
dev: Graph and execution failures now stay executable or failed in-place instead of handing errored tasks to in-review.

View File

@@ -76,4 +76,81 @@ describe("executor graph execute self-requeue gate", () => {
expect.anything(), expect.anything(),
); );
}); });
it("moves in-review graph failures with incomplete steps back to todo for resume", async () => {
resetExecutorMocks();
const store = createMockStore();
const live = task({
id: "FN-7228",
column: "in-review",
status: "failed",
error: "Workflow graph terminated with failure at node 'parse'",
steps: [
{ name: "Preflight", status: "in-progress" },
{ name: "Implement", status: "in-progress" },
{ name: "Testing & Verification", status: "pending" },
],
});
store.getTask.mockResolvedValue(live);
const executor = new TaskExecutor(store, "/tmp/test");
/*
* FNXC:WorkflowLifecycle 2026-06-29-11:12:
* FN-7228/FN-7229 proved that restart-time graph failures can surface after a
* stale handoff put the card in `in-review` with unfinished steps. Review is
* not an error bucket; bounce that shape back to `todo` preserving step
* progress so the engine can resume the correct unfinished step.
*/
await (executor as any).handleGraphFailure(live, {
disposition: "failed",
outcome: "failure",
visitedNodeIds: ["parse"],
context: { "node:parse:value": "parse-error" },
});
expect(store.updateTask).toHaveBeenCalledWith(
live.id,
expect.objectContaining({ status: null, error: null }),
undefined,
);
expect(store.moveTask).toHaveBeenCalledWith(
live.id,
"todo",
expect.objectContaining({ preserveProgress: true, moveSource: "engine", recoveryRehome: true }),
);
expect(store.handoffToReview).not.toHaveBeenCalled();
});
it("does not hand generic graph failures to review", async () => {
resetExecutorMocks();
const store = createMockStore();
const live = task({
id: "FN-7229",
column: "in-progress",
steps: [
{ name: "Preflight", status: "done" },
{ name: "Implement", status: "in-progress" },
],
});
store.getTask.mockResolvedValue(live);
const executor = new TaskExecutor(store, "/tmp/test");
await (executor as any).handleGraphFailure(live, {
disposition: "failed",
outcome: "failure",
visitedNodeIds: ["parse"],
context: { "node:parse:value": "parse-error" },
});
expect(store.updateTask).toHaveBeenCalledWith(
live.id,
expect.objectContaining({
status: "failed",
error: expect.stringContaining("Workflow graph terminated with failure at node 'parse'"),
}),
undefined,
);
expect(store.handoffToReview).not.toHaveBeenCalled();
expect(store.moveTask).not.toHaveBeenCalledWith(live.id, "in-review", expect.anything());
});
}); });

View File

@@ -2934,6 +2934,7 @@ describe("SelfHealingManager", () => {
}); });
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({ (store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
taskStuckTimeoutMs: 60_000, taskStuckTimeoutMs: 60_000,
autoMerge: true,
}); });
const staleUpdatedAt = new Date(Date.now() - 6 * 60_000).toISOString(); const staleUpdatedAt = new Date(Date.now() - 6 * 60_000).toISOString();
@@ -4638,6 +4639,43 @@ describe("SelfHealingManager", () => {
managerWithRecovery.stop(); managerWithRecovery.stop();
}); });
it("moves failed in-review tasks with incomplete steps back to todo immediately after restart", async () => {
const managerWithRecovery = new SelfHealingManager(store, {
rootDir: "/tmp/test-project",
});
(store.getSettings as ReturnType<typeof vi.fn>).mockResolvedValue({
taskStuckTimeoutMs: 60_000,
});
(store.listTasks as ReturnType<typeof vi.fn>).mockResolvedValue([
{
id: "FN-7229",
column: "in-review",
paused: false,
status: "failed",
autoMerge: true,
error: "Workflow graph terminated with failure at node 'parse'",
updatedAt: new Date().toISOString(),
steps: [
{ name: "Preflight", status: "done" },
{ name: "Documentation & Delivery", status: "in-progress" },
],
workflowStepResults: [],
log: [],
},
]);
const result = await managerWithRecovery.recoverStaleIncompleteReviewTasks();
expect(result).toBe(1);
expect(store.moveTask).toHaveBeenCalledWith("FN-7229", "todo", {
preserveProgress: true,
moveSource: "engine",
recoveryRehome: true,
});
managerWithRecovery.stop();
});
}); });
describe("recoverStuckMergeDeadlocks", () => { describe("recoverStuckMergeDeadlocks", () => {

View File

@@ -1991,11 +1991,13 @@ export class TaskExecutor {
/** /**
* Stable handoff reasons used on task:handoff audit events. * Stable handoff reasons used on task:handoff audit events.
* Keep values greppable for executor/self-healing forensics: review-handoff-requested, * Keep values greppable for executor/self-healing forensics: review-handoff-requested,
* completed-task-recovered, worktree-liveness-failed, step-session-completed, * completed-task-recovered, step-session-completed, paused-after-completion,
* step-session-failed, transient-retries-exhausted, paused-after-completion, * fn_task_done, fn_task_done-retry-completed.
* fn_task_done, fn_task_done-retry-completed, max-task-done-retries-exhausted, *
* execution-failed, implicit-fn_task_done-refused, invariant-check-failed, * FNXC:WorkflowLifecycle 2026-06-29-11:20:
* fn_task_done-refused. * Failed execution is not a review handoff. Error paths must either requeue
* executable work for resume or fail in-place; `in-review` is reserved for
* clean completion handoffs.
*/ */
private async handoffTaskToReview(task: Task, reason: string, runId = this.getRunContextFor(task.id)?.runId): Promise<Task> { private async handoffTaskToReview(task: Task, reason: string, runId = this.getRunContextFor(task.id)?.runId): Promise<Task> {
const agentId = this.getRunContextFor(task.id)?.agentId; const agentId = this.getRunContextFor(task.id)?.agentId;
@@ -7666,6 +7668,9 @@ export class TaskExecutor {
if (failedNode === "parse" && failureValue === "pin-mismatch" && await this.routeResetParsePinMismatchToRetry(live)) { if (failedNode === "parse" && failureValue === "pin-mismatch" && await this.routeResetParsePinMismatchToRetry(live)) {
return; return;
} }
if (await this.routeGraphFailureToExecutionResume(live, failedNode ?? "unknown", failureValue)) {
return;
}
if (live.column !== "in-progress") { if (live.column !== "in-progress") {
const benignMessage = `Workflow graph run ended after task already advanced to '${live.column}' — no further action needed`; const benignMessage = `Workflow graph run ended after task already advanced to '${live.column}' — no further action needed`;
executorLog.log(`${task.id}: ${benignMessage}`); executorLog.log(`${task.id}: ${benignMessage}`);
@@ -7719,7 +7724,6 @@ export class TaskExecutor {
// FN-5704-style loop of re-running the graph from scratch. // FN-5704-style loop of re-running the graph from scratch.
await this.store.updateTask(task.id, { error: message, status: "failed" }, this.getRunContextFor(task.id)); await this.store.updateTask(task.id, { error: message, status: "failed" }, this.getRunContextFor(task.id));
await this.persistTokenUsage(task.id); await this.persistTokenUsage(task.id);
await this.handoffTaskToReview(live, "workflow-graph-failed");
} catch (err) { } catch (err) {
executorLog.error( executorLog.error(
`${task.id}: failed to park graph-failed task: ${err instanceof Error ? err.message : String(err)}`, `${task.id}: failed to park graph-failed task: ${err instanceof Error ? err.message : String(err)}`,
@@ -7727,6 +7731,47 @@ export class TaskExecutor {
} }
} }
private async routeGraphFailureToExecutionResume(
live: TaskDetail,
failedNode: string,
failureValue: string | undefined,
): Promise<boolean> {
/*
* FNXC:WorkflowLifecycle 2026-06-29-11:08:
* A workflow graph failure is not a completion handoff. FN-7228/FN-7229 showed
* restart-time parse failures and incomplete steps being parked in `in-review`
* with errors, which blocks the engine from resuming the correct unfinished
* step. Keep executable work in the executable queue: clear graph failure
* markers and move review-column rows with unfinished work back to `todo`
* preserving step progress. Generic graph failures that remain in-progress
* are left failed in-place by the caller; they must never be handed to review.
*/
if (live.deletedAt) return false;
if (live.paused || live.userPaused === true) return false;
if (live.column === "done" || live.column === "archived") return false;
const incompleteSteps = hasNonTerminalWorkflowSteps(live);
if (live.column !== "in-review" && !(incompleteSteps && live.column === "todo")) return false;
const message = incompleteSteps
? `Workflow graph failed at node '${failedNode}'${failureValue ? ` (${failureValue})` : ""} with incomplete steps — moved back to todo for execution resume`
: `Workflow graph failed at node '${failedNode}'${failureValue ? ` (${failureValue})` : ""} before a clean review handoff — moved back to todo for workflow retry`;
executorLog.warn(`${live.id}: ${message}`);
await this.store.logEntry(live.id, message, undefined, this.getRunContextFor(live.id));
await this.store.updateTask(live.id, {
status: null,
error: null,
}, this.getRunContextFor(live.id));
if (live.column !== "todo") {
await this.store.moveTask(live.id, "todo", {
preserveProgress: true,
moveSource: "engine",
recoveryRehome: true,
});
}
await this.persistTokenUsage(live.id);
return true;
}
private async routeResetParsePinMismatchToRetry(live: TaskDetail): Promise<boolean> { private async routeResetParsePinMismatchToRetry(live: TaskDetail): Promise<boolean> {
/* /*
FNXC:WorkflowReset 2026-06-29-10:04: FNXC:WorkflowReset 2026-06-29-10:04:
@@ -8409,10 +8454,9 @@ export class TaskExecutor {
paused: false, paused: false,
pausedByAgentId: null, pausedByAgentId: null,
}); });
await this.store.logEntry(task.id, `${failureMessage} — moved to in-review for inspection`, undefined, this.getRunContextFor(task.id)); await this.store.logEntry(task.id, `${failureMessage} — execution failed after worktree liveness retry budget was exhausted`, undefined, this.getRunContextFor(task.id));
await this.persistTokenUsage(task.id); await this.persistTokenUsage(task.id);
await this.handoffTaskToReview(task, "worktree-liveness-failed"); executorLog.log(`✗ ${task.id} worktree liveness failed`);
executorLog.log(`✗ ${task.id} worktree liveness failed — moved to in-review`);
} }
this.options.onError?.(task, new Error(failureMessage)); this.options.onError?.(task, new Error(failureMessage));
return; return;
@@ -8828,9 +8872,11 @@ export class TaskExecutor {
} else { } else {
const failedSteps = results.filter(r => !r.success); const failedSteps = results.filter(r => !r.success);
const errorSummary = failedSteps.map(r => `Step ${r.stepIndex}: ${r.error || "unknown error"}`).join("; "); const errorSummary = failedSteps.map(r => `Step ${r.stepIndex}: ${r.error || "unknown error"}`).join("; ");
await this.store.updateTask(task.id, { status: "failed", error: errorSummary }); await this.store.updateTask(task.id, { status: null, error: null });
await this.handoffTaskToReview(task, "step-session-failed"); await this.store.logEntry(task.id, `Step-session failed — requeued for execution resume: ${errorSummary}`, undefined, this.getRunContextFor(task.id));
executorLog.log(`✗ ${task.id} step-session failed → in-review: ${errorSummary}`); this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true });
executorLog.log(`✗ ${task.id} step-session failed → todo resume: ${errorSummary}`);
this.options.onError?.(task, new Error(errorSummary)); this.options.onError?.(task, new Error(errorSummary));
} }
}; };
@@ -8930,8 +8976,7 @@ export class TaskExecutor {
if (accumulatedStepTokenUsage) { if (accumulatedStepTokenUsage) {
await this.store.updateTask(task.id, { tokenUsage: accumulatedStepTokenUsage }); await this.store.updateTask(task.id, { tokenUsage: accumulatedStepTokenUsage });
} }
await this.handoffTaskToReview(task, "transient-retries-exhausted"); executorLog.log(`✗ ${task.id} transient retries exhausted — failed in execution`);
executorLog.log(`✗ ${task.id} transient retries exhausted → in-review`);
this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage)); this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage));
} else { } else {
if (accumulatedStepTokenUsage) { if (accumulatedStepTokenUsage) {
@@ -8942,9 +8987,10 @@ export class TaskExecutor {
} }
executorLog.error(`✗ ${task.id} step-session execution failed:`, errorDetail); executorLog.error(`✗ ${task.id} step-session execution failed:`, errorDetail);
await this.store.logEntry(task.id, `Step-session execution failed: ${errorMessage}`, errorStack ?? errorDetail, this.getRunContextFor(task.id)); await this.store.logEntry(task.id, `Step-session execution failed: ${errorMessage}`, errorStack ?? errorDetail, this.getRunContextFor(task.id));
await this.store.updateTask(task.id, { status: "failed", error: errorMessage }); await this.store.updateTask(task.id, { status: null, error: null });
await this.handoffTaskToReview(task, "step-session-failed"); this.markGraphExecuteSelfRequeued(task.id);
executorLog.log(`✗ ${task.id} step-session execution failed → in-review`); await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true });
executorLog.log(`✗ ${task.id} step-session execution failed → todo resume`);
this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage)); this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage));
} }
} finally { } finally {
@@ -9949,10 +9995,9 @@ export class TaskExecutor {
executorLog.log(`✗ ${task.id} failed after ${MAX_TASK_DONE_SESSION_RETRIES} retries — requeued to todo (${nextRequeueCount}/${MAX_TASK_DONE_REQUEUE_RETRIES})`); executorLog.log(`✗ ${task.id} failed after ${MAX_TASK_DONE_SESSION_RETRIES} retries — requeued to todo (${nextRequeueCount}/${MAX_TASK_DONE_REQUEUE_RETRIES})`);
} else { } else {
await this.store.updateTask(task.id, { status: "failed", error: errorMessage }); await this.store.updateTask(task.id, { status: "failed", error: errorMessage });
await this.store.logEntry(task.id, `${errorMessage} — moved to in-review for inspection`, undefined, this.getRunContextFor(task.id)); await this.store.logEntry(task.id, `${errorMessage} — execution failed after task-done retry budget was exhausted`, undefined, this.getRunContextFor(task.id));
await this.persistTokenUsage(task.id); await this.persistTokenUsage(task.id);
await this.handoffTaskToReview(task, "max-task-done-retries-exhausted"); executorLog.log(`✗ ${task.id} failed after ${MAX_TASK_DONE_SESSION_RETRIES} retries — no fn_task_done`);
executorLog.log(`✗ ${task.id} failed after ${MAX_TASK_DONE_SESSION_RETRIES} retries — no fn_task_done → in-review`);
} }
this.options.onError?.(task, new Error(errorMessage)); this.options.onError?.(task, new Error(errorMessage));
} }
@@ -10624,8 +10669,7 @@ export class TaskExecutor {
nextRecoveryAt: null, nextRecoveryAt: null,
}); });
await this.persistTokenUsage(task.id); await this.persistTokenUsage(task.id);
await this.handoffTaskToReview(task, "transient-retries-exhausted"); executorLog.log(`✗ ${task.id} transient retries exhausted — failed in execution`);
executorLog.log(`✗ ${task.id} transient retries exhausted → in-review`);
this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage)); this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage));
return; return;
} }
@@ -10636,8 +10680,7 @@ export class TaskExecutor {
await this.store.logEntry(task.id, `Execution failed: ${terminalError}`, errorStack ?? errorDetail, this.getRunContextFor(task.id)); await this.store.logEntry(task.id, `Execution failed: ${terminalError}`, errorStack ?? errorDetail, this.getRunContextFor(task.id));
await this.store.updateTask(task.id, { status: "failed", error: terminalError }); await this.store.updateTask(task.id, { status: "failed", error: terminalError });
await this.persistTokenUsage(task.id); await this.persistTokenUsage(task.id);
await this.handoffTaskToReview(task, "execution-failed"); executorLog.log(`✗ ${task.id} execution failed`);
executorLog.log(`✗ ${task.id} execution failed → in-review`);
this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage)); this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage));
} }
} finally { } finally {
@@ -11804,9 +11847,8 @@ export class TaskExecutor {
branch: null, branch: null,
sessionFile: null, sessionFile: null,
}); });
await this.store.logEntry(task.id, `${refusal.message} — moved to in-review for inspection`, undefined, this.getRunContextFor(task.id)); await this.store.logEntry(task.id, `${refusal.message} — execution failed because implicit fn_task_done was refused`, undefined, this.getRunContextFor(task.id));
await this.persistTokenUsage(task.id); await this.persistTokenUsage(task.id);
await this.handoffTaskToReview(task, "implicit-fn_task_done-refused");
} }
this.deleteActiveSession(task.id); this.deleteActiveSession(task.id);
@@ -11906,17 +11948,9 @@ export class TaskExecutor {
branch: null, branch: null,
sessionFile: null, sessionFile: null,
}); });
await store.logEntry(taskId, `${refusalMessage} — moved to in-review for inspection`, undefined, this.getRunContextFor(task.id)); await store.logEntry(taskId, `${refusalMessage} — invariant-check retry budget exhausted`, undefined, this.getRunContextFor(task.id));
await this.persistTokenUsage(taskId); await this.persistTokenUsage(taskId);
await store.handoffToReview(taskId, { executorLog.log(`✗ ${taskId} failed invariant check`);
ownerAgentId: this.getRunContextFor(task.id)?.agentId ?? null,
evidence: {
reason: "invariant-check-failed",
runId: this.getRunContextFor(task.id)?.runId,
agentId: this.getRunContextFor(task.id)?.agentId,
},
});
executorLog.log(`✗ ${taskId} failed invariant check — moved to in-review`);
} }
return { return {
@@ -11964,17 +11998,9 @@ export class TaskExecutor {
branch: null, branch: null,
sessionFile: null, sessionFile: null,
}); });
await store.logEntry(taskId, `${refusalMessage} — moved to in-review for inspection`, undefined, this.getRunContextFor(task.id)); await store.logEntry(taskId, `${refusalMessage} — fn_task_done refusal retry budget exhausted`, undefined, this.getRunContextFor(task.id));
await this.persistTokenUsage(taskId); await this.persistTokenUsage(taskId);
await store.handoffToReview(taskId, { executorLog.log(`✗ ${taskId} fn_task_done refusal (${taskDoneRefusal.refusalClass})`);
ownerAgentId: this.getRunContextFor(task.id)?.agentId ?? null,
evidence: {
reason: "fn_task_done-refused",
runId: this.getRunContextFor(task.id)?.runId,
agentId: this.getRunContextFor(task.id)?.agentId,
},
});
executorLog.log(`✗ ${taskId} fn_task_done refusal (${taskDoneRefusal.refusalClass}) — moved to in-review for inspection`);
} }
return { return {
@@ -16676,6 +16702,10 @@ export interface PseudoPauseResult {
matched?: string; matched?: string;
} }
function hasNonTerminalWorkflowSteps(task: Pick<TaskDetail, "steps">): boolean {
return task.steps.length > 0 && task.steps.some((step) => step.status !== "done" && step.status !== "skipped");
}
/** /**
* Detect whether the last assistant text output looks like a "pseudo-pause" — * Detect whether the last assistant text output looks like a "pseudo-pause" —
* where the agent ended a turn by asking for permission or summarizing progress * where the agent ended a turn by asking for permission or summarizing progress

View File

@@ -6328,14 +6328,21 @@ export class SelfHealingManager {
const now = Date.now(); const now = Date.now();
const tasks = await this.store.listTasks({ column: "in-review", slim: true }); const tasks = await this.store.listTasks({ column: "in-review", slim: true });
/*
* FNXC:WorkflowLifecycle 2026-06-29-11:27:
* Restart recovery must not leave errored review-column cards with unfinished
* steps. FN-7228/FN-7229 persisted `column:"in-review"` plus incomplete steps
* after graph failures; failed rows should re-enter `todo` immediately with
* progress preserved instead of waiting for the stale timeout.
*/
const staleIncomplete = tasks.filter((task) => const staleIncomplete = tasks.filter((task) =>
task.column === "in-review" && task.column === "in-review" &&
allowsAutoMergeProcessing(task, settings) && allowsAutoMergeProcessing(task, settings) &&
!task.paused && !task.paused &&
!task.status && (!task.status || task.status === "failed") &&
task.steps.length > 0 && task.steps.length > 0 &&
task.steps.some((step) => NON_TERMINAL_STEP_STATUSES.has(step.status)) && task.steps.some((step) => NON_TERMINAL_STEP_STATUSES.has(step.status)) &&
now - new Date(task.columnMovedAt ?? task.updatedAt).getTime() >= timeoutMs (task.status === "failed" || now - new Date(task.columnMovedAt ?? task.updatedAt).getTime() >= timeoutMs)
); );
if (staleIncomplete.length === 0) return 0; if (staleIncomplete.length === 0) return 0;
@@ -6345,11 +6352,13 @@ export class SelfHealingManager {
let recovered = 0; let recovered = 0;
for (const task of staleIncomplete) { for (const task of staleIncomplete) {
try { try {
const failedReviewRow = task.status === "failed";
const proof = await this.evaluateBackwardMoveTripleProof(task, { const proof = await this.evaluateBackwardMoveTripleProof(task, {
stage: "stale-incomplete-review", stage: "stale-incomplete-review",
graceMs: timeoutMs, graceMs: failedReviewRow ? 0 : timeoutMs,
stalenessAnchor: task.columnMovedAt ?? task.updatedAt, stalenessAnchor: task.columnMovedAt ?? task.updatedAt,
reason: "stale-incomplete-review-candidate", reason: failedReviewRow ? "failed-incomplete-review-candidate" : "stale-incomplete-review-candidate",
extra: { failedReviewRow },
}); });
if (!proof.ok) { if (!proof.ok) {
await this.emitBackwardMoveNoAction(task, "stale-incomplete-review", "task:stale-incomplete-review-no-action", proof); await this.emitBackwardMoveNoAction(task, "stale-incomplete-review", "task:stale-incomplete-review-no-action", proof);