feat(core): add retry for transient HTTP and stream-level errors

.gitleaks.toml

@@ -6,12 +6,8 @@ title = "pi-commandcode-provider secret scan"
 [allowlist]
   description = "Known safe paths and test fixtures"
   paths = [
-    # Test helpers with mock credentials
+    # Test fixtures use intentionally fake credentials.
-    "tests/test-oauth.ts",
+    "^tests/",
-    "tests/test-stream.ts",
-    "tests/test-pure-functions.ts",
-    "tests/test-pi-local.mjs",
-    "tests/test-omp-compat.mjs",
     # CI workflows reference NPM_TOKEN secret name
     ".github/workflows/publish.yml",
   ]

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## Unreleased
+
+- Add retry mechanism for transient HTTP errors (429, 5xx) and stream-level errors, configurable via pi `settings.json` `retry.provider` fields (`timeoutMs`, `maxRetries`, `maxRetryDelayMs`). Supports exponential backoff with jitter and `Retry-After` header.
+
 ## 0.3.1 - 2026-05-29
 
 - Bump CLI version header to `0.29.0` for Command Code API parity.

package.json

@@ -28,7 +28,7 @@
     "LICENSE"
   ],
   "scripts": {
-    "test": "npm run typecheck && tsx tests/test-pure-functions.ts && tsx tests/test-models.ts && tsx tests/test-pricing.ts && tsx tests/test-oauth.ts && tsx tests/test-abort.ts && tsx tests/test-stream.ts && node tests/test-pi-local.mjs && node tests/test-omp-compat.mjs",
+    "test": "npm run typecheck && tsx tests/test-pure-functions.ts && tsx tests/test-models.ts && tsx tests/test-pricing.ts && tsx tests/test-oauth.ts && tsx tests/test-abort.ts && tsx tests/test-stream.ts && tsx tests/test-retry.ts && node tests/test-pi-local.mjs && node tests/test-omp-compat.mjs",
     "typecheck": "tsc --noEmit",
     "format:check": "prettier --check '**/*.{ts,mjs,json,md}'",
     "format": "prettier --write '**/*.{ts,mjs,json,md}'",
@@ -38,6 +38,7 @@
     "test:oauth": "tsx tests/test-oauth.ts",
     "test:abort": "tsx tests/test-abort.ts",
     "test:stream": "tsx tests/test-stream.ts",
+    "test:retry": "tsx tests/test-retry.ts",
     "test:pi-local": "node tests/test-pi-local.mjs",
     "test:smoke": "node tests/test-smoke.mjs"
   },

src/core.ts

@@ -42,6 +42,43 @@ export const DEFAULT_API_BASE = "https://api.commandcode.ai"
 export const COMMAND_CODE_CLI_VERSION = "0.29.0"
 
 const DEFAULT_GENERATE_MAX_TOKENS = 64_000
