feat(runs): retry transient provider errors with backoff

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));
 }
 

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<string, Run>();
   const byKey = new Map<string, string>();
   const ledger = new Map<string, string>();
@@ -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";

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 *`,

src/runs/run-store.ts

@@ -102,3 +102,17 @@ export function errorParks(run: Pick<Run, "errorAttempts" | "maxAttempts" | "att
 export function leaseLapsed(run: Pick<Run, "status" | "leaseExpiresAt">, 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);
+}

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");
+});

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<void> => new Promise((resolve) => setTimeout(resolve, ms));
 

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);