diff --git a/docs-site/src/content/docs/reference/configuration/providers.md b/docs-site/src/content/docs/reference/configuration/providers.md index f7e8f16abe..5ef5a8eb07 100644 --- a/docs-site/src/content/docs/reference/configuration/providers.md +++ b/docs-site/src/content/docs/reference/configuration/providers.md @@ -184,6 +184,7 @@ Providers can expose a built-in shorthand, such as `agy` for `google-antigravity | `terminalContinuationGuard?` | `boolean` | Opt in an `openai-chat` provider to one bounded internal re-ask when an actionable turn announces work, then cleanly stops without a tool call. Defaults to `false`; explicit `false` behaves like omission. Combo attempts and routed compaction turns are excluded, and non-`openai-chat` adapters ignore this option. | | `responsesItemIdRepair?` | `{ message?: string[]; reasoning?: string[]; repairMissingTerminalIds?: boolean; repairInvalidIds?: boolean }` | Disabled-by-default downstream SSE repair for exact placeholder ids, missing terminal ids, and (with `repairInvalidIds`) message/reasoning ids missing the canonical `msg_`/`rs_` prefix. Function-call ids are never rewritten. Built-in DeepSeek enables the last two by default. | | `responsesSnapshotRepair?` | `boolean` | Disabled-by-default client-facing repair for sparse Responses lifecycle snapshots in SSE and JSON. Fills missing canonical status, output, and tool metadata while raw inspection and persistence remain unchanged. | +| `webSearchBridge?` | `{ enabled?: boolean; backend?: "ollama"; maxSearches?: number; timeoutMs?: number; endpoint?: string }` | Key-auth `openai-responses` passthrough providers only. Off by default. Codex always declares the hosted `web_search` tool, and the passthrough relays it on the assumption the destination executes it. A gateway that does not (Ollama Cloud GLM/DeepSeek) answers with a `function_call` named `web_search` that nothing runs, and the undeclared-tool guard ends the turn. With `enabled: true` OpenCodex intercepts that call, runs the search itself, feeds the result back to the same upstream, and shows Codex a hosted `web_search_call` cell. Never armed for `authMode: "forward"` (ChatGPT already searches) or for a provider that executes hosted search upstream. Only the `ollama` backend has an executor; the other ids in the union are accepted and stay inert. The `ollama` backend reuses this provider's own API key on `POST /api/web_search`, so the origin must be `https://ollama.com` unless the operator names `endpoint` explicitly. Streaming turns only; a turn that mixes `web_search` with another client tool call fails closed rather than dropping the client's call. Defaults: `maxSearches: 3` (1..10), `timeoutMs: 60000` (1000..600000). | | `retryOn429?` | `{ enabled?: boolean; attempts?: number; intervalMs?: number; maxIntervalMs?: number; respectRetryAfter?: boolean }` | API-key providers only (`authMode: "key"`). Opt-in same-target 429 retry: when `retryOn429` is absent the feature is off; object presence enables it unless `enabled: false`. On 429 the proxy waits (upstream `Retry-After` or the fixed interval) and replays the identical request on the same key before any key failover — across the main text-turn recovery loop, the Responses passthrough wire, the image/video bridge, the web-search sidecar, and terminal continuations. Only pre-stream HTTP 429 responses are eligible for replay; custom `runTurn` transports are outside the HTTP retry loop. `attempts` counts same-key replays after the first 429 (total sends = `attempts` + 1) and is one request-wide budget shared by the main recovery loop, the terminal-guard continuation, and bridge retries. Exhausting `attempts` only stops further same-key replays: normal key failover or final-error handling then applies per the available targets — on the key-auth passthrough wire there is no failover, so the exhausted 429 surfaces as-is. Codex itself never retries 429, so this is the only defense for single-key providers. Defaults: `enabled: true`, `attempts: 3`, `intervalMs: 5000`, `maxIntervalMs: 60000` (any single wait is capped at `maxIntervalMs`, itself capped at 600000), `respectRetryAfter: true`. | | `transientRetryOn5xx?` | `{ enabled?: boolean; attempts?: number }` | Key-auth `openai-chat` providers only. Opt-in retry for pre-stream transient upstream statuses (500, 502, 503, 504, 520, 521, 522): absent means off, object presence enables it unless `enabled: false`. Covers the initial Responses request, the terminal-guard continuation, and native `/v1/chat/completions`. `attempts` is the TOTAL number of upstream sends allowed for one request including the first (1..10, default 3) — it is one budget shared with connection-reset recovery, so `3` means at most three real requests reach the provider. Waits use a fixed 400 ms exponential backoff capped at 5 s and honor `Retry-After`. Separate from `retryOn429`, which handles rate limiting; mid-stream failures are never replayed. | | `autoToolChoiceOnlyModels?` | `string[]` | Models whose `tool_choice` accepts only `auto` or `none`; forced choices are downgraded. | diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 8587431f7d..7e67b920ab 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1271,6 +1271,7 @@ "web-search-backend-union.test.ts": "web-search", "web-search-candidates.test.ts": "web-search", "web-search-parse.test.ts": "web-search", + "web-search-passthrough-bridge.test.ts": "web-search", "web-search-progress-stream.test.ts": "web-search", "web-search-sources.test.ts": "web-search", "web-search-timeout-contract.test.ts": "web-search", diff --git a/src/config.ts b/src/config.ts index 8cfaf63391..1489483e54 100644 --- a/src/config.ts +++ b/src/config.ts @@ -73,6 +73,7 @@ import { MODEL_ADAPTER_OVERRIDE_ALLOWED, OPENAI_PROVIDER_TIER_VERSION, pinnedWireAdapter, + PROVIDER_WEB_SEARCH_BRIDGE_BACKENDS, UPSTREAM_HTTP_VERSION_VALUES, type OcxClaudeCodeConfig, type OcxConfig, @@ -497,6 +498,47 @@ export function requestPacingConfigError(value: unknown): string | null { return "requestPacing must contain enabled and a valid requestsPerMinute/minIntervalMs provider rule or model overrides"; } +/** + * Bounds for the opt-in passthrough web-search bridge (`providers..webSearchBridge`, + * #3761). Strict for the same reason `retryOn429` is: a misspelled key here would silently + * leave the bridge disarmed while the operator believes they enabled it. `endpoint` is only + * shape-checked here; `planPassthroughWebSearchBridge` re-validates the origin before any key + * is sent to it, because config validation is not an authorization boundary. + */ +const providerWebSearchBridgeSchema = z.object({ + enabled: z.boolean().optional(), + backend: z.enum(PROVIDER_WEB_SEARCH_BRIDGE_BACKENDS).optional(), + maxSearches: z.number().int().min(1).max(10).optional(), + timeoutMs: z.number().int().min(1_000).max(600_000).optional(), + endpoint: z.string().min(1).optional(), +}).strict(); + +export function providerWebSearchBridgeConfigError(value: unknown): string | null { + if (value === undefined) return null; + if (!value || typeof value !== "object" || Array.isArray(value)) { + return "webSearchBridge must be a plain object"; + } + const parsed = providerWebSearchBridgeSchema.safeParse(value); + if (!parsed.success) { + return "webSearchBridge accepts only enabled (boolean), backend " + + `(${PROVIDER_WEB_SEARCH_BRIDGE_BACKENDS.join("|")}), maxSearches (1..10), ` + + "timeoutMs (1000..600000), and endpoint (absolute http(s) URL)"; + } + const endpoint = parsed.data.endpoint; + if (endpoint !== undefined) { + let url: URL; + try { + url = new URL(endpoint); + } catch { + return "webSearchBridge.endpoint must be an absolute http(s) URL"; + } + if (url.protocol !== "https:" && url.protocol !== "http:") { + return "webSearchBridge.endpoint must be an absolute http(s) URL"; + } + } + return null; +} + const fastWireSchema = z.object({ kind: z.string(), canonicalToWire: z.record(z.string().trim(), z.string().trim()), @@ -600,6 +642,10 @@ const providerConfigSchema = z.object({ repairInvalidIds: z.boolean().optional(), }).strict().optional(), responsesSnapshotRepair: z.boolean().optional(), + // Invalid blocks degrade to "absent" rather than failing the whole config load: an unusable + // bridge block must never send an operator through invalid-config recovery for an opt-in + // feature that is off by default. The management write boundary still rejects it loudly. + webSearchBridge: providerWebSearchBridgeSchema.optional().catch(undefined), xaiResponsesXSearch: z.boolean().optional(), xaiResponsesDefaultVersion: z.number().int().positive().optional().catch(undefined), }).passthrough(); diff --git a/src/server/auth-cors.ts b/src/server/auth-cors.ts index 476fd3a4ae..d103721f35 100644 --- a/src/server/auth-cors.ts +++ b/src/server/auth-cors.ts @@ -6,6 +6,7 @@ import { codexAutoStartEnabled, modelPreferHostedToolsConfigError, providerModelCostsConfigError, + providerWebSearchBridgeConfigError, requestPacingConfigError, retryOn429PolicyConfigError, sanitizeModelCostsForDisplay, @@ -650,6 +651,10 @@ export function providerManagementConfigError(name: unknown, provider: unknown): if (requestPacingError) { return `provider ${JSON.stringify(redactSecretString(name))} ${requestPacingError}`; } + const webSearchBridgeError = providerWebSearchBridgeConfigError(raw.webSearchBridge); + if (webSearchBridgeError) { + return `provider ${JSON.stringify(redactSecretString(name))} ${webSearchBridgeError}`; + } const upstreamHttpVersionError = upstreamHttpVersionConfigError(raw.upstreamHttpVersion); if (upstreamHttpVersionError) { return `provider ${JSON.stringify(redactSecretString(name))} ${upstreamHttpVersionError}`; @@ -847,6 +852,7 @@ const PROVIDER_CONFIG_FIELD_POLICY = { xaiResponsesDefaultVersion: "runtime", supportsResponsesCustomTools: "editor", responsesSnapshotRepair: "editor", + webSearchBridge: "editor", reasoningEffortMap: "editor", modelReasoningEffortMap: "editor", reasoningWireFormat: "editor", diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index 73be9a86dd..1b2bf3db58 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -150,6 +150,11 @@ import { } from "../../oauth/generic-account-failover"; import { resolveCopilotApiBaseUrl } from "../../oauth/github-copilot"; import { buildWebSearchTool, planWebSearch, runWithWebSearch, shouldResolveOpenAiWebSearchSidecar } from "../../web-search"; +import { + createOllamaBridgeExecutor, + createPassthroughWebSearchBridgeStream, + planPassthroughWebSearchBridge, +} from "../../web-search/passthrough-bridge"; import { buildImageTool, buildVideoTool, planImageBridge, planVideoBridge, runWithImageBridge, clampImageMaxRounds, IMAGE_GEN_TOOL_NAME, VIDEO_GEN_TOOL_NAME } from "../../images"; import { describeImagesInPlace, isModelTextOnly, planVisionSidecar, resolveOpenAiVisionModel, shouldResolveOpenAiVisionSidecar, stripImagesInPlace } from "../../vision"; import { createAdapterEventQueue, preflightAdapterEvents, type AdapterEventQueue } from "../../adapters/run-turn-queue"; @@ -5762,15 +5767,60 @@ async function handleResponsesInner( route.provider, route.modelId, ); + // #3761: opt-in hosted-web-search bridge. Codex always declares the hosted web_search tool, + // and this branch relays that declaration on the assumption the destination executes it. + // A KEY-auth gateway that does not (Ollama Cloud GLM) answers with a function_call named + // web_search that nothing runs, and the undeclared-tool guard below ends the turn. When the + // provider opts in, the bridge intercepts that one call, runs the search, continues the + // conversation upstream, and hands back ordinary Responses SSE — so every rewrite below, + // including the guard itself, still inspects the client-facing stream. Default OFF: without + // the opt-in this is one planner call and the relay is byte-identical to before. + const webSearchBridgePlan = planPassthroughWebSearchBridge(parsed, route.provider, { + isPassthrough: true, + stream: parsed.stream === true, + }); + // The bridge wraps the RAW upstream body, so terminal repair below still owns the single + // client-facing terminal — the bridge drops the terminal of every intercepted leg. + const upstreamSseBody = webSearchBridgePlan + ? createPassthroughWebSearchBridgeStream({ + plan: webSearchBridgePlan, + firstLeg: upstreamResponse.body, + requestBody: request.body, + // Continuation legs replay the same built request with the executed search appended. + // The first leg already passed the recovery ladder, the outbound size ceiling, and the + // host circuit; a KEY-auth destination has no OAuth refresh to replay on a later leg. + send: (continuationBody: string) => fetchWithHeaderTimeout( + request.url, + { method: request.method, headers: request.headers, body: continuationBody }, + upstream.signal, + connectMs, + true, + providerFetch(route.provider, options.codexWsRuntimeIdentity, { + dispatchOverride: oauthDispatch(request), + providerName: route.providerName, + modelId: route.modelId, + }), + false, + ), + execute: createOllamaBridgeExecutor(webSearchBridgePlan, route.provider.apiKey ?? ""), + // Appending a search result can push the continuation past the ceiling the first leg + // was admitted under, so the same limit is re-applied before every later send. + checkOutboundBody: (continuationBody: string) => { + const result = checkOutboundBodySize(continuationBody, config.maxUpstreamBodyBytes); + return result.admitted ? undefined : describeOutboundBodyRefusal(result); + }, + signal: upstream.signal, + }) + : upstreamResponse.body; const passthroughSseBody = terminalRepairPolicy ? relayResponsesSseWithTerminalRepair( - upstreamResponse.body, + upstreamSseBody, upstream, terminalRepairPolicy, translatorBudget, options.responsesTerminalRepairScheduler, ) - : upstreamResponse.body; + : upstreamSseBody; const repairConfig = route.provider.responsesItemIdRepair; // Grok Build renders deltas live but reconstructs its durable assistant // turn from the completed response snapshot. Native Responses streams diff --git a/src/types.ts b/src/types.ts index f759406fe0..d4b937d040 100644 --- a/src/types.ts +++ b/src/types.ts @@ -101,6 +101,8 @@ export type { ResponsesItemIdRepairConfig, RateLimitRetryPolicy, TransientRetryPolicy, + ProviderWebSearchBridgeBackend, + ProviderWebSearchBridgeConfig, ProviderCostOverlay, RequestPacingRule, ProviderRequestPacingConfig, @@ -111,6 +113,8 @@ export type { OcxProviderConfig, } from "./types/provider"; +export { PROVIDER_WEB_SEARCH_BRIDGE_BACKENDS } from "./types/provider"; + export type { CodexAccount, CodexAccountCredentials, diff --git a/src/types/provider.ts b/src/types/provider.ts index b51230d6d1..0beb5d3371 100644 --- a/src/types/provider.ts +++ b/src/types/provider.ts @@ -86,6 +86,57 @@ export interface RateLimitRetryPolicy { respectRetryAfter?: boolean; } +/** + * Backend ids admitted by `providers..webSearchBridge.backend`. Only `"ollama"` has a + * shipped executor; every other id is explicit-only and inert, the same contract the top-level + * `webSearchSidecar` uses for backends whose executor has not landed. Naming one of them keeps + * the bridge disarmed rather than silently falling back to a different search provider — in + * particular it never auto-selects a paid Luna or Exa search. + */ +export const PROVIDER_WEB_SEARCH_BRIDGE_BACKENDS = [ + "ollama", + "openai", + "anthropic", + "xai", + "gemini", + "exa", +] as const; + +export type ProviderWebSearchBridgeBackend = typeof PROVIDER_WEB_SEARCH_BRIDGE_BACKENDS[number]; + +/** + * Opt-in hosted-web-search bridge for a KEY-auth Responses passthrough provider + * (`providers..webSearchBridge`), default OFF (#3761). + * + * Codex always declares the hosted `{type:"web_search"}` tool. On the passthrough the proxy + * treats that as "the destination runs search itself" and relays it unchanged, which is true for + * the ChatGPT backend and for xAI but false for an OpenAI-shaped key gateway such as Ollama + * Cloud: the model answers with a `function_call` named `web_search` that nothing executes, + * and the undeclared-tool guard ends the turn. With this block enabled the proxy intercepts that + * call, runs the configured search backend itself, feeds the result back upstream, and shows + * Codex a hosted `web_search_call` cell. + * + * Never armed for `authMode: "forward"` (ChatGPT) or for a provider that executes hosted search + * upstream; see `planPassthroughWebSearchBridge` in `src/web-search/passthrough-bridge.ts`. + */ +export interface ProviderWebSearchBridgeConfig { + /** Master switch. Absent or false keeps today's relay-and-fail behavior exactly. */ + enabled?: boolean; + /** Which executor runs the search. Absent disarms the bridge; there is no implicit default. */ + backend?: ProviderWebSearchBridgeBackend; + /** Searches executed per turn before the bridge refuses further ones (1..10, default 3). */ + maxSearches?: number; + /** Per-search deadline in milliseconds (1000..600000, default 60000). */ + timeoutMs?: number; + /** + * Absolute search-API URL. Required to use the `ollama` backend against anything other than + * the canonical `https://ollama.com` origin, which is the only origin derived automatically. + * The bridge sends the PROVIDER's own API key to this URL, so an operator setting it is + * authorizing that key for this destination. + */ + endpoint?: string; +} + /** * User-configured display price for one model (USD per 1M tokens). * Mirrors the `Cost4` shape used by the usage cost estimator; structurally @@ -550,6 +601,11 @@ export interface OcxProviderConfig { * SSE/JSON; raw inspection state remains authoritative. */ responsesSnapshotRepair?: boolean; + /** + * Opt-in hosted-web-search bridge for this KEY-auth Responses passthrough provider (#3761). + * Absent or disabled leaves the passthrough byte-identical to today. + */ + webSearchBridge?: ProviderWebSearchBridgeConfig; /** * Provider-wide mapping from Codex effort labels to upstream `reasoning_effort` values. * Map a label to the reserved value `"__omit__"` to send no reasoning field at all for that diff --git a/src/web-search/ollama-executor.ts b/src/web-search/ollama-executor.ts new file mode 100644 index 0000000000..7ebddb6806 --- /dev/null +++ b/src/web-search/ollama-executor.ts @@ -0,0 +1,127 @@ +/** + * Execute ONE web search via the Ollama web-search API — the executor behind + * `providers..webSearchBridge.backend: "ollama"` (#3761). + * + * Documented contract: POST /api/web_search with a bearer key returns + * {results: [{title, url, content}]}; `max_results` defaults to 5 and caps at 10 + * (https://docs.ollama.com/web-search). Like Exa, this lane returns ranked results + * rather than a prose answer, so the outcome text is a digest the routed model + * synthesizes from. + * + * The key is the PROVIDER's own API key: an operator who enables the bridge is reusing + * their Ollama Cloud route key on a second Ollama endpoint. That is why the planner + * refuses to derive a non-canonical origin on its own. + * + * Never throws; every error string passes redactSecretString and scrubs the literal key. + */ +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"; +import { redactSecretString } from "../lib/redact"; +import { MAX_SIDECAR_RESPONSE_BYTES, type WebSearchSource } from "./parse"; +import type { SidecarOutcome } from "./executor"; + +/** Documented ceiling for the API's own `max_results`; a larger value is rejected upstream. */ +export const OLLAMA_WEB_SEARCH_MAX_RESULTS = 5; +const OLLAMA_SNIPPET_CHARS = 1000; + +function isRec(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} + +export async function runOllamaWebSearch( + query: string, + apiKey: string, + endpoint: string, + timeoutMs: number, + abortSignal?: AbortSignal, +): Promise { + if (!apiKey) { + return { text: "", sources: [], error: "ollama web-search backend selected without a provider apiKey" }; + } + // The executor KNOWS the secret, so pattern-based redaction is not enough: scrub the + // literal value before anything derived from an upstream body is returned. + const scrub = (value: string) => + redactSecretString(value.split(apiKey).join("[redacted-provider-key]")); + const linkedSignal = signalWithTimeout(timeoutMs, abortSignal); + const sidecarExit = sidecarEnter("web-search"); + const startedAt = Date.now(); + try { + const res = await fetchWithResetRetry( + recovery => fetch(endpoint, applyUpstreamRecoveryInit({ + method: "POST", + headers: { "Content-Type": "application/json", Authorization: `Bearer ${apiKey}` }, + body: JSON.stringify({ query, max_results: OLLAMA_WEB_SEARCH_MAX_RESULTS }), + signal: linkedSignal.signal, + // Bun forwards custom headers across redirects, so a redirect would leak the key. + redirect: "manual", + }, recovery)), + { abortSignal: linkedSignal.signal, label: "ollama-web-search-bridge" }, + ); + const detachBodyGuard = cancelBodyOnAbort(res.body, linkedSignal.signal); + try { + let bounded: Awaited> | null = null; + try { + bounded = await readBoundedResponseBytes(res, { + maxBytes: MAX_SIDECAR_RESPONSE_BYTES, + signal: linkedSignal.signal, + }); + } catch { + const reason = linkedSignal.signal.reason; + if (linkedSignal.signal.aborted && reason instanceof Error && reason.name === "TimeoutError") { + throw reason; + } + // A body-read failure degrades to the status-only outcome below. + } + if (bounded?.oversized) { + const prefix = res.ok ? "ollama web-search response" : `ollama web-search HTTP ${res.status} response`; + return { text: "", sources: [], error: `${prefix} exceeded byte bound` }; + } + const text = bounded ? new TextDecoder().decode(bounded.bytes) : ""; + if (!res.ok) { + // Scrub BEFORE truncating: slicing first can cut the literal key at the boundary + // and leave an unscrubbable prefix in the surviving text. + return { text: "", sources: [], error: `ollama web-search HTTP ${res.status}: ${scrub(text).slice(0, 200)}` }; + } + let payload: unknown = null; + try { + payload = JSON.parse(text); + } catch { + // The mapper owns the stable malformed/empty JSON outcome. + } + return mapOllamaSearchResponse(payload); + } finally { + detachBodyGuard(); + } + } catch (error) { + const kind = error instanceof Error && error.name === "TimeoutError" ? "timeout" : "connect_error"; + console.warn(`[web-search] ollama bridge ${kind} (${Date.now() - startedAt}ms)`); + return { text: "", sources: [], error: scrub(error instanceof Error ? error.message : String(error)) }; + } finally { + sidecarExit(); + linkedSignal.cleanup(); + } +} + +/** Map an Ollama /api/web_search payload to a digest the routed model can synthesize from. */ +export function mapOllamaSearchResponse(payload: unknown): SidecarOutcome { + if (!isRec(payload) || !Array.isArray(payload.results)) { + return { text: "", sources: [], error: "ollama web-search returned a non-JSON or shapeless body" }; + } + const sources: WebSearchSource[] = []; + const lines: string[] = []; + const seen = new Set(); + for (const result of payload.results) { + if (!isRec(result) || typeof result.url !== "string" || result.url.length === 0) continue; + if (seen.has(result.url)) continue; + seen.add(result.url); + const title = typeof result.title === "string" && result.title.length > 0 ? result.title : result.url; + sources.push({ url: result.url, ...(title !== result.url ? { title } : {}) }); + const snippet = typeof result.content === "string" ? result.content.trim().slice(0, OLLAMA_SNIPPET_CHARS) : ""; + lines.push(`- ${title}: ${snippet || "(no excerpt)"} [${result.url}]`); + } + if (lines.length === 0) return { text: "", sources: [], error: "ollama web-search returned no results" }; + return { text: `Search results:\n${lines.join("\n")}`, sources }; +} + diff --git a/src/web-search/passthrough-bridge.ts b/src/web-search/passthrough-bridge.ts new file mode 100644 index 0000000000..6212e25c82 --- /dev/null +++ b/src/web-search/passthrough-bridge.ts @@ -0,0 +1,761 @@ +/** + * Hosted-web-search bridge for the KEY-auth Responses passthrough (#3761). + * + * The Codex App always declares the hosted "{type:'web_search'}" tool. On the passthrough the + * proxy reads that declaration as "the destination executes search itself" and relays it + * unchanged, which is correct for the ChatGPT backend and for xAI. It is wrong for an + * OpenAI-shaped KEY gateway that does not run the hosted tool: Ollama Cloud GLM answers with a + * plain "{type:'function_call', name:'web_search'}", nothing on either side executes it, and the + * undeclared-tool guard ends the turn because a hosted declaration never authorizes a client + * function name. + * + * This module is the opt-in repair, armed only by "providers..webSearchBridge.enabled". + * It intercepts that one call out of the upstream stream, runs the configured search backend + * itself, feeds the call and its result back to the SAME upstream in a fresh POST, and shows + * Codex the hosted "web_search_call" cell it already understands. The offending function_call + * never reaches the client, and "web_search" is never added to the guard's allowed names -- + * doing that would authorize a call nobody can execute rather than removing it. + * + * Deliberate boundaries of this first slice: + * - Streaming SSE turns only. A non-streaming turn stays on the existing path. + * - A leg that mixes the search call with any OTHER client tool call fails closed with an + * explicit error. Answering both would need the raw mixed-tool continuation contract the + * 2.47 track deferred (devlog/_plan/260907_track2_protocol/040_hosted_search_disposition.md), + * and silently half-doing it would drop the client's own tool call. +* - Continuation legs use a direct send rather than the core recovery ladder: the first leg +* still goes through it, and a KEY-auth destination has no OAuth refresh path to replay. + * The caller's outbound body ceiling is re-applied to every continuation body. + * - The client stream is renumbered (sequence_number and output_index) because events are both + * dropped and injected; a plain relay cannot preserve upstream numbering through that. + * + * The stream this module produces is ordinary Responses SSE and is handed back to the core relay, + * so the undeclared-tool guard, the provider payload rewrites, terminal-outcome recording, and the + * continuation cache all still apply to it. That is what keeps the guard's authority intact over + * every OTHER call an upstream emits: the bridge removes only the web_search call it executes. + */ +import { nextSseBlock, sseDataPayload } from "../server/sse-payload-rewrite"; +import { toolChoiceToolPredicate } from "../types"; +import type { OcxParsedRequest, OcxProviderConfig, ProviderWebSearchBridgeBackend } from "../types"; +import type { SidecarOutcome } from "./executor"; +import { buildWebSearchTool, WEB_SEARCH_TOOL_NAME } from "./synthetic-tool"; +import { safeWebSearchSources } from "./sources"; +import { runOllamaWebSearch } from "./ollama-executor"; + +/** Canonical Ollama Cloud origin. The only origin the "ollama" backend derives on its own. */ +export const OLLAMA_CLOUD_ORIGIN = "https://ollama.com"; +const OLLAMA_WEB_SEARCH_PATH = "/api/web_search"; + +const DEFAULT_BRIDGE_MAX_SEARCHES = 3; +const DEFAULT_BRIDGE_TIMEOUT_MS = 60_000; +/** Queries honored from one call's "queries" array; the rest are ignored rather than billed. */ +const MAX_QUERIES_PER_CALL = 3; +/** Hard ceiling on retained client-visible items before the terminal snapshot rewrite is skipped. */ +const MAX_RETAINED_OUTPUT_ITEMS = 500; +/** Refuse to buffer an unbounded partial SSE event from a misbehaving upstream. */ +const MAX_SSE_BUFFER_CHARS = 8 * 1024 * 1024; + +export const WEB_SEARCH_BRIDGE_MIXED_TOOLS_ERROR_CODE = "web_search_bridge_mixed_tools"; +export const WEB_SEARCH_BRIDGE_ERROR_CODE = "web_search_bridge_failed"; + +/** Item types whose calls the CLIENT has to execute; any of them alongside a search is mixed. */ +const CLIENT_EXECUTED_ITEM_TYPES = new Set([ + "function_call", + "custom_tool_call", + "local_shell_call", + "tool_search_call", + "computer_call", +]); + +export interface PassthroughWebSearchBridgePlan { + /** Resolved executor id. Only "ollama" has a shipped executor today. */ + backend: ProviderWebSearchBridgeBackend; + /** Absolute search-API URL the executor posts to. */ + endpoint: string; + /** Searches actually executed per turn before further calls are refused. */ + maxSearches: number; + /** Per-search deadline in milliseconds. */ + timeoutMs: number; +} + +function isRecord(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} + +function originOf(value: string | undefined): string | undefined { + if (!value) return undefined; + try { + const url = new URL(value); + if (url.protocol !== "https:" && url.protocol !== "http:") return undefined; + return url.origin; + } catch { + return undefined; + } +} + +/** + * Resolve the search endpoint for the "ollama" backend. + * + * An explicit "endpoint" is the operator's own authorization: they are naming the destination + * that receives this provider's API key. Without one, the origin must be canonical Ollama Cloud + * -- a renamed row pointing at an arbitrary host must not silently receive the key just because + * its adapter happens to be openai-responses. + */ +export function resolveOllamaWebSearchEndpoint( + provider: OcxProviderConfig, +): string | undefined { + const configured = provider.webSearchBridge?.endpoint; + if (configured !== undefined) { + return originOf(configured) === undefined ? undefined : configured; + } + return originOf(provider.baseUrl) === OLLAMA_CLOUD_ORIGIN + ? OLLAMA_CLOUD_ORIGIN + OLLAMA_WEB_SEARCH_PATH + : undefined; +} + +/** + * Decide whether this passthrough turn may run the web-search bridge. + * + * Fails closed on every axis. In particular it never arms for "authMode: 'forward'": that is the + * ChatGPT backend speaking Codex's own protocol with the caller's own credential, and it executes + * hosted search upstream. A provider that runs hosted search itself (xAI) also stays on the + * existing relay, because arming here would replace a real provider-side search with ours. + * + * This is a NEW planner rather than a relaxation of "isPassthrough" in planWebSearch: the sidecar + * rewrites normalized messages, while this path must preserve the raw Responses conversation. + */ +export function planPassthroughWebSearchBridge( + parsed: OcxParsedRequest, + provider: OcxProviderConfig, + options: { isPassthrough: boolean; stream: boolean }, +): PassthroughWebSearchBridgePlan | undefined { + if (!options.isPassthrough || !options.stream) return undefined; + if (!parsed._webSearch) return undefined; + // Never spend a forwarded ChatGPT credential on a proxy-run search, and never pre-empt a + // provider that executes the hosted tool itself. + if (provider.authMode !== "key") return undefined; + const bridge = provider.webSearchBridge; + if (!bridge || bridge.enabled !== true) return undefined; + // A tool_choice that excludes web search excludes the bridge too; the model may not search. + if (!toolChoiceToolPredicate(parsed.options.toolChoice)(buildWebSearchTool())) return undefined; + // Explicit-only, and inert for every backend whose executor has not shipped. + if (bridge.backend !== "ollama") return undefined; + const endpoint = resolveOllamaWebSearchEndpoint(provider); + if (!endpoint) return undefined; + const maxSearches = Number.isInteger(bridge.maxSearches) + && bridge.maxSearches! >= 1 + && bridge.maxSearches! <= 10 + ? bridge.maxSearches! + : DEFAULT_BRIDGE_MAX_SEARCHES; + const timeoutMs = Number.isInteger(bridge.timeoutMs) + && bridge.timeoutMs! >= 1_000 + && bridge.timeoutMs! <= 600_000 + ? bridge.timeoutMs! + : DEFAULT_BRIDGE_TIMEOUT_MS; + return { backend: "ollama", endpoint, maxSearches, timeoutMs }; +} + +/** One intercepted search call, carried from the upstream stream into the next request body. */ +export interface InterceptedSearchCall { + callId: string; + argumentsText: string; + /** Upstream item id of the call this bridge answered; replayed on the continuation item. */ + sourceItemId?: string; + /** Client-facing hosted cell opened in place of the intercepted call. */ + cellItemId: string; + cellOutputIndex: number; + /** Slot reserved in the terminal snapshot so the cell keeps its streamed position. */ + retainedSlot?: number; +} + +export type PassthroughWebSearchBridgeExecutor = ( + queries: string[], + signal?: AbortSignal, +) => Promise; + +export interface PassthroughWebSearchBridgeStreamOptions { + plan: PassthroughWebSearchBridgePlan; + /** The already-open first upstream leg, obtained through the normal core send path. */ + firstLeg: ReadableStream; + /** The exact outbound body that produced the first leg; continuation legs extend it. */ + requestBody: string; + /** Sends one continuation leg and resolves with its response. */ + send: (body: string) => Promise; + execute: PassthroughWebSearchBridgeExecutor; + /** + * Re-applies the caller's outbound body ceiling to a continuation body. Returns a refusal + * message when the extended body may not be sent, or undefined when it is admitted. + */ + checkOutboundBody?: (body: string) => string | undefined; + signal?: AbortSignal; +} + +function parseQueries(argumentsText: string): string[] { + let parsed: unknown; + try { + parsed = JSON.parse(argumentsText); + } catch { + // A non-JSON argument blob is still a search intent; treat the raw text as the query. + const trimmed = argumentsText.trim(); + return trimmed.length > 0 ? [trimmed.slice(0, 1_000)] : []; + } + if (!isRecord(parsed)) return []; + const queries: string[] = []; + const push = (value: unknown): void => { + if (typeof value !== "string") return; + const trimmed = value.trim(); + if (trimmed.length === 0 || queries.includes(trimmed)) return; + if (queries.length < MAX_QUERIES_PER_CALL) queries.push(trimmed.slice(0, 1_000)); + }; + push(parsed.query); + if (Array.isArray(parsed.queries)) for (const entry of parsed.queries) push(entry); + return queries; +} + +function isWebSearchCallItem(item: unknown): boolean { + if (!isRecord(item)) return false; + if (item.type !== "function_call" && item.type !== "custom_tool_call") return false; + // A namespaced "ns__web_search" is a different tool identity that the client declared and + // executes itself; intercepting it would steal a call the client owns. + if (typeof item.namespace === "string") return false; + return item.name === WEB_SEARCH_TOOL_NAME; +} + +function isClientExecutedItem(item: unknown): boolean { + return isRecord(item) && typeof item.type === "string" && CLIENT_EXECUTED_ITEM_TYPES.has(item.type); +} + +/** Yield complete SSE event blocks. */ +async function* readSseBlocks( + body: ReadableStream, +): AsyncGenerator<{ block: string }> { + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + const drain = function* (): Generator<{ block: string }> { + let next: ReturnType; + while ((next = nextSseBlock(buffer))) { + buffer = next.rest; + yield { block: next.block }; + } + }; + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) { + buffer += decoder.decode(); + yield* drain(); + if (buffer.length > 0) { + yield { block: buffer }; + } + return; + } + buffer += decoder.decode(value, { stream: true }); + if (buffer.length > MAX_SSE_BUFFER_CHARS) { + throw new Error("upstream SSE event exceeded the web-search bridge buffer bound"); + } + yield* drain(); + } + } finally { + reader.cancel().catch(() => {}); + } +} +interface LegDecision { + kind: "end" | "continue" | "fail"; + searches: InterceptedSearchCall[]; + message?: string; + code?: string; +} + +/** One client-executed call event held until the leg's fate is known. */ +interface HeldCallEvent { + payload: Record; + upstreamIndex?: number; +} + +/** + * Stateful client-stream builder for one bridged turn. + * + * Owns the two numbering spaces the client sees. Upstream indices are per-leg and include items + * this bridge removes or injects, so every emitted event is remapped onto one monotonic client + * sequence. Client output_index is assigned at EMIT time, which is what keeps a held call's index + * consistent with the order the client actually receives. + */ +class BridgeStreamState { + /** Client-facing sequence_number, rewritten on every emitted payload. */ + private sequence = 0; + /** Next unused client output_index. */ + private outputIndex = 0; + /** Client-visible finished items, used to rebuild the terminal snapshot after an injection. */ + private readonly retainedItems: unknown[] = []; + private retainedItemsComplete = true; + private injected = false; + + /** Per-leg upstream output_index -> client output_index. */ + private indexMap = new Map(); + /** Upstream output_index -> the intercepted call that owns it, so parallel calls stay distinct. */ + private suppressedSearches = new Map(); + /** Item ids of intercepted calls, for events that carry item_id but no output_index. */ + private suppressedItemIds = new Map(); + private searches: InterceptedSearchCall[] = []; + /** + * Client-executed calls are withheld until the leg's fate is known. Emitting one and THEN + * failing the turn would let Codex start running a tool for a turn that never completes. + */ + private heldCalls: HeldCallEvent[] = []; + private heldIndexes = new Set(); + private heldItemIds = new Set(); + private terminalPayload: Record | undefined; + + beginLeg(): void { + this.indexMap = new Map(); + this.suppressedSearches = new Map(); + this.suppressedItemIds = new Map(); + this.searches = []; + this.heldCalls = []; + this.heldIndexes = new Set(); + this.heldItemIds = new Set(); + this.terminalPayload = undefined; + } + + get sawClientExecutedCall(): boolean { + return this.heldCalls.length > 0; + } + + private clientIndexFor(upstreamIndex: number): number { + const existing = this.indexMap.get(upstreamIndex); + if (existing !== undefined) return existing; + const assigned = this.outputIndex++; + this.indexMap.set(upstreamIndex, assigned); + return assigned; + } + + /** Reserve a retained-snapshot slot so an item injected later keeps its streamed position. */ + private reserveRetainedSlot(): number | undefined { + if (!this.retainedItemsComplete) return undefined; + if (this.retainedItems.length >= MAX_RETAINED_OUTPUT_ITEMS) { + this.retainedItemsComplete = false; + return undefined; + } + return this.retainedItems.push(undefined) - 1; + } + + private retain(item: unknown, slot?: number): void { + if (!this.retainedItemsComplete) return; + if (slot !== undefined) { + this.retainedItems[slot] = item; + return; + } + if (this.retainedItems.length >= MAX_RETAINED_OUTPUT_ITEMS) { + this.retainedItemsComplete = false; + return; + } + this.retainedItems.push(item); + } + + private render(type: string, data: Record): string { + return "event: " + type + "\n" + + "data: " + JSON.stringify({ ...data, type, sequence_number: this.sequence++ }); + } + + failureFrames(code: string, message: string): string[] { + const failure = { type: "upstream_error", code, message }; + return [ + this.render("response.failed", { + response: { status: "failed", error: failure, last_error: failure }, + }), + "data: [DONE]", + ]; + } + + searchEndFrames(call: InterceptedSearchCall, queries: string[], outcome: SidecarOutcome): string[] { + const sources = safeWebSearchSources(outcome.sources); + const first = queries[0] ?? ""; + const item = { + type: "web_search_call", + id: call.cellItemId, + status: outcome.error ? "failed" : "completed", + action: { type: "search", query: first, queries: queries.length > 0 ? queries : [first] }, + ...(sources.length > 0 ? { sources } : {}), + }; + this.retain(item, call.retainedSlot); + return [this.render("response.output_item.done", { output_index: call.cellOutputIndex, item })]; + } + + /** + * Translate one upstream block into the blocks the client should receive now. + * + * Search-call events are replaced in place by the hosted cell's opening frame. Client-executed + * call events are withheld. The terminal is held: whether it ends the turn is only decided once + * the whole leg has been read. + */ + consume(block: string, isFirstLeg: boolean): string[] { + const data = sseDataPayload(block); + if (data === null) return [block]; + if (data === "[DONE]") return []; + let payload: unknown; + try { + payload = JSON.parse(data); + } catch { + return [block]; + } + if (!isRecord(payload) || typeof payload.type !== "string") return [block]; + + // A continuation leg opens its own response lifecycle; the client already has one. + if ((payload.type === "response.created" || payload.type === "response.in_progress") && !isFirstLeg) { + return []; + } + if (payload.type === "response.completed" + || payload.type === "response.incomplete" + || payload.type === "response.failed") { + this.terminalPayload = payload; + return []; + } + + const upstreamIndex = typeof payload.output_index === "number" ? payload.output_index : undefined; + const itemId = typeof payload.item_id === "string" ? payload.item_id : undefined; + + if (payload.type === "response.output_item.added" && isRecord(payload.item)) { + const item = payload.item; + if (isWebSearchCallItem(item)) { + // Open the hosted cell exactly where the intercepted call stood, so a search that is not + // the last item of the turn keeps its position instead of being appended after it. + const cellOutputIndex = this.outputIndex++; + const intercepted: InterceptedSearchCall = { + callId: typeof item.call_id === "string" ? item.call_id : "", + sourceItemId: typeof item.id === "string" ? item.id : undefined, + argumentsText: typeof item.arguments === "string" ? item.arguments : "", + cellItemId: "ws_" + crypto.randomUUID(), + cellOutputIndex, + retainedSlot: this.reserveRetainedSlot(), + }; + if (upstreamIndex !== undefined) this.suppressedSearches.set(upstreamIndex, intercepted); + if (intercepted.sourceItemId) this.suppressedItemIds.set(intercepted.sourceItemId, intercepted); + this.searches.push(intercepted); + this.injected = true; + return [this.render("response.output_item.added", { + output_index: cellOutputIndex, + item: { type: "web_search_call", id: intercepted.cellItemId, status: "in_progress" }, + })]; + } + if (isClientExecutedItem(item)) { + if (upstreamIndex !== undefined) this.heldIndexes.add(upstreamIndex); + if (typeof item.id === "string") this.heldItemIds.add(item.id); + this.heldCalls.push({ payload, ...(upstreamIndex === undefined ? {} : { upstreamIndex }) }); + return []; + } + } + + const pending = (upstreamIndex === undefined ? undefined : this.suppressedSearches.get(upstreamIndex)) + ?? (itemId === undefined ? undefined : this.suppressedItemIds.get(itemId)); + if (pending) { + // Argument deltas and the matching done frame belong to a call the client never sees; + // the done frame still carries the authoritative complete arguments. + if (payload.type === "response.output_item.done" && isRecord(payload.item)) { + const args = payload.item.arguments; + if (typeof args === "string" && args.length > 0) pending.argumentsText = args; + } + if (payload.type === "response.function_call_arguments.done" && typeof payload.arguments === "string") { + if (payload.arguments.length > 0) pending.argumentsText = payload.arguments; + } + return []; + } + + if ((upstreamIndex !== undefined && this.heldIndexes.has(upstreamIndex)) + || (itemId !== undefined && this.heldItemIds.has(itemId))) { + this.heldCalls.push({ payload, ...(upstreamIndex === undefined ? {} : { upstreamIndex }) }); + return []; + } + + const rewritten: Record = { ...payload }; + if (upstreamIndex !== undefined) rewritten.output_index = this.clientIndexFor(upstreamIndex); + if (payload.type === "response.output_item.done") this.retain(payload.item); + return [this.render(payload.type, rewritten)]; + } + + /** Release the withheld client tool calls once the turn is known to end here. */ + flushHeldCalls(): string[] { + const blocks: string[] = []; + for (const held of this.heldCalls) { + const rewritten: Record = { ...held.payload }; + if (held.upstreamIndex !== undefined) { + rewritten.output_index = this.clientIndexFor(held.upstreamIndex); + } + if (held.payload.type === "response.output_item.done") this.retain(held.payload.item); + blocks.push(this.render(String(held.payload.type), rewritten)); + } + this.heldCalls = []; + return blocks; + } + + /** Decide what the leg's terminal means once the whole leg has been read. */ + decide(remainingLegs: number): LegDecision { + if (this.searches.length === 0) return { kind: "end", searches: [] }; + if (this.sawClientExecutedCall) { + return { + kind: "fail", + searches: this.searches, + code: WEB_SEARCH_BRIDGE_MIXED_TOOLS_ERROR_CODE, + message: "routed provider requested web_search alongside another client tool in one turn; " + + "the web-search bridge cannot answer both without dropping the client's call", + }; + } + const terminalType = this.terminalPayload?.type; + if (terminalType === "response.failed" || terminalType === "response.incomplete") { + return { kind: "end", searches: [] }; + } + if (remainingLegs <= 0) { + return { + kind: "fail", + searches: this.searches, + code: WEB_SEARCH_BRIDGE_ERROR_CODE, + message: "web-search bridge exhausted its continuation budget for this turn", + }; + } + return { kind: "continue", searches: this.searches }; + } + + /** + * Flush the held terminal. When searches were injected the snapshot is rebuilt from the items + * the client actually received, so response.output matches the streamed turn instead of + * showing only the final leg. + */ + terminalFrames(): string[] { + const held = this.terminalPayload; + if (!held) return ["data: [DONE]"]; + const payload: Record = { ...held }; + if (this.injected && this.retainedItemsComplete && isRecord(payload.response)) { + payload.response = { + ...payload.response, + output: this.retainedItems.filter(item => item !== undefined), + }; + } + return [this.render(String(held.type), payload), "data: [DONE]"]; + } +} + +/** Append one executed search turn to the raw Responses body for the next leg. */ +export function appendBridgeSearchTurn( + requestBody: string, + turns: readonly { call: InterceptedSearchCall; output: string }[], +): string | undefined { + let parsed: unknown; + try { + parsed = JSON.parse(requestBody); + } catch { + return undefined; + } + if (!isRecord(parsed) || !Array.isArray(parsed.input)) return undefined; + const input = [...parsed.input]; + for (const turn of turns) { + input.push({ + type: "function_call", + ...(turn.call.sourceItemId ? { id: turn.call.sourceItemId } : {}), + call_id: turn.call.callId, + name: WEB_SEARCH_TOOL_NAME, + arguments: turn.call.argumentsText || "{}", + }); + input.push({ + type: "function_call_output", + call_id: turn.call.callId, + output: turn.output, + }); + } + return JSON.stringify({ ...parsed, input, stream: true }); +} + +/** + * Bind the shipped executor for a plan. Each query in one call is a separate upstream search; + * their digests are merged so the model receives a single tool result for the call it made. + */ +export function createOllamaBridgeExecutor( + plan: PassthroughWebSearchBridgePlan, + apiKey: string, +): PassthroughWebSearchBridgeExecutor { + return async (queries, signal) => { + const texts: string[] = []; + const sources: SidecarOutcome["sources"] = []; + const errors: string[] = []; + for (const query of queries) { + if (signal?.aborted) break; + const outcome = await runOllamaWebSearch(query, apiKey, plan.endpoint, plan.timeoutMs, signal); + if (outcome.error) { + errors.push(outcome.error); + continue; + } + texts.push(queries.length > 1 ? "Results for \"" + query + "\":\n" + outcome.text : outcome.text); + for (const source of outcome.sources) { + if (!sources.some(existing => existing.url === source.url)) sources.push(source); + } + } + if (texts.length === 0) { + return { text: "", sources: [], error: errors[0] ?? "web search produced no results" }; + } + return { text: texts.join("\n\n"), sources }; + }; +} + +/** + * Run one bridged turn as a client-facing SSE stream. + * + * The first leg is already open -- it came through the core send path with its full recovery, + * circuit, and body-size handling. Every later leg is a direct re-POST of the same outbound body + * extended with the executed search, which is exactly what a KEY-auth Responses continuation is. + */ +async function* bridgeStreamBlocks( + options: PassthroughWebSearchBridgeStreamOptions, + aborted: () => boolean, +): AsyncGenerator { + const state = new BridgeStreamState(); + let requestBody = options.requestBody; + let leg: ReadableStream = options.firstLeg; + let isFirstLeg = true; + let searchesExecuted = 0; + // One continuation leg per allowed search, plus one final leg for the answer itself. + let legsRemaining = options.plan.maxSearches + 1; + + const emit = function* (blocks: readonly string[]): Generator { + for (const block of blocks) yield block + "\n\n"; + }; + + for (;;) { + state.beginLeg(); + try { + for await (const { block } of readSseBlocks(leg)) { + yield* emit(state.consume(block, isFirstLeg)); + if (aborted()) return; + } + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + yield* emit(state.failureFrames( + WEB_SEARCH_BRIDGE_ERROR_CODE, + "web-search bridge upstream read failed: " + message, + )); + return; + } + isFirstLeg = false; + if (aborted()) return; + + const decision = state.decide(legsRemaining); + if (decision.kind === "fail") { + // Close any cell this leg opened, or Codex keeps a "Searching the web" spinner running + // under a failed turn (the same reason src/bridge.ts closes a dangling search on teardown). + for (const call of decision.searches) { + yield* emit(state.searchEndFrames(call, [], { + text: "", + sources: [], + error: decision.message!, + })); + } + // The withheld client call is deliberately dropped: the turn is ending as failed, and + // releasing a tool call Codex would start executing is exactly what must not happen. + yield* emit(state.failureFrames(decision.code!, decision.message!)); + return; + } + if (decision.kind === "end") { + yield* emit(state.flushHeldCalls()); + yield* emit(state.terminalFrames()); + return; + } + + const turns: { call: InterceptedSearchCall; output: string }[] = []; + for (const call of decision.searches) { + const queries = parseQueries(call.argumentsText); + let outcome: SidecarOutcome; + if (aborted()) return; + if (searchesExecuted >= options.plan.maxSearches) { + outcome = { + text: "", + sources: [], + error: "no further web searches are available for this turn", + }; + } else if (queries.length === 0) { + outcome = { text: "", sources: [], error: "web_search was called without a usable query" }; + } else { + searchesExecuted += 1; + outcome = await options.execute(queries, options.signal); + } + yield* emit(state.searchEndFrames(call, queries, outcome)); + turns.push({ + call, + // The model needs a readable result either way; an executor error is reported as the + // tool result rather than as a turn failure, so it can still answer without the search. + output: outcome.error ? "Web search failed: " + outcome.error : outcome.text, + }); + } + + const nextBody = appendBridgeSearchTurn(requestBody, turns); + if (nextBody === undefined) { + yield* emit(state.failureFrames( + WEB_SEARCH_BRIDGE_ERROR_CODE, + "web-search bridge could not extend the outbound request body", + )); + return; + } + // The first leg was admitted by the caller's outbound ceiling; appending a search result can + // push the continuation past it, so re-check rather than sending an unbounded body. + const refusal = options.checkOutboundBody?.(nextBody); + if (refusal) { + yield* emit(state.failureFrames(WEB_SEARCH_BRIDGE_ERROR_CODE, refusal)); + return; + } + requestBody = nextBody; + legsRemaining -= 1; + if (aborted()) return; + + let next: Response; + try { + next = await options.send(requestBody); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + yield* emit(state.failureFrames( + WEB_SEARCH_BRIDGE_ERROR_CODE, + "web-search bridge continuation send failed: " + message, + )); + return; + } + if (!next.ok || !next.body || aborted()) { + next.body?.cancel().catch(() => {}); + if (aborted()) return; + yield* emit(state.failureFrames( + WEB_SEARCH_BRIDGE_ERROR_CODE, + "web-search bridge continuation returned HTTP " + next.status, + )); + return; + } + leg = next.body; + } +} + +/** + * Build the client-facing SSE body for a bridged turn. + * + * Pull-driven so a slow client applies backpressure to the upstream leg instead of letting the + * proxy buffer the whole turn. Cancelling the client stream latches a local abort, so no further + * search is billed and no further continuation is sent once the consumer is gone. + */ +export function createPassthroughWebSearchBridgeStream( + options: PassthroughWebSearchBridgeStreamOptions, +): ReadableStream { + let cancelled = false; + const aborted = (): boolean => cancelled || options.signal?.aborted === true; + const iterator = bridgeStreamBlocks(options, aborted)[Symbol.asyncIterator](); + const encoder = new TextEncoder(); + return new ReadableStream({ + async pull(controller) { + try { + const next = await iterator.next(); + if (next.done) { + controller.close(); + return; + } + controller.enqueue(encoder.encode(next.value)); + } catch (error) { + controller.error(error); + } + }, + cancel(reason) { + cancelled = true; + void iterator.return?.(reason); + }, + }); +} diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index f62377f3e7..37204ab593 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1106,6 +1106,7 @@ "web-search-backend-union.test.ts": "web-search", "web-search-candidates.test.ts": "web-search", "web-search-parse.test.ts": "web-search", + "web-search-passthrough-bridge.test.ts": "web-search", "web-search-progress-stream.test.ts": "web-search", "web-search-sources.test.ts": "web-search", "web-search-timeout-contract.test.ts": "web-search", diff --git a/tests/web-search/web-search-passthrough-bridge.test.ts b/tests/web-search/web-search-passthrough-bridge.test.ts new file mode 100644 index 0000000000..9e7029bb53 --- /dev/null +++ b/tests/web-search/web-search-passthrough-bridge.test.ts @@ -0,0 +1,657 @@ +/** + * #3761: the Codex App always declares the hosted web_search tool, and the KEY-auth Responses + * passthrough relayed that declaration as though the destination executed it. Ollama Cloud GLM + * does not, so it answered with a plain function_call named web_search that nothing ran, and the + * undeclared-tool guard ended the turn. + * + * These pin the opt-in bridge: OFF reproduces the reported abort, ON removes the call from the + * client stream and continues the conversation upstream, and an unrelated undeclared tool still + * fails closed through the bridged stream. + */ +import { describe, expect, test } from "bun:test"; +import { + appendBridgeSearchTurn, + createPassthroughWebSearchBridgeStream, + planPassthroughWebSearchBridge, + resolveOllamaWebSearchEndpoint, + WEB_SEARCH_BRIDGE_ERROR_CODE, + WEB_SEARCH_BRIDGE_MIXED_TOOLS_ERROR_CODE, + type PassthroughWebSearchBridgePlan, +} from "../../src/web-search/passthrough-bridge"; +import { mapOllamaSearchResponse } from "../../src/web-search/ollama-executor"; +import { UNDECLARED_TOOL_CALL_ERROR_CODE } from "../../src/server/responses-undeclared-tool-guard"; +import { handleResponses } from "../../src/server/responses"; +import type { OcxConfig, OcxParsedRequest, OcxProviderConfig, ProviderWebSearchBridgeConfig } from "../../src/types"; + +/** One SSE event block without its blank-line delimiter. */ +function frame(type: string, payload: Record): string { + return "event: " + type + "\ndata: " + JSON.stringify({ type, ...payload }); +} + +function sseBody(...blocks: string[]): string { + return blocks.concat("data: [DONE]").join("\n\n") + "\n\n"; +} + +function streamFromText(text: string): ReadableStream { + const chunk = new TextEncoder().encode(text); + let sent = false; + return new ReadableStream({ + pull(controller) { + if (sent) { + controller.close(); + return; + } + sent = true; + controller.enqueue(chunk); + }, + }); +} + +/** Every parsed data payload the client received, in order. */ +function clientEvents(body: string): Record[] { + return body + .split(/\r?\n/) + .filter(line => line.startsWith("data:")) + .map(line => line.slice(5).trim()) + .filter(payload => payload.length > 0 && payload !== "[DONE]") + .map(payload => JSON.parse(payload) as Record); +} + +function providerFixture( + bridge?: ProviderWebSearchBridgeConfig, + overrides: Partial = {}, +): OcxProviderConfig { + return { + adapter: "openai-responses", + baseUrl: "https://ollama.com/v1", + authMode: "key", + apiKey: "fixture-key", + ...(bridge ? { webSearchBridge: bridge } : {}), + ...overrides, + } as OcxProviderConfig; +} + +function parsedFixture(overrides: Record = {}): OcxParsedRequest { + return { + modelId: "glm-4.7", + options: {}, + stream: true, + _webSearch: { type: "web_search" }, + ...overrides, + } as unknown as OcxParsedRequest; +} + +const armed: ProviderWebSearchBridgeConfig = { enabled: true, backend: "ollama" }; + +describe("planPassthroughWebSearchBridge arming", () => { + test("arms for an enabled ollama-backed key provider on the canonical origin", () => { + const plan = planPassthroughWebSearchBridge(parsedFixture(), providerFixture(armed), { + isPassthrough: true, + stream: true, + }); + expect(plan).toEqual({ + backend: "ollama", + endpoint: "https://ollama.com/api/web_search", + maxSearches: 3, + timeoutMs: 60_000, + }); + }); + + test("stays disarmed without the opt-in", () => { + const off: (ProviderWebSearchBridgeConfig | undefined)[] = [ + undefined, + { backend: "ollama" }, + { enabled: false, backend: "ollama" }, + ]; + for (const bridge of off) { + expect(planPassthroughWebSearchBridge(parsedFixture(), providerFixture(bridge), { + isPassthrough: true, + stream: true, + })).toBeUndefined(); + } + }); + + test("never arms for forwarded ChatGPT auth or a stored OAuth credential", () => { + for (const authMode of ["forward", "oauth"] as const) { + expect(planPassthroughWebSearchBridge( + parsedFixture(), + providerFixture(armed, { authMode }), + { isPassthrough: true, stream: true }, + )).toBeUndefined(); + } + }); + + test("stays disarmed off the passthrough, without hosted web_search, and for non-streaming turns", () => { + const provider = providerFixture(armed); + expect(planPassthroughWebSearchBridge(parsedFixture(), provider, { isPassthrough: false, stream: true })) + .toBeUndefined(); + expect(planPassthroughWebSearchBridge(parsedFixture({ _webSearch: undefined }), provider, { + isPassthrough: true, + stream: true, + })).toBeUndefined(); + expect(planPassthroughWebSearchBridge(parsedFixture(), provider, { isPassthrough: true, stream: false })) + .toBeUndefined(); + }); + + test("a tool_choice that excludes search excludes the bridge", () => { + expect(planPassthroughWebSearchBridge( + parsedFixture({ options: { toolChoice: { type: "function", name: "exec" } } }), + providerFixture(armed), + { isPassthrough: true, stream: true }, + )).toBeUndefined(); + }); + + test("backends without a shipped executor stay inert rather than falling back", () => { + for (const backend of ["openai", "anthropic", "xai", "gemini", "exa"] as const) { + expect(planPassthroughWebSearchBridge( + parsedFixture(), + providerFixture({ enabled: true, backend }), + { isPassthrough: true, stream: true }, + )).toBeUndefined(); + } + }); + + test("the ollama backend refuses a non-canonical origin unless the operator names the endpoint", () => { + const renamed = providerFixture(armed, { baseUrl: "https://gateway.example/v1" }); + expect(resolveOllamaWebSearchEndpoint(renamed)).toBeUndefined(); + expect(planPassthroughWebSearchBridge(parsedFixture(), renamed, { isPassthrough: true, stream: true })) + .toBeUndefined(); + + const operatorSet = providerFixture( + { enabled: true, backend: "ollama", endpoint: "https://search.internal/api/web_search" }, + { baseUrl: "https://gateway.example/v1" }, + ); + const plan = planPassthroughWebSearchBridge(parsedFixture(), operatorSet, { + isPassthrough: true, + stream: true, + }); + expect(plan?.endpoint).toBe("https://search.internal/api/web_search"); + }); + + test("out-of-range bounds fall back to the documented defaults", () => { + const plan = planPassthroughWebSearchBridge( + parsedFixture(), + providerFixture({ enabled: true, backend: "ollama", maxSearches: 99, timeoutMs: 1 }), + { isPassthrough: true, stream: true }, + ); + expect(plan?.maxSearches).toBe(3); + expect(plan?.timeoutMs).toBe(60_000); + }); +}); + +const plan: PassthroughWebSearchBridgePlan = { + backend: "ollama", + endpoint: "https://ollama.com/api/web_search", + maxSearches: 3, + timeoutMs: 60_000, +}; + +const searchCall = { + type: "function_call", + id: "fc_1", + call_id: "call_1", + name: "web_search", + arguments: "{\"query\":\"opencodex release\"}", +}; + +const preamble = { + type: "message", + id: "msg_1", + role: "assistant", + content: [{ type: "output_text", text: "Let me look that up." }], +}; + +const answer = { + type: "message", + id: "msg_2", + role: "assistant", + content: [{ type: "output_text", text: "The current release is 2.50.0." }], +}; + +/** A leg that asks for one search, preceded by a normal assistant message. */ +function searchLeg(): string { + return sseBody( + frame("response.created", { response: { id: "resp_1", status: "in_progress" } }), + frame("response.output_item.added", { output_index: 0, item: { ...preamble, content: [] } }), + frame("response.output_item.done", { output_index: 0, item: preamble }), + frame("response.output_item.added", { output_index: 1, item: { ...searchCall, arguments: "" } }), + frame("response.function_call_arguments.done", { + output_index: 1, + item_id: "fc_1", + arguments: searchCall.arguments, + }), + frame("response.output_item.done", { output_index: 1, item: searchCall }), + frame("response.completed", { + response: { id: "resp_1", status: "completed", output: [preamble, searchCall] }, + }), + ); +} + +function answerLeg(): string { + return sseBody( + frame("response.created", { response: { id: "resp_2", status: "in_progress" } }), + frame("response.output_item.added", { output_index: 0, item: { ...answer, content: [] } }), + frame("response.output_item.done", { output_index: 0, item: answer }), + frame("response.completed", { + response: { id: "resp_2", status: "completed", output: [answer] }, + }), + ); +} + +const initialBody = JSON.stringify({ + model: "glm-4.7", + stream: true, + input: [{ role: "user", content: [{ type: "input_text", text: "what is the latest release?" }] }], + tools: [{ type: "web_search" }], +}); + +describe("the bridged client stream", () => { + test("replaces the web_search function_call with a hosted cell and continues upstream", async () => { + const sent: string[] = []; + const executed: string[][] = []; + const stream = createPassthroughWebSearchBridgeStream({ + plan, + firstLeg: streamFromText(searchLeg()), + requestBody: initialBody, + send: async (body) => { + sent.push(body); + return new Response(streamFromText(answerLeg()), { + headers: { "content-type": "text/event-stream" }, + }); + }, + execute: async (queries) => { + executed.push(queries); + return { text: "opencodex 2.50.0 shipped", sources: [{ url: "https://example.test/rel", title: "Releases" }] }; + }, + }); + + const body = await new Response(stream).text(); + const events = clientEvents(body); + + // The call Codex cannot execute never reaches it; the hosted cell does. + expect(body).not.toContain("\"name\":\"web_search\""); + expect(body).not.toContain("\"type\":\"function_call\""); + const added = events.find(event => + event.type === "response.output_item.added" + && (event.item as Record).type === "web_search_call"); + const done = events.find(event => + event.type === "response.output_item.done" + && (event.item as Record).type === "web_search_call"); + expect(added).toBeDefined(); + expect(done).toBeDefined(); + const addedItem = added!.item as Record; + const doneItem = done!.item as Record; + expect(addedItem.status).toBe("in_progress"); + expect(String(addedItem.id)).toStartWith("ws_"); + expect(doneItem.id).toBe(addedItem.id); + expect(doneItem.status).toBe("completed"); + expect(doneItem.action).toEqual({ + type: "search", + query: "opencodex release", + queries: ["opencodex release"], + }); + expect(doneItem.sources).toEqual([{ url: "https://example.test/rel", title: "Releases" }]); + expect(executed).toEqual([["opencodex release"]]); + + // The second upstream body carries the executed call and its result. + expect(sent).toHaveLength(1); + const continuation = JSON.parse(sent[0]!) as { input: Record[]; stream: boolean }; + expect(continuation.stream).toBe(true); + const call = continuation.input.find(item => item.type === "function_call"); + const output = continuation.input.find(item => item.type === "function_call_output"); + expect(call).toMatchObject({ call_id: "call_1", name: "web_search", arguments: searchCall.arguments }); + expect(output).toMatchObject({ call_id: "call_1", output: "opencodex 2.50.0 shipped" }); + + // Both legs land in one monotonic client numbering, and the terminal snapshot matches it. + const indexes = events + .filter(event => event.type === "response.output_item.added") + .map(event => event.output_index); + expect(indexes).toEqual([0, 1, 2]); + const completed = events.filter(event => event.type === "response.completed"); + expect(completed).toHaveLength(1); + const finalOutput = (completed[0]!.response as { output: Record[] }).output; + expect(finalOutput.map(item => item.type)).toEqual(["message", "web_search_call", "message"]); + const sequences = events.map(event => event.sequence_number as number); + expect(sequences).toEqual([...sequences].sort((a, b) => a - b)); + }); + + test("a turn with no search is relayed untouched and never re-sends", async () => { + let sends = 0; + const stream = createPassthroughWebSearchBridgeStream({ + plan, + firstLeg: streamFromText(answerLeg()), + requestBody: initialBody, + send: async () => { + sends += 1; + return new Response(null, { status: 500 }); + }, + execute: async () => { + throw new Error("must not execute a search for a turn that did not ask for one"); + }, + }); + + const body = await new Response(stream).text(); + expect(sends).toBe(0); + expect(body).not.toContain("web_search_call"); + expect(body).toContain("response.completed"); + expect(body).toContain("The current release is 2.50.0."); + expect(body.trimEnd().endsWith("data: [DONE]")).toBe(true); + }); + + test("a search mixed with another client tool call fails closed instead of dropping it", async () => { + let sends = 0; + const clientCall = { + type: "function_call", + id: "fc_2", + call_id: "call_2", + name: "exec", + arguments: "{}", + }; + const mixedLeg = sseBody( + frame("response.output_item.added", { output_index: 0, item: { ...searchCall, arguments: "" } }), + frame("response.output_item.done", { output_index: 0, item: searchCall }), + frame("response.output_item.added", { output_index: 1, item: { ...clientCall, arguments: "" } }), + frame("response.output_item.done", { output_index: 1, item: clientCall }), + frame("response.completed", { + response: { id: "resp_1", status: "completed", output: [searchCall, clientCall] }, + }), + ); + + const stream = createPassthroughWebSearchBridgeStream({ + plan, + firstLeg: streamFromText(mixedLeg), + requestBody: initialBody, + send: async () => { + sends += 1; + return new Response(null, { status: 500 }); + }, + execute: async () => ({ text: "unused", sources: [] }), + }); + + const body = await new Response(stream).text(); + expect(sends).toBe(0); + // The client tool call is withheld and dropped: releasing it under a failed turn would let + // Codex start running exec for a turn that never completes. + expect(body).not.toContain("\"name\":\"exec\""); + const failed = clientEvents(body).find(event => event.type === "response.failed"); + expect(failed).toBeDefined(); + const error = (failed!.response as { error: Record }).error; + expect(error.code).toBe(WEB_SEARCH_BRIDGE_MIXED_TOOLS_ERROR_CODE); + expect(String(error.message)).toContain("another client tool"); + // The opened hosted cell is closed as failed rather than left spinning. + const cell = clientEvents(body).find(event => + event.type === "response.output_item.done" + && (event.item as Record).type === "web_search_call"); + expect((cell!.item as Record).status).toBe("failed"); + }); + + test("a search that is not the last item keeps its streamed position", async () => { + // The model searches first and keeps talking; the hosted cell must open where the call stood. + const leg = sseBody( + frame("response.output_item.added", { output_index: 0, item: { ...searchCall, arguments: "" } }), + frame("response.output_item.done", { output_index: 0, item: searchCall }), + frame("response.output_item.added", { output_index: 1, item: { ...preamble, content: [] } }), + frame("response.output_item.done", { output_index: 1, item: preamble }), + frame("response.completed", { + response: { id: "resp_1", status: "completed", output: [searchCall, preamble] }, + }), + ); + const stream = createPassthroughWebSearchBridgeStream({ + plan, + firstLeg: streamFromText(leg), + requestBody: initialBody, + send: async () => new Response(streamFromText(answerLeg()), { + headers: { "content-type": "text/event-stream" }, + }), + execute: async () => ({ text: "a result", sources: [] }), + }); + + const events = clientEvents(await new Response(stream).text()); + const added = events.filter(event => event.type === "response.output_item.added"); + expect(added.map(event => (event.item as Record).type)) + .toEqual(["web_search_call", "message", "message"]); + expect(added.map(event => event.output_index)).toEqual([0, 1, 2]); + + // The terminal snapshot keeps the same order the client saw, not the order of completion. + const completed = events.find(event => event.type === "response.completed"); + const output = (completed!.response as { output: Record[] }).output; + expect(output.map(item => item.type)).toEqual(["web_search_call", "message", "message"]); + }); + + test("a continuation body over the outbound ceiling is refused instead of sent", async () => { + let sends = 0; + const stream = createPassthroughWebSearchBridgeStream({ + plan, + firstLeg: streamFromText(searchLeg()), + requestBody: initialBody, + send: async () => { + sends += 1; + return new Response(null, { status: 500 }); + }, + execute: async () => ({ text: "a result", sources: [] }), + checkOutboundBody: () => "outbound body is too large", + }); + + const body = await new Response(stream).text(); + expect(sends).toBe(0); + const failed = clientEvents(body).find(event => event.type === "response.failed"); + const error = (failed!.response as { error: Record }).error; + expect(error.code).toBe(WEB_SEARCH_BRIDGE_ERROR_CODE); + expect(String(error.message)).toContain("outbound body is too large"); + }); + + test("a cancelled client stream bills no further search and sends no continuation", async () => { + let sends = 0; + let executes = 0; + const controller = new AbortController(); + const stream = createPassthroughWebSearchBridgeStream({ + plan, + firstLeg: streamFromText(searchLeg()), + requestBody: initialBody, + send: async () => { + sends += 1; + return new Response(streamFromText(answerLeg()), { + headers: { "content-type": "text/event-stream" }, + }); + }, + execute: async () => { + executes += 1; + return { text: "a result", sources: [] }; + }, + signal: controller.signal, + }); + + controller.abort(); + await new Response(stream).text(); + expect(executes).toBe(0); + expect(sends).toBe(0); + }); + + + test("the search budget is bounded and the turn terminates rather than looping", async () => { + const executed: string[][] = []; + let sends = 0; + const stream = createPassthroughWebSearchBridgeStream({ + plan: { ...plan, maxSearches: 1 }, + firstLeg: streamFromText(searchLeg()), + requestBody: initialBody, + send: async () => { + sends += 1; + return new Response(streamFromText(searchLeg()), { + headers: { "content-type": "text/event-stream" }, + }); + }, + execute: async (queries) => { + executed.push(queries); + return { text: "one result", sources: [] }; + }, + }); + + const body = await new Response(stream).text(); + // One executed search, one refusal cell, then a bounded terminal failure. + expect(executed).toHaveLength(1); + expect(sends).toBe(2); + const failed = clientEvents(body).find(event => event.type === "response.failed"); + expect((failed!.response as { error: Record }).error.code) + .toBe(WEB_SEARCH_BRIDGE_ERROR_CODE); + }); + + test("an executor failure is reported as the tool result, not as a dead turn", async () => { + const sent: string[] = []; + const stream = createPassthroughWebSearchBridgeStream({ + plan, + firstLeg: streamFromText(searchLeg()), + requestBody: initialBody, + send: async (body) => { + sent.push(body); + return new Response(streamFromText(answerLeg()), { + headers: { "content-type": "text/event-stream" }, + }); + }, + execute: async () => ({ text: "", sources: [], error: "ollama web-search HTTP 401" }), + }); + + const body = await new Response(stream).text(); + const done = clientEvents(body).find(event => + event.type === "response.output_item.done" + && (event.item as Record).type === "web_search_call"); + expect((done!.item as Record).status).toBe("failed"); + const continuation = JSON.parse(sent[0]!) as { input: Record[] }; + const output = continuation.input.find(item => item.type === "function_call_output"); + expect(String(output!.output)).toContain("Web search failed: ollama web-search HTTP 401"); + expect(body).toContain("The current release is 2.50.0."); + }); +}); + +describe("bridge helpers", () => { + test("appendBridgeSearchTurn refuses a body whose input is not an array", () => { + expect(appendBridgeSearchTurn("not json", [])).toBeUndefined(); + expect(appendBridgeSearchTurn(JSON.stringify({ input: "prompt" }), [])).toBeUndefined(); + }); + + test("mapOllamaSearchResponse digests results and rejects a shapeless body", () => { + expect(mapOllamaSearchResponse({ nope: true }).error).toBeDefined(); + expect(mapOllamaSearchResponse({ results: [] }).error).toBeDefined(); + const mapped = mapOllamaSearchResponse({ + results: [{ title: "Releases", url: "https://example.test/rel", content: "2.50.0 is out" }], + }); + expect(mapped.error).toBeUndefined(); + expect(mapped.sources).toEqual([{ url: "https://example.test/rel", title: "Releases" }]); + expect(mapped.text).toContain("2.50.0 is out"); + }); +}); + +describe("the reported turn, end to end through handleResponses", () => { + function config(bridge?: ProviderWebSearchBridgeConfig): OcxConfig { + return { + port: 0, + defaultProvider: "fixture", + providers: { + fixture: { + adapter: "openai-responses", + baseUrl: "https://ollama.com/v1", + authMode: "key", + apiKey: "fixture-key", + ...(bridge ? { webSearchBridge: bridge } : {}), + }, + }, + } as unknown as OcxConfig; + } + + // Codex's own shape: the hosted web_search declaration plus ordinary client function tools. + const clientRequest = JSON.stringify({ + model: "fixture/glm-4.7", + stream: true, + input: [{ role: "user", content: [{ type: "input_text", text: "what is the latest release?" }] }], + tools: [ + { type: "web_search" }, + { type: "function", name: "wait", parameters: { type: "object" } }, + ], + }); + + async function post( + ocxConfig: OcxConfig, + legs: string[], + ): Promise<{ body: string; outbound: string[]; searches: number }> { + const savedFetch = globalThis.fetch; + const outbound: string[] = []; + let searches = 0; + let leg = 0; + globalThis.fetch = (async (input: unknown, init?: RequestInit) => { + const url = typeof input === "string" + ? input + : input instanceof URL ? input.href : (input as Request).url; + if (url.includes("/api/web_search")) { + searches += 1; + return new Response(JSON.stringify({ + results: [{ title: "Releases", url: "https://example.test/rel", content: "opencodex 2.50.0" }], + }), { headers: { "content-type": "application/json" } }); + } + outbound.push(String(init?.body ?? "")); + const text = legs[Math.min(leg, legs.length - 1)]!; + leg += 1; + return new Response(text, { headers: { "content-type": "text/event-stream" } }); + }) as unknown as typeof fetch; + try { + const response = await handleResponses(new Request("http://localhost/v1/responses", { + method: "POST", + headers: { "content-type": "application/json" }, + body: clientRequest, + }), ocxConfig, { model: "", provider: "" }); + return { body: await response.text(), outbound, searches }; + } finally { + globalThis.fetch = savedFetch; + } + } + + test("without the opt-in the reported abort still happens", async () => { + const result = await post(config(), [searchLeg()]); + expect(result.searches).toBe(0); + expect(result.outbound).toHaveLength(1); + expect(result.body).toContain("response.failed"); + expect(result.body).toContain(UNDECLARED_TOOL_CALL_ERROR_CODE); + expect(result.body).not.toContain("web_search_call"); + }); + + test("with the opt-in the client sees a hosted search cell and the answer", async () => { + const result = await post(config(armed), [searchLeg(), answerLeg()]); + + expect(result.searches).toBe(1); + expect(result.body).not.toContain(UNDECLARED_TOOL_CALL_ERROR_CODE); + expect(result.body).toContain("\"type\":\"web_search_call\""); + expect(result.body).not.toContain("\"name\":\"web_search\""); + expect(result.body).toContain("The current release is 2.50.0."); + + // The search result reached the SECOND upstream body as a native tool result. + expect(result.outbound).toHaveLength(2); + const continuation = JSON.parse(result.outbound[1]!) as { input: Record[] }; + const output = continuation.input.find(item => item.type === "function_call_output"); + expect(output).toBeDefined(); + expect(String(output!.output)).toContain("opencodex 2.50.0"); + expect(continuation.input.some(item => + item.type === "function_call" && item.name === "web_search")).toBe(true); + }); + + test("an unrelated undeclared tool still fails closed through the bridged stream", async () => { + const strayCall = { + type: "function_call", + id: "fc_9", + call_id: "call_9", + name: "frobnicate", + arguments: "{}", + }; + const strayLeg = sseBody( + frame("response.output_item.added", { output_index: 0, item: { ...strayCall, arguments: "" } }), + frame("response.output_item.done", { output_index: 0, item: strayCall }), + frame("response.completed", { + response: { id: "resp_1", status: "completed", output: [strayCall] }, + }), + ); + + const result = await post(config(armed), [strayLeg]); + expect(result.searches).toBe(0); + expect(result.body).toContain("response.failed"); + expect(result.body).toContain(UNDECLARED_TOOL_CALL_ERROR_CODE); + expect(result.body).toContain("frobnicate"); + }); +});