feat(FN-775): add recoverable-retry with bounded exponential backoff
- Add state to tasks: persisted in DB, gates scheduler pickup to prevent immediate retry of transient failures - Introduce shared recovery-policy module with bounded exponential backoff (1s → 60s, max 5 attempts) - Wire recovery policy into executor, scheduler, and triage so all agents respect the same retry cadence - Persist retry state (attempt count, next eligible time) in task metadata via store and DB schema - Add DB migration for new retry columns and update schema tests - Update README with recovery policy documentation - Refactor dashboard Header component and styles, consolidate header tests - Fix session-files route tests to align with updated route signatures
This commit is contained in:
@@ -3551,7 +3551,12 @@ describe("TaskExecutor usage limit detection", () => {
|
||||
});
|
||||
|
||||
expect(onUsageLimitHitSpy).not.toHaveBeenCalled();
|
||||
expect(store.logEntry).toHaveBeenCalledWith("FN-001", "Transient error (will retry): connection refused");
|
||||
// Recovery policy: first transient error → retry 1/3 with backoff
|
||||
expect(store.logEntry).toHaveBeenCalledWith("FN-001", expect.stringContaining("Transient error (retry 1/3"));
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-001", expect.objectContaining({
|
||||
recoveryRetryCount: 1,
|
||||
nextRecoveryAt: expect.any(String),
|
||||
}));
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
|
||||
expect(store.updateTask).not.toHaveBeenCalledWith(
|
||||
"FN-001",
|
||||
@@ -3663,6 +3668,190 @@ describe("TaskExecutor usage limit detection", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("TaskExecutor bounded recovery retries", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
it("increments recoveryRetryCount on successive transient failures", async () => {
|
||||
const store = createMockStore();
|
||||
const onError = vi.fn();
|
||||
|
||||
mockedCreateHaiAgent.mockRejectedValue(new Error("upstream connect error"));
|
||||
|
||||
const executor = new TaskExecutor(store, "/tmp/test", { onError });
|
||||
|
||||
// First failure: count goes from undefined to 1
|
||||
await executor.execute({
|
||||
id: "FN-001",
|
||||
title: "Test",
|
||||
description: "Test",
|
||||
column: "in-progress",
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-001", expect.objectContaining({
|
||||
recoveryRetryCount: 1,
|
||||
nextRecoveryAt: expect.any(String),
|
||||
}));
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
|
||||
// Second failure: count goes from 1 to 2
|
||||
vi.clearAllMocks();
|
||||
mockedCreateHaiAgent.mockRejectedValue(new Error("upstream connect error"));
|
||||
await executor.execute({
|
||||
id: "FN-001",
|
||||
title: "Test",
|
||||
description: "Test",
|
||||
column: "in-progress",
|
||||
recoveryRetryCount: 1,
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-001", expect.objectContaining({
|
||||
recoveryRetryCount: 2,
|
||||
}));
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("escalates to failure when recovery retries are exhausted", async () => {
|
||||
const store = createMockStore();
|
||||
const onError = vi.fn();
|
||||
|
||||
mockedCreateHaiAgent.mockRejectedValue(new Error("socket hang up"));
|
||||
|
||||
const executor = new TaskExecutor(store, "/tmp/test", { onError });
|
||||
|
||||
// Task already has 3 retries (max) — next failure should escalate
|
||||
await executor.execute({
|
||||
id: "FN-001",
|
||||
title: "Test",
|
||||
description: "Test",
|
||||
column: "in-progress",
|
||||
recoveryRetryCount: 3,
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-001", expect.objectContaining({
|
||||
status: "failed",
|
||||
error: "socket hang up",
|
||||
recoveryRetryCount: null,
|
||||
nextRecoveryAt: null,
|
||||
}));
|
||||
expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "todo");
|
||||
expect(onError).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does NOT consume retry budget for paused tasks", async () => {
|
||||
const store = createMockStore();
|
||||
|
||||
const executor = new TaskExecutor(store, "/tmp/test", {});
|
||||
|
||||
// Simulate a paused abort — the executor checks pausedAborted set
|
||||
const task = {
|
||||
id: "FN-001",
|
||||
title: "Test",
|
||||
description: "Test",
|
||||
column: "in-progress" as const,
|
||||
recoveryRetryCount: 1,
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
};
|
||||
|
||||
// Simulate: task gets paused mid-execution → abort error
|
||||
mockedCreateHaiAgent.mockRejectedValue(new Error("Aborted"));
|
||||
(executor as any).pausedAborted.add("FN-001");
|
||||
|
||||
await executor.execute(task);
|
||||
|
||||
// Should NOT update recoveryRetryCount
|
||||
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", expect.objectContaining({
|
||||
recoveryRetryCount: expect.any(Number),
|
||||
}));
|
||||
});
|
||||
|
||||
it("does NOT consume retry budget for stuck-task-detector kills", async () => {
|
||||
const store = createMockStore();
|
||||
|
||||
const executor = new TaskExecutor(store, "/tmp/test", {});
|
||||
|
||||
mockedCreateHaiAgent.mockRejectedValue(new Error("Aborted"));
|
||||
(executor as any).stuckAborted.add("FN-001");
|
||||
|
||||
await executor.execute({
|
||||
id: "FN-001",
|
||||
title: "Test",
|
||||
description: "Test",
|
||||
column: "in-progress",
|
||||
recoveryRetryCount: 2,
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
// Should NOT update recoveryRetryCount
|
||||
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", expect.objectContaining({
|
||||
recoveryRetryCount: expect.any(Number),
|
||||
}));
|
||||
});
|
||||
|
||||
it("clears recovery metadata after successful run completes", async () => {
|
||||
const store = createMockStore();
|
||||
|
||||
// Mock successful agent session
|
||||
const mockSession = {
|
||||
prompt: vi.fn().mockResolvedValue(undefined),
|
||||
dispose: vi.fn(),
|
||||
state: { error: undefined },
|
||||
};
|
||||
mockedCreateHaiAgent.mockResolvedValue({ session: mockSession } as any);
|
||||
|
||||
const executor = new TaskExecutor(store, "/tmp/test", {});
|
||||
|
||||
await executor.execute({
|
||||
id: "FN-001",
|
||||
title: "Test",
|
||||
description: "Test",
|
||||
column: "in-progress",
|
||||
recoveryRetryCount: 2,
|
||||
nextRecoveryAt: new Date().toISOString(),
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
// moveTask to in-review clears recovery metadata (via store's column transition logic)
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-review");
|
||||
});
|
||||
});
|
||||
|
||||
describe("Per-task model overrides", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
|
||||
@@ -15,6 +15,7 @@ import { executorLog, reviewerLog } from "./logger.js";
|
||||
import { isUsageLimitError, checkSessionError, type UsageLimitPauser } from "./usage-limit-detector.js";
|
||||
import { isTransientError } from "./transient-error-detector.js";
|
||||
import { withRateLimitRetry } from "./rate-limit-retry.js";
|
||||
import { computeRecoveryDecision, formatDelay, MAX_RECOVERY_RETRIES } from "./recovery-policy.js";
|
||||
import type { StuckTaskDetector } from "./stuck-task-detector.js";
|
||||
|
||||
// Re-export for backward compatibility (tests import from executor.ts)
|
||||
@@ -723,10 +724,35 @@ export class TaskExecutor {
|
||||
if (this.options.usageLimitPauser && isUsageLimitError(err.message)) {
|
||||
await this.options.usageLimitPauser.onUsageLimitHit("executor", task.id, err.message);
|
||||
} else if (isTransientError(err.message)) {
|
||||
// Transient network/infrastructure error — retry instead of failing
|
||||
executorLog.warn(`⚡ ${task.id} transient error — moving to todo for retry: ${err.message}`);
|
||||
await this.store.logEntry(task.id, `Transient error (will retry): ${err.message}`);
|
||||
await this.store.moveTask(task.id, "todo");
|
||||
// Transient network/infrastructure error — use bounded recovery policy
|
||||
const decision = computeRecoveryDecision({
|
||||
recoveryRetryCount: task.recoveryRetryCount,
|
||||
nextRecoveryAt: task.nextRecoveryAt,
|
||||
});
|
||||
|
||||
if (decision.shouldRetry) {
|
||||
const attempt = decision.nextState.recoveryRetryCount;
|
||||
const delay = formatDelay(decision.delayMs);
|
||||
executorLog.warn(`⚡ ${task.id} transient error — retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}: ${err.message}`);
|
||||
await this.store.logEntry(task.id, `Transient error (retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}): ${err.message}`);
|
||||
await this.store.updateTask(task.id, {
|
||||
recoveryRetryCount: decision.nextState.recoveryRetryCount,
|
||||
nextRecoveryAt: decision.nextState.nextRecoveryAt,
|
||||
});
|
||||
await this.store.moveTask(task.id, "todo");
|
||||
return;
|
||||
}
|
||||
|
||||
// Recovery budget exhausted — escalate to real failure
|
||||
executorLog.error(`✗ ${task.id} transient error retries exhausted (${MAX_RECOVERY_RETRIES} attempts): ${err.message}`);
|
||||
await this.store.logEntry(task.id, `Transient error retries exhausted after ${MAX_RECOVERY_RETRIES} attempts: ${err.message}`);
|
||||
await this.store.updateTask(task.id, {
|
||||
status: "failed",
|
||||
error: err.message,
|
||||
recoveryRetryCount: null,
|
||||
nextRecoveryAt: null,
|
||||
});
|
||||
this.options.onError?.(task, err);
|
||||
return;
|
||||
}
|
||||
executorLog.error(`✗ ${task.id} execution failed:`, err.message);
|
||||
|
||||
160
packages/engine/src/recovery-policy.test.ts
Normal file
160
packages/engine/src/recovery-policy.test.ts
Normal file
@@ -0,0 +1,160 @@
|
||||
import { describe, it, expect, vi, afterEach } from "vitest";
|
||||
import {
|
||||
computeRecoveryDecision,
|
||||
formatDelay,
|
||||
MAX_RECOVERY_RETRIES,
|
||||
BASE_DELAY_MS,
|
||||
MAX_DELAY_MS,
|
||||
BACKOFF_MULTIPLIER,
|
||||
} from "./recovery-policy.js";
|
||||
|
||||
describe("computeRecoveryDecision", () => {
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("returns shouldRetry=true on first failure (count=0)", () => {
|
||||
const decision = computeRecoveryDecision({});
|
||||
expect(decision.shouldRetry).toBe(true);
|
||||
expect(decision.exhausted).toBe(false);
|
||||
expect(decision.nextState.recoveryRetryCount).toBe(1);
|
||||
expect(decision.nextState.nextRecoveryAt).toBeDefined();
|
||||
expect(decision.delayMs).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it("increments recovery count on each attempt", () => {
|
||||
const d1 = computeRecoveryDecision({ recoveryRetryCount: 0 });
|
||||
expect(d1.nextState.recoveryRetryCount).toBe(1);
|
||||
|
||||
const d2 = computeRecoveryDecision({ recoveryRetryCount: 1 });
|
||||
expect(d2.nextState.recoveryRetryCount).toBe(2);
|
||||
|
||||
const d3 = computeRecoveryDecision({ recoveryRetryCount: 2 });
|
||||
expect(d3.nextState.recoveryRetryCount).toBe(3);
|
||||
});
|
||||
|
||||
it("exhausts after MAX_RECOVERY_RETRIES attempts", () => {
|
||||
const decision = computeRecoveryDecision({
|
||||
recoveryRetryCount: MAX_RECOVERY_RETRIES,
|
||||
});
|
||||
expect(decision.shouldRetry).toBe(false);
|
||||
expect(decision.exhausted).toBe(true);
|
||||
expect(decision.nextState.recoveryRetryCount).toBeUndefined();
|
||||
expect(decision.nextState.nextRecoveryAt).toBeUndefined();
|
||||
expect(decision.delayMs).toBe(0);
|
||||
});
|
||||
|
||||
it("also exhausts when count exceeds max (overflow safety)", () => {
|
||||
const decision = computeRecoveryDecision({
|
||||
recoveryRetryCount: 999,
|
||||
});
|
||||
expect(decision.shouldRetry).toBe(false);
|
||||
expect(decision.exhausted).toBe(true);
|
||||
});
|
||||
|
||||
it("uses exponential backoff with increasing delays", () => {
|
||||
// Use fixed random for deterministic test
|
||||
vi.spyOn(Math, "random").mockReturnValue(0.5); // No jitter when random=0.5
|
||||
|
||||
const d1 = computeRecoveryDecision({});
|
||||
const d2 = computeRecoveryDecision({ recoveryRetryCount: 1 });
|
||||
const d3 = computeRecoveryDecision({ recoveryRetryCount: 2 });
|
||||
|
||||
// Base: 60s, then 120s, then 240s (capped at 300s)
|
||||
expect(d1.delayMs).toBe(BASE_DELAY_MS); // 60s × 2^0 = 60s
|
||||
expect(d2.delayMs).toBe(BASE_DELAY_MS * BACKOFF_MULTIPLIER); // 60s × 2^1 = 120s
|
||||
expect(d3.delayMs).toBe(BASE_DELAY_MS * BACKOFF_MULTIPLIER ** 2); // 60s × 2^2 = 240s
|
||||
});
|
||||
|
||||
it("caps delay at MAX_DELAY_MS", () => {
|
||||
vi.spyOn(Math, "random").mockReturnValue(0.5);
|
||||
|
||||
// With high retry count, delay should be capped
|
||||
const decision = computeRecoveryDecision({ recoveryRetryCount: 2 });
|
||||
expect(decision.delayMs).toBeLessThanOrEqual(MAX_DELAY_MS * 1.1); // Allow for jitter
|
||||
});
|
||||
|
||||
it("applies jitter (±10%) to delays", () => {
|
||||
// Zero jitter
|
||||
vi.spyOn(Math, "random").mockReturnValue(0.5);
|
||||
const noJitter = computeRecoveryDecision({});
|
||||
|
||||
// Max positive jitter
|
||||
vi.spyOn(Math, "random").mockReturnValue(1.0);
|
||||
const maxJitter = computeRecoveryDecision({});
|
||||
|
||||
// Max negative jitter
|
||||
vi.spyOn(Math, "random").mockReturnValue(0.0);
|
||||
const minJitter = computeRecoveryDecision({});
|
||||
|
||||
// All should be within ±10% of base delay
|
||||
const base = BASE_DELAY_MS;
|
||||
expect(noJitter.delayMs).toBe(base);
|
||||
expect(maxJitter.delayMs).toBeGreaterThan(base);
|
||||
expect(maxJitter.delayMs).toBeLessThanOrEqual(base * 1.1);
|
||||
expect(minJitter.delayMs).toBeLessThan(base);
|
||||
expect(minJitter.delayMs).toBeGreaterThanOrEqual(base * 0.9);
|
||||
});
|
||||
|
||||
it("sets nextRecoveryAt to a future ISO timestamp", () => {
|
||||
const before = Date.now();
|
||||
const decision = computeRecoveryDecision({});
|
||||
const after = Date.now();
|
||||
|
||||
const recoveryTime = new Date(decision.nextState.nextRecoveryAt!).getTime();
|
||||
expect(recoveryTime).toBeGreaterThanOrEqual(before + decision.delayMs - 1);
|
||||
expect(recoveryTime).toBeLessThanOrEqual(after + decision.delayMs + 1);
|
||||
});
|
||||
|
||||
it("treats undefined recoveryRetryCount as 0", () => {
|
||||
const decision = computeRecoveryDecision({ recoveryRetryCount: undefined });
|
||||
expect(decision.shouldRetry).toBe(true);
|
||||
expect(decision.nextState.recoveryRetryCount).toBe(1);
|
||||
});
|
||||
|
||||
it("clears recovery metadata when exhausted", () => {
|
||||
const decision = computeRecoveryDecision({
|
||||
recoveryRetryCount: MAX_RECOVERY_RETRIES,
|
||||
nextRecoveryAt: new Date().toISOString(),
|
||||
});
|
||||
expect(decision.nextState.recoveryRetryCount).toBeUndefined();
|
||||
expect(decision.nextState.nextRecoveryAt).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("formatDelay", () => {
|
||||
it("formats seconds under 60 as Ns", () => {
|
||||
expect(formatDelay(5000)).toBe("5s");
|
||||
expect(formatDelay(30000)).toBe("30s");
|
||||
expect(formatDelay(59000)).toBe("59s");
|
||||
});
|
||||
|
||||
it("formats exact minutes as Nm", () => {
|
||||
expect(formatDelay(60000)).toBe("1m");
|
||||
expect(formatDelay(120000)).toBe("2m");
|
||||
expect(formatDelay(300000)).toBe("5m");
|
||||
});
|
||||
|
||||
it("formats non-exact minutes as seconds", () => {
|
||||
expect(formatDelay(90000)).toBe("90s");
|
||||
expect(formatDelay(150000)).toBe("150s");
|
||||
});
|
||||
|
||||
it("handles zero", () => {
|
||||
expect(formatDelay(0)).toBe("0s");
|
||||
});
|
||||
});
|
||||
|
||||
describe("constants", () => {
|
||||
it("MAX_RECOVERY_RETRIES is 3", () => {
|
||||
expect(MAX_RECOVERY_RETRIES).toBe(3);
|
||||
});
|
||||
|
||||
it("BASE_DELAY_MS is 60 seconds", () => {
|
||||
expect(BASE_DELAY_MS).toBe(60_000);
|
||||
});
|
||||
|
||||
it("MAX_DELAY_MS is 300 seconds (5 minutes)", () => {
|
||||
expect(MAX_DELAY_MS).toBe(300_000);
|
||||
});
|
||||
});
|
||||
125
packages/engine/src/recovery-policy.ts
Normal file
125
packages/engine/src/recovery-policy.ts
Normal file
@@ -0,0 +1,125 @@
|
||||
/**
|
||||
* Recovery Policy — bounded exponential-backoff retry for recoverable executor/triage failures.
|
||||
*
|
||||
* This module provides a **pure decision function** that computes whether a transient
|
||||
* failure should be retried, and if so, what the updated recovery state should be.
|
||||
*
|
||||
* **Design boundary:**
|
||||
* - `recovery-policy.ts` handles **inter-poll** recoverable retries — tasks moved back
|
||||
* to todo/triage with backoff, gated by `nextRecoveryAt` in the scheduler/triage poller.
|
||||
* - `withRateLimitRetry()` in `rate-limit-retry.ts` handles **intra-session** rate-limit
|
||||
* retries — immediate retry within the same agent session with exponential backoff.
|
||||
* - `transient-error-detector.ts` provides the low-level error classifier (`isTransientError`,
|
||||
* `classifyError`). This module consumes those classifiers but does not replace them.
|
||||
*
|
||||
* **Retry semantics:**
|
||||
* - Up to `MAX_RECOVERY_RETRIES` attempts with exponential backoff.
|
||||
* - Base delay: 60 seconds, multiplied by 2^attempt, capped at 300 seconds.
|
||||
* - ±10% jitter to avoid thundering-herd effects.
|
||||
* - Recovery metadata (`recoveryRetryCount`, `nextRecoveryAt`) is persisted on the task
|
||||
* so retries survive engine restarts.
|
||||
* - Exhausted retry budgets escalate to a real failure (task marked failed or error set).
|
||||
*
|
||||
* **Not retried via this policy:**
|
||||
* - Usage-limit errors (handled by `UsageLimitPauser` with global pause)
|
||||
* - User pauses (handled by pause flow)
|
||||
* - Stuck-task-detector kills (handled by stuck flow)
|
||||
* - Dependency-abort cleanups (handled by dep-abort flow)
|
||||
* - Merge-conflict retries (handled by `mergeRetries` separately)
|
||||
*/
|
||||
|
||||
// ── Constants ────────────────────────────────────────────────────────
|
||||
|
||||
/** Maximum number of recovery retry attempts before escalating to failure. */
|
||||
export const MAX_RECOVERY_RETRIES = 3;
|
||||
|
||||
/** Base delay in milliseconds for the first retry (60 seconds). */
|
||||
export const BASE_DELAY_MS = 60_000;
|
||||
|
||||
/** Maximum delay cap in milliseconds (300 seconds = 5 minutes). */
|
||||
export const MAX_DELAY_MS = 300_000;
|
||||
|
||||
/** Backoff multiplier (2x exponential). */
|
||||
export const BACKOFF_MULTIPLIER = 2;
|
||||
|
||||
// ── Types ────────────────────────────────────────────────────────────
|
||||
|
||||
export interface RecoveryState {
|
||||
recoveryRetryCount?: number;
|
||||
nextRecoveryAt?: string;
|
||||
}
|
||||
|
||||
export interface RecoveryDecision {
|
||||
/** Whether the task should be retried (moved back to todo/triage). */
|
||||
shouldRetry: boolean;
|
||||
/** Whether the retry budget is exhausted (terminal failure). */
|
||||
exhausted: boolean;
|
||||
/** Updated recovery state to persist on the task. */
|
||||
nextState: RecoveryState;
|
||||
/** Computed delay in milliseconds (for logging). Zero when exhausted. */
|
||||
delayMs: number;
|
||||
}
|
||||
|
||||
// ── Decision function ────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Compute whether a recoverable failure should be retried and what the
|
||||
* updated recovery state should be.
|
||||
*
|
||||
* This is a **pure function** — it does not call TaskStore or perform I/O.
|
||||
* The caller is responsible for persisting `nextState` via `store.updateTask()`.
|
||||
*
|
||||
* @param currentState - Current recovery metadata from the task
|
||||
* @returns A decision describing whether to retry or escalate
|
||||
*/
|
||||
export function computeRecoveryDecision(
|
||||
currentState: RecoveryState,
|
||||
): RecoveryDecision {
|
||||
const currentCount = currentState.recoveryRetryCount ?? 0;
|
||||
const nextCount = currentCount + 1;
|
||||
|
||||
if (nextCount > MAX_RECOVERY_RETRIES) {
|
||||
// Budget exhausted — escalate to real failure
|
||||
return {
|
||||
shouldRetry: false,
|
||||
exhausted: true,
|
||||
nextState: { recoveryRetryCount: undefined, nextRecoveryAt: undefined },
|
||||
delayMs: 0,
|
||||
};
|
||||
}
|
||||
|
||||
// Exponential backoff: base × 2^(attempt-1), capped at max
|
||||
const rawDelay = Math.min(
|
||||
BASE_DELAY_MS * BACKOFF_MULTIPLIER ** (nextCount - 1),
|
||||
MAX_DELAY_MS,
|
||||
);
|
||||
|
||||
// ±10% jitter to avoid thundering herd
|
||||
const jitter = rawDelay * 0.1 * (2 * Math.random() - 1);
|
||||
const delayMs = Math.max(0, Math.round(rawDelay + jitter));
|
||||
|
||||
const nextRecoveryAt = new Date(Date.now() + delayMs).toISOString();
|
||||
|
||||
return {
|
||||
shouldRetry: true,
|
||||
exhausted: false,
|
||||
nextState: {
|
||||
recoveryRetryCount: nextCount,
|
||||
nextRecoveryAt,
|
||||
},
|
||||
delayMs,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Format a retry delay for human-readable logging.
|
||||
*
|
||||
* @param delayMs - Delay in milliseconds
|
||||
* @returns Human-readable string like "60s" or "120s"
|
||||
*/
|
||||
export function formatDelay(delayMs: number): string {
|
||||
const seconds = Math.round(delayMs / 1000);
|
||||
if (seconds < 60) return `${seconds}s`;
|
||||
const minutes = Math.round(seconds / 60);
|
||||
return seconds % 60 === 0 ? `${minutes}m` : `${seconds}s`;
|
||||
}
|
||||
@@ -924,4 +924,86 @@ describe("Scheduler", () => {
|
||||
expect(mockMissionStore.activateSlice).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("recovery due-time gating (nextRecoveryAt)", () => {
|
||||
it("skips todo tasks whose nextRecoveryAt is in the future", async () => {
|
||||
const future = new Date(Date.now() + 60_000).toISOString();
|
||||
const task = createMockTask({
|
||||
id: "FN-010",
|
||||
column: "todo",
|
||||
nextRecoveryAt: future,
|
||||
recoveryRetryCount: 1,
|
||||
});
|
||||
|
||||
const store = createMockStore({
|
||||
listTasks: vi.fn().mockResolvedValue([task]),
|
||||
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4 }),
|
||||
});
|
||||
|
||||
const onSchedule = vi.fn();
|
||||
const scheduler = new Scheduler(store, { onSchedule });
|
||||
scheduler.start();
|
||||
await scheduler.schedule();
|
||||
scheduler.stop();
|
||||
|
||||
// Should NOT have been started
|
||||
expect(onSchedule).not.toHaveBeenCalled();
|
||||
expect(store.moveTask).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("picks up todo tasks whose nextRecoveryAt has elapsed", async () => {
|
||||
const past = new Date(Date.now() - 1000).toISOString();
|
||||
const task = createMockTask({
|
||||
id: "FN-011",
|
||||
column: "todo",
|
||||
nextRecoveryAt: past,
|
||||
recoveryRetryCount: 1,
|
||||
});
|
||||
|
||||
// Mock filesystem validation: task dir exists, PROMPT.md exists and non-empty
|
||||
(existsSync as any).mockReturnValue(true);
|
||||
(readFile as any).mockResolvedValue("# Task\n\nSome content\n## File Scope\n- foo.ts\n");
|
||||
|
||||
const store = createMockStore({
|
||||
listTasks: vi.fn().mockResolvedValue([task]),
|
||||
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4 }),
|
||||
parseFileScopeFromPrompt: vi.fn().mockResolvedValue([]),
|
||||
getTasksDir: vi.fn().mockReturnValue("/test/project/.fusion/tasks"),
|
||||
});
|
||||
|
||||
const onSchedule = vi.fn();
|
||||
const scheduler = new Scheduler(store, { onSchedule });
|
||||
// Call schedule() directly without start() to avoid scheduling guard race
|
||||
(scheduler as any).running = true;
|
||||
await scheduler.schedule();
|
||||
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-011", "in-progress");
|
||||
});
|
||||
|
||||
it("picks up todo tasks without nextRecoveryAt normally", async () => {
|
||||
const task = createMockTask({
|
||||
id: "FN-012",
|
||||
column: "todo",
|
||||
// No nextRecoveryAt — should be picked up normally
|
||||
});
|
||||
|
||||
(existsSync as any).mockReturnValue(true);
|
||||
(readFile as any).mockResolvedValue("# Task\n\nSome content\n## File Scope\n- foo.ts\n");
|
||||
|
||||
const store = createMockStore({
|
||||
listTasks: vi.fn().mockResolvedValue([task]),
|
||||
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4 }),
|
||||
parseFileScopeFromPrompt: vi.fn().mockResolvedValue([]),
|
||||
getTasksDir: vi.fn().mockReturnValue("/test/project/.fusion/tasks"),
|
||||
});
|
||||
|
||||
const onSchedule = vi.fn();
|
||||
const scheduler = new Scheduler(store, { onSchedule });
|
||||
// Call schedule() directly without start() to avoid scheduling guard race
|
||||
(scheduler as any).running = true;
|
||||
await scheduler.schedule();
|
||||
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-012", "in-progress");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -410,7 +410,13 @@ export class Scheduler {
|
||||
);
|
||||
if (available <= 0) return;
|
||||
|
||||
const todo = tasks.filter((t) => t.column === "todo" && !t.paused);
|
||||
const now = Date.now();
|
||||
const todo = tasks.filter((t) => {
|
||||
if (t.column !== "todo" || t.paused) return false;
|
||||
// Skip tasks with a recovery backoff that hasn't elapsed yet
|
||||
if (t.nextRecoveryAt && new Date(t.nextRecoveryAt).getTime() > now) return false;
|
||||
return true;
|
||||
});
|
||||
if (todo.length === 0) return;
|
||||
|
||||
/**
|
||||
|
||||
@@ -10,14 +10,21 @@ import { fileURLToPath } from "node:url";
|
||||
import { dirname, join } from "node:path";
|
||||
import { mkdir, writeFile, rm } from "node:fs/promises";
|
||||
|
||||
const { mockReviewStep } = vi.hoisted(() => ({
|
||||
const { mockReviewStep, mockCreateKbAgent } = vi.hoisted(() => ({
|
||||
mockReviewStep: vi.fn(),
|
||||
mockCreateKbAgent: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("./reviewer.js", () => ({
|
||||
reviewStep: mockReviewStep,
|
||||
}));
|
||||
|
||||
vi.mock("./pi.js", () => ({
|
||||
createKbAgent: mockCreateKbAgent,
|
||||
describeModel: vi.fn().mockReturnValue("mock-model"),
|
||||
promptWithFallback: vi.fn().mockReturnValue("mock-prompt"),
|
||||
}));
|
||||
|
||||
const __dirname = dirname(fileURLToPath(import.meta.url));
|
||||
|
||||
function createMockStore(overrides: Partial<TaskStore> = {}): TaskStore {
|
||||
@@ -26,7 +33,7 @@ function createMockStore(overrides: Partial<TaskStore> = {}): TaskStore {
|
||||
listTasks: vi.fn().mockResolvedValue([]),
|
||||
createTask: vi.fn(),
|
||||
moveTask: vi.fn(),
|
||||
updateTask: vi.fn(),
|
||||
updateTask: vi.fn().mockResolvedValue(undefined),
|
||||
deleteTask: vi.fn(),
|
||||
mergeTask: vi.fn(),
|
||||
getSettings: vi.fn().mockResolvedValue({
|
||||
@@ -718,4 +725,147 @@ describe("taskCreate tool model inheritance", () => {
|
||||
validatorModelId: undefined,
|
||||
}));
|
||||
});
|
||||
|
||||
describe("bounded recovery retries for triage", () => {
|
||||
it("sets recoveryRetryCount and nextRecoveryAt on first transient error via specifyTask", async () => {
|
||||
const task = {
|
||||
id: "FN-200",
|
||||
description: "Test triage task",
|
||||
column: "triage",
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
} as unknown as Task;
|
||||
|
||||
const store = createMockStore({
|
||||
getTask: vi.fn().mockResolvedValue({ ...task, attachments: [] }),
|
||||
});
|
||||
|
||||
const processor = new TriageProcessor(store, "/test/root", {
|
||||
pollIntervalMs: 100_000,
|
||||
});
|
||||
|
||||
// Mock createKbAgent to throw a transient error
|
||||
mockCreateKbAgent.mockRejectedValue(new Error("upstream connect error"));
|
||||
|
||||
await processor.specifyTask(task);
|
||||
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-200", expect.objectContaining({
|
||||
recoveryRetryCount: 1,
|
||||
nextRecoveryAt: expect.any(String),
|
||||
}));
|
||||
});
|
||||
|
||||
it("escalates to error state when triage retries are exhausted via specifyTask", async () => {
|
||||
const task = {
|
||||
id: "FN-201",
|
||||
description: "Test triage task",
|
||||
column: "triage",
|
||||
recoveryRetryCount: 3, // Already at max
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
} as unknown as Task;
|
||||
|
||||
const onSpecifyError = vi.fn();
|
||||
const store = createMockStore({
|
||||
getTask: vi.fn().mockResolvedValue({ ...task, attachments: [] }),
|
||||
});
|
||||
|
||||
const processor = new TriageProcessor(store, "/test/root", {
|
||||
pollIntervalMs: 100_000,
|
||||
onSpecifyError,
|
||||
});
|
||||
|
||||
mockCreateKbAgent.mockRejectedValue(new Error("connection reset"));
|
||||
|
||||
await processor.specifyTask(task);
|
||||
|
||||
// Should set error and clear recovery metadata
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-201", expect.objectContaining({
|
||||
error: expect.stringContaining("Specification failed after 3 transient errors"),
|
||||
recoveryRetryCount: null,
|
||||
nextRecoveryAt: null,
|
||||
}));
|
||||
expect(onSpecifyError).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("recovery due-time gating (nextRecoveryAt)", () => {
|
||||
it("skips triage tasks whose nextRecoveryAt is in the future", async () => {
|
||||
const future = new Date(Date.now() + 60_000).toISOString();
|
||||
const task = {
|
||||
id: "FN-100",
|
||||
description: "Test triage task",
|
||||
column: "triage",
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
nextRecoveryAt: future,
|
||||
recoveryRetryCount: 1,
|
||||
} as unknown as Task;
|
||||
|
||||
const store = createMockStore({
|
||||
listTasks: vi.fn().mockResolvedValue([task]),
|
||||
});
|
||||
|
||||
const processor = new TriageProcessor(store, "/test/root", {
|
||||
pollIntervalMs: 100_000, // long interval so only manual poll runs
|
||||
});
|
||||
|
||||
// Spy on specifyTask to ensure it's NOT called for gated tasks
|
||||
const specifySpy = vi.spyOn(processor, "specifyTask");
|
||||
|
||||
processor.start();
|
||||
// Wait a tick for the initial poll
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
processor.stop();
|
||||
|
||||
expect(specifySpy).not.toHaveBeenCalled();
|
||||
specifySpy.mockRestore();
|
||||
});
|
||||
|
||||
it("processes triage tasks whose nextRecoveryAt has elapsed", async () => {
|
||||
const past = new Date(Date.now() - 1000).toISOString();
|
||||
const task = {
|
||||
id: "FN-101",
|
||||
description: "Test triage task past",
|
||||
column: "triage",
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
nextRecoveryAt: past,
|
||||
recoveryRetryCount: 1,
|
||||
} as unknown as Task;
|
||||
|
||||
const store = createMockStore({
|
||||
listTasks: vi.fn().mockResolvedValue([task]),
|
||||
});
|
||||
|
||||
const processor = new TriageProcessor(store, "/test/root", {
|
||||
pollIntervalMs: 100_000,
|
||||
});
|
||||
|
||||
const specifySpy = vi.spyOn(processor, "specifyTask").mockResolvedValue(undefined);
|
||||
|
||||
processor.start();
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
processor.stop();
|
||||
|
||||
expect(specifySpy).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-101" }));
|
||||
specifySpy.mockRestore();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -23,6 +23,7 @@ import {
|
||||
} from "./usage-limit-detector.js";
|
||||
import { isTransientError } from "./transient-error-detector.js";
|
||||
import { withRateLimitRetry } from "./rate-limit-retry.js";
|
||||
import { computeRecoveryDecision, formatDelay, MAX_RECOVERY_RETRIES } from "./recovery-policy.js";
|
||||
|
||||
export const TRIAGE_SYSTEM_PROMPT = `You are a task specification agent for "kb", an AI-orchestrated task board.
|
||||
|
||||
@@ -373,8 +374,11 @@ export class TriageProcessor {
|
||||
this.wasEnginePaused = false;
|
||||
|
||||
const tasks = await this.store.listTasks();
|
||||
const now = Date.now();
|
||||
const triageTasks = tasks.filter(
|
||||
(t) => t.column === "triage" && !this.processing.has(t.id) && !t.paused,
|
||||
(t) => t.column === "triage" && !this.processing.has(t.id) && !t.paused
|
||||
// Skip tasks with a recovery backoff that hasn't elapsed yet
|
||||
&& !(t.nextRecoveryAt && new Date(t.nextRecoveryAt).getTime() > now),
|
||||
);
|
||||
|
||||
for (const task of triageTasks) {
|
||||
@@ -685,12 +689,35 @@ export class TriageProcessor {
|
||||
err.message,
|
||||
);
|
||||
} else if (isTransientError(err.message)) {
|
||||
// Transient network/infrastructure error — don't mark as failed, allow retry
|
||||
triageLog.warn(`⚡ ${task.id} transient error during triage — will retry: ${err.message}`);
|
||||
await this.store.logEntry(task.id, `Transient error during specification (will retry): ${err.message}`).catch(() => {});
|
||||
// Restore status so triage picks it up again on next pass
|
||||
const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : undefined;
|
||||
await this.store.updateTask(task.id, { status: restoreStatus }).catch(() => {});
|
||||
// Transient network/infrastructure error — use bounded recovery policy
|
||||
const decision = computeRecoveryDecision({
|
||||
recoveryRetryCount: task.recoveryRetryCount,
|
||||
nextRecoveryAt: task.nextRecoveryAt,
|
||||
});
|
||||
|
||||
if (decision.shouldRetry) {
|
||||
const attempt = decision.nextState.recoveryRetryCount;
|
||||
const delay = formatDelay(decision.delayMs);
|
||||
triageLog.warn(`⚡ ${task.id} transient error during triage — retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}: ${err.message}`);
|
||||
await this.store.logEntry(task.id, `Transient error during specification (retry ${attempt}/${MAX_RECOVERY_RETRIES} in ${delay}): ${err.message}`).catch(() => {});
|
||||
const restoreStatus = task.status === "needs-respecify" ? "needs-respecify" : undefined;
|
||||
await this.store.updateTask(task.id, {
|
||||
status: restoreStatus,
|
||||
recoveryRetryCount: decision.nextState.recoveryRetryCount,
|
||||
nextRecoveryAt: decision.nextState.nextRecoveryAt,
|
||||
}).catch(() => {});
|
||||
return;
|
||||
}
|
||||
|
||||
// Recovery budget exhausted — freeze in triage with error for manual intervention
|
||||
triageLog.error(`✗ ${task.id} transient error retries exhausted (${MAX_RECOVERY_RETRIES} attempts): ${err.message}`);
|
||||
await this.store.logEntry(task.id, `Specification failed after ${MAX_RECOVERY_RETRIES} transient errors: ${err.message}`).catch(() => {});
|
||||
await this.store.updateTask(task.id, {
|
||||
error: `Specification failed after ${MAX_RECOVERY_RETRIES} transient errors: ${err.message}`,
|
||||
recoveryRetryCount: null,
|
||||
nextRecoveryAt: null,
|
||||
}).catch(() => {});
|
||||
this.options.onSpecifyError?.(task, err);
|
||||
return;
|
||||
}
|
||||
// For re-specification, restore needs-respecify status so it can be retried
|
||||
|
||||
Reference in New Issue
Block a user