From 57bcdc7396962e7d21f997c98ddf21c0e0d06158 Mon Sep 17 00:00:00 2001 From: jizhejiang Date: Tue, 2 Jun 2026 15:51:19 +0800 Subject: [PATCH] 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. --- .gitleaks.toml | 1 + src/core.ts | 234 +++++++++++++++++++++++--------------------- src/types.ts | 2 +- tests/test-retry.ts | 134 +++++++++++++++++++++---- 4 files changed, 244 insertions(+), 127 deletions(-) diff --git a/.gitleaks.toml b/.gitleaks.toml index fbc6932..5e0e276 100644 --- a/.gitleaks.toml +++ b/.gitleaks.toml @@ -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", diff --git a/src/core.ts b/src/core.ts index 88bce2d..cacd562 100644 --- a/src/core.ts +++ b/src/core.ts @@ -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 | 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" diff --git a/src/types.ts b/src/types.ts index d0f5b5c..7604e2c 100644 --- a/src/types.ts +++ b/src/types.ts @@ -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 /** diff --git a/tests/test-retry.ts b/tests/test-retry.ts index fd9caaf..44330e5 100644 --- a/tests/test-retry.ts +++ b/tests/test-retry.ts @@ -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.