diff --git a/src/harness/pi-harness.ts b/src/harness/pi-harness.ts index 73ff2ba74..62263e8f8 100644 --- a/src/harness/pi-harness.ts +++ b/src/harness/pi-harness.ts @@ -863,9 +863,61 @@ function piAssistantError(session: AssistantTextSession): string | null { return formatPiAssistantError(lastAssistant.errorMessage); } +/** + * Provider error types that a retry can plausibly survive — the provider is + * alive but refusing (rate limit), degraded (overload / 5xx), or slow + * (timeout). Everything else (authentication, permission, invalid request…) + * fails identically on the next attempt, so those keep the non-retryable + * classification (#602). + */ +const RETRYABLE_PROVIDER_ERROR_TYPES = new Set([ + "rate_limit_error", + "overloaded_error", + "timeout_error", + "api_error", + "server_error", + "internal_server_error", + "service_unavailable", + "temporarily_unavailable", + "model_service_overloaded", +]); + +function piProviderErrorType(raw: string | undefined): string | null { + const message = raw?.trim(); + if (!message) return null; + const jsonAt = message.indexOf("{"); + if (jsonAt < 0) return null; + try { + const parsed = JSON.parse(message.slice(jsonAt)) as { error?: { type?: unknown } }; + const providerType = typeof parsed.error?.type === "string" ? parsed.error.type.trim() : ""; + return providerType || null; + } catch { + return null; + } +} + +/** + * Classify a provider stop-error for retry purposes: transient provider + * failures (rate limit, overload, 5xx, timeout) become plain errors so the + * runs worker retries them with backoff; permanent ones (authentication, + * permission, bad request) stay non-retryable — a second attempt would fail + * identically (#602). + */ +function piTurnErrorFor(raw: string | undefined): Error { + const message = formatPiAssistantError(raw); + const providerType = piProviderErrorType(raw); + if (providerType !== null && RETRYABLE_PROVIDER_ERROR_TYPES.has(providerType)) { + // Same formatted message as the non-retryable path — the class carries + // the retry decision, the copy stays what operators grep for. + return new Error(message); + } + return new NonRetryableTurnError(message); +} + export function piLastAssistantTextOrThrow(session: AssistantTextSession): string | undefined { - const err = piAssistantError(session); - if (err) throw new NonRetryableTurnError(err); + const lastAssistant = [...session.messages].reverse().find((m) => m.role === "assistant") as + { stopReason?: string; errorMessage?: string } | undefined; + if (lastAssistant?.stopReason === "error") throw piTurnErrorFor(lastAssistant.errorMessage); return session.getLastAssistantText(); } @@ -874,8 +926,9 @@ export function piTurnError(session: AssistantTextSession, thrown: unknown, mess messagesBefore === undefined ? session : ({ messages: session.messages.slice(messagesBefore) } as AssistantTextSession); - const detailed = piAssistantError(fresh); - if (detailed) return new NonRetryableTurnError(detailed); + const lastAssistant = [...fresh.messages].reverse().find((m) => m.role === "assistant") as + { stopReason?: string; errorMessage?: string } | undefined; + if (lastAssistant?.stopReason === "error") return piTurnErrorFor(lastAssistant.errorMessage); return thrown instanceof Error ? thrown : new Error(String(thrown)); } diff --git a/src/runs/memory-run-store.ts b/src/runs/memory-run-store.ts index 62e5d7e4c..d33b0dd6b 100644 --- a/src/runs/memory-run-store.ts +++ b/src/runs/memory-run-store.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { retryBackoffMs } from './run-store.ts'; import { EventEmitter } from "node:events"; import type { EnqueueInput, EnqueueResult, ReapEvent, Run, RunDeliveryState, RunStore } from "./run-store.ts"; import { isTerminal, leaseLapsed } from "./run-store.ts"; @@ -9,8 +10,11 @@ export interface MemoryRuntime { ledger: ToolLedger; } -export function createMemoryRunStore(opts?: { maxClaims?: number }): MemoryRuntime { +export function createMemoryRunStore( + opts?: { maxClaims?: number; retryBackoffMs?: (nextErrorAttempt: number) => number }, +): MemoryRuntime { const maxClaims = opts?.maxClaims ?? Number.POSITIVE_INFINITY; + const backoff = opts?.retryBackoffMs ?? retryBackoffMs; const runs = new Map(); const byKey = new Map(); const ledger = new Map(); @@ -66,8 +70,16 @@ export function createMemoryRunStore(opts?: { maxClaims?: number }): MemoryRunti }, async claim(workerId, ttlMs) { + const now = Date.now(); const pending = [...runs.values()] - .filter((r) => r.status === "pending" && !sessionHasRunning(r.sessionId)) + .filter( + (r) => + r.status === "pending" && + !sessionHasRunning(r.sessionId) && + // A pending row's leaseExpiresAt is the error-retry not-before + // (see retire); claim only once it has passed (#602). + (r.leaseExpiresAt === null || r.leaseExpiresAt <= now), + ) .sort((a, b) => a.createdAt - b.createdAt); const run = pending[0]; if (!run) return null; @@ -77,6 +89,7 @@ export function createMemoryRunStore(opts?: { maxClaims?: number }): MemoryRunti async claimById(runId, workerId, ttlMs) { const run = runs.get(runId); if (!run || run.status !== "pending" || sessionHasRunning(run.sessionId)) return null; + if (run.leaseExpiresAt !== null && run.leaseExpiresAt > Date.now()) return null; return lease(run, workerId, ttlMs); }, @@ -239,6 +252,13 @@ export function createMemoryRunStore(opts?: { maxClaims?: number }): MemoryRunti const overClaimed = run.attempts >= maxClaims; if (retry && run.errorAttempts < run.maxAttempts && !overClaimed) { run.status = "pending"; + // ERROR retries wait out the backoff before they are claimable again + // (#602); the claim paths read this as a not-before, not a lease. + // Lease-expiry requeues (a suspected worker crash, countsAsError=false) + // keep their immediate retry — they have their own poison-pill budget. + if (opts?.countsAsError) { + run.leaseExpiresAt = Date.now() + backoff(run.errorAttempts); + } return { requeued: true, applied: true }; } run.status = "failed"; diff --git a/src/runs/postgres-run-store.ts b/src/runs/postgres-run-store.ts index adb3280d5..47c8ba56e 100644 --- a/src/runs/postgres-run-store.ts +++ b/src/runs/postgres-run-store.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { retryBackoffMs } from './run-store.ts'; import { EventEmitter } from "node:events"; import { createPgPool } from "../persistence/pg-pool.ts"; import type { TurnResult } from "../types.ts"; @@ -128,11 +129,17 @@ export function createPostgresRunStore(connectionString: string, opts?: { maxCla const errorAttemptsAfter = run.errorAttempts + (countsAsError ? 1 : 0); const overClaimed = run.attempts >= maxClaims; if (retry && errorAttemptsAfter < run.maxAttempts && !overClaimed) { + // A pending row's lease_expires_at is the error-retry not-before + // (#602): claims hold off until it passes, so a rate-limited provider + // isn't re-hit back-to-back maxAttempts times. + const notBefore = countsAsError + ? Date.now() + retryBackoffMs(Math.max(1, errorAttemptsAfter)) + : null; const { rowCount } = await q( - `UPDATE runs SET status='pending', lease_token=NULL, lease_expires_at=NULL, worker_id=NULL, + `UPDATE runs SET status='pending', lease_token=NULL, lease_expires_at=$5, worker_id=NULL, error_attempts=error_attempts+$4 WHERE id=$1 AND lease_token=$2 AND status='running' AND ($3::bigint IS NULL OR lease_expires_at <= $3)`, - [run.id, run.leaseToken, ifExpiredAt, countsAsError ? 1 : 0], + [run.id, run.leaseToken, ifExpiredAt, countsAsError ? 1 : 0, notBefore], ); return { requeued: rowCount > 0, applied: rowCount > 0 }; } @@ -177,9 +184,10 @@ export function createPostgresRunStore(connectionString: string, opts?: { maxCla WHERE id = ( SELECT id FROM runs WHERE status='pending' AND session_id NOT IN (SELECT session_id FROM runs WHERE status='running') + AND (lease_expires_at IS NULL OR lease_expires_at <= $5) ORDER BY created_at ASC, seq ASC FOR UPDATE SKIP LOCKED LIMIT 1 ) RETURNING *`, - [token, now + ttlMs, workerId, now], + [token, now + ttlMs, workerId, now, now], ); return rows[0] ? rowToRun(rows[0]) : null; } catch (err) { @@ -197,6 +205,7 @@ export function createPostgresRunStore(connectionString: string, opts?: { maxCla attempts=attempts+1, started_at=COALESCE(started_at,$4) WHERE id = ( SELECT id FROM runs WHERE id=$5 AND status='pending' + AND (lease_expires_at IS NULL OR lease_expires_at <= $4) AND session_id NOT IN (SELECT session_id FROM runs WHERE status='running') FOR UPDATE SKIP LOCKED LIMIT 1 ) RETURNING *`, diff --git a/src/runs/run-store.ts b/src/runs/run-store.ts index ae21eecd6..3f330ab2b 100644 --- a/src/runs/run-store.ts +++ b/src/runs/run-store.ts @@ -102,3 +102,17 @@ export function errorParks(run: Pick, asOf: number): boolean { return run.status === "running" && run.leaseExpiresAt !== null && run.leaseExpiresAt <= asOf; } + +/** + * Delay before a run's next attempt after an ERROR-driven requeue (#602). + * + * Retries used to be immediate: a rate-limited provider got maxAttempts + * requests back-to-back, feeding the very limit that caused the failure. + * Exponential from 5s (attempt 2 → 5s, 3 → 10s, …) capped at 60s — enough + * spacing for transient provider blips without making a genuinely-failing + * run linger. Applied by both run stores on the pending requeue, honored by + * the claim paths via the pending row's lease_expires_at (not-before). + */ +export function retryBackoffMs(nextErrorAttempt: number): number { + return Math.min(5_000 * 2 ** Math.max(0, nextErrorAttempt - 1), 60_000); +} diff --git a/test/pi-retryable-errors.test.ts b/test/pi-retryable-errors.test.ts new file mode 100644 index 000000000..886d9984b --- /dev/null +++ b/test/pi-retryable-errors.test.ts @@ -0,0 +1,111 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; + +import { NonRetryableTurnError } from "../src/core/turn-error.ts"; +import { + piLastAssistantTextOrThrow, + piTurnError, +} from "../src/harness/pi-harness.ts"; +import { retryBackoffMs } from "../src/runs/run-store.ts"; +import { createMemoryRunStore } from "../src/runs/memory-run-store.ts"; +import type { OrchestratorInput } from "../src/core/orchestrator.ts"; +import type { Principal } from "../src/types.ts"; + +const actor: Principal = { id: "internal:U1", type: "internal" }; +function turn(text: string): OrchestratorInput { + return { + actor, + conversation: { kind: "dm", threadRef: "t-602", audience: [actor] }, + origin: { kind: "direct" }, + text, + }; +} + +/** A pi AssistantTextSession whose last assistant message stopped with `error`. */ +function errorSession(errorMessage: string): { + getLastAssistantText: () => string; + messages: unknown[]; +} { + return { + getLastAssistantText: () => "", + messages: [{ role: "assistant", stopReason: "error", errorMessage }], + }; +} + +test("a rate-limit provider error is retryable, not terminal (#602)", () => { + // The exact failure shape from the issue's runs table: a rate limit that + // used to park the run after one attempt because every provider error was + // classified NonRetryableTurnError. + const session = errorSession( + '{"error":{"type":"rate_limit_error","message":"[1113][Insufficient balance]"}}', + ); + assert.throws(() => piLastAssistantTextOrThrow(session as never), (err: Error) => { + assert.equal(err instanceof NonRetryableTurnError, false, "rate limit must be retryable"); + assert.match(err.message, /rate_limit_error/); + return true; + }); + const turnErr = piTurnError(session as never, new Error("wrapped")); + assert.equal(turnErr instanceof NonRetryableTurnError, false); +}); + +test("transient provider errors (overload, 5xx, timeout) are retryable (#602)", () => { + for (const type of ["overloaded_error", "api_error", "server_error", "timeout_error"]) { + const session = errorSession(`{"error":{"type":"${type}","message":"brief provider hiccup"}}`); + const err = piTurnError(session as never, new Error("wrapped")); + assert.equal( + err instanceof NonRetryableTurnError, + false, + `${type} must be retryable`, + ); + } +}); + +test("permanent provider errors stay non-retryable (#602)", () => { + for (const type of ["authentication_error", "permission_error", "invalid_request_error"]) { + const session = errorSession(`{"error":{"type":"${type}","message":"bad key"}}`); + const err = piTurnError(session as never, new Error("wrapped")); + assert.ok(err instanceof NonRetryableTurnError, `${type} must stay non-retryable`); + assert.match(err.message, new RegExp(type)); + } +}); + +test("non-JSON provider errors keep the historical non-retryable class", () => { + const session = errorSession("plain provider failure text"); + assert.ok(piTurnError(session as never, new Error("wrapped")) instanceof NonRetryableTurnError); +}); + +test("retryBackoffMs spaces error retries exponentially (#602)", () => { + assert.equal(retryBackoffMs(1), 5_000); + assert.equal(retryBackoffMs(2), 10_000); + assert.equal(retryBackoffMs(3), 20_000); + assert.equal(retryBackoffMs(9), 60_000, "capped at 60s"); +}); + +test("an error-requeued run is not claimable until the backoff passes (#602)", async () => { + const { runs } = createMemoryRunStore(); + const { run } = await runs.enqueue({ sessionId: "s-602", request: turn("hi") }); + + const first = await runs.claim("w1", 60_000); + assert.ok(first, "first claim succeeds"); + + const requeued = await runs.fail(run!.id, first!.leaseToken!, "Model provider API error", { + retry: true, + }); + assert.equal(requeued.requeued, true); + + // Inside the backoff window the retry is NOT claimable — the retry used to + // be immediate, re-hitting a rate-limited provider back-to-back. + const tooSoon = await runs.claim("w1", 60_000); + assert.equal(tooSoon, null, "backed-off retry must not be claimable"); + + // claimById honors the same not-before. + const byId = await runs.claimById(run!.id, "w1", 60_000); + assert.equal(byId, null, "claimById honors the backoff window"); +}); + +test("a fresh run (no error) is claimable immediately (#602)", async () => { + const { runs } = createMemoryRunStore(); + await runs.enqueue({ sessionId: "s-602b", request: turn("hi") }); + const claimed = await runs.claim("w1", 60_000); + assert.ok(claimed, "no backoff on the first attempt"); +}); diff --git a/test/run-store.test.ts b/test/run-store.test.ts index 00b6b291e..3f0b86902 100644 --- a/test/run-store.test.ts +++ b/test/run-store.test.ts @@ -18,7 +18,11 @@ function turn(text: string, surface?: string): OrchestratorInput { } type Backend = { name: string; make: () => { runs: RunStore; ledger: ToolLedger } }; -const backends: Backend[] = [{ name: "memory", make: () => createMemoryRunStore() }]; +// Zero error-retry backoff keeps this suite's immediate claim-after-fail +// cadence; the backoff itself is covered in pi-retryable-errors.test.ts. +const backends: Backend[] = [ + { name: "memory", make: () => createMemoryRunStore({ retryBackoffMs: () => 0 }) }, +]; const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); diff --git a/test/worker-reaper.test.ts b/test/worker-reaper.test.ts index fbd89ca69..c25c0e615 100644 --- a/test/worker-reaper.test.ts +++ b/test/worker-reaper.test.ts @@ -39,7 +39,7 @@ test("reaper requeues a run whose lease expired (crashed worker)", async () => { }); test("a run parks once the ERROR budget (error_attempts) is exhausted", async () => { - const { runs } = createMemoryRunStore(); + const { runs } = createMemoryRunStore({ retryBackoffMs: () => 0 }); const r = (await runs.enqueue({ sessionId: "s1", request: turn, maxAttempts: 2 })).run; let claimed = await runs.claim("w1", 10_000); @@ -101,7 +101,7 @@ test("with maxClaims set, repeated lease-expiry reaps PARK the poison pill inste }); test("a concrete error parks with its own message even when over the claim cap", async () => { - const { runs } = createMemoryRunStore({ maxClaims: 2 }); + const { runs } = createMemoryRunStore({ maxClaims: 2, retryBackoffMs: () => 0 }); const r = (await runs.enqueue({ sessionId: "s1", request: turn, maxAttempts: 99 })).run; let claimed = await runs.claim("w1", 10_000);