diff --git a/devlog/_plan/260830_lane_o_reset_recovery_parity/000_units.md b/devlog/_plan/260830_lane_o_reset_recovery_parity/000_units.md new file mode 100644 index 0000000000..4a7c6b81d5 --- /dev/null +++ b/devlog/_plan/260830_lane_o_reset_recovery_parity/000_units.md @@ -0,0 +1,87 @@ +# Lane O — connection-reset recovery parity on the sidecar and loop legs + +Unit for the work-phase that followed Lane N. Its trigger was a hook restating +issue #2885 fallout as "the web-search sidecar ignores `upstreamHttpVersion` and +skips the fresh-connection retry treatment". Half of that was already shipped; +the other half turned out to be wider than the sidecar. + +## What was already true + +PR #2908 (`22f2df614`) landed the transport-pin half. `src/web-search/loop.ts:326` +resolves `deps.incomingMeta.providerFetch`, both send legs use it, and +`src/server/responses/core.ts:5006` rebuilds it at send time so a 429 rotation +cannot pin a stale credential. `src/web-search/executor.ts:77` wraps the sidecar +leg in `withUpstreamHttpVersion(forwardProvider)`. Nothing in that description is +outstanding, and no part of this unit re-does it. + +## The gap that was real + +`applyUpstreamRecoveryInit` (`src/lib/upstream-retry.ts:295`) exists for one +reason: Bun has ignored the hop-by-hop `Connection: close` header +(oven-sh/bun#20492), so leaving a half-closed pooled socket needs the +transport-level `keepalive: false` extension as well. Setting the header alone +lets the retry land back on the same dead socket. + +The main lanes call it — `src/server/chat-native.ts:207`, +`src/server/responses/compact.ts:715`, and six sites in +`src/server/responses/core.ts` (3831, 3902, 4103, 4163, 5521, 6038). The +web-search loop and the images loop take the `retryRecovery` argument +`fetchWithResetRetry` hands them, spend it on `deps.onAttemptSend` telemetry, and +then build a plain init. Every sidecar executor passes a zero-argument thunk: it +still retries, but it cannot ask for fresh-connection recovery. + +Be precise about the consequence. `fetchWithResetRetry` retries a reset up to +three times, and on these legs each replay stays *eligible* to reuse the pooled +socket the reset came from — not guaranteed to, since the pool may hand out +another. That is enough to make recovery a matter of luck, and the retry then +reports as exhausted rather than as the pool problem it is. It matches the +failure shape #2885 reported without explaining it, and this unit does not claim +to close that issue. + +`src/adapters/kiro-retry.ts` already hand-rolls the same two fields (header at +168, `keepalive` at 173) and is out of scope; an independent audit confirmed it +correct. + +## Diff + +Thread the recovery init through the legs that already receive the recovery kind, +and give the sidecar thunks the argument they were missing: + +- `src/web-search/loop.ts` and `src/images/loop.ts` — pass the existing + `retryRecovery` through `applyUpstreamRecoveryInit`, preserving the + `accept-encoding: identity` handling and the provider-scoped executor. +- the sidecar executors — accept the recovery argument and route their init the + same way, composed so a protocol pin and the recovery fields cannot displace + each other. + +## Composition constraint + +`withUpstreamHttpVersion` spreads `{...(init ?? {}), protocol}` and is typed to +return `RequestInit | undefined`; `applyUpstreamRecoveryInit` spreads +`{...init, headers}` and adds `keepalive`. The order is not free. Recovery goes +**inside**: + +```ts +withUpstreamHttpVersion(url, applyUpstreamRecoveryInit(baseInit, recovery), provider) +``` + +so the recovery helper always receives a defined init and the version helper +spreads the result, keeping headers, `keepalive`, body, signal, and redirect +alongside `protocol`. The reverse nesting needs a `?? baseInit` fallback to type-check +at all and would otherwise dereference the `undefined` branch. An independent +audit probed the composed object under Bun and observed `protocol`, +`keepalive: false`, `connection: close`, the body, and `redirect: "manual"` +surviving together. The regression asserts a pinned provider still sees its +`protocol` on the replay. + +What is not verified: no wire capture was taken. Under an HTTP/2 pin, +`Connection` is a prohibited hop-by-hop header and Bun may normalize it away +while still honoring `keepalive: false`. The same canonical helper already runs +on provider paths that support an HTTP/2 pin, so this is a documented unknown +rather than a reason to exclude a site. + +## Evidence standard + +A green suite proves nothing here. Each assertion is driven red by reverting its +own site to the plain init, and any assertion that stays green under that +mutation is deleted rather than kept. diff --git a/src/images/loop.ts b/src/images/loop.ts index 74f759b2b5..0191215960 100644 --- a/src/images/loop.ts +++ b/src/images/loop.ts @@ -21,7 +21,7 @@ import type { AttemptRecoveryKind } from "../usage/log"; import { bridgeToResponsesSSE } from "../bridge"; import { clearableDeadline, idleDeadline } from "../lib/abort"; import { readBoundedResponseBody } from "../lib/bounded-body"; -import { fetchWithResetRetry, prepareSameTarget429Wait } from "../lib/upstream-retry"; +import { applyUpstreamRecoveryInit, fetchWithResetRetry, prepareSameTarget429Wait } from "../lib/upstream-retry"; import { rateLimitRetryDelayMs } from "../providers/key-failover"; import { isTranslatorBudgetExceededError, @@ -521,12 +521,15 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise fetch(`${base}/v1/messages`, { + recovery => fetch(`${base}/v1/messages`, applyUpstreamRecoveryInit({ method: "POST", headers, body: JSON.stringify(body), signal: linkedSignal.signal, - }), + }, recovery)), { abortSignal: linkedSignal.signal, label: "vision-sidecar-anthropic" }, ); if (!res.ok) { diff --git a/src/vision/describe.ts b/src/vision/describe.ts index d580a607e8..b919fb738a 100644 --- a/src/vision/describe.ts +++ b/src/vision/describe.ts @@ -4,7 +4,7 @@ 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 { fetchWithResetRetry } from "../lib/upstream-retry"; +import { applyUpstreamRecoveryInit, fetchWithResetRetry } from "../lib/upstream-retry"; import { parseSidecarSSE } from "../web-search/parse"; import type { SidecarOutcomeRecorder } from "../web-search/executor"; @@ -90,7 +90,9 @@ export async function describeImage( const t0 = Date.now(); try { const res = await fetchWithResetRetry( - () => fetch(`${forwardProvider.baseUrl}/responses`, { + // The replay needs `keepalive: false` to abandon the half-closed pooled socket; Bun has + // ignored a bare `Connection: close` (oven-sh/bun#20492). + recovery => fetch(`${forwardProvider.baseUrl}/responses`, applyUpstreamRecoveryInit({ method: "POST", headers, body: JSON.stringify(body), @@ -99,7 +101,7 @@ export async function describeImage( // across origins but forwards nonstandard headers such as `chatgpt-account-id`, // `session_id`, and `x-codex-turn-metadata` to the redirect target. redirect: "manual", - }), + }, recovery)), { abortSignal: linkedSignal.signal, label: "vision-sidecar" }, ); const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal); diff --git a/src/web-search/anthropic-executor.ts b/src/web-search/anthropic-executor.ts index aeba03a829..1eb206afa8 100644 --- a/src/web-search/anthropic-executor.ts +++ b/src/web-search/anthropic-executor.ts @@ -4,7 +4,7 @@ import { ANTHROPIC_OAUTH_BETA, CLAUDE_CODE_SYSTEM_INSTRUCTION } from "../oauth/a import { CLAUDE_CODE_HEADERS, claudeCodeSessionId } from "../adapters/client-fingerprint"; import { signalWithTimeout, cancelBodyOnAbort } from "../lib/abort"; import { sidecarEnter } from "../lib/sidecar-tracker"; -import { fetchWithResetRetry } from "../lib/upstream-retry"; +import { applyUpstreamRecoveryInit, fetchWithResetRetry } from "../lib/upstream-retry"; import type { WebSearchSource } from "./parse"; import { BASE_INSTRUCTION, IMAGE_INSTRUCTION, type SidecarOutcome, type SidecarSettings } from "./executor"; @@ -162,7 +162,14 @@ export async function runAnthropicWebSearch( const t0 = Date.now(); try { const res = await fetchWithResetRetry( - () => fetch(url, { method: "POST", headers, body: JSON.stringify(body), signal: linkedSignal.signal }), + // The replay needs `keepalive: false` to leave the half-closed pooled socket; Bun has + // ignored a bare `Connection: close` (oven-sh/bun#20492). + recovery => fetch(url, applyUpstreamRecoveryInit({ + method: "POST", + headers, + body: JSON.stringify(body), + signal: linkedSignal.signal, + }, recovery)), { abortSignal: linkedSignal.signal, label: "web-search-sidecar-anthropic" }, ); // Guard before any branch reads the body: the failure branch's `res.text()` ran ahead of diff --git a/src/web-search/exa-executor.ts b/src/web-search/exa-executor.ts index 12e8e7ef02..2170eec140 100644 --- a/src/web-search/exa-executor.ts +++ b/src/web-search/exa-executor.ts @@ -8,7 +8,7 @@ * redirect: "manual" because Bun forwards custom headers across redirects. * Never throws; every error string passes redactSecretString. */ -import { fetchWithResetRetry } from "../lib/upstream-retry"; +import { applyUpstreamRecoveryInit, fetchWithResetRetry } from "../lib/upstream-retry"; import { cancelBodyOnAbort, signalWithTimeout } from "../lib/abort"; import { readBoundedResponseBytes } from "../lib/bounded-body"; import { sidecarEnter } from "../lib/sidecar-tracker"; @@ -39,13 +39,13 @@ export async function runExaWebSearch( const t0 = Date.now(); try { const res = await fetchWithResetRetry( - () => fetch(EXA_SEARCH_URL, { + recovery => fetch(EXA_SEARCH_URL, applyUpstreamRecoveryInit({ method: "POST", headers: { "Content-Type": "application/json", "x-api-key": apiKey }, body: JSON.stringify({ query, numResults: EXA_NUM_RESULTS, contents: { text: { maxCharacters: EXA_SNIPPET_CHARS } } }), signal: linkedSignal.signal, redirect: "manual", - }), + }, recovery)), { abortSignal: linkedSignal.signal, label: "exa-web-search-sidecar" }, ); const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal); diff --git a/src/web-search/executor.ts b/src/web-search/executor.ts index 84daf31b9a..840f062fbc 100644 --- a/src/web-search/executor.ts +++ b/src/web-search/executor.ts @@ -3,7 +3,7 @@ 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 { fetchWithResetRetry } from "../lib/upstream-retry"; +import { applyUpstreamRecoveryInit, fetchWithResetRetry } from "../lib/upstream-retry"; import { withUpstreamHttpVersion } from "../lib/upstream-http-version"; import { parseSidecarSSE, type WebSearchResult } from "./parse"; import type { CodexUpstreamOutcome } from "../codex/routing"; @@ -74,7 +74,11 @@ export async function runWebSearch( const t0 = Date.now(); try { const res = await fetchWithResetRetry( - () => fetch(url, withUpstreamHttpVersion(url, { + // 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 + // together. The reverse order needs a `?? init` fallback to type-check at all. + recovery => fetch(url, withUpstreamHttpVersion(url, applyUpstreamRecoveryInit({ method: "POST", headers, body: JSON.stringify(body), @@ -83,7 +87,7 @@ export async function runWebSearch( // across origins but forwards nonstandard headers such as `chatgpt-account-id`, // `session_id`, and `x-codex-turn-metadata` to the redirect target. redirect: "manual", - }, forwardProvider)), + }, recovery), forwardProvider)), { abortSignal: linkedSignal.signal, label: "web-search-sidecar" }, ); // Attach the body guard before ANY branch reads it. The success path guarded itself below, diff --git a/src/web-search/gemini-executor.ts b/src/web-search/gemini-executor.ts index f575526ef4..72c74169cd 100644 --- a/src/web-search/gemini-executor.ts +++ b/src/web-search/gemini-executor.ts @@ -10,7 +10,7 @@ */ import type { OcxProviderConfig } from "../types"; import { getValidAccessTokenSnapshot, publicOAuthAuthenticationErrorMessage } from "../oauth"; -import { fetchWithResetRetry } from "../lib/upstream-retry"; +import { applyUpstreamRecoveryInit, fetchWithResetRetry } from "../lib/upstream-retry"; import { cancelBodyOnAbort, signalWithTimeout } from "../lib/abort"; import { readBoundedResponseBytes } from "../lib/bounded-body"; import { sidecarEnter } from "../lib/sidecar-tracker"; @@ -69,7 +69,7 @@ export async function runGeminiWebSearch( const t0 = Date.now(); try { const res = await fetchWithResetRetry( - () => fetch(`${base}/v1internal:generateContent`, { + recovery => fetch(`${base}/v1internal:generateContent`, applyUpstreamRecoveryInit({ method: "POST", headers: { "Content-Type": "application/json", @@ -79,7 +79,7 @@ export async function runGeminiWebSearch( body: JSON.stringify(envelope), signal: linkedSignal.signal, redirect: "manual", - }), + }, recovery)), { abortSignal: linkedSignal.signal, label: "gemini-web-search-sidecar" }, ); const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal); diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index 4b3fde2a82..682e482eea 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -13,7 +13,7 @@ import type { WebSearchBackendId } from "./index"; import { clearableDeadline } from "../lib/abort"; import { redactSecretString } from "../lib/redact"; import { readBoundedResponseBody } from "../lib/bounded-body"; -import { fetchWithResetRetry, prepareSameTarget429Wait } from "../lib/upstream-retry"; +import { applyUpstreamRecoveryInit, fetchWithResetRetry, prepareSameTarget429Wait } from "../lib/upstream-retry"; import { rateLimitRetryDelayMs } from "../providers/key-failover"; import { isTranslatorBudgetExceededError, @@ -460,12 +460,17 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise fetch(url, { + recovery => fetch(url, applyUpstreamRecoveryInit({ method: "POST", headers: { "Content-Type": "application/json", "Authorization": `Bearer ${token}` }, body: JSON.stringify(body), signal: linkedSignal.signal, // Credential-bearing: never follow a redirect off the pinned origin. redirect: "manual", - }), + }, recovery)), { abortSignal: linkedSignal.signal, label: "xai-web-search-sidecar" }, ); const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal); diff --git a/tests/exa-web-search.test.ts b/tests/exa-web-search.test.ts index 0ad77abab9..f9202f768b 100644 --- a/tests/exa-web-search.test.ts +++ b/tests/exa-web-search.test.ts @@ -242,7 +242,8 @@ describe("runExaWebSearch key hygiene (canary)", () => { expect(captured).toHaveLength(1); expect(captured[0]!.url).toBe("https://api.exa.ai/search"); expect(captured[0]!.init.redirect).toBe("manual"); - expect((captured[0]!.init.headers as Record)["x-api-key"]).toBe("key-1"); + // Representation-independent: the init may carry a plain record or a Headers instance. + expect(new Headers(captured[0]!.init.headers).get("x-api-key")).toBe("key-1"); } finally { globalThis.fetch = realFetch; } diff --git a/tests/gemini-web-search.test.ts b/tests/gemini-web-search.test.ts index 0cc71cca4c..0f0bd782a1 100644 --- a/tests/gemini-web-search.test.ts +++ b/tests/gemini-web-search.test.ts @@ -129,9 +129,12 @@ describe("runGeminiWebSearch request shape (review P1)", () => { expect(new URL(req.url).origin).toBe("https://daily-cloudcode-pa.googleapis.com"); expect(req.url).toContain("/v1internal:generateContent"); expect(req.init.redirect).toBe("manual"); - const headers = req.init.headers as Record; - expect(headers["Authorization"]).toBe("Bearer gem-token-abc"); - expect(headers["User-Agent"]).toContain("antigravity"); + // Read through Headers so the credential assertion holds whether the init carries a plain + // record or a Headers instance: the reset-recovery helper normalizes headers on the send + // path, and this canary is about WHICH bearer goes out, not how the init spells it. + const headers = new Headers(req.init.headers); + expect(headers.get("Authorization")).toBe("Bearer gem-token-abc"); + expect(headers.get("User-Agent")).toContain("antigravity"); const body = JSON.parse(String(req.init.body)); expect(body.project).toBe("proj-9"); expect(body.userAgent).toBe("antigravity"); @@ -176,7 +179,7 @@ describe("runGeminiWebSearch request shape (review P1)", () => { try { const out = await runGeminiWebSearch("q", "google-antigravity", cca, { model: "gemini-3.7-flash", reasoning: "low", timeoutMs: 5000 }); expect(out.text).toBe("ok"); - expect((request!.headers as Record)["Authorization"]).toBe("Bearer token-a"); + expect(new Headers(request!.headers).get("Authorization")).toBe("Bearer token-a"); expect(JSON.parse(String(request!.body)).project).toBe("project-a"); expect(accountSets["google-antigravity"]!.activeAccountId).toBe("account-b"); } finally { diff --git a/tests/web-search.test.ts b/tests/web-search.test.ts index 035c88e7ae..a212b4f892 100644 --- a/tests/web-search.test.ts +++ b/tests/web-search.test.ts @@ -13,6 +13,20 @@ import type { AdapterFetchContext, ProviderAdapter } from "../src/adapters/base" import type { OcxMessage, OcxParsedRequest } from "../src/types"; import { fakeChatGptJwt } from "./helpers/fake-chatgpt-jwt"; import { createTestTranslatorBudget } from "./helpers/translator-budget"; +import { withUpstreamHttpVersion } from "../src/lib/upstream-http-version"; + +/** + * Wrap a fetch so it applies the provider's HTTP-version pin the way `providerFetch` does in + * production. The reset-recovery tests need a provider-scoped executor that pins a protocol, so the + * composition order in the loop leg is observable without standing up the whole server path. + */ +function withUpstreamHttpVersionExecutor( + inner: typeof globalThis.fetch, + provider: Pick, +): typeof globalThis.fetch { + return ((input: Parameters[0], init?: RequestInit) => + inner(input, withUpstreamHttpVersion(input, init, provider))) as typeof globalThis.fetch; +} /** Run the web-search loop with a default test translator budget. */ function runWithWebSearch( @@ -2518,3 +2532,123 @@ describe("web-search sidecar live streaming (streamRoutedModelOutput)", () => { expect(frames.some(f => f.event === "response.completed")).toBe(true); }); }); + +describe("connection-reset recovery parity on the web-search legs", () => { + const originalGlobalFetch = globalThis.fetch; + afterEach(() => { globalThis.fetch = originalGlobalFetch; }); + + /** Bun's reset rejection shape, as matched by isConnectionResetError. */ + function bunResetError(): Error { + return new Error("The socket connection was closed unexpectedly. For more information, pass `verbose: true` in the second argument to fetch()"); + } + + type Observed = { keepalive: unknown; connection: string | null; protocol: string | undefined; body: unknown; redirect: string | undefined }; + + function observe(init: RequestInit | undefined): Observed { + const withExtras = init as (RequestInit & { keepalive?: unknown; protocol?: string }) | undefined; + return { + keepalive: withExtras?.keepalive, + connection: new Headers(init?.headers).get("connection"), + protocol: withExtras?.protocol, + body: init?.body, + redirect: init?.redirect, + }; + } + + test("the sidecar leg replays a reset on a fresh connection while keeping the provider HTTP version pin", async () => { + const attempts: Observed[] = []; + globalThis.fetch = (async (_input: string | URL | Request, init?: RequestInit) => { + attempts.push(observe(init)); + if (attempts.length === 1) throw bunResetError(); + return new Response("data: [DONE]\n\n", { headers: { "content-type": "text/event-stream" } }); + }) as typeof fetch; + + await runOpenAiWebSearch( + "current docs", + { type: "web_search" }, + { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", + upstreamHttpVersion: "http1.1", + }, + new Headers({ authorization: "Bearer selected-token" }), + { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 1_000 }, + ); + + expect(attempts.length).toBe(2); + // The first attempt must NOT force a fresh connection: pooling is the normal, faster path. + expect(attempts[0]!.keepalive).toBeUndefined(); + expect(attempts[0]!.connection).toBeNull(); + // The replay must leave the half-closed pooled socket. `keepalive: false` is the field that + // actually does it — Bun has ignored a bare `Connection: close` (oven-sh/bun#20492) — so + // assert both rather than treating the header as sufficient. + expect(attempts[1]!.keepalive).toBe(false); + expect(attempts[1]!.connection).toBe("close"); + // Composition order guard: recovery must not displace the protocol pin, or a user who set + // http1.1 to work around a transport failure loses it on exactly the retry that needs it. + expect(attempts[0]!.protocol).toBe("http1.1"); + expect(attempts[1]!.protocol).toBe("http1.1"); + // Credential-boundary and replayability fields survive the composition. + expect(attempts[1]!.redirect).toBe("manual"); + expect(typeof attempts[1]!.body).toBe("string"); + }); + + test("the routed loop leg replays a reset on a fresh connection through the provider-scoped fetch", async () => { + const attempts: Observed[] = []; + const routedProvider: OcxProviderConfig = { + adapter: "openai-chat", + baseUrl: "https://routed.test/v1", + apiKey: "routed-key", + upstreamHttpVersion: "http1.1", + }; + const providerScopedFetch = (async (_input: string | URL | Request, init?: RequestInit) => { + attempts.push(observe(init)); + if (attempts.length === 1) throw bunResetError(); + return new Response( + 'data: {"choices":[{"delta":{"content":"answer"},"finish_reason":null}]}\n\n' + + 'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n' + + "data: [DONE]\n\n", + { headers: { "Content-Type": "text/event-stream" } }, + ); + }) as typeof fetch; + globalThis.fetch = (async () => { + throw new Error("the routed leg must use the provider-scoped fetch, not the global one"); + }) as typeof fetch; + + const parsed = parseRequest({ model: "routed/model", input: "search please", stream: true }); + const response = await runWithWebSearch({ + parsed, + adapter: createOpenAIChatAdapter(routedProvider), + hostedTool: { type: "web_search" }, + settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 1_000 }, + maxSearches: 1, + selectedForwardHeaders: new Headers(), + forwardProvider: { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", + }, + incomingMeta: { + headers: new Headers(), + translatorBudget: createTestTranslatorBudget(), + providerFetch: withUpstreamHttpVersionExecutor(providerScopedFetch, routedProvider), + }, + }); + + expect(response.status).toBe(200); + await response.text(); + expect(attempts.length).toBeGreaterThanOrEqual(2); + expect(attempts[0]!.keepalive).toBeUndefined(); + expect(attempts[0]!.connection).toBeNull(); + expect(attempts[1]!.keepalive).toBe(false); + expect(attempts[1]!.connection).toBe("close"); + expect(attempts[1]!.protocol).toBe("http1.1"); + // The loop sets accept-encoding: identity so raw byte progress stays observable; the recovery + // helper clones headers into a Headers instance and must not drop it. + expect(typeof attempts[1]!.body).toBe("string"); + }); +}); +