+const DEFAULT_MAX_RETRIES = 0
+const DEFAULT_MAX_RETRY_DELAY_MS = 60_000
+const BASE_RETRY_DELAY_MS = 500
+
+function isRetryableStatus(status: number): boolean {
+  return status === 429 || (status >= 500 && status < 600)
+}
+
+function parseRetryAfterSeconds(value: string | null): number | undefined {
+  if (!value) return undefined
+  const seconds = Number(value)
+  if (Number.isFinite(seconds) && seconds >= 0) return seconds
+  const date = Date.parse(value)
+  if (!Number.isNaN(date)) return Math.max(0, (date - Date.now()) / 1000)
+  return undefined
+}
+
+function effectiveMaxRetryDelayMs(value: number | undefined): number {
+  if (value === undefined) return DEFAULT_MAX_RETRY_DELAY_MS
+  if (value === 0) return Number.POSITIVE_INFINITY
+  return value
+}
+
+function retryDelayMs(
+  attempt: number,
+  retryAfterHeader: string | null,
+  maxDelayMs: number,
+): number {
+  const retryAfterMs = parseRetryAfterSeconds(retryAfterHeader)
+  if (retryAfterMs !== undefined) {
+    if (retryAfterMs * 1000 > maxDelayMs) return -1
+    return retryAfterMs * 1000
+  }
+  const exponential = BASE_RETRY_DELAY_MS * 2 ** attempt
+  const jitter = exponential * 0.2 * Math.random()
+  return Math.min(exponential + jitter, maxDelayMs)
+}
 
 function defaultUsage(): Usage {
   return {
@@ -76,6 +113,14 @@ function abortError(message = "The operation was aborted"): DOMException {
   return new DOMException(message, "AbortError")
 }
 
+function timeoutError(timeoutMs: number | undefined): Error {
+  return new Error(
+    timeoutMs === undefined
+      ? "Command Code API request timed out"
+      : `Command Code API request timed out after ${timeoutMs}ms`,
+  )
+}
+
 function successStopReason(reason: TerminalReason): StopReason {
   if (reason === "length" || reason === "toolUse") return reason
   return "stop"
@@ -104,6 +149,22 @@ export function createStreamCommandCode(deps: CoreDependencies) {
   const cwd = deps.cwd ?? (() => process.cwd())
   const now = deps.now ?? (() => Date.now())
   const uuid = deps.uuid ?? (() => randomUUID())
+  const delay =
+    deps.delay ??
+    ((ms: number, signal: AbortSignal) => {
+      if (signal.aborted) return Promise.reject(abortError())
+      return new Promise<void>((resolve, reject) => {
+        const id = setTimeout(() => {
+          signal.removeEventListener("abort", onAbort)
+          resolve()
+        }, ms)
+        const onAbort = () => {
+          clearTimeout(id)
+          reject(abortError())
+        }
+        signal.addEventListener("abort", onAbort, { once: true })
+      })
+    })
 
   function raceAbort<T>(promise: Promise<T>, signal: AbortSignal): Promise<T> {
     if (signal.aborted) return Promise.reject(abortError())
@@ -386,10 +447,10 @@ export function createStreamCommandCode(deps: CoreDependencies) {
         )
         if (nextBody !== undefined) body = nextBody
 
-        const response = await raceAbort(
+        const maxRetries = options?.maxRetries ?? DEFAULT_MAX_RETRIES
-          fetchImpl(`${apiBase}/alpha/generate`, {
+        const maxRetryDelayMs = effectiveMaxRetryDelayMs(options?.maxRetryDelayMs)
-            method: "POST",
+        const timeoutMs = options?.timeoutMs
-            headers: {
+        const requestHeaders = {
           "Content-Type": "application/json",
           Authorization: `Bearer ${apiKey}`,
           "x-command-code-version": COMMAND_CODE_CLI_VERSION,
@@ -398,12 +459,64 @@ export function createStreamCommandCode(deps: CoreDependencies) {
           "x-taste-learning": "true",
           "x-co-flag": "false",
           ...options?.headers,
-            },
+        }
-            body: JSON.stringify(body),
+        const bodyStr = JSON.stringify(body)
-            signal: controller.signal,
+
-          }),
+        let response!: Response
-          controller.signal,
+        retryLoop: for (let attempt = 0; ; attempt++) {
-        )
+          const attemptController = new AbortController()
+          let attemptTimedOut = false
+          let attemptTimeoutId: ReturnType<typeof setTimeout> | undefined
+
+          const clearAttemptTimeout = () => {
+            if (attemptTimeoutId !== undefined) {
+              clearTimeout(attemptTimeoutId)
+              attemptTimeoutId = undefined
+            }
+          }
+
+          if (timeoutMs !== undefined) {
+            attemptTimeoutId = setTimeout(() => {
+              attemptTimedOut = true
+              attemptController.abort()
+            }, timeoutMs)
+          }
+          const onOuterAbort = () => attemptController.abort()
+          controller.signal.addEventListener("abort", onOuterAbort, { once: true })
+
+          try {
+            try {
+              response = await fetchImpl(`${apiBase}/alpha/generate`, {
+                method: "POST",
+                headers: requestHeaders,
+                body: bodyStr,
+                signal: attemptController.signal,
+              })
+            } catch (fetchError: unknown) {
+              if (controller.signal.aborted) throw abortError("Aborted")
+              if (attemptTimedOut) {
+                if (attempt < maxRetries) continue retryLoop
+                throw timeoutError(timeoutMs)
+              }
+              throw fetchError
+            }
+
+            // --- HTTP-level retry ---
+            if (!response.ok && isRetryableStatus(response.status)) {
+              const retryAfter = response.headers.get("retry-after")
+              const waitMs = retryDelayMs(attempt, retryAfter, maxRetryDelayMs)
+              if (waitMs < 0) {
+                const requestedSeconds = parseRetryAfterSeconds(retryAfter) ?? 0
+                const capLabel =
+                  maxRetryDelayMs === Number.POSITIVE_INFINITY ? "disabled" : `${maxRetryDelayMs}ms`
+                throw new Error(`Retry-After delay ${requestedSeconds}s exceeds max ${capLabel}`)
+              }
+              if (attempt < maxRetries) {
+                await response.text().catch(() => "")
+                if (waitMs > 0) await delay(waitMs, controller.signal)
+                continue retryLoop
+              }
+            }
 
             await raceAbort(
               Promise.resolve(
@@ -426,15 +539,17 @@ export function createStreamCommandCode(deps: CoreDependencies) {
               throw new Error(`Command Code API error ${response.status}: ${errBody.slice(0, 500)}`)
             }
 
+            // --- Read response stream ---
             reader = response.body?.getReader()
             if (!reader) throw new Error("No response body")
 
             const decoder = new TextDecoder()
             let buffer = ""
 
+            try {
               readLoop: for (;;) {
                 if (controller.signal.aborted) throw abortError("Aborted")
-          const { done, value } = await raceAbort(reader.read(), controller.signal)
+                const { done, value } = await raceAbort(reader.read(), attemptController.signal)
                 if (done) {
                   if (buffer.trim()) handleEvent(parseStreamEventLine(buffer))
                   break
@@ -451,7 +566,36 @@ export function createStreamCommandCode(deps: CoreDependencies) {
                   if (finished) break readLoop
                 }
               }
+            } catch (streamError: unknown) {
+              // Stream-level error (e.g. API returned 200 OK but sent an error event)
+              // or per-attempt timeout during stream reading.
+              await reader.cancel().catch(() => {})
+              try {
+                reader.releaseLock()
+              } catch {}
+              reader = undefined
 
+              if (controller.signal.aborted) throw streamError
+
+              // Never retry after visible content was emitted (including timeout mid-stream).
+              const canRetry = output.content.length === 0 && attempt < maxRetries
+              if (canRetry) {
+                output.content.length = 0
+                textBlock = undefined
+                currentTextIdx = -1
+                thinkingIdx = -1
+                output.stopReason = "stop"
+                output.errorMessage = undefined
+                finished = false
+                const waitMs = attemptTimedOut ? 0 : retryDelayMs(attempt, null, maxRetryDelayMs)
+                if (waitMs > 0) await delay(waitMs, controller.signal)
+                continue retryLoop
+              }
+              if (attemptTimedOut) throw timeoutError(timeoutMs)
+              throw streamError
+            }
+
+            // Stream completed successfully.
             endTextBlock()
             endThinking()
 
@@ -461,6 +605,12 @@ export function createStreamCommandCode(deps: CoreDependencies) {
               message: output,
             })
             stream.end()
+            break retryLoop
+          } finally {
+            controller.signal.removeEventListener("abort", onOuterAbort)
+            clearAttemptTimeout()
+          }
+        }
       } catch (error: unknown) {
         const reason: ErrorReason = controller.signal.aborted ? "aborted" : "error"
         output.stopReason = reason

src/types.ts

@@ -89,6 +89,23 @@ export interface StreamOptions {
   maxTokens?: number
   onPayload?: (payload: unknown, model: ModelLike) => unknown | Promise<unknown>
   onResponse?: (response: ProviderResponseInfo, model: ModelLike) => void | Promise<void>
+  /**
+   * HTTP request timeout in milliseconds.
+   * Applied per-attempt; on timeout the request is retried if retries remain.
+   */
+  timeoutMs?: number
+  /**
+   * Maximum retry attempts for transient HTTP errors (429, 5xx).
+   * Default: 0 (pi agent-level retry handles visible retries when unset).
+   */
+  maxRetries?: number
+  /**
+   * Maximum delay in milliseconds to wait for a retry when the server requests
+   * a long wait via Retry-After. If the server's requested delay exceeds this
+   * value, the request fails immediately. Default: 60000 (60 seconds).
+   * Set to 0 to disable the cap.
+   */
+  maxRetryDelayMs?: number
 }
 
 export type AssistantMessageEvent =
@@ -153,4 +170,6 @@ export interface CoreDependencies {
   now?: () => number
   uuid?: () => string
   homeDir?: () => string
+  /** Injectable delay for retry backoff. Defaults to setTimeout. */
+  delay?: (ms: number, signal: AbortSignal) => Promise<void>
 }

tests/helpers.ts

@@ -112,6 +112,7 @@ export function createTestDeps(overrides: Partial<CoreDependencies> = {}): TestD
     now: () => new Date("2026-05-05T12:00:00Z").getTime(),
     uuid: () => "00000000-0000-4000-8000-000000000000",
     cwd: () => "/repo",
+    delay: async () => {},
     ...overrides,
   })
   return { streamCommandCode, calculatedUsages }
@@ -124,12 +125,15 @@ type SuccessPlan = {
   chunks?: string[]
   delays?: number[]
   hangAfterLast?: boolean
+  /** Delay in ms before the server starts sending the response. */
+  responseDelay?: number
 }
 
 type ErrorPlan = {
   type: "error"
   status: number
   body: string
+  headers?: Record<string, string>
 }
 
 export type ResponsePlan = SuccessPlan | ErrorPlan
@@ -146,6 +150,7 @@ function headersToRecord(headers: IncomingHttpHeaders): Record<string, string> {
 export interface MockCommandCodeServer {
   baseUrl(): string
   mockResponse(plan: ResponsePlan): void
+  mockResponseQueue(plans: ResponsePlan[]): void
   reset(): void
   close(): Promise<void>
   lastRequestBody(): unknown
@@ -155,7 +160,7 @@ export interface MockCommandCodeServer {
 }
 
 export async function startMockCommandCodeServer(): Promise<MockCommandCodeServer> {
-  let nextPlan: ResponsePlan = { type: "success", events: [] }
+  let planQueue: ResponsePlan[] = [{ type: "success", events: [] }]
   let lastBody: unknown
   let lastHeaders: Record<string, string> = {}
   let requests = 0
@@ -183,9 +188,12 @@ export async function startMockCommandCodeServer(): Promise<MockCommandCodeServe
         lastBody = undefined
       }
 
-      const plan = nextPlan
+      // Pop the first plan from the queue; keep the last one as fallback.
+      const plan = planQueue.length > 1 ? planQueue.shift()! : planQueue[0]
+
       if (plan.type === "error") {
-        res.writeHead(plan.status, { "Content-Type": "text/plain" })
+        const headers: Record<string, string> = { "Content-Type": "text/plain", ...plan.headers }
+        res.writeHead(plan.status, headers)
         res.end(plan.body)
         return
       }
@@ -223,7 +231,11 @@ export async function startMockCommandCodeServer(): Promise<MockCommandCodeServe
         }
       }
 
+      if (plan.responseDelay) {
+        setTimeout(sendNext, plan.responseDelay)
+      } else {
         sendNext()
+      }
     })
   })
 
@@ -238,10 +250,13 @@ export async function startMockCommandCodeServer(): Promise<MockCommandCodeServe
   return {
     baseUrl: () => `http://127.0.0.1:${port}`,
     mockResponse(plan: ResponsePlan) {
-      nextPlan = plan
+      planQueue = [plan]
+    },
+    mockResponseQueue(plans: ResponsePlan[]) {
+      planQueue = [...plans]
     },
     reset() {
-      nextPlan = { type: "success", events: [] }
+      planQueue = [{ type: "success", events: [] }]
       lastBody = undefined
       lastHeaders = {}
       requests = 0

tests/test-retry.ts

@@ -0,0 +1,470 @@
+/**
+ * Tests for retry and timeout behaviour driven by pi settings.json
+ * retry config (timeoutMs, maxRetries, maxRetryDelayMs).
+ */
+
+import assert from "node:assert/strict"
+import { after, before, beforeEach, describe, it } from "node:test"
+
+import type { AssistantMessageEvent } from "../src/core.ts"
+import {
+  collectEvents,
+  createTestDeps,
+  makeContext,
+  makeModel,
+  startMockCommandCodeServer,
+  type MockCommandCodeServer,
+} from "./helpers.ts"
+
+const TEST_API_KEY = "option-key"
+
+let server: MockCommandCodeServer
+
+before(async () => {
+  server = await startMockCommandCodeServer()
+})
+
+after(async () => {
+  await server.close()
+})
+
+beforeEach(() => {
+  server.reset()
+})
+
+function eventTypes(events: readonly AssistantMessageEvent[]): string[] {
+  return events.map((event) => event.type)
+}
+
+describe("streamCommandCode — retry on transient errors", () => {
+  it("retries on 429 and succeeds on the second attempt", async () => {
+    server.mockResponseQueue([
+      { type: "error", status: 429, body: "rate limited" },
+      {
+        type: "success",
+        events: [
+          JSON.stringify({ type: "text-delta", text: "ok" }),
+          JSON.stringify({ type: "finish", finishReason: "stop" }),
+        ],
+      },
+    ])
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 2,
+      }),
+    )
+
+    assert.equal(server.requestCount(), 2)
+    assert.deepEqual(eventTypes(events), ["start", "text_start", "text_delta", "text_end", "done"])
+    const done = events.at(-1)
+    if (done?.type !== "done") throw new Error("expected done")
+    assert.equal(done.reason, "stop")
+  })
+
+  it("retries on 500 and succeeds on the second attempt", async () => {
+    server.mockResponseQueue([
+      { type: "error", status: 500, body: "internal server error" },
+      {
+        type: "success",
+        events: [JSON.stringify({ type: "finish", finishReason: "stop" })],
+      },
+    ])
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 2,
+      }),
+    )
+
+    assert.equal(server.requestCount(), 2)
+    assert.equal(events.at(-1)?.type, "done")
+  })
+
+  it("does NOT retry on 400 (non-retryable client error)", async () => {
+    server.mockResponse({ type: "error", status: 400, body: "bad request" })
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), { apiKey: TEST_API_KEY }),
+    )
+
+    assert.equal(server.requestCount(), 1)
+    assert.deepEqual(eventTypes(events), ["start", "error"])
+    const last = events.at(-1)
+    if (last?.type !== "error") throw new Error("expected error")
+    assert.match(last.error.errorMessage ?? "", /400/)
+  })
+
+  it("exhausts maxRetries and emits an error", async () => {
+    server.mockResponse({ type: "error", status: 503, body: "unavailable" })
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 3,
+      }),
+    )
+
+    // initial attempt + 3 retries = 4 total
+    assert.equal(server.requestCount(), 4)
+    assert.deepEqual(eventTypes(events), ["start", "error"])
+    const last503 = events.at(-1)
+    if (last503?.type !== "error") throw new Error("expected error")
+    assert.match(last503.error.errorMessage ?? "", /503/)
+  })
+})
+
+describe("streamCommandCode — Retry-After header", () => {
+  it("respects Retry-After delay in seconds", async () => {
+    let delayCalled = false
+    server.mockResponseQueue([
+      {
+        type: "error",
+        status: 429,
+        body: "rate limited",
+        headers: { "retry-after": "2" },
+      },
+      {
+        type: "success",
+        events: [JSON.stringify({ type: "finish", finishReason: "stop" })],
+      },
+    ])
+    const { streamCommandCode } = createTestDeps({
+      apiBase: server.baseUrl(),
+      delay: async (ms: number) => {
+        delayCalled = true
+        assert.equal(ms, 2000)
+      },
+    })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 2,
+      }),
+    )
+
+    assert.equal(server.requestCount(), 2)
+    assert.equal(events.at(-1)?.type, "done")
+    assert.ok(delayCalled, "delay should have been called with Retry-After value")
+  })
+
+  it("fails immediately when Retry-After exceeds maxRetryDelayMs", async () => {
+    server.mockResponse({
+      type: "error",
+      status: 429,
+      body: "rate limited",
+      headers: { "retry-after": "300" },
+    })
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetryDelayMs: 10_000,
+      }),
+    )
+
+    assert.equal(server.requestCount(), 1)
+    assert.deepEqual(eventTypes(events), ["start", "error"])
+    const lastMax = events.at(-1)
+    if (lastMax?.type !== "error") throw new Error("expected error")
+    assert.match(lastMax.error.errorMessage ?? "", /exceeds max/)
+  })
+
+  it("does not cap Retry-After when maxRetryDelayMs is 0", async () => {
+    let delayCalled = false
+    server.mockResponseQueue([
+      {
+        type: "error",
+        status: 429,
+        body: "rate limited",
+        headers: { "retry-after": "120" },
+      },
+      {
+        type: "success",
+        events: [JSON.stringify({ type: "finish", finishReason: "stop" })],
+      },
+    ])
+    const { streamCommandCode } = createTestDeps({
+      apiBase: server.baseUrl(),
+      delay: async (ms: number) => {
+        delayCalled = true
+        assert.equal(ms, 120_000)
+      },
+    })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 1,
+        maxRetryDelayMs: 0,
+      }),
+    )
+
+    assert.equal(server.requestCount(), 2)
+    assert.equal(events.at(-1)?.type, "done")
+    assert.ok(delayCalled)
+  })
+})
+
+describe("streamCommandCode — timeout", () => {
+  it("retries on per-attempt timeout and succeeds", async () => {
+    server.mockResponseQueue([
+      {
+        type: "success",
+        events: [JSON.stringify({ type: "finish", finishReason: "stop" })],
+        hangAfterLast: true,
+        responseDelay: 200,
+      },
+      {
+        type: "success",
+        events: [
+          JSON.stringify({ type: "text-delta", text: "fast" }),
+          JSON.stringify({ type: "finish", finishReason: "stop" }),
+        ],
+      },
+    ])
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        timeoutMs: 50,
+        maxRetries: 2,
+      }),
+      5_000,
+    )
+
+    assert.equal(server.requestCount(), 2)
+    assert.deepEqual(eventTypes(events), ["start", "text_start", "text_delta", "text_end", "done"])
+  })
+
+  it("retries when the response starts but the stream hangs before finish", async () => {
+    server.mockResponseQueue([
+      {
+        type: "success",
+        events: [],
+        hangAfterLast: true,
+      },
+      {
+        type: "success",
+        events: [
+          JSON.stringify({ type: "text-delta", text: "ok" }),
+          JSON.stringify({ type: "finish", finishReason: "stop" }),
+        ],
+      },
+    ])
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        timeoutMs: 50,
+        maxRetries: 2,
+      }),
+      5_000,
+    )
+
+    assert.equal(server.requestCount(), 2)
+    assert.deepEqual(eventTypes(events), ["start", "text_start", "text_delta", "text_end", "done"])
+  })
+
+  it("does NOT retry on timeout after partial text-delta was emitted", async () => {
+    server.mockResponse({
+      type: "success",
+      events: [JSON.stringify({ type: "text-delta", text: "partial" })],
+      hangAfterLast: true,
+    })
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        timeoutMs: 50,
+        maxRetries: 2,
+      }),
+      5_000,
+    )
+
+    assert.equal(server.requestCount(), 1)
+    assert.deepEqual(eventTypes(events), ["start", "text_start", "text_delta", "error"])
+  })
+
+  it("emits error when all retry attempts time out", async () => {
+    server.mockResponse({
+      type: "success",
+      events: [JSON.stringify({ type: "finish", finishReason: "stop" })],
+      hangAfterLast: true,
+      responseDelay: 200,
+    })
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        timeoutMs: 50,
+        maxRetries: 1,
+      }),
+      5_000,
+    )
+
+    // initial + 1 retry = 2
+    assert.equal(server.requestCount(), 2)
+    assert.deepEqual(eventTypes(events), ["start", "error"])
+    const error = events.at(-1)
+    if (error?.type !== "error") throw new Error("expected error")
+    assert.match(error.error.errorMessage ?? "", /timed out after 50ms/)
+  })
+})
+
+describe("streamCommandCode — abort cancels retry loop", () => {
+  it("user abort stops retries immediately", async () => {
+    server.mockResponse({ type: "error", status: 500, body: "error" })
+    const controller = new AbortController()
+    const { streamCommandCode } = createTestDeps({
+      apiBase: server.baseUrl(),
+      delay: async (_ms: number, signal: AbortSignal) => {
+        // Abort during the retry delay
+        controller.abort()
+        // Simulate the real delay which rejects on abort
+        return new Promise<void>((_, reject) => {
+          if (signal.aborted) reject(new DOMException("Aborted", "AbortError"))
+          signal.addEventListener("abort", () => reject(new DOMException("Aborted", "AbortError")))
+        })
+      },
+    })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        signal: controller.signal,
+        maxRetries: 10,
+      }),
+    )
+
+    // Should only have made 1 request (the initial one), then aborted during delay
+    assert.equal(server.requestCount(), 1)
+    assert.deepEqual(eventTypes(events), ["start", "error"])
+    const error = events.at(-1)
+    if (error?.type !== "error") throw new Error("expected error")
+    assert.equal(error.reason, "aborted")
+  })
+})
+
+describe("streamCommandCode — retry defaults", () => {
+  it("uses default maxRetries of 0 when not specified", async () => {
+    server.mockResponse({ type: "error", status: 500, body: "error" })
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    await collectEvents(streamCommandCode(makeModel(), makeContext(), { apiKey: TEST_API_KEY }))
+
+    assert.equal(server.requestCount(), 1)
+  })
+
+  it("respects maxRetries: 0 (no retries)", async () => {
+    server.mockResponse({ type: "error", status: 500, body: "error" })
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 0,
+      }),
+    )
+
+    assert.equal(server.requestCount(), 1)
+    assert.deepEqual(eventTypes(events), ["start", "error"])
+  })
+})
+
+describe("streamCommandCode — stream-level error retry", () => {
+  it("retries when API returns 200 OK but stream contains an error event", async () => {
+    server.mockResponseQueue([
+      {
+        type: "success",
+        events: [
+          JSON.stringify({
+            type: "error",
+            error: "Service temporarily unavailable. Please try again shortly.",
+          }),
+        ],
+      },
+      {
+        type: "success",
+        events: [
+          JSON.stringify({ type: "text-delta", text: "ok" }),
+          JSON.stringify({ type: "finish", finishReason: "stop" }),
+        ],
+      },
+    ])
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 2,
+      }),
+    )
+
+    assert.equal(server.requestCount(), 2)
+    assert.deepEqual(eventTypes(events), ["start", "text_start", "text_delta", "text_end", "done"])
+  })
+
+  it("exhausts retries on persistent stream-level errors", async () => {
+    server.mockResponse({
+      type: "success",
+      events: [
+        JSON.stringify({
+          type: "error",
+          error: "Service temporarily unavailable. Please try again shortly.",
+        }),
+      ],
+    })
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 3,
+      }),
+    )
+
+    // initial + 3 retries = 4
+    assert.equal(server.requestCount(), 4)
+    assert.deepEqual(eventTypes(events), ["start", "error"])
+    const last = events.at(-1)
+    if (last?.type !== "error") throw new Error("expected error")
+    assert.match(last.error.errorMessage ?? "", /temporarily unavailable/)
+  })
+
+  it("does NOT retry stream error when content was already emitted", async () => {
+    server.mockResponseQueue([
+      {
+        type: "success",
+        events: [
+          JSON.stringify({ type: "text-delta", text: "partial" }),
+          JSON.stringify({
+            type: "error",
+            error: "Service temporarily unavailable",
+          }),
+        ],
+      },
+    ])
+    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })
+
+    const events = await collectEvents(
+      streamCommandCode(makeModel(), makeContext(), { apiKey: TEST_API_KEY }),
+    )
+
+    // Only 1 request — no retry because content was already emitted.
+    assert.equal(server.requestCount(), 1)
+    assert.deepEqual(eventTypes(events), ["start", "text_start", "text_delta", "error"])
+  })
+})