diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 906114492db..a610f9e2eb1 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1543,7 +1543,8 @@ "devin-stated-reset-retry.test.ts": "providers", "ci-structure-gate.test.ts": "ci-workflows", "responses-code-mode-patch-compile.test.ts": "responses", - "gui-codex-usage-score-parity.test.ts": "gui" + "gui-codex-usage-score-parity.test.ts": "gui", + "web-search-sidecar-429.test.ts": "web-search" }, "migrated": [ "adapters", diff --git a/src/web-search/executor.ts b/src/web-search/executor.ts index d6514acfd49..33b6a1eb6cc 100644 --- a/src/web-search/executor.ts +++ b/src/web-search/executor.ts @@ -3,7 +3,14 @@ import { FORWARD_HEADERS } from "../adapters/openai-responses"; import { signalWithTimeout, cancelBodyOnAbort } from "../lib/abort"; import { redactSecretString } from "../lib/redact"; import { sidecarEnter } from "../lib/sidecar-tracker"; -import { applyUpstreamRecoveryInit, fetchWithResetRetry } from "../lib/upstream-retry"; +import { + applyUpstreamRecoveryInit, + fetchWithResetRetry, + releaseResponseBodyBestEffort, + retryBackoffDelayMs, + sleepWithAbort, + RETRY_AFTER_CEILING_MS, +} from "../lib/upstream-retry"; import { withUpstreamHttpVersion } from "../lib/upstream-http-version"; import { parseSidecarSSE, type WebSearchResult } from "./parse"; import type { CodexUpstreamOutcome } from "../codex/routing"; @@ -35,6 +42,22 @@ export const IMAGE_INSTRUCTION = /** A search result, or an `error` string when the search couldn't run (surfaced as a tool result). */ export type SidecarOutcome = WebSearchResult & { error?: string }; + +/** + * Bounded same-search 429 replays for the sidecar POST. + * + * The forward backend throttles burst sidecar traffic, and without a replay the 429 becomes a + * failed tool result that poisons the query for the whole turn (see failedQueries in loop.ts). + * 1 initial send + 2 replays; Retry-After is honored as a lower bound and capped by + * RETRY_AFTER_CEILING_MS (an instruction past the ceiling ends with the 429 instead of + * parking the search). Each wait releases the unread 429 body first so sockets do not + * accumulate under a rate-limit storm. Abort or timeout ends the wait through the existing + * catch, exactly like an abort during the SSE parse. + */ +const SIDECAR_429_MAX_ATTEMPTS = 3; +const SIDECAR_429_BASE_DELAY_MS = 1_000; +const SIDECAR_429_MAX_DELAY_MS = 10_000; + export type SidecarOutcomeRecorder = (outcome: CodexUpstreamOutcome) => void; /** @@ -79,7 +102,7 @@ export async function runWebSearch( const sidecarExit = sidecarEnter("web-search"); const t0 = Date.now(); try { - const res = await fetchWithResetRetry( + const sendOnce = () => fetchWithResetRetry( // Recovery nests INSIDE the version helper: applyUpstreamRecoveryInit then always receives a // defined init, and withUpstreamHttpVersion spreads the result, so `protocol` and the // recovery fields (`connection: close` + Bun's transport-level `keepalive: false`) survive @@ -96,6 +119,22 @@ export async function runWebSearch( }, recovery), forwardProvider)), { replaySafe: true, abortSignal: linkedSignal.signal, label: "web-search-sidecar" }, ); + let res = await sendOnce(); + for (let attempt = 0; res.status === 429 && attempt + 1 < SIDECAR_429_MAX_ATTEMPTS; attempt++) { + const delay = retryBackoffDelayMs(attempt, { + baseDelayMs: SIDECAR_429_BASE_DELAY_MS, + maxDelayMs: SIDECAR_429_MAX_DELAY_MS, + headers: res.headers, + retryAfterIsLowerBound: true, + }); + // A deadline, not a clamp: an instruction past the ceiling ends the search with the + // 429 instead of parking it at a provider that already said it would refuse. + if (delay > RETRY_AFTER_CEILING_MS) break; + console.warn(`[web-search] sidecar HTTP 429 — retrying (${attempt + 2}/${SIDECAR_429_MAX_ATTEMPTS}) after ${delay}ms`); + await releaseResponseBodyBestEffort(res.body, linkedSignal.signal); + await sleepWithAbort(delay, linkedSignal.signal); + res = await sendOnce(); + } // Attach the body guard before ANY branch reads it. The success path guarded itself below, // but the failure branch's `res.text()` runs first, so a cancel landing between fetch // resolution and reader attach orphaned the internal rejection (found investigating #1419). diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index ade9fea9a71..c4bd7f4bceb 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1375,5 +1375,6 @@ "devin-stated-reset-retry.test.ts": "providers", "ci-structure-gate.test.ts": "ci-workflows", "responses-code-mode-patch-compile.test.ts": "responses", - "gui-codex-usage-score-parity.test.ts": "gui" + "gui-codex-usage-score-parity.test.ts": "gui", + "web-search-sidecar-429.test.ts": "web-search" } diff --git a/tests/web-search/web-search-sidecar-429.test.ts b/tests/web-search/web-search-sidecar-429.test.ts new file mode 100644 index 00000000000..6956b855333 --- /dev/null +++ b/tests/web-search/web-search-sidecar-429.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, test } from "bun:test"; +import { runWebSearch as runOpenAiWebSearch } from "../../src/web-search/executor"; +import { listOpenAiForwardSidecarCandidates } from "../../src/providers/openai-sidecar"; +import type { OcxConfig } from "../../src/types"; + +function testConfig(overrides: Partial = {}): OcxConfig { + return { + port: 10100, + defaultProvider: "routed", + providers: {}, + ...overrides, + }; +} + +describe("web-search sidecar 429 replays", () => { + function sidecarProvider() { + const cfg = testConfig({ + providers: { + openai: { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", + }, + }, + }); + return listOpenAiForwardSidecarCandidates(cfg)[0]!.provider; + } + + function sseDone(): Response { + return new Response("data: [DONE]\n\n", { headers: { "content-type": "text/event-stream" } }); + } + + function searchWith(fetchImpl: () => Promise) { + globalThis.fetch = fetchImpl as unknown as typeof fetch; + return runOpenAiWebSearch( + "current docs", + { type: "web_search" }, + sidecarProvider(), + new Headers({ authorization: "Bearer selected-token" }), + { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, + ); + } + + test("a burst 429 is replayed and the recovered answer is returned", async () => { + let calls = 0; + const outcome = await searchWith(async () => { + calls += 1; + if (calls === 1) return new Response("rate limited", { status: 429 }); + return sseDone(); + }); + expect(calls).toBe(2); + expect(outcome.error).toBeUndefined(); + }); + + test("a persistent 429 ends with the 429 after bounded attempts", async () => { + let calls = 0; + const outcome = await searchWith(async () => { + calls += 1; + return new Response("rate limited", { status: 429 }); + }); + expect(calls).toBe(3); + expect(outcome.error).toContain("429"); + }); + + test("a Retry-After past the ceiling ends with the 429 without parking", async () => { + let calls = 0; + const outcome = await searchWith(async () => { + calls += 1; + return new Response("slow down", { status: 429, headers: { "retry-after": "120" } }); + }); + expect(calls).toBe(1); + expect(outcome.error).toContain("429"); + }); +});