diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index ec88c2eb22..4557af6c03 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -5000,9 +5000,21 @@ async function handleResponsesInner( // through web-search instead of being swallowed. runTurn adapters never enter this branch. if (canRunWebSearch && wsPlan) { parsed.context.tools = [...(parsed.context.tools ?? []), buildWebSearchTool()]; + // Resolve the mutable route at send time: a 429 rotation replaces route.provider, so retaining + // one pre-rotation providerFetch would keep the old credential and transport pin. + const routedProviderFetch = ((input: Parameters[0], init?: RequestInit) => + providerFetch(route.provider, options.codexWsRuntimeIdentity, { + providerName: route.providerName, + modelId: route.modelId, + })(input, init)) as typeof globalThis.fetch; const wsResponse = await runWithWebSearch({ parsed, adapter, - incomingMeta: { headers: selectedForwardHeaders, abortSignal: options.abortSignal, translatorBudget }, + incomingMeta: { + headers: selectedForwardHeaders, + abortSignal: options.abortSignal, + translatorBudget, + providerFetch: routedProviderFetch, + }, backend: wsPlan.backend, forwardProvider: wsPlan.forwardSidecar?.provider, anthropicSidecar: wsPlan.anthropicSidecar, diff --git a/src/web-search/executor.ts b/src/web-search/executor.ts index 672ba72a74..84daf31b9a 100644 --- a/src/web-search/executor.ts +++ b/src/web-search/executor.ts @@ -4,6 +4,7 @@ import { signalWithTimeout, cancelBodyOnAbort } from "../lib/abort"; import { redactSecretString } from "../lib/redact"; import { sidecarEnter } from "../lib/sidecar-tracker"; import { fetchWithResetRetry } from "../lib/upstream-retry"; +import { withUpstreamHttpVersion } from "../lib/upstream-http-version"; import { parseSidecarSSE, type WebSearchResult } from "./parse"; import type { CodexUpstreamOutcome } from "../codex/routing"; @@ -73,7 +74,7 @@ export async function runWebSearch( const t0 = Date.now(); try { const res = await fetchWithResetRetry( - () => fetch(url, { + () => fetch(url, withUpstreamHttpVersion(url, { method: "POST", headers, body: JSON.stringify(body), @@ -82,7 +83,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)), { 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/loop.ts b/src/web-search/loop.ts index 749e97f93d..4b3fde2a82 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -323,6 +323,7 @@ export interface WebSearchLoopDeps { */ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise { const translatorBudget = deps.incomingMeta.translatorBudget; + const routedProviderFetch = deps.incomingMeta.providerFetch ?? globalThis.fetch; const { parsed, selectedForwardHeaders, forwardProvider, hostedTool, settings, maxSearches, abortSignal, recordSidecarOutcome } = deps; const backend = deps.backend ?? "openai"; const anthropicSidecar = deps.anthropicSidecar; @@ -426,6 +427,7 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise { const originalFetch = globalThis.fetch; afterEach(() => { globalThis.fetch = originalFetch; }); +test("issue #2885 — Zhipu-shaped web-search routing preserves the provider HTTP version pin", async () => { + let routedProtocol: string | undefined; + let routedBody: Record | undefined; + const zhipuProvider = { + adapter: "openai-chat", + baseUrl: "https://zhipu.test/v4", + apiKey: "zhipu-key", + upstreamHttpVersion: "http1.1", + models: ["glm-4.7"], + liveModels: false, + fetch: (async (_input: RequestInfo | URL, init?: RequestInit) => { + routedProtocol = (init as RequestInit & { protocol?: string } | undefined)?.protocol; + routedBody = JSON.parse(String(init?.body)) as Record; + return new Response( + 'data: {"choices":[{"delta":{"content":"done"},"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, + } satisfies OcxProviderConfig & { fetch: typeof fetch }; + const cfg: OcxConfig = { + port: 10100, + defaultProvider: "zhipu", + providers: { + zhipu: zhipuProvider, + openai: { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", + }, + }, + }; + globalThis.fetch = (async () => { + throw new Error("the routed web-search leg must use the provider fetch"); + }) as typeof fetch; + + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", + headers: { + "Content-Type": "application/json", + authorization: `Bearer ${fakeChatGptJwt({ chatgpt_account_id: "acct-zhipu-search" })}`, + "chatgpt-account-id": "acct-zhipu-search", + }, + body: JSON.stringify({ + model: "zhipu/glm-4.7", + input: "Search current docs", + stream: true, + tools: [{ type: "web_search" }], + }), + }), cfg, { model: "", provider: "" }); + + expect(response.status).toBe(200); + const frames = await collectSse(response.body!); + expect(frames.some(frame => frame.event === "response.completed")).toBe(true); + expect((routedBody?.tools as { function?: { name?: string } }[] | undefined) + ?.some(tool => tool.function?.name === "web_search")).toBe(true); + expect(routedProtocol).toBe("http1.1"); +}); + +test("web-search adapters receive the provider-scoped fetch executor", async () => { + let routedProtocol: string | undefined; + const pinnedProvider = { + adapter: "openai-chat", + baseUrl: "https://routed.test/v1", + apiKey: "routed-key", + upstreamHttpVersion: "http1.1", + fetch: (async (_input: RequestInfo | URL, init?: RequestInit) => { + routedProtocol = (init as RequestInit & { protocol?: string } | undefined)?.protocol; + return new Response("wire", { status: 200 }); + }) as typeof fetch, + } satisfies OcxProviderConfig & { fetch: typeof fetch }; + const adapter: ProviderAdapter = { + name: "executor-aware", + buildRequest: (_parsed, incoming) => { + expect(incoming.providerFetch).toBeDefined(); + return { url: "https://routed.test/v1/chat/completions", method: "POST", headers: {}, body: "{}" }; + }, + fetchResponse: (request, ctx) => ctx!.executor!(request.url, { + method: request.method, + headers: request.headers, + body: request.body, + signal: ctx?.abortSignal, + }), + async *parseStream() { + yield { type: "text_delta", text: "done" }; + yield { type: "done" }; + }, + }; + + const response = await runWithWebSearch({ + parsed: parseRequest({ model: "routed/model", input: "hi", stream: true, tools: [{ type: "web_search" }] }), + adapter, + incomingMeta: { + headers: new Headers(), + translatorBudget: createTestTranslatorBudget(), + providerFetch: providerFetch(pinnedProvider), + }, + forwardProvider, + hostedTool: { type: "web_search" }, + selectedForwardHeaders: new Headers({ authorization: "Bearer token" }), + settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, + maxSearches: 1, + }); + + expect(response.status).toBe(200); + await collectSse(response.body!); + expect(routedProtocol).toBe("http1.1"); +}); + test("OpenAI web-search execution uses the pinned canonical URL and selected credentials", async () => { const cfg = config({ providers: { @@ -397,16 +510,18 @@ test("OpenAI web-search execution uses the pinned canonical URL and selected cre const candidate = listOpenAiForwardSidecarCandidates(cfg)[0]!; let observedUrl = ""; let observedHeaders = new Headers(); + let observedProtocol: string | undefined; globalThis.fetch = (async (input: string | URL | Request, init?: RequestInit) => { observedUrl = typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; observedHeaders = new Headers(init?.headers); + observedProtocol = (init as RequestInit & { protocol?: string } | undefined)?.protocol; return new Response("data: [DONE]\n\n", { headers: { "content-type": "text/event-stream" } }); }) as typeof fetch; await runOpenAiWebSearch( "current docs", { type: "web_search" }, - candidate.provider, + { ...candidate.provider, upstreamHttpVersion: "http1.1" }, new Headers({ authorization: "Bearer selected-token", "chatgpt-account-id": "selected-account" }), { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 1_000 }, ); @@ -414,6 +529,7 @@ test("OpenAI web-search execution uses the pinned canonical URL and selected cre expect(observedUrl).toBe("https://chatgpt.com/backend-api/codex/responses"); expect(observedHeaders.get("authorization")).toBe("Bearer selected-token"); expect(observedHeaders.get("chatgpt-account-id")).toBe("selected-account"); + expect(observedProtocol).toBe("http1.1"); }); async function collectSse(stream: ReadableStream): Promise<{ event?: string; data: Record }[]> {