fix(core): address PR review on retry defaults and timeouts

Align provider maxRetries default with pi (0), treat maxRetryDelayMs 0 as no cap, keep per-attempt timeout through stream reads, and fix gitleaks test fixtures.
2026-06-02 16:15:19 +08:00
parent c3107dbc1c
commit 57bcdc7396
4 changed files with 244 additions and 127 deletions
@@ -8,6 +8,7 @@ title = "pi-commandcode-provider secret scan"
  paths = [
    # Test helpers with mock credentials
    "tests/test-oauth.ts",
+    "tests/test-retry.ts",
    "tests/test-stream.ts",
    "tests/test-pure-functions.ts",
    "tests/test-pi-local.mjs",
@@ -42,7 +42,7 @@ 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 = 2
+const DEFAULT_MAX_RETRIES = 0
 const DEFAULT_MAX_RETRY_DELAY_MS = 60_000
 const BASE_RETRY_DELAY_MS = 500

@@ -59,6 +59,12 @@ function parseRetryAfterSeconds(value: string | null): number | undefined {
  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,
@@ -434,7 +440,7 @@ export function createStreamCommandCode(deps: CoreDependencies) {
        if (nextBody !== undefined) body = nextBody

        const maxRetries = options?.maxRetries ?? DEFAULT_MAX_RETRIES
-        const maxRetryDelayMs = options?.maxRetryDelayMs ?? DEFAULT_MAX_RETRY_DELAY_MS
+        const maxRetryDelayMs = effectiveMaxRetryDelayMs(options?.maxRetryDelayMs)
        const timeoutMs = options?.timeoutMs
        const requestHeaders = {
          "Content-Type": "application/json",
@@ -453,6 +459,14 @@ export function createStreamCommandCode(deps: CoreDependencies) {
          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
@@ -463,127 +477,129 @@ export function createStreamCommandCode(deps: CoreDependencies) {
          controller.signal.addEventListener("abort", onOuterAbort, { once: true })

          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 (
-              timeoutMs !== undefined &&
-              attemptController.signal.aborted &&
-              attempt < maxRetries
-            ) {
-              continue
+            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 && attempt < maxRetries) {
+                continue retryLoop
+              }
+              throw fetchError
            }
-            throw fetchError
-          } finally {
-            controller.signal.removeEventListener("abort", onOuterAbort)
-            if (attemptTimeoutId !== undefined) clearTimeout(attemptTimeoutId)
-          }

-          // --- HTTP-level retry ---
-          if (!response.ok && isRetryableStatus(response.status) && attempt < maxRetries) {
-            const retryAfter = response.headers.get("retry-after")
-            const waitMs = retryDelayMs(attempt, retryAfter, maxRetryDelayMs)
-            if (waitMs < 0) {
-              const requestedSeconds = parseRetryAfterSeconds(retryAfter) ?? 0
-              throw new Error(
-                `Retry-After delay ${requestedSeconds}s exceeds max ${maxRetryDelayMs}ms`,
-              )
+            // --- 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 response.text().catch(() => "")
-            if (waitMs > 0) await delay(waitMs, controller.signal)
-            continue retryLoop
-          }

-          await raceAbort(
-            Promise.resolve(
-              options?.onResponse?.(
-                {
-                  status: response.status,
-                  headers: headersToRecord(response.headers),
-                },
-                model,
+            await raceAbort(
+              Promise.resolve(
+                options?.onResponse?.(
+                  {
+                    status: response.status,
+                    headers: headersToRecord(response.headers),
+                  },
+                  model,
+                ),
              ),
-            ),
-            controller.signal,
-          )
-
-          if (!response.ok) {
-            const errBody = await raceAbort(
-              response.text().catch(() => ""),
              controller.signal,
            )
-            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)
-              if (done) {
-                if (buffer.trim()) handleEvent(parseStreamEventLine(buffer))
-                break
-              }
-              if (controller.signal.aborted) throw abortError("Aborted")
-
-              buffer += decoder.decode(value, { stream: true })
-              const lines = buffer.split("\n")
-              buffer = lines.pop() ?? ""
-
-              for (const line of lines) {
-                if (controller.signal.aborted) throw abortError("Aborted")
-                handleEvent(parseStreamEventLine(line))
-                if (finished) break readLoop
-              }
+            if (!response.ok) {
+              const errBody = await raceAbort(
+                response.text().catch(() => ""),
+                controller.signal,
+              )
+              throw new Error(`Command Code API error ${response.status}: ${errBody.slice(0, 500)}`)
            }
-          } 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(() => {})
+
+            // --- Read response stream ---
+            reader = response.body?.getReader()
+            if (!reader) throw new Error("No response body")
+
+            const decoder = new TextDecoder()
+            let buffer = ""
+
            try {
-              reader.releaseLock()
-            } catch {}
-            reader = undefined
+              readLoop: for (;;) {
+                if (controller.signal.aborted) throw abortError("Aborted")
+                const { done, value } = await raceAbort(reader.read(), attemptController.signal)
+                if (done) {
+                  if (buffer.trim()) handleEvent(parseStreamEventLine(buffer))
+                  break
+                }
+                if (controller.signal.aborted) throw abortError("Aborted")

-            if (controller.signal.aborted) throw streamError
+                buffer += decoder.decode(value, { stream: true })
+                const lines = buffer.split("\n")
+                buffer = lines.pop() ?? ""

-            const canRetry =
-              (output.content.length === 0 || attemptTimedOut) && attempt < maxRetries
-            if (canRetry) {
-              // Reset state for the next attempt.
-              output.content.length = 0
-              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
+                for (const line of lines) {
+                  if (controller.signal.aborted) throw abortError("Aborted")
+                  handleEvent(parseStreamEventLine(line))
+                  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
+              }
+              throw streamError
            }
-            throw streamError
+
+            // Stream completed successfully.
+            endTextBlock()
+            endThinking()
+
+            stream.push({
+              type: "done",
+              reason: successStopReason(output.stopReason),
+              message: output,
+            })
+            stream.end()
+            break retryLoop
+          } finally {
+            controller.signal.removeEventListener("abort", onOuterAbort)
+            clearAttemptTimeout()
          }
-
-          // Stream completed successfully.
-          endTextBlock()
-          endThinking()
-
-          stream.push({
-            type: "done",
-            reason: successStopReason(output.stopReason),
-            message: output,
-          })
-          stream.end()
-          break retryLoop
        }
      } catch (error: unknown) {
        const reason: ErrorReason = controller.signal.aborted ? "aborted" : "error"
@@ -96,7 +96,7 @@ export interface StreamOptions {
  timeoutMs?: number
  /**
   * Maximum retry attempts for transient HTTP errors (429, 5xx).
-   * Default: 2.
+   * Default: 0 (pi agent-level retry handles visible retries when unset).
   */
  maxRetries?: number
  /**
@@ -16,6 +16,8 @@ import {
  type MockCommandCodeServer,
 } from "./helpers.ts"

+const TEST_API_KEY = "option-key"
+
 let server: MockCommandCodeServer

 before(async () => {
@@ -49,7 +51,10 @@ describe("streamCommandCode — retry on transient errors", () => {
    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })

    const events = await collectEvents(
-      streamCommandCode(makeModel(), makeContext(), { apiKey: "mock-key" }),
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 2,
+      }),
    )

    assert.equal(server.requestCount(), 2)
@@ -70,7 +75,10 @@ describe("streamCommandCode — retry on transient errors", () => {
    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })

    const events = await collectEvents(
-      streamCommandCode(makeModel(), makeContext(), { apiKey: "mock-key" }),
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 2,
+      }),
    )

    assert.equal(server.requestCount(), 2)
@@ -82,7 +90,7 @@ describe("streamCommandCode — retry on transient errors", () => {
    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })

    const events = await collectEvents(
-      streamCommandCode(makeModel(), makeContext(), { apiKey: "mock-key" }),
+      streamCommandCode(makeModel(), makeContext(), { apiKey: TEST_API_KEY }),
    )

    assert.equal(server.requestCount(), 1)
@@ -98,7 +106,7 @@ describe("streamCommandCode — retry on transient errors", () => {

    const events = await collectEvents(
      streamCommandCode(makeModel(), makeContext(), {
-        apiKey: "mock-key",
+        apiKey: TEST_API_KEY,
        maxRetries: 3,
      }),
    )
@@ -136,7 +144,10 @@ describe("streamCommandCode — Retry-After header", () => {
    })

    const events = await collectEvents(
-      streamCommandCode(makeModel(), makeContext(), { apiKey: "mock-key" }),
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 2,
+      }),
    )

    assert.equal(server.requestCount(), 2)
@@ -155,7 +166,7 @@ describe("streamCommandCode — Retry-After header", () => {

    const events = await collectEvents(
      streamCommandCode(makeModel(), makeContext(), {
-        apiKey: "mock-key",
+        apiKey: TEST_API_KEY,
        maxRetryDelayMs: 10_000,
      }),
    )
@@ -166,6 +177,41 @@ describe("streamCommandCode — Retry-After header", () => {
    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", () => {
@@ -189,8 +235,9 @@ describe("streamCommandCode — timeout", () => {

    const events = await collectEvents(
      streamCommandCode(makeModel(), makeContext(), {
-        apiKey: "mock-key",
+        apiKey: TEST_API_KEY,
        timeoutMs: 50,
+        maxRetries: 2,
      }),
      5_000,
    )
@@ -199,6 +246,57 @@ describe("streamCommandCode — timeout", () => {
    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",
@@ -210,7 +308,7 @@ describe("streamCommandCode — timeout", () => {

    const events = await collectEvents(
      streamCommandCode(makeModel(), makeContext(), {
-        apiKey: "mock-key",
+        apiKey: TEST_API_KEY,
        timeoutMs: 50,
        maxRetries: 1,
      }),
@@ -242,7 +340,7 @@ describe("streamCommandCode — abort cancels retry loop", () => {

    const events = await collectEvents(
      streamCommandCode(makeModel(), makeContext(), {
-        apiKey: "mock-key",
+        apiKey: TEST_API_KEY,
        signal: controller.signal,
        maxRetries: 10,
      }),
@@ -258,14 +356,13 @@ describe("streamCommandCode — abort cancels retry loop", () => {
 })

 describe("streamCommandCode — retry defaults", () => {
-  it("uses default maxRetries of 2 when not specified", async () => {
+  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: "mock-key" }))
+    await collectEvents(streamCommandCode(makeModel(), makeContext(), { apiKey: TEST_API_KEY }))

-    // initial + 2 retries = 3
-    assert.equal(server.requestCount(), 3)
+    assert.equal(server.requestCount(), 1)
  })

  it("respects maxRetries: 0 (no retries)", async () => {
@@ -274,7 +371,7 @@ describe("streamCommandCode — retry defaults", () => {

    const events = await collectEvents(
      streamCommandCode(makeModel(), makeContext(), {
-        apiKey: "mock-key",
+        apiKey: TEST_API_KEY,
        maxRetries: 0,
      }),
    )
@@ -307,7 +404,10 @@ describe("streamCommandCode — stream-level error retry", () => {
    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })

    const events = await collectEvents(
-      streamCommandCode(makeModel(), makeContext(), { apiKey: "mock-key" }),
+      streamCommandCode(makeModel(), makeContext(), {
+        apiKey: TEST_API_KEY,
+        maxRetries: 2,
+      }),
    )

    assert.equal(server.requestCount(), 2)
@@ -328,7 +428,7 @@ describe("streamCommandCode — stream-level error retry", () => {

    const events = await collectEvents(
      streamCommandCode(makeModel(), makeContext(), {
-        apiKey: "mock-key",
+        apiKey: TEST_API_KEY,
        maxRetries: 3,
      }),
    )
@@ -357,7 +457,7 @@ describe("streamCommandCode — stream-level error retry", () => {
    const { streamCommandCode } = createTestDeps({ apiBase: server.baseUrl() })

    const events = await collectEvents(
-      streamCommandCode(makeModel(), makeContext(), { apiKey: "mock-key" }),
+      streamCommandCode(makeModel(), makeContext(), { apiKey: TEST_API_KEY }),
    )

    // Only 1 request — no retry because content was already emitted.