From 83a1a94353d2d8dffeca21ded9f1c56f0c2e66c9 Mon Sep 17 00:00:00 2001 From: bitkyc08-arch Date: Sun, 30 Aug 2026 01:58:10 +0900 Subject: [PATCH] fix(upstream): apply fresh-connection recovery on the sidecar and loop retry legs `applyUpstreamRecoveryInit` exists because Bun has ignored a bare hop-by-hop `Connection: close` (oven-sh/bun#20492), so leaving a half-closed pooled socket needs the transport-level `keepalive: false` it adds. The main lanes call it (chat-native, compact, six sites in responses/core). Nine legs did not. The web-search and images loops received the `retryRecovery` argument `fetchWithResetRetry` hands them and spent it on `onAttemptSend` telemetry, then built a plain init. The seven sidecar and vision executors passed zero-argument thunks: they retried, but could not ask for fresh-connection recovery. Every replay on those legs therefore stayed eligible for the same dead socket the reset came from, so recovery was luck, and the retry reported as exhausted rather than as the pool problem it was. All nine now route their init through the helper. On the one leg that also pins a provider HTTP version, recovery nests inside `withUpstreamHttpVersion` so `protocol` and the recovery fields survive together; the reverse order needs a `?? init` fallback to type-check and would drop one of them. Two regressions in tests/web-search.test.ts observe each attempt's init. Both were driven red four ways: reverting the sidecar leg, reverting the loop leg, dropping only `keepalive` while keeping the header, and mis-nesting the composition so the protocol pin is lost. The last mutation also showed the pre-existing #2885 test covers only the routed leg, which is why the new sidecar assertion checks the pin. Normalizing headers changed one observable: three credential canaries in the gemini and exa suites read `init.headers` as a plain record and broke. Those reads now go through `new Headers(...)`, which is representation-independent; each was re-proven to still fail on a substituted bearer, so the invariant is unchanged. Deliberately does not close #2885: this is retry-path parity, not an explanation of the Bun 1.4.0 versus 1.3.14 difference. No wire capture was taken, so whether Bun normalizes the prohibited `Connection` header away under an HTTP/2 pin is unverified; `keepalive: false` is the field that does the work either way. --- .../000_units.md | 87 ++++++++++++ src/images/loop.ts | 9 +- src/vision/anthropic-describe.ts | 6 +- src/vision/describe.ts | 8 +- src/web-search/anthropic-executor.ts | 11 +- src/web-search/exa-executor.ts | 6 +- src/web-search/executor.ts | 10 +- src/web-search/gemini-executor.ts | 6 +- src/web-search/loop.ts | 11 +- src/web-search/xai-executor.ts | 6 +- tests/exa-web-search.test.ts | 3 +- tests/gemini-web-search.test.ts | 11 +- tests/web-search.test.ts | 134 ++++++++++++++++++ 13 files changed, 277 insertions(+), 31 deletions(-) create mode 100644 devlog/_plan/260830_lane_o_reset_recovery_parity/000_units.md 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"); + }); +}); +