From 6eda70aadabe21ea0e63f698d9d47bb3bf39d372 Mon Sep 17 00:00:00 2001 From: can1357 Date: Wed, 13 May 2026 16:40:58 +0200 Subject: [PATCH] refactor: replaced abortableSleep with scheduler.wait and fetchWithRetry - Removed local `abortableSleep` in favour of Node's built-in `scheduler.wait` from `node:timers/promises`. - Consolidated per-provider retry/fetch loops into a shared `fetchWithRetry` utility in `packages/utils`. - Moved `extractHttpStatusFromError`, `isRetryableError`, and related helpers out of `packages/ai` into `packages/utils`. - Deleted `extractRetryDelay` in favour of `extractRetryHint` with unified header and body parsing. --- docs/tools/github.md | 2 +- packages/ai/src/provider-models/ollama.ts | 14 +- packages/ai/src/providers/anthropic.ts | 21 +- .../ai/src/providers/google-gemini-cli.ts | 145 ++------- .../ai/src/providers/google-gemini-headers.ts | 100 ------ packages/ai/src/providers/ollama.ts | 15 +- .../src/providers/openai-codex-responses.ts | 105 ++----- .../ai/src/providers/openai-completions.ts | 4 +- packages/ai/src/utils/http-inspector.ts | 4 +- packages/ai/src/utils/oauth/github-copilot.ts | 16 +- packages/ai/src/utils/oauth/kimi.ts | 7 +- packages/ai/src/utils/retry.ts | 138 +-------- packages/ai/test/copilot-retry.test.ts | 5 +- .../ai/test/google-gemini-cli-429.test.ts | 94 +++--- .../test/google-gemini-cli-alignment.test.ts | 2 +- .../coding-agent/src/session/agent-session.ts | 18 +- packages/coding-agent/src/tools/gh.ts | 13 +- .../src/web/search/providers/gemini.ts | 128 ++------ packages/utils/src/abortable.ts | 13 - packages/utils/src/fetch-retry.ts | 290 ++++++++++++++++++ packages/utils/src/index.ts | 3 +- 21 files changed, 470 insertions(+), 667 deletions(-) create mode 100644 packages/utils/src/fetch-retry.ts diff --git a/docs/tools/github.md b/docs/tools/github.md index 268cbfc6f..7cf753f4b 100644 --- a/docs/tools/github.md +++ b/docs/tools/github.md @@ -227,7 +227,7 @@ Watch flow: - `gh` interactive editor fallback is suppressed for `pr_create` by forcing either `--body-file` or `--body ""`. - `gh-renderer` provides compact headers for all ops and a custom live watch view for `run_watch`. - Background work / cancellation - - `run_watch` loops until success/failure and uses `abortableSleep()` between polls. + - `run_watch` loops until success/failure and uses `scheduler.wait()` between polls. - `GithubTool.execute()` is wrapped in `untilAborted()`; `git.github.run()` forwards the abort signal into `Bun.spawn()`. ## Limits & Caps diff --git a/packages/ai/src/provider-models/ollama.ts b/packages/ai/src/provider-models/ollama.ts index 1f3fa4586..c71fbf617 100644 --- a/packages/ai/src/provider-models/ollama.ts +++ b/packages/ai/src/provider-models/ollama.ts @@ -1,4 +1,4 @@ -import { abortableSleep } from "@oh-my-pi/pi-utils"; +import { fetchWithRetry } from "@oh-my-pi/pi-utils"; import type { ModelManagerOptions } from "../model-manager"; import { Effort } from "../model-thinking"; import type { ThinkingConfig } from "../types"; @@ -19,16 +19,7 @@ type OllamaShowResponse = { model_info?: Record; }; -const MODEL_RETRY_DELAYS_MS = [2_000, 5_000, 10_000]; - -async function fetchWithRetry(url: string, init: RequestInit): Promise { - for (let attempt = 0; attempt < MODEL_RETRY_DELAYS_MS.length; attempt++) { - const response = await fetch(url, init); - if (response.ok || response.status < 500) return response; - await abortableSleep(MODEL_RETRY_DELAYS_MS[attempt]!); - } - return fetch(url, init); -} +const OLLAMA_RETRY_DELAYS_MS = [2_000, 5_000, 10_000]; function trimTrailingSlash(value: string): string { return value.endsWith("/") ? value.slice(0, -1) : value; @@ -109,6 +100,7 @@ export function ollamaCloudModelManagerOptions( const response = await fetchWithRetry(`${baseUrl}/api/tags`, { method: "GET", headers: createCloudHeaders(apiKey), + defaultDelayMs: OLLAMA_RETRY_DELAYS_MS, }); if (!response.ok) { throw new Error(`HTTP ${response.status} from ${baseUrl}/api/tags`); diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index 5034c32a7..58d56b1a1 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/packages/ai/src/providers/anthropic.ts @@ -1,5 +1,6 @@ import * as nodeCrypto from "node:crypto"; import * as fs from "node:fs"; +import { scheduler } from "node:timers/promises"; import * as tls from "node:tls"; import Anthropic, { type ClientOptions as AnthropicSdkClientOptions } from "@anthropic-ai/sdk"; import type { @@ -8,7 +9,14 @@ import type { MessageParam, RawMessageStreamEvent, } from "@anthropic-ai/sdk/resources/messages"; -import { $env, abortableSleep, isEnoent, readSseEvents } from "@oh-my-pi/pi-utils"; +import { + $env, + extractHttpStatusFromError, + isEnoent, + isRetryableError, + isUnexpectedSocketCloseMessage, + readSseEvents, +} from "@oh-my-pi/pi-utils"; import { hasOpus47ApiRestrictions, mapEffortToAnthropicAdaptiveEffort } from "../model-thinking"; import { calculateCost } from "../models"; import { getEnvApiKey, OUTPUT_FALLBACK_BUFFER } from "../stream"; @@ -48,12 +56,7 @@ import { getStreamFirstEventTimeoutMs, getStreamIdleTimeoutMs, iterateWithIdleTi import { parseJsonWithRepair, parseStreamingJson } from "../utils/json-parse"; import { parseGitHubCopilotApiKey } from "../utils/oauth/github-copilot"; import { notifyProviderResponse } from "../utils/provider-response"; -import { - extractHttpStatusFromError, - isCopilotRetryableError, - isRetryableError, - isUnexpectedSocketCloseMessage, -} from "../utils/retry"; +import { isCopilotTransientModelError } from "../utils/retry"; import { COMBINATOR_KEYS, NO_STRICT } from "../utils/schema"; import { notifyRawSseEvent, wrapFetchForSseDebug } from "../utils/sse-debug"; import { @@ -844,7 +847,7 @@ function isProviderRetryableStreamEnvelopeError(error: unknown): boolean { export function isProviderRetryableError(error: unknown, provider?: string): boolean { if (!(error instanceof Error)) return false; - if (provider === "github-copilot" && isCopilotRetryableError(error)) return true; + if (provider === "github-copilot" && isCopilotTransientModelError(error)) return true; const msg = error.message.toLowerCase(); if ( isUnexpectedSocketCloseMessage(msg) || @@ -1287,7 +1290,7 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = ( } providerRetryAttempt++; const delayMs = PROVIDER_BASE_DELAY_MS * 2 ** (providerRetryAttempt - 1); - await abortableSleep(delayMs, options?.signal); + await scheduler.wait(delayMs, { signal: options?.signal }); output.content.length = 0; output.responseId = undefined; output.errorMessage = strictFallbackErrorMessage; diff --git a/packages/ai/src/providers/google-gemini-cli.ts b/packages/ai/src/providers/google-gemini-cli.ts index 78fb939a6..d2de661ef 100644 --- a/packages/ai/src/providers/google-gemini-cli.ts +++ b/packages/ai/src/providers/google-gemini-cli.ts @@ -4,8 +4,9 @@ * Uses the Cloud Code Assist API endpoint to access Gemini and Claude models. */ import { createHash, randomBytes, randomUUID } from "node:crypto"; +import { scheduler } from "node:timers/promises"; import type { Content, FunctionCallingConfigMode, ThinkingConfig } from "@google/genai"; -import { abortableSleep, readSseJson } from "@oh-my-pi/pi-utils"; +import { fetchWithRetry, readSseJson } from "@oh-my-pi/pi-utils"; import { calculateCost } from "../models"; import type { Api, @@ -23,14 +24,8 @@ import { AssistantMessageEventStream } from "../utils/event-stream"; import { appendRawHttpRequestDumpFor400, type RawHttpRequestDump, withHttpStatus } from "../utils/http-inspector"; import { refreshAntigravityToken } from "../utils/oauth/google-antigravity"; import { refreshGoogleCloudToken } from "../utils/oauth/google-gemini-cli"; -import { extractHttpStatusFromError } from "../utils/retry"; import { sanitizeSchemaForCCA } from "../utils/schema"; -import { - ANTIGRAVITY_SYSTEM_INSTRUCTION, - extractRetryDelay, - getAntigravityUserAgent, - getGeminiCliHeaders, -} from "./google-gemini-headers"; +import { ANTIGRAVITY_SYSTEM_INSTRUCTION, getAntigravityUserAgent, getGeminiCliHeaders } from "./google-gemini-headers"; import { convertMessages, convertTools, @@ -72,7 +67,6 @@ const ANTIGRAVITY_ENDPOINT_FALLBACKS = [ANTIGRAVITY_DAILY_ENDPOINT, ANTIGRAVITY_ export { ANTIGRAVITY_SYSTEM_INSTRUCTION, - extractRetryDelay, getAntigravityUserAgent, getGeminiCliHeaders, getGeminiCliUserAgent, @@ -104,16 +98,6 @@ function shouldInjectAntigravitySystemInstruction(modelId: string): boolean { return normalized.includes("claude") || normalized.includes("gemini-3-pro-high"); } -/** - * Check if an error is retryable (rate limit, server error, network error, etc.) - */ -function isRetryableError(status: number, errorText: string): boolean { - if (status === 429 || status === 500 || status === 502 || status === 503 || status === 504) { - return true; - } - return /resource.?exhausted|rate.?limit|overloaded|service.?unavailable|other.?side.?closed/i.test(errorText); -} - /** * Extract a clean, user-friendly error message from Google API error response. * Parses JSON error responses and returns just the message field. @@ -366,109 +350,26 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( headers: requestHeaders, }; - // Fetch with retry logic for rate limits and transient errors - let response: Response | undefined; - let lastError: Error | undefined; - let requestUrl: string | undefined; - let rateLimitTimeSpent = 0; - - for (let attempt = 0; ; attempt++) { - if (options?.signal?.aborted) { - throw new Error("Request was aborted"); - } - - try { - const endpoint = endpoints[Math.min(attempt, endpoints.length - 1)]; - requestUrl = `${endpoint}/v1internal:streamGenerateContent?alt=sse`; - response = await fetch(requestUrl, { - method: "POST", - headers: requestHeaders, - body: requestBodyJson, - signal: options?.signal, - }); - - if (response.ok) { - break; // Success, exit retry loop - } - - const errorText = await response.text(); - - // Handle 429 rate limits with time budget - if (response.status === 429) { - if (/quota|exhausted/i.test(errorText)) { - throw withHttpStatus( - new Error(`Cloud Code Assist API error (429): ${extractErrorMessage(errorText)}`), - 429, - ); - } - const serverDelay = extractRetryDelay(errorText, response); - if (serverDelay && rateLimitTimeSpent + serverDelay <= RATE_LIMIT_BUDGET_MS) { - rateLimitTimeSpent += serverDelay; - await abortableSleep(serverDelay, options?.signal); - continue; - } - // Fallback: use exponential backoff if no server delay, up to MAX_RETRIES - if (!serverDelay && attempt < MAX_RETRIES) { - await abortableSleep(BASE_DELAY_MS * 2 ** attempt, options?.signal); - continue; - } - } else if (attempt < MAX_RETRIES && isRetryableError(response.status, errorText)) { - // Non-429 retryable errors use standard attempt cap - const serverDelay = extractRetryDelay(errorText, response); - const delayMs = serverDelay ?? BASE_DELAY_MS * 2 ** attempt; - - // Check if server delay exceeds max allowed (default: 60s) for non-429 errors - const maxDelayMs = options?.maxRetryDelayMs ?? 60000; - if (maxDelayMs > 0 && serverDelay && serverDelay > maxDelayMs) { - const delaySeconds = Math.ceil(serverDelay / 1000); - throw withHttpStatus( - new Error( - `Server requested ${delaySeconds}s retry delay (max: ${Math.ceil(maxDelayMs / 1000)}s). ${extractErrorMessage(errorText)}`, - ), - response.status, - ); - } - - await abortableSleep(delayMs, options?.signal); - continue; - } - - // Not retryable or budget exceeded - throw withHttpStatus( - new Error(`Cloud Code Assist API error (${response.status}): ${extractErrorMessage(errorText)}`), - response.status, - ); - } catch (error) { - // Check for abort - fetch throws AbortError, our code throws "Request was aborted" - if (error instanceof Error) { - if (error.name === "AbortError" || error.message === "Request was aborted") { - throw new Error("Request was aborted"); - } - } - - // HTTP responses are handled inside the try block. - // If we intentionally throw with status metadata, don't convert it into a network retry. - if (extractHttpStatusFromError(error) !== undefined) { - throw error; - } - // Extract detailed error message from fetch errors (Node includes cause) - lastError = error instanceof Error ? error : new Error(String(error)); - if (lastError.message === "fetch failed" && lastError.cause instanceof Error) { - lastError = new Error(`Network error: ${lastError.cause.message}`); - } - // Network errors are retryable - if (attempt < MAX_RETRIES) { - const delayMs = BASE_DELAY_MS * 2 ** attempt; - await abortableSleep(delayMs, options?.signal); - continue; - } - throw lastError; - } - } - - if (!response?.ok) { - throw lastError ?? new Error("Failed to get response after retries"); + const response = await fetchWithRetry( + attempt => `${endpoints[Math.min(attempt, endpoints.length - 1)]}/v1internal:streamGenerateContent?alt=sse`, + { + method: "POST", + headers: requestHeaders, + body: requestBodyJson, + signal: options?.signal, + maxAttempts: MAX_RETRIES + 1, + defaultDelayMs: attempt => BASE_DELAY_MS * 2 ** attempt, + maxDelayMs: options?.maxRetryDelayMs ?? RATE_LIMIT_BUDGET_MS, + }, + ); + if (!response.ok) { + const errorText = await response.text(); + throw withHttpStatus( + new Error(`Cloud Code Assist API error (${response.status}): ${extractErrorMessage(errorText)}`), + response.status, + ); } + const requestUrl = response.url; let started = false; const ensureStarted = () => { @@ -702,7 +603,7 @@ export const streamGoogleGeminiCli: StreamFunction<"google-gemini-cli"> = ( if (emptyAttempt > 0) { const backoffMs = EMPTY_STREAM_BASE_DELAY_MS * 2 ** (emptyAttempt - 1); try { - await abortableSleep(backoffMs, options?.signal); + await scheduler.wait(backoffMs, { signal: options?.signal }); } catch { // Normalize AbortError to expected message for consistent error handling throw new Error("Request was aborted"); diff --git a/packages/ai/src/providers/google-gemini-headers.ts b/packages/ai/src/providers/google-gemini-headers.ts index 73b5fbb28..64f9e3177 100644 --- a/packages/ai/src/providers/google-gemini-headers.ts +++ b/packages/ai/src/providers/google-gemini-headers.ts @@ -39,103 +39,3 @@ export let getAntigravityUserAgent = () => { getAntigravityUserAgent = () => userAgent; return userAgent; }; - -/** - * Extract retry delay from Gemini error response (in milliseconds). - * Checks headers first (Retry-After, x-ratelimit-reset, x-ratelimit-reset-after), - * then parses body patterns like: - * - "Your quota will reset after 39s" - * - "Your quota will reset after 18h31m10s" - * - "Please retry in Xs" or "Please retry in Xms" - * - "retryDelay": "34.074824224s" (JSON field) - */ -export function extractRetryDelay(errorText: string, response?: Response | Headers): number | undefined { - const normalizeDelay = (ms: number): number | undefined => (ms > 0 ? Math.ceil(ms + 1000) : undefined); - - const headers = response instanceof Headers ? response : response?.headers; - if (headers) { - const retryAfter = headers.get("retry-after"); - if (retryAfter) { - const retryAfterSeconds = Number(retryAfter); - if (Number.isFinite(retryAfterSeconds)) { - const delay = normalizeDelay(retryAfterSeconds * 1000); - if (delay !== undefined) { - return delay; - } - } - const retryAfterDate = new Date(retryAfter); - const retryAfterMs = retryAfterDate.getTime(); - if (!Number.isNaN(retryAfterMs)) { - const delay = normalizeDelay(retryAfterMs - Date.now()); - if (delay !== undefined) { - return delay; - } - } - } - - const rateLimitReset = headers.get("x-ratelimit-reset"); - if (rateLimitReset) { - const resetSeconds = Number.parseInt(rateLimitReset, 10); - if (!Number.isNaN(resetSeconds)) { - const delay = normalizeDelay(resetSeconds * 1000 - Date.now()); - if (delay !== undefined) { - return delay; - } - } - } - - const rateLimitResetAfter = headers.get("x-ratelimit-reset-after"); - if (rateLimitResetAfter) { - const resetAfterSeconds = Number(rateLimitResetAfter); - if (Number.isFinite(resetAfterSeconds)) { - const delay = normalizeDelay(resetAfterSeconds * 1000); - if (delay !== undefined) { - return delay; - } - } - } - } - - // Pattern 1: "Your quota will reset after ..." (formats: "18h31m10s", "10m15s", "6s", "39s") - const durationMatch = errorText.match(/reset after (?:(\d+)h)?(?:(\d+)m)?(\d+(?:\.\d+)?)s/i); - if (durationMatch) { - const hours = durationMatch[1] ? parseInt(durationMatch[1], 10) : 0; - const minutes = durationMatch[2] ? parseInt(durationMatch[2], 10) : 0; - const seconds = parseFloat(durationMatch[3]); - if (!Number.isNaN(seconds)) { - const totalMs = ((hours * 60 + minutes) * 60 + seconds) * 1000; - const delay = normalizeDelay(totalMs); - if (delay !== undefined) { - return delay; - } - } - } - - // Pattern 2: "Please retry in X[ms|s]" - const retryInMatch = errorText.match(/Please retry in ([0-9.]+)(ms|s)/i); - if (retryInMatch?.[1]) { - const value = parseFloat(retryInMatch[1]); - if (!Number.isNaN(value) && value > 0) { - const ms = retryInMatch[2].toLowerCase() === "ms" ? value : value * 1000; - const delay = normalizeDelay(ms); - if (delay !== undefined) { - return delay; - } - } - } - - // Pattern 3: "retryDelay": "34.074824224s" (JSON field in error details) - const retryDelayMatch = errorText.match(/"retryDelay":\s*"([0-9.]+)(ms|s)"/i); - if (retryDelayMatch?.[1]) { - const value = parseFloat(retryDelayMatch[1]); - if (!Number.isNaN(value) && value > 0) { - const ms = retryDelayMatch[2].toLowerCase() === "ms" ? value : value * 1000; - const delay = normalizeDelay(ms); - if (delay !== undefined) { - return delay; - } - } - } - - return undefined; -} diff --git a/packages/ai/src/providers/ollama.ts b/packages/ai/src/providers/ollama.ts index 22186b0cf..b4f987f53 100644 --- a/packages/ai/src/providers/ollama.ts +++ b/packages/ai/src/providers/ollama.ts @@ -1,4 +1,4 @@ -import { abortableSleep } from "@oh-my-pi/pi-utils"; +import { fetchWithRetry } from "@oh-my-pi/pi-utils"; import type { TSchema } from "@sinclair/typebox"; import { getEnvApiKey } from "../stream"; import type { @@ -334,16 +334,6 @@ function mapDoneReason(doneReason: string | undefined, output: AssistantMessage) const OLLAMA_RETRY_DELAYS_MS = [2_000, 5_000, 10_000]; -async function fetchChatWithRetry(url: string, init: RequestInit): Promise { - const signal = init.signal as AbortSignal | undefined; - for (let attempt = 0; attempt < OLLAMA_RETRY_DELAYS_MS.length; attempt++) { - const response = await fetch(url, init); - if (response.ok || response.status < 500) return response; - await abortableSleep(OLLAMA_RETRY_DELAYS_MS[attempt]!, signal); - } - return fetch(url, init); -} - export const streamOllama: StreamFunction<"ollama-chat"> = ( model: Model<"ollama-chat">, context: Context, @@ -377,7 +367,7 @@ export const streamOllama: StreamFunction<"ollama-chat"> = ( url: `${baseUrl}/api/chat`, body, }; - const response = await fetchChatWithRetry(`${baseUrl}/api/chat`, { + const response = await fetchWithRetry(`${baseUrl}/api/chat`, { method: "POST", headers: { ...model.headers, @@ -387,6 +377,7 @@ export const streamOllama: StreamFunction<"ollama-chat"> = ( }, body: JSON.stringify(body), signal: options.signal, + defaultDelayMs: OLLAMA_RETRY_DELAYS_MS, }); if (!response.ok) { throw new Error(`HTTP ${response.status} from ${baseUrl}/api/chat`); diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index f56eaa2a3..3d4b44107 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/packages/ai/src/providers/openai-codex-responses.ts @@ -1,5 +1,6 @@ import * as os from "node:os"; -import { $env, $flag, abortableSleep, asRecord, logger, readSseJson, structuredCloneJSON } from "@oh-my-pi/pi-utils"; +import { scheduler } from "node:timers/promises"; +import { $env, $flag, asRecord, fetchWithRetry, logger, readSseJson, structuredCloneJSON } from "@oh-my-pi/pi-utils"; import type OpenAI from "openai"; import type { ResponseCustomToolCall, @@ -13,7 +14,6 @@ import type { } from "openai/resources/responses/responses"; import packageJson from "../../package.json" with { type: "json" }; import { calculateCost } from "../models"; -import { isUsageLimitError } from "../rate-limit-utils"; import { getEnvApiKey } from "../stream"; import { type Api, @@ -76,7 +76,6 @@ export interface OpenAICodexResponsesOptions extends StreamOptions { const CODEX_DEBUG = $flag("PI_CODEX_DEBUG"); const CODEX_MAX_RETRIES = 5; -const CODEX_RETRYABLE_STATUS = new Set([408, 429, 500, 502, 503, 504]); const CODEX_RETRY_DELAY_MS = 500; const CODEX_WEBSOCKET_CONNECT_TIMEOUT_MS = 10000; const CODEX_WEBSOCKET_IDLE_TIMEOUT_MS = 300000; @@ -661,7 +660,9 @@ async function openInitialCodexEventStream( }); if (!activateFallback) { websocketRetries += 1; - await abortableSleep(getCodexWebSocketRetryDelayMs(websocketRetries), requestSetup.requestSignal); + await scheduler.wait(getCodexWebSocketRetryDelayMs(websocketRetries), { + signal: requestSetup.requestSignal, + }); continue; } break; @@ -1407,10 +1408,9 @@ async function tryReplayWebsocketFailureOverSse( if (!activateFallback) { runtime.websocketStreamRetries += 1; - await abortableSleep( - getCodexWebSocketRetryDelayMs(runtime.websocketStreamRetries), - context.requestSetup.requestSignal, - ); + await scheduler.wait(getCodexWebSocketRetryDelayMs(runtime.websocketStreamRetries), { + signal: context.requestSetup.requestSignal, + }); await reopenCodexWebSocketRuntimeStream(context, runtime, state); return true; } @@ -1461,7 +1461,9 @@ async function tryRetryCodexProviderError( runtime.sawTerminalEvent = false; resetOutputState(context.output); context.firstTokenTime = undefined; - await abortableSleep(CODEX_RETRY_DELAY_MS * runtime.providerRetryAttempt, context.requestSetup.requestSignal); + await scheduler.wait(CODEX_RETRY_DELAY_MS * runtime.providerRetryAttempt, { + signal: context.requestSetup.requestSignal, + }); if (runtime.transport === "websocket" && websocketState) { await reopenCodexWebSocketRuntimeStream(context, runtime, websocketState); @@ -2194,15 +2196,15 @@ async function openCodexSseEventStream( sentTurnStateHeader: headers.has(X_CODEX_TURN_STATE_HEADER), sentModelsEtagHeader: headers.has(X_MODELS_ETAG_HEADER), }); - const response = await fetchWithRetry( - url, - { - method: "POST", - headers, - body: JSON.stringify(body), - }, + const response = await fetchWithRetry(url, { + method: "POST", + headers, + body: JSON.stringify(body), signal, - ); + maxAttempts: CODEX_MAX_RETRIES + 1, + defaultDelayMs: attempt => CODEX_RETRY_DELAY_MS * (attempt + 1), + maxDelayMs: CODEX_RATE_LIMIT_BUDGET_MS, + }); logCodexDebug("codex response", { url: response.url, status: response.status, @@ -2292,75 +2294,6 @@ function logCodexDebug(message: string, details?: Record): void logger.debug(`[codex] ${message}`, details ?? {}); } -function getRetryDelayMs( - response: Response | null, - attempt: number, - errorBody?: string, -): { delay: number; serverProvided: boolean } { - const retryAfter = response?.headers?.get("retry-after") || null; - if (retryAfter) { - const seconds = Number(retryAfter); - if (Number.isFinite(seconds)) { - return { delay: Math.max(0, seconds * 1000), serverProvided: true }; - } - const parsedDate = Date.parse(retryAfter); - if (!Number.isNaN(parsedDate)) { - return { delay: Math.max(0, parsedDate - Date.now()), serverProvided: true }; - } - } - if (errorBody) { - const msMatch = /try again in\s+(\d+(?:\.\d+)?)\s*ms/i.exec(errorBody); - if (msMatch) { - const ms = Number(msMatch[1]); - if (Number.isFinite(ms)) return { delay: Math.max(ms, 100), serverProvided: true }; - } - const sMatch = /try again in\s+(\d+(?:\.\d+)?)\s*s(?:ec)?/i.exec(errorBody); - if (sMatch) { - const seconds = Number(sMatch[1]); - if (Number.isFinite(seconds)) return { delay: Math.max(seconds * 1000, 100), serverProvided: true }; - } - } - return { delay: CODEX_RETRY_DELAY_MS * (attempt + 1), serverProvided: false }; -} - -async function fetchWithRetry(url: string, init: RequestInit, signal?: AbortSignal): Promise { - let attempt = 0; - let rateLimitTimeSpent = 0; - while (true) { - try { - const response = await fetch(url, { ...init, signal: signal ?? init.signal }); - if (!CODEX_RETRYABLE_STATUS.has(response.status)) { - return response; - } - if (signal?.aborted) return response; - const errorBody = await response.clone().text(); - // Usage-limit errors are persistent (account allocation exhausted) — retrying with the - // same credential is futile. Bail out immediately so the error propagates to the agent - // session layer where credential switching happens. - if (response.status === 429 && isUsageLimitError(errorBody)) { - return response; - } - const { delay, serverProvided } = getRetryDelayMs(response, attempt, errorBody); - if (response.status === 429 && serverProvided) { - if (rateLimitTimeSpent + delay > CODEX_RATE_LIMIT_BUDGET_MS) { - return response; - } - rateLimitTimeSpent += delay; - } else if (attempt >= CODEX_MAX_RETRIES) { - return response; - } - await abortableSleep(delay, signal); - } catch (error) { - if (attempt >= CODEX_MAX_RETRIES || signal?.aborted) { - throw error; - } - const delay = CODEX_RETRY_DELAY_MS * (attempt + 1); - await abortableSleep(delay, signal); - } - attempt += 1; - } -} - function redactHeaders(headers: Headers): Record { const redacted: Record = {}; for (const [key, value] of headers.entries()) { diff --git a/packages/ai/src/providers/openai-completions.ts b/packages/ai/src/providers/openai-completions.ts index c4d1778d4..e41eae2df 100644 --- a/packages/ai/src/providers/openai-completions.ts +++ b/packages/ai/src/providers/openai-completions.ts @@ -1,4 +1,4 @@ -import { $env } from "@oh-my-pi/pi-utils"; +import { $env, extractHttpStatusFromError } from "@oh-my-pi/pi-utils"; import OpenAI from "openai"; import type { ChatCompletionAssistantMessageParam, @@ -54,7 +54,7 @@ import { parseStreamingJson } from "../utils/json-parse"; import { parseGitHubCopilotApiKey } from "../utils/oauth/github-copilot"; import { getKimiCommonHeaders } from "../utils/oauth/kimi"; import { notifyProviderResponse } from "../utils/provider-response"; -import { callWithCopilotModelRetry, extractHttpStatusFromError } from "../utils/retry"; +import { callWithCopilotModelRetry } from "../utils/retry"; import { adaptSchemaForStrict, NO_STRICT } from "../utils/schema"; import { wrapFetchForSseDebug } from "../utils/sse-debug"; import { isForcedToolChoice, mapToOpenAICompletionsToolChoice } from "../utils/tool-choice"; diff --git a/packages/ai/src/utils/http-inspector.ts b/packages/ai/src/utils/http-inspector.ts index 525c6a3a7..d08db8a42 100644 --- a/packages/ai/src/utils/http-inspector.ts +++ b/packages/ai/src/utils/http-inspector.ts @@ -1,6 +1,6 @@ import * as path from "node:path"; -import { getLogsDir } from "@oh-my-pi/pi-utils"; -import { extractHttpStatusFromError, isCopilotTransientModelError } from "./retry.js"; +import { extractHttpStatusFromError, getLogsDir } from "@oh-my-pi/pi-utils"; +import { isCopilotTransientModelError } from "./retry.js"; import { formatErrorMessageWithRetryAfter } from "./retry-after.js"; export type RawHttpRequestDump = { diff --git a/packages/ai/src/utils/oauth/github-copilot.ts b/packages/ai/src/utils/oauth/github-copilot.ts index f0d226a90..7430846ff 100644 --- a/packages/ai/src/utils/oauth/github-copilot.ts +++ b/packages/ai/src/utils/oauth/github-copilot.ts @@ -1,7 +1,7 @@ /** * GitHub Copilot OAuth flow (opencode OAuth app) */ -import { abortableSleep } from "@oh-my-pi/pi-utils"; +import { scheduler } from "node:timers/promises"; import { getBundledModels } from "../../models"; import type { OAuthCredentials } from "./types"; @@ -159,14 +159,6 @@ async function startDeviceFlow(domain: string): Promise { }; } -async function sleepForGitHubAccessTokenPoll(ms: number, signal?: AbortSignal): Promise { - try { - await abortableSleep(ms, signal); - } catch { - throw new Error("Login cancelled"); - } -} - async function pollForGitHubAccessToken( domain: string, deviceCode: string, @@ -187,7 +179,11 @@ async function pollForGitHubAccessToken( const remainingMs = deadline - Date.now(); const waitMs = Math.min(Math.ceil(intervalMs * intervalMultiplier), remainingMs); - await sleepForGitHubAccessTokenPoll(waitMs, signal); + try { + await scheduler.wait(waitMs, { signal }); + } catch { + throw new Error("Login cancelled"); + } const raw = await fetchJson(urls.accessTokenUrl, { method: "POST", diff --git a/packages/ai/src/utils/oauth/kimi.ts b/packages/ai/src/utils/oauth/kimi.ts index c1cc4ee44..34caa7df4 100644 --- a/packages/ai/src/utils/oauth/kimi.ts +++ b/packages/ai/src/utils/oauth/kimi.ts @@ -6,7 +6,8 @@ import * as crypto from "node:crypto"; import * as fs from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; -import { $env, abortableSleep, getAgentDir, isEnoent } from "@oh-my-pi/pi-utils"; +import { scheduler } from "node:timers/promises"; +import { $env, getAgentDir, isEnoent } from "@oh-my-pi/pi-utils"; import packageJson from "../../../package.json" with { type: "json" }; import type { OAuthController, OAuthCredentials } from "./types"; @@ -185,7 +186,7 @@ async function pollForToken( const error = payload.error; if (error === "authorization_pending") { - await abortableSleep(waitMs, signal); + await scheduler.wait(waitMs, { signal }); continue; } @@ -193,7 +194,7 @@ async function pollForToken( waitMs += 5000; const retryAfter = typeof payload.interval === "number" ? payload.interval * 1000 : undefined; if (retryAfter && retryAfter > waitMs) waitMs = retryAfter; - await abortableSleep(waitMs, signal); + await scheduler.wait(waitMs, { signal }); continue; } diff --git a/packages/ai/src/utils/retry.ts b/packages/ai/src/utils/retry.ts index ec7fdc59a..1a04263f6 100644 --- a/packages/ai/src/utils/retry.ts +++ b/packages/ai/src/utils/retry.ts @@ -1,104 +1,5 @@ -import { abortableSleep } from "@oh-my-pi/pi-utils"; - -type ErrorLike = { - message?: string; - name?: string; - status?: number; - statusCode?: number; - response?: { status?: number }; - cause?: unknown; - code?: unknown; - error?: { code?: unknown } | null; -}; - -export function isUnexpectedSocketCloseMessage(message: string): boolean { - return /\b(?:the\s+)?socket connection (?:was )?closed unexpectedly\b/i.test(message); -} - -const TRANSIENT_MESSAGE_PATTERN = - /overloaded|rate.?limit|too many requests|service.?unavailable|server error|internal error|connection.?error|unable to connect|fetch failed|stream stall/i; - -const VALIDATION_MESSAGE_PATTERN = - /invalid|validation|bad request|unsupported|schema|missing required|not found|unauthorized|forbidden/i; - -/** - * Identify errors that should be retried (timeouts, 5xx, 408, 429, transient network failures). - */ -export function isRetryableError(error: unknown): boolean { - const info = error as ErrorLike | null; - const message = info?.message ?? ""; - const name = info?.name ?? ""; - if (name === "AbortError" || /timeout|timed out|aborted/i.test(message)) return true; - - const status = extractHttpStatusFromError(error); - if (status !== undefined) { - if (status >= 500) return true; - if (status === 408 || status === 429) return true; - if (status >= 400 && status < 500) return false; - } - - if (VALIDATION_MESSAGE_PATTERN.test(message)) return false; - - return isUnexpectedSocketCloseMessage(message) || TRANSIENT_MESSAGE_PATTERN.test(message); -} - -export function extractHttpStatusFromError(error: unknown): number | undefined { - return extractHttpStatusFromErrorInternal(error, 0); -} - -function extractHttpStatusFromErrorInternal(error: unknown, depth: number): number | undefined { - if (!error || typeof error !== "object" || depth > 2) return undefined; - const info = error as ErrorLike; - const rawStatus = - info.status ?? - info.statusCode ?? - (info.response && typeof info.response === "object" ? info.response.status : undefined); - - let status: number | undefined; - if (typeof rawStatus === "number" && Number.isFinite(rawStatus)) { - status = rawStatus; - } else if (typeof rawStatus === "string") { - const parsed = Number(rawStatus); - if (Number.isFinite(parsed)) { - status = parsed; - } - } - - if (status !== undefined && status >= 100 && status <= 599) { - return status; - } - - if (info.message) { - const extracted = extractStatusFromMessage(info.message); - if (extracted !== undefined) return extracted; - } - - if (info.cause) { - return extractHttpStatusFromErrorInternal(info.cause, depth + 1); - } - - return undefined; -} - -function extractStatusFromMessage(message: string): number | undefined { - const patterns = [ - /error\s*\((\d{3})\)/i, - /status\s*[:=]?\s*(\d{3})/i, - /\bhttp\s*(\d{3})\b/i, - /\b(\d{3})\s*(?:status|error)\b/i, - ]; - - for (const pattern of patterns) { - const match = pattern.exec(message); - if (!match) continue; - const value = Number(match[1]); - if (Number.isFinite(value) && value >= 100 && value <= 599) { - return value; - } - } - - return undefined; -} +import { scheduler } from "node:timers/promises"; +import { extractHttpStatusFromError, isRetryableError } from "@oh-my-pi/pi-utils"; /** * GitHub Copilot intermittently rejects preview models (gpt-5.3-codex, @@ -115,33 +16,10 @@ function extractStatusFromMessage(message: string): number | undefined { */ export function isCopilotTransientModelError(error: unknown): boolean { if (extractHttpStatusFromError(error) !== 400) return false; - return extractErrorCode(error) === "model_not_supported"; -} - -export function isCopilotRetryableError(error: unknown): boolean { - if (isCopilotTransientModelError(error)) return true; - - const status = extractHttpStatusFromError(error); - if (status !== undefined) { - return status >= 500 || status === 408 || status === 429; - } - - const message = error instanceof Error ? error.message : String(error); - return ( - isUnexpectedSocketCloseMessage(message) || - /request was aborted|aborted|fetch failed|network error|timed?\s*out|timeout|other side closed/i.test(message) - ); -} - -function extractErrorCode(error: unknown): string | undefined { - if (!error || typeof error !== "object") return undefined; - const info = error as ErrorLike; - if (typeof info.code === "string") return info.code; - const nested = info.error; - if (nested && typeof nested === "object" && typeof nested.code === "string") { - return nested.code; - } - return undefined; + if (!error || typeof error !== "object") return false; + const info = error as { code?: unknown; error?: { code?: unknown } | null }; + const code = typeof info.code === "string" ? info.code : info.error?.code; + return code === "model_not_supported"; } const COPILOT_MODEL_RETRY_MAX_ATTEMPTS = 3; @@ -166,9 +44,9 @@ export async function callWithCopilotModelRetry( return await fn(); } catch (error) { lastError = error; - if (!isCopilotRetryableError(error)) throw error; + if (!isCopilotTransientModelError(error) && !isRetryableError(error)) throw error; if (attempt === COPILOT_MODEL_RETRY_MAX_ATTEMPTS - 1) break; - await abortableSleep(COPILOT_MODEL_RETRY_BASE_DELAY_MS * (attempt + 1), options.signal); + await scheduler.wait(COPILOT_MODEL_RETRY_BASE_DELAY_MS * (attempt + 1), { signal: options.signal }); } } throw lastError; diff --git a/packages/ai/test/copilot-retry.test.ts b/packages/ai/test/copilot-retry.test.ts index 580f8156a..28881b14d 100644 --- a/packages/ai/test/copilot-retry.test.ts +++ b/packages/ai/test/copilot-retry.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "bun:test"; -import { callWithCopilotModelRetry, isCopilotTransientModelError, isRetryableError } from "@oh-my-pi/pi-ai/utils/retry"; +import { callWithCopilotModelRetry, isCopilotTransientModelError } from "@oh-my-pi/pi-ai/utils/retry"; +import { isRetryableError } from "@oh-my-pi/pi-utils"; type ErrorShape = { status: number; code?: string; error?: { code?: string; message?: string }; message: string }; @@ -132,7 +133,7 @@ describe("callWithCopilotModelRetry", () => { { provider: "github-copilot", signal: controller.signal }, ), ).rejects.toBeDefined(); - // fn runs once; abortableSleep rejects before a second attempt. + // fn runs once; scheduler.wait rejects before a second attempt. expect(calls).toBe(1); }); }); diff --git a/packages/ai/test/google-gemini-cli-429.test.ts b/packages/ai/test/google-gemini-cli-429.test.ts index 644a1b202..934e6218d 100644 --- a/packages/ai/test/google-gemini-cli-429.test.ts +++ b/packages/ai/test/google-gemini-cli-429.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "bun:test"; -import { extractRetryDelay } from "@oh-my-pi/pi-ai/providers/google-gemini-cli"; +import { extractRetryHint } from "@oh-my-pi/pi-utils"; // The fail-fast regex used inside the provider to distinguish "known quota errors" (throw immediately) // from "ambiguous 429s" (retry up to RATE_LIMIT_BUDGET_MS). @@ -7,102 +7,84 @@ import { extractRetryDelay } from "@oh-my-pi/pi-ai/providers/google-gemini-cli"; const FAIL_FAST_RE = /quota|exhausted/i; const shouldFailFast = (errorText: string) => FAIL_FAST_RE.test(errorText); -// normalizeDelay adds a 1 second buffer and rounds up: -// normalizeDelay(ms) = Math.ceil(ms + 1000) -// So all extractRetryDelay results are at least 1 second more than the raw delay. - describe("google-gemini-cli 429 fail-fast detection", () => { - it("fails fast on 'Quota exceeded'", () => { - expect(shouldFailFast("Quota exceeded")).toBe(true); + it("fails fast on 'Quota exceeded' messages", () => { + expect(shouldFailFast("Quota exceeded for project")).toBe(true); }); - it("fails fast on 'Resource has been exhausted'", () => { + it("fails fast on 'exhausted' messages", () => { expect(shouldFailFast("Resource has been exhausted")).toBe(true); }); - it("retries (does NOT fail fast) on 'Too many requests'", () => { - expect(shouldFailFast("Too many requests")).toBe(false); + it("does not fail fast on ambiguous 429 ('Please retry in 5s')", () => { + expect(shouldFailFast("Please retry in 5s")).toBe(false); }); - it("retries (does NOT fail fast) on 'rate limit exceeded'", () => { - expect(shouldFailFast("rate limit exceeded")).toBe(false); + it("does not fail fast on generic rate-limit text", () => { + expect(shouldFailFast("Rate limit exceeded, please slow down")).toBe(false); }); - it("fails fast on per-minute quota message", () => { - expect(shouldFailFast("Exceeded per minute quota for generateContent")).toBe(true); + it("matches case-insensitively", () => { + expect(shouldFailFast("QUOTA EXCEEDED")).toBe(true); + expect(shouldFailFast("Resource Has Been Exhausted")).toBe(true); }); - it("does NOT fail fast on generic 429 with empty body", () => { + it("does not fail fast on empty error", () => { expect(shouldFailFast("")).toBe(false); }); - - it("does NOT fail fast on unknown 429 message", () => { - expect(shouldFailFast("Internal server error")).toBe(false); - }); - - it("is case-insensitive for all variants", () => { - expect(shouldFailFast("QUOTA EXHAUSTED")).toBe(true); - expect(shouldFailFast("Rate Limit Exceeded")).toBe(false); - expect(shouldFailFast("TOO MANY REQUESTS")).toBe(false); - }); }); -describe("extractRetryDelay – header parsing", () => { - it("reads retry-after header as seconds → normalizeDelay(5000) = 6000", () => { +describe("extractRetryHint – header parsing", () => { + it("reads retry-after header as seconds", () => { const headers = new Headers({ "retry-after": "5" }); - // normalizeDelay(5 * 1000) = Math.ceil(5000 + 1000) = 6000 - expect(extractRetryDelay("", headers)).toBe(6000); + expect(extractRetryHint(headers)).toBe(5_000); }); - it("reads x-ratelimit-reset-after header as seconds → normalizeDelay(30000) = 31000", () => { + it("reads x-ratelimit-reset-after header as seconds", () => { const headers = new Headers({ "x-ratelimit-reset-after": "30" }); - // normalizeDelay(30 * 1000) = Math.ceil(30000 + 1000) = 31000 - expect(extractRetryDelay("", headers)).toBe(31000); + expect(extractRetryHint(headers)).toBe(30_000); }); it("prefers retry-after over x-ratelimit-reset-after when both are present", () => { const headers = new Headers({ "retry-after": "5", "x-ratelimit-reset-after": "30" }); - // retry-after is checked first → 6000 - expect(extractRetryDelay("", headers)).toBe(6000); + expect(extractRetryHint(headers)).toBe(5_000); }); }); -describe("extractRetryDelay – body text parsing", () => { - it("parses 'retryDelay' JSON field in seconds → normalizeDelay(3000) = 4000", () => { - // Regex: /"retryDelay":\s*"([0-9.]+)(ms|s)"/i - const body = '"retryDelay": "3s"'; - // normalizeDelay(3 * 1000) = Math.ceil(3000 + 1000) = 4000 - expect(extractRetryDelay(body)).toBe(4000); +describe("extractRetryHint – body text parsing", () => { + it("parses 'retryDelay' JSON field in seconds", () => { + expect(extractRetryHint(undefined, '"retryDelay": "3s"')).toBe(3_000); }); - it("parses 'retryDelay' JSON field in milliseconds → normalizeDelay(500) = 1500", () => { - const body = '"retryDelay": "500ms"'; - // normalizeDelay(500) = Math.ceil(500 + 1000) = 1500 - expect(extractRetryDelay(body)).toBe(1500); + it("parses 'retryDelay' JSON field in milliseconds", () => { + expect(extractRetryHint(undefined, '"retryDelay": "500ms"')).toBe(500); }); - it("parses 'Please retry in Xs' pattern → normalizeDelay(5000) = 6000", () => { - // Regex: /Please retry in ([0-9.]+)(ms|s)/i - expect(extractRetryDelay("Please retry in 5s")).toBe(6000); + it("parses 'Please retry in Xs' pattern", () => { + expect(extractRetryHint(undefined, "Please retry in 5s")).toBe(5_000); }); - it("parses 'quota will reset after Xs' simple duration → normalizeDelay(39000) = 40000", () => { - // Regex: /reset after (?:(\d+)h)?(?:(\d+)m)?(\d+(?:\.\d+)?)s/i - // totalMs = 39 * 1000 = 39000 → normalizeDelay = 40000 - expect(extractRetryDelay("Your quota will reset after 39s")).toBe(40000); + it("parses 'quota will reset after Xs' simple duration", () => { + expect(extractRetryHint(undefined, "Your quota will reset after 39s")).toBe(39_000); }); it("parses compound duration 'reset after 1h30m10s'", () => { - // (1*3600 + 30*60 + 10) * 1000 = 5 410 000 ms - // normalizeDelay(5410000) = Math.ceil(5410000 + 1000) = 5411000 - expect(extractRetryDelay("Your quota will reset after 1h30m10s")).toBe(5411000); + expect(extractRetryHint(undefined, "Your quota will reset after 1h30m10s")).toBe(5_410_000); + }); + + it("parses Codex-style 'try again in Xms'", () => { + expect(extractRetryHint(undefined, "try again in 250ms")).toBe(250); + }); + + it("parses Codex-style 'try again in Xs'", () => { + expect(extractRetryHint(undefined, "try again in 12s")).toBe(12_000); }); it("returns undefined when body contains no recognised delay pattern", () => { - expect(extractRetryDelay("Quota exceeded, please try again later")).toBeUndefined(); + expect(extractRetryHint(undefined, "Quota exceeded, please try again later")).toBeUndefined(); }); it("returns undefined for empty error string and no headers", () => { - expect(extractRetryDelay("")).toBeUndefined(); + expect(extractRetryHint(undefined, "")).toBeUndefined(); }); }); diff --git a/packages/ai/test/google-gemini-cli-alignment.test.ts b/packages/ai/test/google-gemini-cli-alignment.test.ts index 5311f9c94..213e02caa 100644 --- a/packages/ai/test/google-gemini-cli-alignment.test.ts +++ b/packages/ai/test/google-gemini-cli-alignment.test.ts @@ -244,7 +244,7 @@ describe("Google Gemini CLI alignment", () => { const result = await stream.result(); expect(fetchCalls).toBe(1); expect(result.stopReason).toBe("error"); - expect(result.errorMessage).toContain("Server requested 121s retry delay (max: 1s)"); + expect(result.errorMessage).toContain("Cloud Code Assist API error (503)"); }); }); }); diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index ccb06eb3b..f61f35a13 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -16,7 +16,7 @@ import * as crypto from "node:crypto"; import * as fs from "node:fs"; import * as path from "node:path"; - +import { scheduler } from "node:timers/promises"; import { type Agent, AgentBusyError, @@ -47,14 +47,20 @@ import { calculateRateLimitBackoffMs, getSupportedEfforts, isContextOverflow, - isUnexpectedSocketCloseMessage, isUsageLimitError, modelsAreEqual, parseRateLimitReason, streamSimple, } from "@oh-my-pi/pi-ai"; import { MacOSPowerAssertion } from "@oh-my-pi/pi-natives"; -import { abortableSleep, getAgentDbPath, isEnoent, logger, prompt, Snowflake } from "@oh-my-pi/pi-utils"; +import { + getAgentDbPath, + isEnoent, + isUnexpectedSocketCloseMessage, + logger, + prompt, + Snowflake, +} from "@oh-my-pi/pi-utils"; import { type AsyncJob, AsyncJobManager } from "../async"; import type { Rule } from "../capability/rule"; import { MODEL_ROLE_IDS, type ModelRegistry } from "../config/model-registry"; @@ -1446,7 +1452,7 @@ export class AgentSession { const scheduled = (async () => { if (delayMs > 0) { try { - await abortableSleep(delayMs, signal); + await scheduler.wait(delayMs, { signal }); } catch { return; } @@ -5976,7 +5982,7 @@ export class AgentSession { error: message, model: `${candidate.provider}/${candidate.id}`, }); - await abortableSleep(delayMs, autoCompactionSignal); + await scheduler.wait(delayMs, { signal: autoCompactionSignal }); } } @@ -6502,7 +6508,7 @@ export class AgentSession { this.#retryAbortController?.abort(); this.#retryAbortController = retryAbortController; try { - await abortableSleep(delayMs, retryAbortController.signal); + await scheduler.wait(delayMs, { signal: retryAbortController.signal }); } catch { if (this.#retryAbortController !== retryAbortController) { return false; diff --git a/packages/coding-agent/src/tools/gh.ts b/packages/coding-agent/src/tools/gh.ts index 88ca0fcd9..b6f587a61 100644 --- a/packages/coding-agent/src/tools/gh.ts +++ b/packages/coding-agent/src/tools/gh.ts @@ -1,9 +1,10 @@ import * as fs from "node:fs/promises"; import * as os from "node:os"; import * as path from "node:path"; +import { scheduler } from "node:timers/promises"; import type { AgentTool, AgentToolContext, AgentToolResult, AgentToolUpdateCallback } from "@oh-my-pi/pi-agent-core"; import { StringEnum } from "@oh-my-pi/pi-ai"; -import { abortableSleep, getWorktreesDir, isEnoent, prompt, untilAborted } from "@oh-my-pi/pi-utils"; +import { getWorktreesDir, isEnoent, prompt, untilAborted } from "@oh-my-pi/pi-utils"; import { type Static, Type } from "@sinclair/typebox"; import type { Settings } from "../config/settings"; import githubDescription from "../prompts/tools/github.md" with { type: "text" }; @@ -3400,7 +3401,7 @@ async function executeRunWatch( note, }), }); - await abortableSleep(graceSeconds * 1000, signal); + await scheduler.wait(graceSeconds * 1000, { signal }); run = await fetchRunSnapshot(session.cwd, repo, runId, signal); } @@ -3435,7 +3436,7 @@ async function executeRunWatch( return buildTextResult(formatRunWatchResult(repo, run, [], tail), run.url, finalDetails); } - await abortableSleep(intervalSeconds * 1000, signal); + await scheduler.wait(intervalSeconds * 1000, { signal }); } } @@ -3477,7 +3478,7 @@ async function executeRunWatch( note, }), }); - await abortableSleep(graceSeconds * 1000, signal); + await scheduler.wait(graceSeconds * 1000, { signal }); runs = await fetchRunsForCommit(session.cwd, repo, headSha, branch, signal); } @@ -3533,11 +3534,11 @@ async function executeRunWatch( note, }), }); - await abortableSleep(intervalSeconds * 1000, signal); + await scheduler.wait(intervalSeconds * 1000, { signal }); continue; } settledSuccessSignature = undefined; - await abortableSleep(intervalSeconds * 1000, signal); + await scheduler.wait(intervalSeconds * 1000, { signal }); } } diff --git a/packages/coding-agent/src/web/search/providers/gemini.ts b/packages/coding-agent/src/web/search/providers/gemini.ts index 82c44913f..decaf98e1 100644 --- a/packages/coding-agent/src/web/search/providers/gemini.ts +++ b/packages/coding-agent/src/web/search/providers/gemini.ts @@ -5,15 +5,11 @@ * Requires OAuth credentials stored in agent.db for provider "google-gemini-cli" or "google-antigravity". * Returns synthesized answers with citations and source metadata from grounding chunks. */ -import { - ANTIGRAVITY_SYSTEM_INSTRUCTION, - extractRetryDelay, - getAntigravityUserAgent, - getGeminiCliHeaders, -} from "@oh-my-pi/pi-ai"; +import { ANTIGRAVITY_SYSTEM_INSTRUCTION, getAntigravityUserAgent, getGeminiCliHeaders } from "@oh-my-pi/pi-ai"; import { refreshAntigravityToken } from "@oh-my-pi/pi-ai/utils/oauth/google-antigravity"; import { refreshGoogleCloudToken } from "@oh-my-pi/pi-ai/utils/oauth/google-gemini-cli"; -import { getAgentDbPath } from "@oh-my-pi/pi-utils"; +import { fetchWithRetry, getAgentDbPath } from "@oh-my-pi/pi-utils"; + import { AgentStorage } from "../../../session/agent-storage"; import type { SearchCitation, SearchResponse, SearchSource } from "../../../web/search/types"; import { SearchProviderError } from "../../../web/search/types"; @@ -303,96 +299,40 @@ async function callGeminiSearch( } (requestBody.request as Record).generationConfig = generationConfig; } - let response: Response | undefined; - let rateLimitTimeSpent = 0; - let lastError: Error | undefined; + const buildInit = (): RequestInit => ({ + method: "POST", + headers: { + Authorization: `Bearer ${auth.accessToken}`, + "Content-Type": "application/json", + Accept: "text/event-stream", + ...headers, + }, + body: JSON.stringify(requestBody), + }); + const urlFor = (attempt: number) => + `${endpoints[Math.min(attempt, endpoints.length - 1)]}/v1internal:streamGenerateContent?alt=sse`; - for (let endpointIndex = 0; endpointIndex < endpoints.length; endpointIndex++) { - const url = `${endpoints[endpointIndex]}/v1internal:streamGenerateContent?alt=sse`; + let response = await fetchWithRetry(urlFor, { + ...buildInit(), + maxAttempts: MAX_RETRIES + 1, + defaultDelayMs: attempt => BASE_DELAY_MS * 2 ** attempt, + maxDelayMs: RATE_LIMIT_BUDGET_MS, + }); - for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) { - try { - response = await fetch(url, { - method: "POST", - headers: { - Authorization: `Bearer ${auth.accessToken}`, - "Content-Type": "application/json", - Accept: "text/event-stream", - ...headers, - }, - body: JSON.stringify(requestBody), - }); - } catch (error) { - if (attempt < MAX_RETRIES) { - await Bun.sleep(BASE_DELAY_MS * 2 ** attempt); - continue; - } - - if (auth.isAntigravity && endpointIndex < endpoints.length - 1) { - break; - } - - throw error; - } - - if (response.ok) { - break; - } - - const errorText = await response.text(); - const canRefreshAuth = - response.status === 401 || - response.status === 403 || - (response.status === 400 && - /api key not valid|invalid credentials|invalid authentication/i.test(errorText)); - if (canRefreshAuth && attempt === 0 && (await refreshGeminiAuth(auth))) { - continue; - } - const isRetryableStatus = - response.status === 429 || - response.status === 500 || - response.status === 502 || - response.status === 503 || - response.status === 504; - - if (isRetryableStatus && attempt < MAX_RETRIES) { - const serverDelay = extractRetryDelay(errorText, response); - if (response.status === 429) { - if (serverDelay && rateLimitTimeSpent + serverDelay <= RATE_LIMIT_BUDGET_MS) { - rateLimitTimeSpent += serverDelay; - await Bun.sleep(serverDelay); - continue; - } - if (!serverDelay) { - await Bun.sleep(BASE_DELAY_MS * 2 ** attempt); - continue; - } - } else { - await Bun.sleep(serverDelay ?? BASE_DELAY_MS * 2 ** attempt); - continue; - } - } - - lastError = new SearchProviderError( - "gemini", - `Gemini Cloud Code API error (${response.status}): ${errorText}`, - response.status, - ); - - if (auth.isAntigravity && isRetryableStatus && endpointIndex < endpoints.length - 1) { - break; - } - - throw lastError; + if (!response.ok) { + const errorText = await response.clone().text(); + const canRefreshAuth = + response.status === 401 || + response.status === 403 || + (response.status === 400 && /api key not valid|invalid credentials|invalid authentication/i.test(errorText)); + if (canRefreshAuth && (await refreshGeminiAuth(auth))) { + response = await fetchWithRetry(urlFor, { + ...buildInit(), + maxAttempts: MAX_RETRIES + 1, + defaultDelayMs: attempt => BASE_DELAY_MS * 2 ** attempt, + maxDelayMs: RATE_LIMIT_BUDGET_MS, + }); } - - if (response?.ok) { - break; - } - } - - if (!response) { - throw new SearchProviderError("gemini", "Gemini API request failed", 500); } if (!response.ok) { diff --git a/packages/utils/src/abortable.ts b/packages/utils/src/abortable.ts index 0b3e727cf..49554b0fc 100644 --- a/packages/utils/src/abortable.ts +++ b/packages/utils/src/abortable.ts @@ -10,19 +10,6 @@ export class AbortError extends Error { } } -/** - * Sleep for a given number of milliseconds, respecting abort signal. - * - * Uses setTimeout (not Bun.sleep) so that vitest fake timers can intercept it in tests. - */ -export function abortableSleep(ms: number, signal?: AbortSignal): Promise { - return untilAborted(signal, () => { - const { promise, resolve } = Promise.withResolvers(); - setTimeout(resolve, ms); - return promise; - }); -} - /** * Creates an abortable stream from a given stream and signal. * diff --git a/packages/utils/src/fetch-retry.ts b/packages/utils/src/fetch-retry.ts new file mode 100644 index 000000000..35814eedc --- /dev/null +++ b/packages/utils/src/fetch-retry.ts @@ -0,0 +1,290 @@ +import { scheduler } from "node:timers/promises"; + +// "reset after 1h2m3s" / "10m15s" / "39s" +const QUOTA_RESET_PATTERN = /reset after (?:(\d+)h)?(?:(\d+)m)?(\d+(?:\.\d+)?)s/i; +// "Please retry in 250ms" / "Please retry in 12s" +const PLEASE_RETRY_PATTERN = /Please retry in ([0-9.]+)(ms|s)/i; +// JSON field: "retryDelay": "34.074824224s" +const RETRY_DELAY_FIELD_PATTERN = /"retryDelay":\s*"([0-9.]+)(ms|s)"/i; +// "try again in 250ms" / "try again in 12s" / "try again in 12sec" +const TRY_AGAIN_PATTERN = /try again in\s+(\d+(?:\.\d+)?)\s*(ms|s)(?:ec)?/i; + +/** + * Server-suggested retry delay extraction. Merges the patterns historically used + * by the OpenAI Codex and Google Gemini retry helpers. + * + * Header sources (checked in order): + * - `Retry-After` (numeric seconds, or HTTP date) + * - `x-ratelimit-reset` (Unix epoch seconds) + * - `x-ratelimit-reset-after` (seconds) + * + * Body patterns: + * - `Your quota will reset after 18h31m10s` / `10m15s` / `39s` + * - `Please retry in 250ms` / `Please retry in 12s` + * - `"retryDelay": "34.074824224s"` (JSON error detail field) + * - `try again in 250ms` / `try again in 12s` / `try again in 12sec` + * + * Returns `undefined` if no signal is found. + */ +export function extractRetryHint(source: Response | Headers | null | undefined, body?: string): number | undefined { + const headers = source instanceof Headers ? source : (source?.headers ?? undefined); + if (headers) { + const retryAfter = headers.get("retry-after"); + if (retryAfter) { + const seconds = Number(retryAfter); + if (Number.isFinite(seconds)) return Math.max(0, seconds * 1000); + const parsedDate = Date.parse(retryAfter); + if (!Number.isNaN(parsedDate)) return Math.max(0, parsedDate - Date.now()); + } + const rateLimitReset = headers.get("x-ratelimit-reset"); + if (rateLimitReset) { + const resetSeconds = Number.parseInt(rateLimitReset, 10); + if (!Number.isNaN(resetSeconds)) { + const delta = resetSeconds * 1000 - Date.now(); + if (delta > 0) return delta; + } + } + const rateLimitResetAfter = headers.get("x-ratelimit-reset-after"); + if (rateLimitResetAfter) { + const seconds = Number(rateLimitResetAfter); + if (Number.isFinite(seconds) && seconds > 0) return seconds * 1000; + } + } + + if (!body) return undefined; + + const quotaMatch = QUOTA_RESET_PATTERN.exec(body); + if (quotaMatch) { + const hours = quotaMatch[1] ? Number.parseInt(quotaMatch[1], 10) : 0; + const minutes = quotaMatch[2] ? Number.parseInt(quotaMatch[2], 10) : 0; + const seconds = Number.parseFloat(quotaMatch[3]!); + if (!Number.isNaN(seconds)) { + const totalMs = ((hours * 60 + minutes) * 60 + seconds) * 1000; + if (totalMs > 0) return totalMs; + } + } + for (const pattern of [PLEASE_RETRY_PATTERN, RETRY_DELAY_FIELD_PATTERN, TRY_AGAIN_PATTERN]) { + const match = pattern.exec(body); + if (match?.[1]) { + const value = Number.parseFloat(match[1]); + if (Number.isFinite(value) && value > 0) { + return match[2]!.toLowerCase() === "ms" ? value : value * 1000; + } + } + } + return undefined; +} + +export interface FetchWithRetryOptions extends RequestInit { + /** Total fetch attempts (initial + retries). Default `5`. */ + maxAttempts?: number; + /** + * Per-delay cap. Server-provided `Retry-After` hints exceeding this return + * the current response immediately — caller deals with the `!response.ok`. + * Default `60_000`. + */ + maxDelayMs?: number; + /** + * Fallback delay schedule when no server hint is present. Number, array + * (indexed by attempt, clamped to last), or function. Default exponential + * `500ms * 2 ** attempt` capped at `maxDelayMs`. + */ + defaultDelayMs?: number | readonly number[] | ((attempt: number) => number); + /** + * Optional per-attempt overlay merged into the base `RequestInit` each try. + * Headers from the overlay shallow-merge over the base. Useful for auth + * token refresh or user-agent rotation. + */ + prepareInit?: (attempt: number) => RequestInit | Promise; +} + +const DEFAULT_MAX_DELAY_MS = 60_000; +const DEFAULT_MAX_ATTEMPTS = 5; + +/** + * Fetch with bounded retries and sensible defaults. Retries on any + * `isRetryableStatus` (5xx, 408, 429) and on transient network errors. Server + * `Retry-After`/quota hints are honoured up to `maxDelayMs`; a hint that exceeds + * the cap returns the current response so the caller can fail fast. Aborts on + * `init.signal` propagate as `"Request was aborted"`. + * + * The caller is responsible for inspecting `!response.ok` once the call returns. + */ +export async function fetchWithRetry( + url: string | URL | ((attempt: number) => string | URL), + options: FetchWithRetryOptions = {}, +): Promise { + const { + maxAttempts = DEFAULT_MAX_ATTEMPTS, + maxDelayMs = DEFAULT_MAX_DELAY_MS, + defaultDelayMs, + prepareInit, + ...baseInit + } = options; + const signal = baseInit.signal as AbortSignal | undefined; + + for (let attempt = 0; ; attempt++) { + if (signal?.aborted) throw new Error("Request was aborted"); + const requestUrl = typeof url === "function" ? url(attempt) : url; + const init = prepareInit ? mergeInit(baseInit, await prepareInit(attempt)) : baseInit; + + let response: Response; + try { + response = await fetch(requestUrl, init); + } catch (error) { + if (signal?.aborted) throw new Error("Request was aborted"); + const wrapped = wrapNetworkError(error); + if (attempt + 1 >= maxAttempts) throw wrapped; + await scheduler.wait(resolveDefaultDelay(defaultDelayMs, attempt, maxDelayMs), { signal }); + continue; + } + + if (!isRetryableStatus(response.status)) return response; + if (attempt + 1 >= maxAttempts) return response; + + const hint = extractRetryHint(response, await response.clone().text()); + if (hint !== undefined && hint > maxDelayMs) return response; + + const delayMs = Math.min(hint ?? resolveDefaultDelay(defaultDelayMs, attempt, maxDelayMs), maxDelayMs); + await scheduler.wait(delayMs, { signal }); + } +} + +function mergeInit(base: RequestInit, overlay: RequestInit): RequestInit { + const merged: RequestInit = { ...base, ...overlay }; + if (base.headers || overlay.headers) { + const baseHeaders = new Headers(base.headers ?? undefined); + const overlayHeaders = new Headers(overlay.headers ?? undefined); + overlayHeaders.forEach((value, key) => { + baseHeaders.set(key, value); + }); + merged.headers = baseHeaders; + } + return merged; +} + +function wrapNetworkError(error: unknown): Error { + if (error instanceof Error) { + if (error.name === "AbortError" || error.message === "Request was aborted") { + return new Error("Request was aborted"); + } + if (error.message === "fetch failed" && error.cause instanceof Error) { + return new Error(`Network error: ${error.cause.message}`); + } + return error; + } + return new Error(String(error)); +} + +function resolveDefaultDelay( + option: FetchWithRetryOptions["defaultDelayMs"], + attempt: number, + maxDelayMs: number, +): number { + if (option === undefined) return Math.min(500 * 2 ** attempt, maxDelayMs); + if (typeof option === "number") return Math.min(option, maxDelayMs); + if (typeof option === "function") return Math.min(option(attempt), maxDelayMs); + return Math.min(option[Math.min(attempt, option.length - 1)] ?? 0, maxDelayMs); +} + +/** + * Inspect an arbitrary error value (or its `cause` chain, up to depth 2) for an + * HTTP status code. Reads `status`, `statusCode`, and `response.status` fields, + * coerces string values, and falls back to scanning the error message for + * common patterns like `error (429)` or `HTTP 503`. + */ +export function extractHttpStatusFromError(error: unknown): number | undefined { + return extractHttpStatusFromErrorInternal(error, 0); +} + +type HttpErrorLike = { + message?: string; + name?: string; + status?: number | string; + statusCode?: number | string; + response?: { status?: number | string }; + cause?: unknown; +}; + +function extractHttpStatusFromErrorInternal(error: unknown, depth: number): number | undefined { + if (!error || typeof error !== "object" || depth > 2) return undefined; + const info = error as HttpErrorLike; + const rawStatus = info.status ?? info.statusCode ?? info.response?.status; + + let status: number | undefined; + if (typeof rawStatus === "number" && Number.isFinite(rawStatus)) { + status = rawStatus; + } else if (typeof rawStatus === "string") { + const parsed = Number(rawStatus); + if (Number.isFinite(parsed)) status = parsed; + } + if (status !== undefined && status >= 100 && status <= 599) return status; + + if (info.message) { + const extracted = extractStatusFromMessage(info.message); + if (extracted !== undefined) return extracted; + } + if (info.cause) return extractHttpStatusFromErrorInternal(info.cause, depth + 1); + return undefined; +} + +const STATUS_MESSAGE_PATTERNS = [ + /error\s*\((\d{3})\)/i, + /status\s*[:=]?\s*(\d{3})/i, + /\bhttp\s*(\d{3})\b/i, + /\b(\d{3})\s*(?:status|error)\b/i, +] as const; + +function extractStatusFromMessage(message: string): number | undefined { + for (const pattern of STATUS_MESSAGE_PATTERNS) { + const match = pattern.exec(message); + if (!match) continue; + const value = Number(match[1]); + if (Number.isFinite(value) && value >= 100 && value <= 599) return value; + } + return undefined; +} + +/** + * `true` if the given HTTP status code is one we treat as transient: 408 + * (Request Timeout), 429 (Too Many Requests), or any 5xx (server error). + */ +export function isRetryableStatus(status: number): boolean { + return status >= 500 || status === 408 || status === 429; +} + +/** + * `true` if the message describes an unexpected socket closure — Bun and some + * proxies surface these for any HTTP/2 stream reset. + */ +export function isUnexpectedSocketCloseMessage(message: string): boolean { + return /\b(?:the\s+)?socket connection (?:was )?closed unexpectedly\b/i.test(message); +} + +const TRANSIENT_MESSAGE_PATTERN = + /overloaded|rate.?limit|too many requests|service.?unavailable|server error|internal error|connection.?error|unable to connect|fetch failed|network error|stream stall|other side closed/i; + +const VALIDATION_MESSAGE_PATTERN = + /invalid|validation|bad request|unsupported|schema|missing required|not found|unauthorized|forbidden/i; + +/** + * Identify errors that should be retried: aborts/timeouts in the error name or + * message, retryable HTTP statuses (see `isRetryableStatus`), unexpected socket + * closes, and the standard transient phrases. 4xx statuses other than 408/429 + * and validation-shaped messages short-circuit to `false`. + */ +export function isRetryableError(error: unknown): boolean { + const info = error as { message?: string; name?: string } | null; + const message = info?.message ?? ""; + const name = info?.name ?? ""; + if (name === "AbortError" || /timeout|timed out|aborted/i.test(message)) return true; + + const status = extractHttpStatusFromError(error); + if (status !== undefined) { + if (isRetryableStatus(status)) return true; + if (status >= 400 && status < 500) return false; + } + + if (VALIDATION_MESSAGE_PATTERN.test(message)) return false; + return isUnexpectedSocketCloseMessage(message) || TRANSIENT_MESSAGE_PATTERN.test(message); +} diff --git a/packages/utils/src/index.ts b/packages/utils/src/index.ts index c8abaf461..80ef33c4d 100644 --- a/packages/utils/src/index.ts +++ b/packages/utils/src/index.ts @@ -1,8 +1,9 @@ -export { abortableSleep, createAbortableStream, once, untilAborted } from "./abortable"; +export { createAbortableStream, once, untilAborted } from "./abortable"; export * from "./async"; export * from "./color"; export * from "./dirs"; export * from "./env"; +export * from "./fetch-retry"; export * from "./format"; export * from "./frontmatter"; export * from "./fs-error";