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

feat(FN-4259): complete Step 3 — stop exhausted-task churn

Browse Source 
Fusion-Task-Id: FN-4259
Fusion-Task-Lineage: 7a7ca17b-93ce-4319-8461-7d1462a2c0e1
This commit is contained in:
Fusion
2026-05-13 05:18:29 -07:00
committed by gsxdsm
parent b39ba27338
commit 1bbf57b8e1
4 changed files with 81 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
packages/engine/src/__tests__/stuck-task-detector.test.ts
View File

@@ -546,7 +546,8 @@ describe("StuckTaskDetector", () => {
await customDetector.killAndRetry("FN-001", 60000);
expect(beforeRequeue).toHaveBeenCalledWith("FN-001");
expect(beforeRequeue).toHaveBeenCalledWith("FN-001", "inactivity");
expect(customDetector.trackedCount).toBe(0);
expect(session.dispose).toHaveBeenCalled();
// onStuck should still be called with shouldRequeue=false
expect(onStuck).toHaveBeenCalledWith(
@@ -571,7 +572,7 @@ describe("StuckTaskDetector", () => {
await customDetector.killAndRetry("FN-001", 60000);
expect(beforeRequeue).toHaveBeenCalledWith("FN-001");
expect(beforeRequeue).toHaveBeenCalledWith("FN-001", "inactivity");
expect(onStuck).toHaveBeenCalledWith(
expect.objectContaining({ taskId: "FN-001", shouldRequeue: true }),
);
@@ -1316,4 +1317,31 @@ describe("StuckTaskDetector heartbeat tracking (FN-978)", () => {
expect(detector.getActivitySinceProgress("FN-001")).toBe(0);
expect(detector.trackedCount).toBe(1);
});
it("does not re-track STUCK_LOOP_EXHAUSTED failed tasks", async () => {
const beforeRequeue = vi.fn().mockResolvedValue(false);
const exhaustedStore = createMockStore({
getTask: vi.fn().mockResolvedValue({
id: "FN-001",
status: "failed",
error: "STUCK_LOOP_EXHAUSTED: stuck kill budget exhausted (7/6) after last reason=loop.",
}),
});
const customDetector = new StuckTaskDetector(exhaustedStore, { beforeRequeue });
const session = createMockSession();
vi.useFakeTimers({ shouldAdvanceTime: true });
customDetector.trackTask("FN-001", session);
vi.advanceTimersByTime(61_000);
await customDetector.killAndRetry("FN-001", 60_000);
expect(customDetector.trackedCount).toBe(0);
customDetector.trackTask("FN-001-step-0", createMockSession(), "FN-001");
await Promise.resolve();
await Promise.resolve();
expect((exhaustedStore.getTask as ReturnType<typeof vi.fn>)).toHaveBeenCalledWith("FN-001");
expect(customDetector.trackedCount).toBe(0);
vi.useRealTimers();
});
});

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

@@ -7582,7 +7582,9 @@ and show an appropriate message to the user.\`
return false;
}
// Check attempt ceiling (max 1 compact-and-resume per execute() lifecycle)
// Check attempt ceiling (max 1 compact-and-resume per execute() lifecycle).
// After this fallback, StuckTaskDetector -> SelfHealingManager.checkStuckBudget
// enforces STUCK_LOOP_EXHAUSTED terminalization when retry budget is spent.
const state = this.loopRecoveryState.get(taskId);
if (state && state.attempts >= 1) {
executorLog.log(`${taskId} loop detected but compact ceiling reached — falling back to kill/requeue`);

2
packages/engine/src/runtimes/in-process-runtime.ts
View File

@@ -337,7 +337,7 @@ export class InProcessRuntime
// 5b. Initialize TaskExecutor
this.stuckTaskDetector = new StuckTaskDetector(this.taskStore, {
beforeRequeue: (taskId) => this.selfHealingManager?.checkStuckBudget(taskId) ?? Promise.resolve(true),
beforeRequeue: (taskId, reason) => this.selfHealingManager?.checkStuckBudget(taskId, reason) ?? Promise.resolve(true),
onLoopDetected: (event) => this.executor?.handleLoopDetected(event) ?? Promise.resolve(false),
onStuck: (event) => {
this.triageProcessor?.markStuckAborted(event.taskId);

52
packages/engine/src/stuck-task-detector.ts
View File

@@ -73,7 +73,7 @@ export interface StuckTaskDetectorOptions {
/** Called before re-queuing a killed task. Return false to prevent re-queue
* (caller is responsible for marking the task as terminally failed).
* Used by SelfHealingManager to enforce stuck kill budgets. */
beforeRequeue?: (taskId: string) => Promise<boolean>;
beforeRequeue?: (taskId: string, reason: "inactivity" | "loop") => Promise<boolean>;
/** Pre-kill callback invoked ONLY when reason is "loop".
* Called BEFORE session.dispose() / moveTask("todo") so the caller can
* attempt in-process recovery (e.g. compact-and-resume) without killing
@@ -93,9 +93,10 @@ export class StuckTaskDetector {
private interval: ReturnType<typeof setInterval> | null = null;
private pollIntervalMs: number;
private onStuck?: (event: StuckTaskEvent) => void;
private beforeRequeue?: (taskId: string) => Promise<boolean>;
private beforeRequeue?: (taskId: string, reason: "inactivity" | "loop") => Promise<boolean>;
private onLoopDetected?: (event: StuckTaskEvent) => Promise<boolean>;
private paused = false;
private exhaustedTasks = new Set<string>();
constructor(
private store: TaskStore,
@@ -146,15 +147,51 @@ export class StuckTaskDetector {
* trackingKey is used as-is (single-session mode where they are identical).
*/
trackTask(trackingKey: string, session: DisposableSession, canonicalTaskId?: string): void {
const canonicalId = canonicalTaskId ?? trackingKey;
if (this.exhaustedTasks.has(canonicalId)) {
void this.store.getTask(canonicalId)
.then((task) => {
const isExhausted = task.status === "failed" && task.error?.startsWith("STUCK_LOOP_EXHAUSTED:");
if (isExhausted) {
stuckLog.log(`Skipping tracking for ${trackingKey} (canonical=${canonicalId}) — task is in STUCK_LOOP_EXHAUSTED terminal state`);
return;
}
this.exhaustedTasks.delete(canonicalId);
const now = Date.now();
this.tracked.set(trackingKey, {
session,
lastActivity: now,
lastProgressAt: now,
activitySinceProgress: 0,
canonicalTaskId: canonicalId,
});
stuckLog.log(`Tracking task ${trackingKey} (canonical=${canonicalId}, total tracked: ${this.tracked.size})`);
})
.catch((err) => {
stuckLog.error(`Failed to validate exhausted status for ${canonicalId}; proceeding to track:`, err);
this.exhaustedTasks.delete(canonicalId);
const now = Date.now();
this.tracked.set(trackingKey, {
session,
lastActivity: now,
lastProgressAt: now,
activitySinceProgress: 0,
canonicalTaskId: canonicalId,
});
stuckLog.log(`Tracking task ${trackingKey} (canonical=${canonicalId}, total tracked: ${this.tracked.size})`);
});
return;
}
const now = Date.now();
this.tracked.set(trackingKey, {
session,
lastActivity: now,
lastProgressAt: now,
activitySinceProgress: 0,
canonicalTaskId: canonicalTaskId ?? trackingKey,
canonicalTaskId: canonicalId,
});
stuckLog.log(`Tracking task ${trackingKey} (canonical=${canonicalTaskId ?? trackingKey}, total tracked: ${this.tracked.size})`);
stuckLog.log(`Tracking task ${trackingKey} (canonical=${canonicalId}, total tracked: ${this.tracked.size})`);
}
/**
@@ -343,7 +380,7 @@ export class StuckTaskDetector {
let shouldRequeue = true;
if (this.beforeRequeue) {
try {
shouldRequeue = await this.beforeRequeue(canonicalId);
shouldRequeue = await this.beforeRequeue(canonicalId, reason);
if (!shouldRequeue) {
stuckLog.log(`${canonicalId} exceeded stuck kill budget — not re-queuing`);
}
@@ -394,6 +431,11 @@ export class StuckTaskDetector {
// mark the abort as intentional before the disposed session unwinds.
this.onStuck?.(event);
if (!shouldRequeue) {
this.exhaustedTasks.add(canonicalId);
stuckLog.log(`${canonicalId} untracked due to STUCK_LOOP_EXHAUSTED terminal state (no automatic retries)`);
}
// Dispose the agent session after listeners have marked the abort.
try {
entry.session.dispose();