FN-5889: suppress post-done continuation wedges
Keep completed post-done continuation errors from wedging review tasks in a failed state. - route step-session non-continuable continuation errors through the executor recovery path before marking tasks failed - add self-healing recovery and run-audit events for in-review tasks already wedged by post-done non-continuable errors - extend reliability coverage and architecture/backstop docs for the new post-done wedge handling Files changed: AGENTS.md | 1 + docs/architecture.md | 3 +- packages/engine/src/__tests__/reliability-interactions/post-done-continuation-no-wedge.test.ts | 131 ++++++++++++++++++++- packages/engine/src/executor.ts | 11 +- packages/engine/src/run-audit.ts | 2 + packages/engine/src/self-healing.ts | 97 ++++++++++++++- 6 files changed, 236 insertions(+), 9 deletions(-) Fusion-Task-Id: FN-5889 Fusion-Task-Lineage: c22716f2-cddb-4c98-ac87-0282017e7c82
This commit is contained in:
gsxdsm
@@ -3,9 +3,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { Task, TaskStore } from "@fusion/core";
|
||||
import "../executor-test-helpers.js";
|
||||
import { TaskExecutor } from "../../executor.js";
|
||||
import { SelfHealingManager } from "../../self-healing.js";
|
||||
import { MAX_POST_DONE_NONCONTINUABLE_WEDGE_RECOVERIES, SelfHealingManager } from "../../self-healing.js";
|
||||
import { MAX_RECOVERY_RETRIES } from "../../recovery-policy.js";
|
||||
import { mockedCreateFnAgent, resetExecutorMocks } from "../executor-test-helpers.js";
|
||||
import { mockExecuteAll, mockedCreateFnAgent, resetExecutorMocks } from "../executor-test-helpers.js";
|
||||
|
||||
function makeTask(overrides: Partial<Task> = {}): Task {
|
||||
return {
|
||||
@@ -54,7 +54,10 @@ function createStore(task: Task, settingsOverrides: Record<string, unknown> = {}
|
||||
...settingsOverrides,
|
||||
});
|
||||
(emitter as any).updateTask = vi.fn().mockImplementation(async (_taskId: string, updates: Partial<Task>) => {
|
||||
Object.assign(task, updates, { updatedAt: new Date(Date.now()).toISOString() });
|
||||
const normalized = { ...updates } as Record<string, unknown>;
|
||||
if (normalized.status === null) normalized.status = undefined;
|
||||
if (normalized.error === null) normalized.error = undefined;
|
||||
Object.assign(task, normalized, { updatedAt: new Date(Date.now()).toISOString() });
|
||||
return task;
|
||||
});
|
||||
(emitter as any).moveTask = vi.fn().mockImplementation(async (_taskId: string, column: Task["column"]) => {
|
||||
@@ -92,6 +95,48 @@ function createStore(task: Task, settingsOverrides: Record<string, unknown> = {}
|
||||
return emitter;
|
||||
}
|
||||
|
||||
function createSelfHealingStore(tasks: Task[], settingsOverrides: Record<string, unknown> = {}): TaskStore & EventEmitter {
|
||||
const emitter = new EventEmitter() as TaskStore & EventEmitter;
|
||||
const audits: any[] = [];
|
||||
const taskMap = new Map(tasks.map((task) => [task.id, task]));
|
||||
|
||||
(emitter as any).__audits = audits;
|
||||
(emitter as any).getTask = vi.fn().mockImplementation(async (taskId: string) => taskMap.get(taskId));
|
||||
(emitter as any).listTasks = vi.fn().mockImplementation(async ({ column }: { column?: string } = {}) => {
|
||||
const values = [...taskMap.values()];
|
||||
return column ? values.filter((task) => task.column === column) : values;
|
||||
});
|
||||
(emitter as any).getSettings = vi.fn().mockResolvedValue({
|
||||
autoMerge: true,
|
||||
globalPause: false,
|
||||
enginePaused: false,
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
pollIntervalMs: 15_000,
|
||||
inReviewStallDeadlockThreshold: 3,
|
||||
taskStuckTimeoutMs: 60_000,
|
||||
...settingsOverrides,
|
||||
});
|
||||
(emitter as any).updateTask = vi.fn().mockImplementation(async (taskId: string, updates: Partial<Task>) => {
|
||||
const task = taskMap.get(taskId)!;
|
||||
const normalized = { ...updates } as Record<string, unknown>;
|
||||
if (normalized.status === null) normalized.status = undefined;
|
||||
if (normalized.error === null) normalized.error = undefined;
|
||||
Object.assign(task, normalized, { updatedAt: new Date(Date.now()).toISOString() });
|
||||
return task;
|
||||
});
|
||||
(emitter as any).logEntry = vi.fn().mockImplementation(async (taskId: string, action: string, detail?: string) => {
|
||||
const task = taskMap.get(taskId)!;
|
||||
task.log = task.log ?? [];
|
||||
task.log.push({ timestamp: new Date(Date.now()).toISOString(), action, detail } as any);
|
||||
});
|
||||
(emitter as any).recordRunAuditEvent = vi.fn().mockImplementation(async (event: any) => {
|
||||
audits.push(event);
|
||||
});
|
||||
(emitter as any).emit = emitter.emit.bind(emitter);
|
||||
return emitter;
|
||||
}
|
||||
|
||||
describe("FN-5866 reliability interactions: post-done continuation no wedge", () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
@@ -143,6 +188,39 @@ describe("FN-5866 reliability interactions: post-done continuation no wedge", ()
|
||||
manager.stop();
|
||||
});
|
||||
|
||||
it("keeps completed work cleanly in-review when a post-done step-session continuation is not continuable", async () => {
|
||||
const task = makeTask({
|
||||
id: "FN-5889-STEP-SESSION-WEDGE",
|
||||
steps: [{ name: "Implement", status: "in-progress" as const }],
|
||||
});
|
||||
const store = createStore(task, { runStepsInNewSessions: true });
|
||||
const onComplete = vi.fn();
|
||||
const onError = vi.fn();
|
||||
|
||||
mockExecuteAll.mockImplementation(async () => {
|
||||
task.steps = [{ name: "Implement", status: "done" as const }];
|
||||
task.currentStep = 1;
|
||||
task.log = [
|
||||
...task.log,
|
||||
{ timestamp: new Date(Date.now()).toISOString(), action: "Task marked done by agent" } as any,
|
||||
];
|
||||
throw new Error("Cannot continue from message role: assistant");
|
||||
});
|
||||
|
||||
const executor = new TaskExecutor(store, "/tmp/test", { onComplete, onError, agentStore: { getAgent: vi.fn().mockResolvedValue(null) } as any });
|
||||
await executor.execute(task);
|
||||
|
||||
// Pre-fix root cause: the post-done step-session catch in executor.ts marked
|
||||
// status=failed + handoff directly instead of consulting handleNonContinuableSessionError().
|
||||
expect(task.column).toBe("in-review");
|
||||
expect(task.status).toBeUndefined();
|
||||
expect(task.error).toBeUndefined();
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
expect(onComplete).toHaveBeenCalled();
|
||||
expect(store.handoffToReview).toHaveBeenCalledTimes(1);
|
||||
expect((task.log ?? []).some((entry: any) => entry.action.includes("Post-done session continuation suppressed"))).toBe(true);
|
||||
});
|
||||
|
||||
it("requeues incomplete work with a fresh session when the session is not continuable", async () => {
|
||||
const task = makeTask({
|
||||
id: "FN-5866-INCOMPLETE",
|
||||
@@ -176,6 +254,53 @@ describe("FN-5866 reliability interactions: post-done continuation no wedge", ()
|
||||
expect((task.log ?? []).some((entry: any) => entry.action.includes("Non-continuable session — fresh-session retry"))).toBe(true);
|
||||
});
|
||||
|
||||
it("self-heals already wedged post-done non-continuable failures back to clean in-review", async () => {
|
||||
const wedged = makeTask({
|
||||
id: "FN-5889-WEDGED",
|
||||
column: "in-review",
|
||||
status: "failed",
|
||||
error: "Cannot continue from message role: assistant",
|
||||
steps: [{ name: "Implement", status: "done" as const }],
|
||||
log: [{ timestamp: new Date(Date.now() - 60_000).toISOString(), action: "Task marked done by agent" } as any],
|
||||
});
|
||||
const exhausted = makeTask({
|
||||
id: "FN-5889-EXHAUSTED",
|
||||
column: "in-review",
|
||||
status: "failed",
|
||||
error: "Cannot continue from message role: assistant",
|
||||
completionHandoffLimboRecoveryCount: MAX_POST_DONE_NONCONTINUABLE_WEDGE_RECOVERIES,
|
||||
steps: [{ name: "Implement", status: "done" as const }],
|
||||
log: [{ timestamp: new Date(Date.now() - 60_000).toISOString(), action: "Task marked done by agent" } as any],
|
||||
});
|
||||
const nonMatching = makeTask({
|
||||
id: "FN-5889-NON-MATCH",
|
||||
column: "in-review",
|
||||
status: "failed",
|
||||
error: "Different failure",
|
||||
steps: [{ name: "Implement", status: "in-progress" as const }],
|
||||
log: [{ timestamp: new Date(Date.now() - 60_000).toISOString(), action: "Task marked done by agent" } as any],
|
||||
});
|
||||
const store = createSelfHealingStore([wedged, exhausted, nonMatching]);
|
||||
const manager = new SelfHealingManager(store, { rootDir: "/tmp/repo" });
|
||||
|
||||
expect(await manager.recoverPostDoneNonContinuableWedge()).toBe(1);
|
||||
|
||||
expect(wedged.column).toBe("in-review");
|
||||
expect(wedged.status).toBeUndefined();
|
||||
expect(wedged.error).toBeUndefined();
|
||||
expect(wedged.completionHandoffLimboRecoveryCount).toBe(1);
|
||||
expect((wedged.log ?? []).some((entry: any) => entry.action.includes("Auto-recovered completed-task non-continuable wedge"))).toBe(true);
|
||||
expect(((store as any).__audits as any[]).some((event: any) => event.mutationType === "task:auto-recover-post-done-noncontinuable-wedge" && event.target === wedged.id)).toBe(true);
|
||||
|
||||
expect(exhausted.status).toBe("failed");
|
||||
expect(exhausted.error).toBe("Cannot continue from message role: assistant");
|
||||
expect(((store as any).__audits as any[]).some((event: any) => event.mutationType === "task:auto-recover-post-done-noncontinuable-wedge-exhausted" && event.target === exhausted.id)).toBe(true);
|
||||
|
||||
expect(nonMatching.status).toBe("failed");
|
||||
expect(nonMatching.error).toBe("Different failure");
|
||||
manager.stop();
|
||||
});
|
||||
|
||||
it("falls through to terminal failure after the non-continuable fresh-session retry budget is exhausted", async () => {
|
||||
const task = makeTask({
|
||||
id: "FN-5866-INCOMPLETE-EXHAUSTED",
|
||||
|
||||
@@ -2580,7 +2580,7 @@ export class TaskExecutor {
|
||||
await this.store.logEntry(task.id, diagnosticMessage, errorMessage, this.getRunContextFor(task.id));
|
||||
|
||||
if (liveTask.status === "failed" || liveTask.error) {
|
||||
await this.store.updateTask(task.id, { status: undefined, error: undefined });
|
||||
await this.store.updateTask(task.id, { status: null, error: null });
|
||||
}
|
||||
|
||||
await this.persistTokenUsage(task.id);
|
||||
@@ -4003,12 +4003,15 @@ export class TaskExecutor {
|
||||
executorLog.log(`✗ ${task.id} transient retries exhausted → in-review`);
|
||||
this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage));
|
||||
} else {
|
||||
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.updateTask(task.id, { status: "failed", error: errorMessage });
|
||||
if (accumulatedStepTokenUsage) {
|
||||
await this.store.updateTask(task.id, { tokenUsage: accumulatedStepTokenUsage });
|
||||
}
|
||||
if (await this.handleNonContinuableSessionError(task, false, errorMessage)) {
|
||||
return;
|
||||
}
|
||||
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.updateTask(task.id, { status: "failed", error: errorMessage });
|
||||
await this.handoffTaskToReview(task, "step-session-failed");
|
||||
executorLog.log(`✗ ${task.id} step-session execution failed → in-review`);
|
||||
this.options.onError?.(task, err instanceof Error ? err : new Error(errorMessage));
|
||||
|
||||
@@ -466,6 +466,8 @@ export type DatabaseMutationType =
|
||||
| "task:auto-recover-completion-fanout"
|
||||
| "task:auto-recover-completion-handoff-limbo"
|
||||
| "task:auto-recover-completion-handoff-limbo-exhausted"
|
||||
| "task:auto-recover-post-done-noncontinuable-wedge"
|
||||
| "task:auto-recover-post-done-noncontinuable-wedge-exhausted"
|
||||
| "task:auto-recover-worktree-session-exhausted"
|
||||
| "task:auto-recover-in-progress-limbo"
|
||||
| "task:resume-limbo-escalated"
|
||||
|
||||
@@ -39,7 +39,7 @@ import {
|
||||
isRecoverableMissingWorktreeReviewFailureNoProgress,
|
||||
isRecoverableMissingWorktreeReviewFailureWithProgress,
|
||||
} from "./restart-recovery-coordinator.js";
|
||||
import { classifyError, extractMissingModulePath, isOperatorActionableAgentError, isStaleWorktreeModuleResolutionError } from "./transient-error-detector.js";
|
||||
import { classifyError, extractMissingModulePath, isNonContinuableSessionError, isOperatorActionableAgentError, isStaleWorktreeModuleResolutionError } from "./transient-error-detector.js";
|
||||
import { classifyForeignOnlyContamination, deriveTaskIdFromFusionBranch, inspectBranchConflict, listUniqueBranchCommits } from "./branch-conflicts.js";
|
||||
import { createRunAuditor, generateSyntheticRunId, type DatabaseMutationType, type RunAuditor } from "./run-audit.js";
|
||||
import { AutoRecoveryDispatcher } from "./auto-recovery.js";
|
||||
@@ -73,6 +73,7 @@ const DB_CORRUPTION_NOTIFICATION_COOLDOWN_MS = 60 * 60 * 1000;
|
||||
export const STALE_ACTIVE_BRANCH_EXECUTION_GRACE_MS = 10 * 60_000;
|
||||
export const COMPLETION_HANDOFF_LIMBO_GRACE_MS = 5 * 60_000;
|
||||
export const MAX_COMPLETION_HANDOFF_LIMBO_RECOVERIES = 3;
|
||||
export const MAX_POST_DONE_NONCONTINUABLE_WEDGE_RECOVERIES = 3;
|
||||
const MAX_NO_PROGRESS_RESUME_ATTEMPTS = 2;
|
||||
|
||||
type BranchGroupLandingRecorder = {
|
||||
@@ -838,6 +839,7 @@ export class SelfHealingManager {
|
||||
// not stalled by a leaked `status: "merging"` on an already-done task.
|
||||
{ name: "reconcile-stale-merger-status", fn: () => this.reconcileStaleMergerStatus().then(() => undefined) },
|
||||
{ name: "recover-already-merged-review", fn: () => this.recoverAlreadyMergedReviewTasks().then(() => undefined) },
|
||||
{ name: "recover-post-done-noncontinuable-wedge", fn: () => this.recoverPostDoneNonContinuableWedge().then(() => undefined) },
|
||||
{ name: "recover-completion-handoff-limbo", fn: () => this.recoverCompletionHandoffLimbo().then(() => undefined) },
|
||||
{ name: "recover-branch-misbound-in-review", fn: () => this.recoverBranchMisboundInReviewTasks().then(() => undefined) },
|
||||
{ name: "recover-foreign-only-contamination-in-review", fn: () => this.recoverForeignOnlyContaminatedInReviewTasks().then(() => undefined) },
|
||||
@@ -1697,6 +1699,7 @@ export class SelfHealingManager {
|
||||
{ name: "recover-mergeable-review", fn: () => this.recoverMergeableReviewTasks() },
|
||||
{ name: "recover-merged-review", fn: () => this.recoverMergedReviewTasks() },
|
||||
{ name: "recover-already-merged-review", fn: () => this.recoverAlreadyMergedReviewTasks() },
|
||||
{ name: "recover-post-done-noncontinuable-wedge", fn: () => this.recoverPostDoneNonContinuableWedge() },
|
||||
{ name: "recover-completion-handoff-limbo", fn: () => this.recoverCompletionHandoffLimbo() },
|
||||
{ name: "recover-branch-misbound-in-review", fn: () => this.recoverBranchMisboundInReviewTasks() },
|
||||
{ name: "recover-foreign-only-contamination-in-review", fn: () => this.recoverForeignOnlyContaminatedInReviewTasks() },
|
||||
@@ -6534,6 +6537,98 @@ export class SelfHealingManager {
|
||||
}
|
||||
}
|
||||
|
||||
private getPostDoneNonContinuableEvidence(task: Task): string | null {
|
||||
const candidates: string[] = [];
|
||||
if (typeof task.error === "string" && task.error.trim()) {
|
||||
candidates.push(task.error);
|
||||
}
|
||||
for (const entry of [...(task.log ?? [])].reverse()) {
|
||||
if (typeof entry.outcome === "string" && entry.outcome.trim()) {
|
||||
candidates.push(entry.outcome);
|
||||
}
|
||||
if (typeof entry.action === "string" && entry.action.trim()) {
|
||||
candidates.push(entry.action);
|
||||
}
|
||||
}
|
||||
return candidates.find((value) => isNonContinuableSessionError(value)) ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Recover completed in-review tasks wedged as failed only because a post-done
|
||||
* session continuation hit a non-continuable signature.
|
||||
*
|
||||
* No-op when `settings.autoMerge === false` — PR-based review flow owns lifecycle until human merge.
|
||||
*/
|
||||
async recoverPostDoneNonContinuableWedge(): Promise<number> {
|
||||
try {
|
||||
const settings = await this.store.getSettings();
|
||||
if (settings.globalPause || settings.enginePaused) return 0;
|
||||
if (settings.autoMerge === false) return 0;
|
||||
|
||||
const tasks = await this.store.listTasks({ column: "in-review", slim: false });
|
||||
let recovered = 0;
|
||||
|
||||
for (const task of tasks) {
|
||||
if (task.column !== "in-review" || task.deletedAt) continue;
|
||||
if (task.paused || task.userPaused) continue;
|
||||
if (task.status !== "failed") continue;
|
||||
if (this.options.isTaskActive?.(task.id)) continue;
|
||||
if (!(task.steps ?? []).every((step) => step.status === "done" || step.status === "skipped")) continue;
|
||||
const doneMarker = [...(task.log ?? [])].reverse().find((entry) => entry.action === "Task marked done by agent");
|
||||
if (!doneMarker) continue;
|
||||
if (getTaskHardMergeBlocker({ ...task, status: undefined, error: undefined, steps: task.steps ?? [], workflowStepResults: task.workflowStepResults })) continue;
|
||||
|
||||
const evidence = this.getPostDoneNonContinuableEvidence(task);
|
||||
if (!evidence) continue;
|
||||
|
||||
const currentCount = task.completionHandoffLimboRecoveryCount ?? 0;
|
||||
const audit = createRunAuditor(this.store, {
|
||||
runId: generateSyntheticRunId("self-heal-post-done-noncontinuable", task.id),
|
||||
agentId: "self-healing",
|
||||
taskId: task.id,
|
||||
taskLineageId: task.lineageId,
|
||||
phase: "recover-post-done-noncontinuable-wedge",
|
||||
});
|
||||
|
||||
if (currentCount >= MAX_POST_DONE_NONCONTINUABLE_WEDGE_RECOVERIES) {
|
||||
await audit.database({
|
||||
type: "task:auto-recover-post-done-noncontinuable-wedge-exhausted",
|
||||
target: task.id,
|
||||
metadata: { attempts: currentCount, errorSnippet: evidence.slice(0, 200) },
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
await this.store.updateTask(task.id, {
|
||||
completionHandoffLimboRecoveryCount: currentCount + 1,
|
||||
status: null,
|
||||
error: null,
|
||||
});
|
||||
await this.store.logEntry(
|
||||
task.id,
|
||||
"Auto-recovered completed-task non-continuable wedge — cleared failed status after post-done session continuation error",
|
||||
evidence,
|
||||
);
|
||||
await audit.database({
|
||||
type: "task:auto-recover-post-done-noncontinuable-wedge",
|
||||
target: task.id,
|
||||
metadata: {
|
||||
attempts: currentCount + 1,
|
||||
source: "self-healing-in-review-sweep",
|
||||
errorSnippet: evidence.slice(0, 200),
|
||||
},
|
||||
});
|
||||
recovered += 1;
|
||||
}
|
||||
|
||||
return recovered;
|
||||
} catch (err: unknown) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
log.error(`Post-done non-continuable wedge recovery failed: ${errorMessage}`);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* No-op when `settings.autoMerge === false` — PR-based review flow owns lifecycle until human merge.
|
||||
*/
|
||||
|
||||
Reference in New Issue
Block a user