From 8c246b036f1c7d89e0c9ad6f38b04e1513cfa219 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Fri, 3 Apr 2026 07:58:45 -0700 Subject: [PATCH] feat(FN-775): add recoverable-retry with bounded exponential backoff MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- README.md | 9 + packages/core/src/db-migrate.ts | 7 +- packages/core/src/db.test.ts | 14 +- packages/core/src/db.ts | 13 +- packages/core/src/store.test.ts | 108 +++++++++++ packages/core/src/store.ts | 34 +++- packages/core/src/types.ts | 11 ++ packages/engine/src/executor.test.ts | 191 +++++++++++++++++++- packages/engine/src/executor.ts | 34 +++- packages/engine/src/recovery-policy.test.ts | 160 ++++++++++++++++ packages/engine/src/recovery-policy.ts | 125 +++++++++++++ packages/engine/src/scheduler.test.ts | 82 +++++++++ packages/engine/src/scheduler.ts | 8 +- packages/engine/src/triage.test.ts | 154 +++++++++++++++- packages/engine/src/triage.ts | 41 ++++- 15 files changed, 961 insertions(+), 30 deletions(-) create mode 100644 packages/engine/src/recovery-policy.test.ts create mode 100644 packages/engine/src/recovery-policy.ts diff --git a/README.md b/README.md index 718818ad9..7954a6291 100644 --- a/README.md +++ b/README.md @@ -245,6 +245,15 @@ Each pi agent session gets: - In-memory sessions (no persistence needed) - The user's existing pi auth (API keys from `~/.pi/agent/auth.json`) +### Error Recovery + +The engine automatically recovers from transient infrastructure failures (network blips, proxy errors, connection resets) using bounded exponential backoff: + +- **Recoverable failures** — When a transient error is detected during task execution or triage specification, the task is requeued with an increasing backoff delay (60s → 120s → 240s, capped at 5 minutes). Up to 3 retry attempts are made before the task is marked as permanently failed. +- **Recovery metadata** — Each task stores `recoveryRetryCount` and `nextRecoveryAt` (ISO-8601 timestamp) in SQLite. The scheduler and triage processor skip tasks whose `nextRecoveryAt` is still in the future, ensuring backoff is respected across engine restarts. +- **Budget exhaustion** — After 3 failed recovery attempts, executor tasks are marked as `failed` and triage tasks receive an error message for manual intervention. Recovery metadata is cleared. +- **Separate from other retry mechanisms** — Recovery retries are distinct from `mergeRetries` (merge-conflict resolution), `withRateLimitRetry` (in-session rate-limit backoff), and usage-limit global pauses. User pauses, stuck-task-detector kills, and dependency-abort cleanups do not consume the recovery budget. + ## Model System Fusion provides flexible AI model configuration with support for model presets, per-task overrides, and a hierarchical settings system. diff --git a/packages/core/src/db-migrate.ts b/packages/core/src/db-migrate.ts index e61645080..34e641b1f 100644 --- a/packages/core/src/db-migrate.ts +++ b/packages/core/src/db-migrate.ts @@ -166,13 +166,14 @@ async function migrateTasks(kbDir: string, db: Database): Promise { id, title, description, "column", status, size, reviewLevel, currentStep, worktree, blockedBy, paused, baseBranch, baseCommitSha, modelPresetId, modelProvider, modelId, validatorModelProvider, validatorModelId, - mergeRetries, error, summary, thinkingLevel, createdAt, updatedAt, + mergeRetries, recoveryRetryCount, nextRecoveryAt, + error, summary, thinkingLevel, createdAt, updatedAt, columnMovedAt, dependencies, steps, log, attachments, steeringComments, comments, workflowStepResults, prInfo, issueInfo, mergeDetails, breakIntoSubtasks, enabledWorkflowSteps, modifiedFiles, sliceId ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, - ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? ) `); @@ -211,6 +212,8 @@ async function migrateTasks(kbDir: string, db: Database): Promise { task.validatorModelProvider ?? null, task.validatorModelId ?? null, task.mergeRetries ?? null, + task.recoveryRetryCount ?? null, + task.nextRecoveryAt ?? null, task.error ?? null, task.summary ?? null, task.thinkingLevel ?? null, diff --git a/packages/core/src/db.test.ts b/packages/core/src/db.test.ts index 99db53beb..030e6db61 100644 --- a/packages/core/src/db.test.ts +++ b/packages/core/src/db.test.ts @@ -86,7 +86,7 @@ describe("Database", () => { }); it("seeds schema version", () => { - expect(db.getSchemaVersion()).toBe(6); + expect(db.getSchemaVersion()).toBe(7); }); it("seeds lastModified", () => { @@ -109,7 +109,7 @@ describe("Database", () => { it("is idempotent - calling init() twice does not fail", () => { expect(() => db.init()).not.toThrow(); - expect(db.getSchemaVersion()).toBe(6); + expect(db.getSchemaVersion()).toBe(7); }); it("does not overwrite existing config on re-init", () => { @@ -704,7 +704,7 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 5 (includes v1→v2, v2→v3, v3→v4, and v4→v5 migrations) - expect(db.getSchemaVersion()).toBe(6); + expect(db.getSchemaVersion()).toBe(7); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -729,11 +729,11 @@ describe("schema migrations", () => { const db = new Database(kbDir); db.init(); - expect(db.getSchemaVersion()).toBe(6); + expect(db.getSchemaVersion()).toBe(7); // Re-init should not fail db.init(); - expect(db.getSchemaVersion()).toBe(6); + expect(db.getSchemaVersion()).toBe(7); db.close(); }); @@ -828,7 +828,7 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 5 - expect(db.getSchemaVersion()).toBe(6); + expect(db.getSchemaVersion()).toBe(7); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1038,7 +1038,7 @@ describe("createDatabase factory", () => { const db = createDatabase(kbDir); db.init(); - expect(db.getSchemaVersion()).toBe(6); + expect(db.getSchemaVersion()).toBe(7); expect(db.getLastModified()).toBeGreaterThan(0); db.close(); diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 7805364bd..2a38ad199 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -59,7 +59,7 @@ export function fromJson(json: string | null | undefined): T | undefined { // ── Schema Definition ──────────────────────────────────────────────── -const SCHEMA_VERSION = 6; +const SCHEMA_VERSION = 7; function normalizeTaskComments( steeringComments: SteeringComment[] | undefined, @@ -142,6 +142,8 @@ CREATE TABLE IF NOT EXISTS tasks ( validatorModelProvider TEXT, validatorModelId TEXT, mergeRetries INTEGER, + recoveryRetryCount INTEGER, + nextRecoveryAt TEXT, error TEXT, summary TEXT, thinkingLevel TEXT, @@ -409,8 +411,15 @@ export class Database { }); } + if (version < 7) { + this.applyMigration(7, () => { + this.addColumnIfMissing("tasks", "recoveryRetryCount", "INTEGER"); + this.addColumnIfMissing("tasks", "nextRecoveryAt", "TEXT"); + }); + } + // Future migrations go here: - // if (version < 7) { this.applyMigration(7, () => { ... }); } + // if (version < 8) { this.applyMigration(8, () => { ... }); } } /** diff --git a/packages/core/src/store.test.ts b/packages/core/src/store.test.ts index a8fe83a1a..07d1ba3ef 100644 --- a/packages/core/src/store.test.ts +++ b/packages/core/src/store.test.ts @@ -4630,4 +4630,112 @@ Task with acceptance criteria expect(movedEvents[0].to).toBe("todo"); }); }); + + describe("recovery metadata (recoveryRetryCount / nextRecoveryAt)", () => { + async function createTestTask(overrides: Partial = {}) { + return store.createTask({ description: "recovery test task", ...overrides }); + } + + it("new tasks have no recovery metadata (defaults to undefined)", async () => { + const task = await createTestTask(); + expect(task.recoveryRetryCount).toBeUndefined(); + expect(task.nextRecoveryAt).toBeUndefined(); + }); + + it("updateTask can set and clear recoveryRetryCount and nextRecoveryAt", async () => { + const task = await createTestTask(); + const futureTime = new Date(Date.now() + 60_000).toISOString(); + + // Set + const updated = await store.updateTask(task.id, { + recoveryRetryCount: 2, + nextRecoveryAt: futureTime, + }); + expect(updated.recoveryRetryCount).toBe(2); + expect(updated.nextRecoveryAt).toBe(futureTime); + + // Re-read to verify persistence + const reread = await store.getTask(task.id); + expect(reread.recoveryRetryCount).toBe(2); + expect(reread.nextRecoveryAt).toBe(futureTime); + + // Clear + const cleared = await store.updateTask(task.id, { + recoveryRetryCount: null, + nextRecoveryAt: null, + }); + expect(cleared.recoveryRetryCount).toBeUndefined(); + expect(cleared.nextRecoveryAt).toBeUndefined(); + }); + + it("moveTask to in-review clears recovery metadata", async () => { + const task = await createTestTask(); + await store.moveTask(task.id, "todo"); + await store.updateTask(task.id, { + recoveryRetryCount: 3, + nextRecoveryAt: new Date().toISOString(), + }); + await store.moveTask(task.id, "in-progress"); + const moved = await store.moveTask(task.id, "in-review"); + expect(moved.recoveryRetryCount).toBeUndefined(); + expect(moved.nextRecoveryAt).toBeUndefined(); + }); + + it("moveTask to done clears recovery metadata", async () => { + const task = await createTestTask(); + await store.moveTask(task.id, "todo"); + await store.updateTask(task.id, { + recoveryRetryCount: 1, + nextRecoveryAt: new Date().toISOString(), + }); + await store.moveTask(task.id, "in-progress"); + await store.moveTask(task.id, "in-review"); + const done = await store.moveTask(task.id, "done"); + expect(done.recoveryRetryCount).toBeUndefined(); + expect(done.nextRecoveryAt).toBeUndefined(); + }); + + it("moveTask from in-progress to todo preserves recovery metadata", async () => { + const task = await createTestTask(); + await store.moveTask(task.id, "todo"); + await store.moveTask(task.id, "in-progress"); + + const futureTime = new Date(Date.now() + 60_000).toISOString(); + await store.updateTask(task.id, { + recoveryRetryCount: 2, + nextRecoveryAt: futureTime, + }); + + const moved = await store.moveTask(task.id, "todo"); + expect(moved.recoveryRetryCount).toBe(2); + expect(moved.nextRecoveryAt).toBe(futureTime); + }); + + it("recovery metadata persists across store re-initialization", async () => { + const task = await createTestTask(); + const futureTime = new Date(Date.now() + 60_000).toISOString(); + await store.updateTask(task.id, { + recoveryRetryCount: 5, + nextRecoveryAt: futureTime, + }); + + // Re-create store to simulate restart + store.stopWatching(); + const store2 = new TaskStore(rootDir, globalDir); + await store2.init(); + + const reloaded = await store2.getTask(task.id); + expect(reloaded.recoveryRetryCount).toBe(5); + expect(reloaded.nextRecoveryAt).toBe(futureTime); + store2.stopWatching(); + }); + + it("schema migration: existing rows default to NULL (undefined) for recovery fields", async () => { + // Tasks created before the migration should have undefined recovery fields + const task = await createTestTask(); + const detail = await store.getTask(task.id); + expect(detail.recoveryRetryCount).toBeUndefined(); + expect(detail.nextRecoveryAt).toBeUndefined(); + }); + }); }); diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 5618e7672..6bddee11c 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -178,6 +178,8 @@ export class TaskStore extends EventEmitter { validatorModelProvider: row.validatorModelProvider || undefined, validatorModelId: row.validatorModelId || undefined, mergeRetries: row.mergeRetries ?? undefined, + recoveryRetryCount: row.recoveryRetryCount ?? undefined, + nextRecoveryAt: row.nextRecoveryAt || undefined, error: row.error || undefined, summary: row.summary || undefined, thinkingLevel: row.thinkingLevel || undefined, @@ -225,14 +227,15 @@ export class TaskStore extends EventEmitter { INSERT OR REPLACE INTO tasks ( id, title, description, "column", status, size, reviewLevel, currentStep, worktree, blockedBy, paused, baseBranch, branch, baseCommitSha, modelPresetId, modelProvider, - modelId, validatorModelProvider, validatorModelId, mergeRetries, error, + modelId, validatorModelProvider, validatorModelId, mergeRetries, + recoveryRetryCount, nextRecoveryAt, error, summary, thinkingLevel, createdAt, updatedAt, columnMovedAt, dependencies, steps, log, attachments, steeringComments, comments, workflowStepResults, prInfo, issueInfo, mergeDetails, breakIntoSubtasks, enabledWorkflowSteps, modifiedFiles, missionId, sliceId ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, - ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? ) `).run( task.id, @@ -255,6 +258,8 @@ export class TaskStore extends EventEmitter { task.validatorModelProvider ?? null, task.validatorModelId ?? null, task.mergeRetries ?? null, + task.recoveryRetryCount ?? null, + task.nextRecoveryAt ?? null, task.error ?? null, task.summary ?? null, task.thinkingLevel ?? null, @@ -930,10 +935,15 @@ export class TaskStore extends EventEmitter { task.error = undefined; task.worktree = undefined; task.blockedBy = undefined; + task.recoveryRetryCount = undefined; + task.nextRecoveryAt = undefined; } // Clear transient fields when moving from in-progress to reset columns (todo/triage) - // This ensures failed tasks don't show failed status after being moved for retry + // This ensures failed tasks don't show failed status after being moved for retry. + // Note: recovery metadata (recoveryRetryCount, nextRecoveryAt) is intentionally + // preserved here — the recovery-policy module manages those fields. They are + // only cleared on terminal transitions (in-review, done, archived). if (fromColumn === "in-progress" && (toColumn === "todo" || toColumn === "triage")) { task.status = undefined; task.error = undefined; @@ -941,6 +951,12 @@ export class TaskStore extends EventEmitter { task.blockedBy = undefined; } + // Clear recovery metadata when task reaches in-review (successful completion) + if (toColumn === "in-review") { + task.recoveryRetryCount = undefined; + task.nextRecoveryAt = undefined; + } + // Clear workflow step results when moving from in-review back to todo or in-progress // This ensures fresh workflow step runs on retry if (fromColumn === "in-review" && (toColumn === "todo" || toColumn === "in-progress")) { @@ -959,7 +975,7 @@ export class TaskStore extends EventEmitter { async updateTask( id: string, - updates: { title?: string; description?: string; prompt?: string; worktree?: string; status?: string | null; dependencies?: string[]; blockedBy?: string | null; paused?: boolean; baseBranch?: string; branch?: string; baseCommitSha?: string; size?: "S" | "M" | "L"; reviewLevel?: number; mergeRetries?: number; modelProvider?: string | null; modelId?: string | null; validatorModelProvider?: string | null; validatorModelId?: string | null; error?: string | null; summary?: string | null; workflowStepResults?: import("./types.js").WorkflowStepResult[] | null; modifiedFiles?: string[] | null; missionId?: string | null; sliceId?: string | null }, + updates: { title?: string; description?: string; prompt?: string; worktree?: string; status?: string | null; dependencies?: string[]; blockedBy?: string | null; paused?: boolean; baseBranch?: string; branch?: string; baseCommitSha?: string; size?: "S" | "M" | "L"; reviewLevel?: number; mergeRetries?: number; recoveryRetryCount?: number | null; nextRecoveryAt?: string | null; modelProvider?: string | null; modelId?: string | null; validatorModelProvider?: string | null; validatorModelId?: string | null; error?: string | null; summary?: string | null; workflowStepResults?: import("./types.js").WorkflowStepResult[] | null; modifiedFiles?: string[] | null; missionId?: string | null; sliceId?: string | null }, ): Promise { return this.withTaskLock(id, async () => { // Validate that task doesn't depend on itself @@ -1014,6 +1030,16 @@ export class TaskStore extends EventEmitter { if (updates.size !== undefined) task.size = updates.size; if (updates.reviewLevel !== undefined) task.reviewLevel = updates.reviewLevel; if (updates.mergeRetries !== undefined) task.mergeRetries = updates.mergeRetries; + if (updates.recoveryRetryCount === null) { + task.recoveryRetryCount = undefined; + } else if (updates.recoveryRetryCount !== undefined) { + task.recoveryRetryCount = updates.recoveryRetryCount; + } + if (updates.nextRecoveryAt === null) { + task.nextRecoveryAt = undefined; + } else if (updates.nextRecoveryAt !== undefined) { + task.nextRecoveryAt = updates.nextRecoveryAt; + } if (updates.modelProvider === null) { task.modelProvider = undefined; } else if (updates.modelProvider !== undefined) { diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 8340edd53..a141b3b69 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -475,6 +475,15 @@ export interface Task { workflowStepResults?: WorkflowStepResult[]; /** Number of merge retry attempts made for this task (auto-merge conflict recovery) */ mergeRetries?: number; + /** Number of bounded recovery retry attempts for transient executor/triage failures. + * Distinct from `mergeRetries` (merge-conflict-specific). Incremented by the + * recovery-policy module on each recoverable failure; cleared when work restarts + * cleanly or reaches a terminal column (in-review, done, archived). */ + recoveryRetryCount?: number; + /** ISO-8601 timestamp indicating when the task becomes eligible for the next + * recovery retry. Scheduler and triage processor skip tasks whose + * `nextRecoveryAt` is still in the future. Cleared alongside `recoveryRetryCount`. */ + nextRecoveryAt?: string; /** Thinking level for AI agent sessions — controls reasoning effort (off/minimal/low/medium/high) */ thinkingLevel?: ThinkingLevel; /** Error message from the last failure, if the task failed during execution */ @@ -1008,6 +1017,8 @@ export interface ArchivedTaskEntry { /** Slice ID this task is linked to */ sliceId?: string; mergeRetries?: number; + recoveryRetryCount?: number; + nextRecoveryAt?: string; error?: string; } diff --git a/packages/engine/src/executor.test.ts b/packages/engine/src/executor.test.ts index cbee21d24..9c48cf593 100644 --- a/packages/engine/src/executor.test.ts +++ b/packages/engine/src/executor.test.ts @@ -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(); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 0010bf9ad..ef11a72d6 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -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); diff --git a/packages/engine/src/recovery-policy.test.ts b/packages/engine/src/recovery-policy.test.ts new file mode 100644 index 000000000..3df1ed17a --- /dev/null +++ b/packages/engine/src/recovery-policy.test.ts @@ -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); + }); +}); diff --git a/packages/engine/src/recovery-policy.ts b/packages/engine/src/recovery-policy.ts new file mode 100644 index 000000000..7e4b8791c --- /dev/null +++ b/packages/engine/src/recovery-policy.ts @@ -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`; +} diff --git a/packages/engine/src/scheduler.test.ts b/packages/engine/src/scheduler.test.ts index 5314a65ba..d32c5916c 100644 --- a/packages/engine/src/scheduler.test.ts +++ b/packages/engine/src/scheduler.test.ts @@ -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"); + }); + }); }); diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index bfae06ed3..0641c87a2 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -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; /** diff --git a/packages/engine/src/triage.test.ts b/packages/engine/src/triage.test.ts index 480b239cf..145c9f04f 100644 --- a/packages/engine/src/triage.test.ts +++ b/packages/engine/src/triage.test.ts @@ -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 { @@ -26,7 +33,7 @@ function createMockStore(overrides: Partial = {}): 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(); + }); + }); }); diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 9a09537fc..24b6b4aac 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -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