fix(responses): repair terminal-less bridged search legs

src/server/responses/passthrough-delivery.ts

@@ -380,36 +380,65 @@ export async function deliverPassthroughResponse(
       });
       // Capture the binding that actually served the first leg, after its permitted reselection.
       const webSearchBridgeBinding = requestBindings.get(nativeExchange.request);
-      // 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
+      // Repair must observe the raw first leg before the bridge suppresses an intercepted search
+      // lifecycle. Otherwise a provider that leaves that complete call open never arms repair's
+      // grace timer, so the bridge cannot execute the search or begin its continuation.
+      let passthroughSseBody = terminalRepairPolicy
+        ? relayResponsesSseWithTerminalRepair(
+          upstreamResponse.body,
+          upstream,
+          terminalRepairPolicy,
+          translatorBudget,
+          options.responsesTerminalRepairScheduler,
+        )
+        : upstreamResponse.body;
+      passthroughSseBody = webSearchBridgePlan
         ? createPassthroughWebSearchBridgeStream({
           plan: webSearchBridgePlan,
-          firstLeg: upstreamResponse.body,
+          firstLeg: passthroughSseBody,
           requestBody: nativeExchange.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(
-            nativeExchange.request.url,
-            { method: nativeExchange.request.method, headers: nativeExchange.request.headers, body: continuationBody },
-            upstream.signal,
-            connectMs,
-            true,
-            providerFetch(route.provider, options.codexWsRuntimeIdentity, {
-              // Pacing can outlive a manual selection change. A continuation must retain the
-              // first leg's key and appended search result, never rebuild from the original turn.
-              beforeDispatch: () => {
-                if (webSearchBridgeBinding?.kind !== "api-key"
-                  || !providerApiKeySelectionIsCurrent(config, route.providerName, webSearchBridgeBinding.provider)) {
-                  throw new Error("API key selection changed during a web-search continuation");
-                }
+          send: async (continuationBody: string) => {
+            const continuation = await fetchWithHeaderTimeout(
+              nativeExchange.request.url,
+              { method: nativeExchange.request.method, headers: nativeExchange.request.headers, body: continuationBody },
+              upstream.signal,
+              connectMs,
+              true,
+              providerFetch(route.provider, options.codexWsRuntimeIdentity, {
+                // Pacing can outlive a manual selection change. A continuation must retain the
+                // first leg's key and appended search result, never rebuild from the original turn.
+                beforeDispatch: () => {
+                  if (webSearchBridgeBinding?.kind !== "api-key"
+                    || !providerApiKeySelectionIsCurrent(config, route.providerName, webSearchBridgeBinding.provider)) {
+                    throw new Error("API key selection changed during a web-search continuation");
+                  }
+                },
+                providerName: route.providerName,
+                modelId: route.modelId,
+              }),
+              false,
+            );
+            // A continuation rides the same transport that can leave a complete leg open, so
+            // every repaired leg gets its own grace window — not just the first one.
+            if (!terminalRepairPolicy || !continuation.ok || !continuation.body) return continuation;
+            return new Response(
+              relayResponsesSseWithTerminalRepair(
+                continuation.body,
+                upstream,
+                terminalRepairPolicy,
+                translatorBudget,
+                options.responsesTerminalRepairScheduler,
+              ),
+              {
+                status: continuation.status,
+                statusText: continuation.statusText,
+                headers: continuation.headers,
               },
-              providerName: route.providerName,
-              modelId: route.modelId,
-            }),
-            false,
-          ),
+            );
+          },
           execute: createPassthroughWebSearchBridgeExecutor(webSearchBridgePlan, {
             providerApiKey: route.provider.apiKey ?? "",
             auth: webSearchBridgeAuth,
@@ -433,16 +462,7 @@ export async function deliverPassthroughResponse(
           onFinalize: () => releaseCodexAuthContextProbeLease(openAiSidecar?.authContext),
           signal: upstream.signal,
         })
-        : upstreamResponse.body;
-      const passthroughSseBody = terminalRepairPolicy
-        ? relayResponsesSseWithTerminalRepair(
-          upstreamSseBody,
-          upstream,
-          terminalRepairPolicy,
-          translatorBudget,
-          options.responsesTerminalRepairScheduler,
-        )
-        : upstreamSseBody;
+        : passthroughSseBody;
       const repairConfig = route.provider.responsesItemIdRepair;
       // Grok Build renders deltas live but reconstructs its durable assistant
       // turn from the completed response snapshot. Native Responses streams

structure/providers-and-adapters.md

@@ -159,7 +159,10 @@ the configured entry, reference, revision, resolved key, authentication mode, an
 disabled or removed provider fails the same check. Drift produces the bridge's failed terminal
 without another provider request, and an unchanged binding resends the built request with its
 executed search result appended, never re-entering the initial reselection/rebuild path. Initial
-dispatch keeps its normal reselection policy. `tests/web-search/web-search-passthrough-bridge.test.ts`
+dispatch keeps its normal reselection policy. When the route's registry policy carries a
+terminal-repair grace (`modelResponsesTerminalRepair`), the response body of every successful
+continuation is wrapped by the same repair that saw the raw first leg, so a complete leg the
+destination leaves open still ends that leg on schedule instead of stalling the turn. `tests/web-search/web-search-passthrough-bridge.test.ts`
 covers drift during search, while pacing, and before first-leg headers return, plus successful
 first-dispatch reselection and result preservation.
 

structure/transports/streaming-health.md

@@ -248,6 +248,14 @@ as `response.incomplete`, never synthetic success. The repair shares the per-tur
 budget, preserves backpressure, and composes ahead of item-id/snapshot rewrites so HTTP/SSE and
 WebSocket clients observe the same canonical lifecycle.
 
+When the hosted-search bridge is also armed, repair wraps the raw first leg BEFORE the bridge:
+the bridge suppresses an intercepted `web_search` lifecycle, so a complete call whose leg never
+closes would otherwise leave the grace timer unarmed and the turn stalled. The same wrap applies
+to every continuation leg the bridge's `send` returns — each leg gets its own grace window on the
+shared abort controller — so a terminal-less continuation cannot stall the bridged turn either.
+`tests/web-search/web-search-passthrough-bridge.test.ts` drives both legs through `handleResponses`
+with an injected scheduler and proves search execution, continuation dispatch, and final terminal.
+
 `ws-bridge.ts` preserves upstream `failed` and `incomplete` status values in the final WebSocket
 frame rather than always emitting `response.completed`. If the response status is `failed`, a
 `response.failed` frame is sent; otherwise `response.completed` carries through the original status.

tests/responses/passthrough-abort.test.ts

@@ -62,8 +62,21 @@ describe("passthrough relayWithAbort (RC2, passthrough path)", () => {
     // The captured static policy now supplies the repair decision; the real platform gate and
     // pure native relay invariants below are unchanged.
     expect(sseBranch).toContain("const terminalRepairPolicy = route.staticPolicy.model.responsesTerminalRepair;");
-    expect(sseBranch).toContain("const passthroughSseBody = terminalRepairPolicy");
+    expect(sseBranch).toContain("let passthroughSseBody = terminalRepairPolicy");
     expect(sseBranch).toContain(": upstreamResponse.body;");
+    // Repair has to wrap the raw first leg before the bridge hides its completed web-search call;
+    // otherwise a terminal-less open leg cannot trigger the repair timer and continuation stalls.
+    const terminalRepair = sseBranch.indexOf("relayResponsesSseWithTerminalRepair(");
+    const webSearchBridge = sseBranch.indexOf("createPassthroughWebSearchBridgeStream({");
+    expect(terminalRepair).toBeGreaterThanOrEqual(0);
+    expect(webSearchBridge).toBeGreaterThan(terminalRepair);
+    expect(sseBranch.slice(webSearchBridge)).toContain("firstLeg: passthroughSseBody,");
+    // Continuation legs need the same repair: the same transport can leave a complete leg
+    // open, and an unwrapped continuation body would stall the bridge identically.
+    const sendWrap = sseBranch.slice(webSearchBridge);
+    expect(sendWrap).toContain("send: async (continuationBody: string)");
+    expect(sendWrap.indexOf("relayResponsesSseWithTerminalRepair(\n                continuation.body"))
+      .toBeGreaterThan(sendWrap.indexOf("send: async"));
     // Native tee stays inside the bounded observer. The production owner passes
     // the raw stream and disconnect signal before any client-side rewrite.
     expect(sseBranch).toMatch(/const \[nativeBody, inspectBody\] = teeWithBoundedInspection\(passthroughSseBody, \{ clientGoneSignal \}\)/);

tests/web-search/web-search-passthrough-bridge.test.ts

@@ -26,6 +26,9 @@ import { providerWebSearchBridgeConfigError, validateConfigCandidate } from "../
 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 { providerConfigSeed } from "../../src/providers/derive";
+import { getProviderRegistryEntry } from "../../src/providers/registry";
+import type { ResponsesTerminalRepairScheduler } from "../../src/server/responses-terminal-repair";
 import {
   resetProviderRequestPacingForTest,
   setProviderRequestPacingRuntimeForTest,
@@ -1469,6 +1472,174 @@ describe("the reported turn, end to end through handleResponses", () => {
       item.type === "function_call" && item.name === "web_search")).toBe(true);
   });
 
+  test("a complete but terminal-less leg still repairs, on the first leg AND the continuation", async () => {
+    // Repair is registry-gated, so only a registry-keyed provider arms it: deepseek carries
+    // modelResponsesTerminalRepair for the V4 flash ids. The fixture legs below emit a fully
+    // complete item lifecycle and then stay open — the reported stall — with no terminal and
+    // no [DONE]. Before the fix the repaired first leg could fire the search, but the raw
+    // continuation leg never got a grace window, so the turn still hung.
+    class ManualScheduler implements ResponsesTerminalRepairScheduler {
+      private current = 0;
+      private nextId = 1;
+      private readonly jobs = new Map<number, { at: number; callback: () => void }>();
+      nowMs(): number { return this.current; }
+      schedule(callback: () => void, delayMs: number): unknown {
+        const id = this.nextId++;
+        this.jobs.set(id, { at: this.current + delayMs, callback });
+        return id;
+      }
+      cancel(handle: unknown): void { this.jobs.delete(handle as number); }
+      pending(): number { return this.jobs.size; }
+      advance(ms: number): void {
+        this.current += ms;
+        for (const [id, job] of [...this.jobs.entries()]) {
+          if (job.at > this.current || !this.jobs.delete(id)) continue;
+          job.callback();
+        }
+      }
+    }
+
+    const openSse = (): { stream: ReadableStream<Uint8Array>; push: (text: string) => void; end: () => void } => {
+      const encoder = new TextEncoder();
+      let controller: ReadableStreamDefaultController<Uint8Array> | null = null;
+      return {
+        stream: new ReadableStream<Uint8Array>({ start(next) { controller = next; } }),
+        push(text) { controller?.enqueue(encoder.encode(text)); },
+        end() { try { controller?.close(); } catch { /* already closed */ } },
+      };
+    };
+
+    // Every item must reach a COMPLETE output_item.done or repair never arms — the status
+    // field is what isCompleteItem actually requires.
+    const donePreamble = { ...preamble, status: "completed" };
+    const doneSearchCall = { ...searchCall, status: "completed" };
+    const doneAnswer = { ...answer, status: "completed" };
+    const blocks = (...frames: string[]): string => frames.join("\n\n") + "\n\n";
+    const openSearchLeg = blocks(
+      frame("response.created", { response: { id: "resp_1", status: "in_progress" } }),
+      frame("response.output_item.added", { output_index: 0, item: { ...donePreamble, content: [] } }),
+      frame("response.output_item.done", { output_index: 0, item: donePreamble }),
+      frame("response.output_item.added", { output_index: 1, item: { ...doneSearchCall, 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: doneSearchCall }),
+    );
+    const openAnswerLeg = blocks(
+      frame("response.created", { response: { id: "resp_2", status: "in_progress" } }),
+      frame("response.output_item.added", { output_index: 0, item: { ...doneAnswer, content: [] } }),
+      frame("response.output_item.done", { output_index: 0, item: doneAnswer }),
+    );
+
+    const firstLeg = openSse();
+    const continuationLeg = openSse();
+    const scheduler = new ManualScheduler();
+    const outbound: string[] = [];
+    let searches = 0;
+    const savedFetch = globalThis.fetch;
+    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.exa.ai/search")) {
+        searches += 1;
+        return new Response(JSON.stringify({
+          results: [{ title: "Releases", url: "https://example.test/rel", content: "opencodex 2.50.0", text: "opencodex 2.50.0" }],
+        }), { headers: { "content-type": "application/json" } });
+      }
+      outbound.push(String(init?.body ?? ""));
+      return new Response(outbound.length === 1 ? firstLeg.stream : continuationLeg.stream, {
+        headers: { "content-type": "text/event-stream" },
+      });
+    }) as unknown as typeof fetch;
+    const cfg = {
+      port: 0,
+      defaultProvider: "deepseek",
+      providers: {
+        deepseek: {
+          ...providerConfigSeed(getProviderRegistryEntry("deepseek")!),
+          apiKey: "fixture-key",
+          webSearchBridge: { enabled: true, backend: "exa" },
+        },
+      },
+      webSearchSidecar: { exaApiKey: "exa-canary" },
+    } as unknown as OcxConfig;
+    const releaseSpendHome = acquireOwnedSpendHome();
+    const decoder = new TextDecoder();
+    const readUntil = async (reader: ReadableStreamDefaultReader<Uint8Array>, pattern: string): Promise<string> => {
+      let out = "";
+      while (!out.includes(pattern)) {
+        const { done, value } = await reader.read();
+        if (done) throw new Error(`stream closed before ${pattern}`);
+        out += decoder.decode(value, { stream: true });
+      }
+      return out;
+    };
+    const flush = async (condition: () => boolean): Promise<void> => {
+      for (let attempts = 0; attempts < 50 && !condition(); attempts += 1) await Bun.sleep(0);
+    };
+    try {
+      const response = await handleResponses(new Request("http://localhost/v1/responses", {
+        method: "POST",
+        headers: { "content-type": "application/json", authorization: "Bearer caller-inbound" },
+        body: JSON.stringify({
+          model: "deepseek/deepseek-v4-flash",
+          stream: true,
+          input: [{ role: "user", content: [{ type: "input_text", text: "what is the latest release?" }] }],
+          tools: [{ type: "web_search" }],
+        }),
+      }), cfg, { model: "", provider: "" }, {
+        responsesTerminalRepairScheduler: scheduler,
+      });
+      const reader = response.body!.getReader();
+      try {
+        // First leg: the complete search lifecycle streams through while the leg stays open.
+        firstLeg.push(openSearchLeg);
+        const opened = await readUntil(reader, "web_search_call");
+        expect(opened).toContain("\"type\":\"web_search_call\"");
+        await flush(() => scheduler.pending() === 1);
+        expect(scheduler.pending()).toBe(1);
+        // The grace window is what ends the leg — before it fires, no search may run.
+        expect(searches).toBe(0);
+        scheduler.advance(5_000);
+        await flush(() => searches === 1 && outbound.length === 2);
+        expect(searches).toBe(1);
+        expect(outbound).toHaveLength(2);
+        const continued = JSON.parse(outbound[1]!) as { input: Record<string, unknown>[] };
+        expect(continued.input.some(item => item.type === "function_call_output"
+          && String(item.output).includes("opencodex 2.50.0"))).toBe(true);
+
+        // Continuation leg: a complete answer that also never sends its terminal. Without
+        // repair on send() this is where the turn hangs.
+        continuationLeg.push(openAnswerLeg);
+        await flush(() => scheduler.pending() === 1);
+        expect(scheduler.pending()).toBe(1);
+        scheduler.advance(5_000);
+        const rest = await Promise.race([
+          (async () => {
+            let out = "";
+            for (;;) {
+              const { done, value } = await reader.read();
+              if (done) return out + decoder.decode();
+              out += decoder.decode(value, { stream: true });
+            }
+          })(),
+          new Promise<never>((_, reject) => setTimeout(() => reject(new Error("continuation never repaired")), 5_000)),
+        ]);
+        expect(rest).toContain("response.completed");
+        expect(rest).toContain("The current release is 2.50.0.");
+        expect(rest).toContain("[DONE]");
+      } finally {
+        try { await reader.cancel(); } catch { /* already closed */ }
+        firstLeg.end();
+        continuationLeg.end();
+      }
+    } finally {
+      releaseSpendHome();
+      globalThis.fetch = savedFetch;
+    }
+  });
+
   const selectionChanges: Array<[string, (ocxConfig: OcxConfig) => void]> = [
     ["selection revision with an unchanged key", cfg => {
       cfg.providers.fixture!.apiKeySelectionRevision = "selection-after";