fix(ai): disable hidden provider 429 retries (#4991)
This commit is contained in:
@@ -512,7 +512,7 @@ export const streamAnthropic: StreamFunction<"anthropic-messages", AnthropicOpti
|
||||
const requestOptions = {
|
||||
...(options?.signal ? { signal: options.signal } : {}),
|
||||
...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
|
||||
...(options?.maxRetries !== undefined ? { maxRetries: options.maxRetries } : {}),
|
||||
maxRetries: options?.maxRetries ?? 0,
|
||||
};
|
||||
const response = await client.messages.create({ ...params, stream: true }, requestOptions).asResponse();
|
||||
await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
|
||||
|
||||
@@ -111,7 +111,7 @@ export const streamAzureOpenAIResponses: StreamFunction<"azure-openai-responses"
|
||||
const requestOptions = {
|
||||
...(options?.signal ? { signal: options.signal } : {}),
|
||||
...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
|
||||
...(options?.maxRetries !== undefined ? { maxRetries: options.maxRetries } : {}),
|
||||
maxRetries: options?.maxRetries ?? 0,
|
||||
};
|
||||
const { data: openaiStream, response } = await client.responses.create(params, requestOptions).withResponse();
|
||||
await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
|
||||
|
||||
@@ -63,7 +63,7 @@ export const generateImagesOpenRouter: ImagesFunction<"openrouter-images", Image
|
||||
const requestOptions = {
|
||||
...(options?.signal ? { signal: options.signal } : {}),
|
||||
...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
|
||||
...(options?.maxRetries !== undefined ? { maxRetries: options.maxRetries } : {}),
|
||||
maxRetries: options?.maxRetries ?? 0,
|
||||
};
|
||||
const { data: response, response: rawResponse } = await client.chat.completions
|
||||
.create(params as unknown as ChatCompletionCreateParamsNonStreaming, requestOptions)
|
||||
|
||||
@@ -50,8 +50,9 @@ import { buildBaseOptions } from "./simple-options.ts";
|
||||
|
||||
const DEFAULT_CODEX_BASE_URL = "https://chatgpt.com/backend-api";
|
||||
const JWT_CLAIM_PATH = "https://api.openai.com/auth" as const;
|
||||
const DEFAULT_MAX_RETRIES = 3;
|
||||
const DEFAULT_MAX_RETRIES = 0;
|
||||
const BASE_DELAY_MS = 1000;
|
||||
const DEFAULT_MAX_RETRY_DELAY_MS = 60_000;
|
||||
const CODEX_TOOL_CALL_PROVIDERS = new Set(["openai", "openai-codex", "opencode"]);
|
||||
const WEBSOCKET_MESSAGE_TOO_BIG_CLOSE_CODE = 1009;
|
||||
|
||||
@@ -100,13 +101,54 @@ interface RequestBody {
|
||||
// Retry Helpers
|
||||
// ============================================================================
|
||||
|
||||
function isTerminalRateLimitError(errorText: string): boolean {
|
||||
return /GoUsageLimitError|FreeUsageLimitError|Monthly usage limit reached|available balance|insufficient_quota|out of budget|quota exceeded|billing/i.test(
|
||||
errorText,
|
||||
);
|
||||
}
|
||||
|
||||
function isRetryableError(status: number, errorText: string): boolean {
|
||||
if (status === 429 && isTerminalRateLimitError(errorText)) {
|
||||
return false;
|
||||
}
|
||||
if (status === 429 || status === 500 || status === 502 || status === 503 || status === 504) {
|
||||
return true;
|
||||
}
|
||||
return /rate.?limit|overloaded|service.?unavailable|upstream.?connect|connection.?refused/i.test(errorText);
|
||||
}
|
||||
|
||||
function getRetryAfterDelayMs(headers: Headers): number | undefined {
|
||||
const retryAfterMs = headers.get("retry-after-ms");
|
||||
if (retryAfterMs !== null) {
|
||||
const millis = Number(retryAfterMs);
|
||||
if (Number.isFinite(millis)) {
|
||||
return Math.max(0, millis);
|
||||
}
|
||||
}
|
||||
|
||||
const retryAfter = headers.get("retry-after");
|
||||
if (!retryAfter) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const seconds = Number(retryAfter);
|
||||
if (Number.isFinite(seconds)) {
|
||||
return Math.max(0, seconds * 1000);
|
||||
}
|
||||
|
||||
const date = Date.parse(retryAfter);
|
||||
if (!Number.isNaN(date)) {
|
||||
return Math.max(0, date - Date.now());
|
||||
}
|
||||
|
||||
return undefined;
|
||||
}
|
||||
|
||||
function capRetryDelayMs(delayMs: number, options?: StreamOptions): number {
|
||||
const maxRetryDelayMs = options?.maxRetryDelayMs ?? DEFAULT_MAX_RETRY_DELAY_MS;
|
||||
return maxRetryDelayMs > 0 ? Math.min(delayMs, maxRetryDelayMs) : delayMs;
|
||||
}
|
||||
|
||||
function sleep(ms: number, signal?: AbortSignal): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
if (signal?.aborted) {
|
||||
@@ -256,28 +298,13 @@ export const streamOpenAICodexResponses: StreamFunction<"openai-codex-responses"
|
||||
|
||||
const errorText = await response.text();
|
||||
if (attempt < maxRetries && isRetryableError(response.status, errorText)) {
|
||||
let delayMs = BASE_DELAY_MS * 2 ** attempt;
|
||||
|
||||
const retryAfterMs = response.headers.get("retry-after-ms");
|
||||
if (retryAfterMs !== null) {
|
||||
const millis = Number(retryAfterMs);
|
||||
if (Number.isFinite(millis)) {
|
||||
delayMs = Math.max(0, millis);
|
||||
}
|
||||
} else {
|
||||
const retryAfter = response.headers.get("retry-after");
|
||||
if (retryAfter) {
|
||||
const seconds = Number(retryAfter);
|
||||
if (Number.isFinite(seconds)) {
|
||||
delayMs = Math.max(0, seconds * 1000);
|
||||
} else {
|
||||
const date = Date.parse(retryAfter);
|
||||
if (!Number.isNaN(date)) {
|
||||
delayMs = Math.max(0, date - Date.now());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
const retryAfterDelayMs = getRetryAfterDelayMs(response.headers);
|
||||
const delayMs =
|
||||
retryAfterDelayMs === undefined
|
||||
? BASE_DELAY_MS * 2 ** attempt
|
||||
: response.status === 429
|
||||
? capRetryDelayMs(retryAfterDelayMs, options)
|
||||
: retryAfterDelayMs;
|
||||
|
||||
await sleep(delayMs, options?.signal);
|
||||
continue;
|
||||
|
||||
@@ -149,7 +149,7 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions", OpenA
|
||||
const requestOptions = {
|
||||
...(options?.signal ? { signal: options.signal } : {}),
|
||||
...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
|
||||
...(options?.maxRetries !== undefined ? { maxRetries: options.maxRetries } : {}),
|
||||
maxRetries: options?.maxRetries ?? 0,
|
||||
};
|
||||
const { data: openaiStream, response } = await client.chat.completions
|
||||
.create(params, requestOptions)
|
||||
|
||||
@@ -119,7 +119,7 @@ export const streamOpenAIResponses: StreamFunction<"openai-responses", OpenAIRes
|
||||
const requestOptions = {
|
||||
...(options?.signal ? { signal: options.signal } : {}),
|
||||
...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
|
||||
...(options?.maxRetries !== undefined ? { maxRetries: options.maxRetries } : {}),
|
||||
maxRetries: options?.maxRetries ?? 0,
|
||||
};
|
||||
const { data: openaiStream, response } = await client.responses.create(params, requestOptions).withResponse();
|
||||
await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
|
||||
|
||||
@@ -1131,7 +1131,11 @@ describe("openai-codex streaming", () => {
|
||||
messages: [{ role: "user", content: "Say hello", timestamp: Date.now() }],
|
||||
};
|
||||
|
||||
const resultPromise = streamOpenAICodexResponses(model, context, { apiKey: token, transport: "sse" }).result();
|
||||
const resultPromise = streamOpenAICodexResponses(model, context, {
|
||||
apiKey: token,
|
||||
transport: "sse",
|
||||
maxRetries: 1,
|
||||
}).result();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), expectedDelay);
|
||||
|
||||
@@ -1193,7 +1197,11 @@ describe("openai-codex streaming", () => {
|
||||
messages: [{ role: "user", content: "Say hello", timestamp: Date.now() }],
|
||||
};
|
||||
|
||||
const resultPromise = streamOpenAICodexResponses(model, context, { apiKey: token, transport: "sse" }).result();
|
||||
const resultPromise = streamOpenAICodexResponses(model, context, {
|
||||
apiKey: token,
|
||||
transport: "sse",
|
||||
maxRetries: 3,
|
||||
}).result();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(setTimeoutSpy).toHaveBeenNthCalledWith(1, expect.any(Function), 1000);
|
||||
|
||||
|
||||
86
packages/ai/test/openai-completions-retry.test.ts
Normal file
86
packages/ai/test/openai-completions-retry.test.ts
Normal file
@@ -0,0 +1,86 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { streamOpenAICompletions } from "../src/providers/openai-completions.ts";
|
||||
import type { Context, Model } from "../src/types.ts";
|
||||
|
||||
const mockState = vi.hoisted(() => ({
|
||||
requestOptions: [] as unknown[],
|
||||
}));
|
||||
|
||||
vi.mock("openai", () => {
|
||||
class FakeOpenAI {
|
||||
chat = {
|
||||
completions: {
|
||||
create: (_params: unknown, options: unknown) => {
|
||||
mockState.requestOptions.push(options);
|
||||
const stream = {
|
||||
async *[Symbol.asyncIterator]() {
|
||||
yield {
|
||||
id: "chatcmpl-test",
|
||||
choices: [{ index: 0, delta: { content: "ok" } }],
|
||||
};
|
||||
yield {
|
||||
id: "chatcmpl-test",
|
||||
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
|
||||
};
|
||||
},
|
||||
};
|
||||
const promise = Promise.resolve(stream) as Promise<typeof stream> & {
|
||||
withResponse: () => Promise<{
|
||||
data: typeof stream;
|
||||
response: { status: number; headers: Headers };
|
||||
}>;
|
||||
};
|
||||
promise.withResponse = async () => ({
|
||||
data: stream,
|
||||
response: { status: 200, headers: new Headers() },
|
||||
});
|
||||
return promise;
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
return { default: FakeOpenAI };
|
||||
});
|
||||
|
||||
const model: Model<"openai-completions"> = {
|
||||
id: "test-model",
|
||||
name: "Test Model",
|
||||
api: "openai-completions",
|
||||
provider: "opencode-go",
|
||||
baseUrl: "https://opencode.ai/zen/go/v1",
|
||||
reasoning: false,
|
||||
input: ["text"],
|
||||
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
|
||||
contextWindow: 1000,
|
||||
maxTokens: 100,
|
||||
};
|
||||
|
||||
const context: Context = {
|
||||
systemPrompt: "",
|
||||
messages: [{ role: "user", content: [{ type: "text", text: "hi" }], timestamp: 0 }],
|
||||
tools: [],
|
||||
};
|
||||
|
||||
async function consume(options?: { maxRetries?: number }) {
|
||||
const stream = streamOpenAICompletions(model, context, { apiKey: "test", ...options });
|
||||
for await (const _event of stream) {
|
||||
void _event;
|
||||
}
|
||||
return stream.result();
|
||||
}
|
||||
|
||||
describe("openai-completions provider retries", () => {
|
||||
beforeEach(() => {
|
||||
mockState.requestOptions = [];
|
||||
});
|
||||
|
||||
it("disables SDK retries by default", async () => {
|
||||
await consume();
|
||||
expect(mockState.requestOptions).toEqual([expect.objectContaining({ maxRetries: 0 })]);
|
||||
});
|
||||
|
||||
it("honors explicit provider retry settings", async () => {
|
||||
await consume({ maxRetries: 2 });
|
||||
expect(mockState.requestOptions).toEqual([expect.objectContaining({ maxRetries: 2 })]);
|
||||
});
|
||||
});
|
||||
@@ -101,11 +101,13 @@ Set `PI_SKIP_VERSION_CHECK=1` to disable the Pi version update check. Use `--off
|
||||
| `retry.maxRetries` | number | `3` | Maximum agent-level retry attempts |
|
||||
| `retry.baseDelayMs` | number | `2000` | Base delay for agent-level exponential backoff (2s, 4s, 8s) |
|
||||
| `retry.provider.timeoutMs` | number | SDK default | Provider/SDK request timeout in milliseconds |
|
||||
| `retry.provider.maxRetries` | number | SDK default | Provider/SDK retry attempts |
|
||||
| `retry.provider.maxRetries` | number | `0` | Provider/SDK retry attempts |
|
||||
| `retry.provider.maxRetryDelayMs` | number | `60000` | Max server-requested delay before failing (60s) |
|
||||
|
||||
When a provider requests a retry delay longer than `retry.provider.maxRetryDelayMs` (e.g., Google's "quota will reset after 5h"), the request fails immediately with an informative error instead of waiting silently. Set to `0` to disable the cap.
|
||||
|
||||
Keep `retry.provider.maxRetries` at `0` unless provider-level retries are explicitly needed. Setting it above `0` can make SDK/provider retries handle out-of-usage-limit errors before Pi sees them, which may block the agent until the provider quota resets in some circumstances.
|
||||
|
||||
```json
|
||||
{
|
||||
"retry": {
|
||||
|
||||
@@ -2422,6 +2422,12 @@ export class AgentSession {
|
||||
// Auto-Retry
|
||||
// =========================================================================
|
||||
|
||||
private _isNonRetryableProviderLimitError(errorMessage: string): boolean {
|
||||
return /GoUsageLimitError|FreeUsageLimitError|Monthly usage limit reached|available balance|insufficient_quota|out of budget|quota exceeded|billing/i.test(
|
||||
errorMessage,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if an error is retryable (overloaded, rate limit, server errors).
|
||||
* Context overflow errors are NOT retryable (handled by compaction instead).
|
||||
@@ -2434,6 +2440,7 @@ export class AgentSession {
|
||||
if (isContextOverflow(message, contextWindow)) return false;
|
||||
|
||||
const err = message.errorMessage;
|
||||
if (this._isNonRetryableProviderLimitError(err)) return false;
|
||||
// Match: overloaded_error, provider returned error, rate limit, 429, 500, 502, 503, 504, service unavailable, network/connection errors (including connection lost), WebSocket transport closes/errors, fetch failed, premature stream endings, HTTP/2 closed before response, terminated, retry delay exceeded
|
||||
return /overloaded|provider.?returned.?error|rate.?limit|too many requests|429|500|502|503|504|service.?unavailable|server.?error|internal.?error|network.?error|connection.?error|connection.?refused|connection.?lost|websocket.?closed|websocket.?error|other side closed|fetch failed|upstream.?connect|reset before headers|socket hang up|ended without|stream ended before message_stop|http2 request did not get a response|timed? out|timeout|terminated|retry delay/i.test(
|
||||
err,
|
||||
|
||||
Reference in New Issue
Block a user