fix(web-search): bound withheld tool events and close cells on read failure

src/web-search/passthrough-bridge.ts

@@ -128,6 +128,10 @@ const MAX_QUERIES_PER_CALL = 3;
 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;
+/** Bound both object overhead and serialized payloads withheld before a leg's fate is known. */
+const MAX_HELD_CALL_EVENTS = 1_000;
+/** UTF-16 code units in SSE data payloads, not a byte or total-heap measurement. */
+const MAX_HELD_CALL_CHARS = 8 * 1024 * 1024;
 
 /**
  * Retained for importers that pinned the first slice's contract: a leg mixing the search with
@@ -488,6 +492,7 @@ class BridgeStreamState {
    * failing the turn would let Codex start running a tool for a turn that never completes.
    */
   private heldCalls: HeldCallEvent[] = [];
+  private heldCallChars = 0;
   private heldIndexes = new Set<number>();
   private heldItemIds = new Set<string>();
   private terminalPayload: Record<string, unknown> | undefined;
@@ -497,16 +502,23 @@ class BridgeStreamState {
     this.suppressedSearches = new Map();
     this.suppressedItemIds = new Map();
     this.searches = [];
-    this.heldCalls = [];
-    this.heldIndexes = new Set();
-    this.heldItemIds = new Set();
+    this.dropHeldCalls();
     this.terminalPayload = undefined;
   }
 
   get sawClientExecutedCall(): boolean {
     return this.heldCalls.length > 0;
   }
 
+  private holdCall(payload: Record<string, unknown>, dataChars: number, upstreamIndex?: number): void {
+    if (this.heldCalls.length >= MAX_HELD_CALL_EVENTS
+      || dataChars > MAX_HELD_CALL_CHARS - this.heldCallChars) {
+      throw new Error("upstream client tool events exceeded the web-search bridge buffer bound");
+    }
+    this.heldCalls.push({ payload, ...(upstreamIndex === undefined ? {} : { upstreamIndex }) });
+    this.heldCallChars += dataChars;
+  }
+
   private clientIndexFor(upstreamIndex: number): number {
     const existing = this.indexMap.get(upstreamIndex);
     if (existing !== undefined) return existing;
@@ -626,7 +638,7 @@ class BridgeStreamState {
       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 }) });
+        this.holdCall(payload, data.length, upstreamIndex);
         return [];
       }
     }
@@ -648,7 +660,7 @@ class BridgeStreamState {
 
     if ((upstreamIndex !== undefined && this.heldIndexes.has(upstreamIndex))
       || (itemId !== undefined && this.heldItemIds.has(itemId))) {
-      this.heldCalls.push({ payload, ...(upstreamIndex === undefined ? {} : { upstreamIndex }) });
+      this.holdCall(payload, data.length, upstreamIndex);
       return [];
     }
 
@@ -658,19 +670,20 @@ class BridgeStreamState {
     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<string, unknown> = { ...held.payload };
-      if (held.upstreamIndex !== undefined) {
-        rewritten.output_index = this.clientIndexFor(held.upstreamIndex);
+  /** Release lazily so flushing does not allocate a second full set of serialized events. */
+  *flushHeldCalls(): Generator<string> {
+    try {
+      for (const held of this.heldCalls) {
+        const rewritten: Record<string, unknown> = { ...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);
+        yield this.render(String(held.payload.type), rewritten);
       }
-      if (held.payload.type === "response.output_item.done") this.retain(held.payload.item);
-      blocks.push(this.render(String(held.payload.type), rewritten));
+    } finally {
+      this.dropHeldCalls();
     }
-    this.heldCalls = [];
-    return blocks;
   }
 
   /**
@@ -680,6 +693,18 @@ class BridgeStreamState {
    */
   dropHeldCalls(): void {
     this.heldCalls = [];
+    this.heldCallChars = 0;
+    this.heldIndexes.clear();
+    this.heldItemIds.clear();
+  }
+
+  /** Fail before executing this leg's searches, closing every cell already shown to the client. */
+  *failLegFrames(code: string, message: string): Generator<string> {
+    this.dropHeldCalls();
+    for (const call of this.searches) {
+      yield* this.searchEndFrames(call, [], { text: "", sources: [], error: message });
+    }
+    yield* this.failureFrames(code, message);
   }
 
   /** Decide what the leg's terminal means once the whole leg has been read. */
@@ -979,7 +1004,7 @@ async function* bridgeStreamBlocks(
   // 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<string> {
+  const emit = function* (blocks: Iterable<string>): Generator<string> {
     for (const block of blocks) yield block + "\n\n";
   };
 
@@ -991,8 +1016,9 @@ async function* bridgeStreamBlocks(
         if (aborted()) return;
       }
     } catch (error) {
+      if (aborted()) return;
       const message = error instanceof Error ? error.message : String(error);
-      yield* emit(state.failureFrames(
+      yield* emit(state.failLegFrames(
         WEB_SEARCH_BRIDGE_ERROR_CODE,
         "web-search bridge upstream read failed: " + message,
       ));
@@ -1003,18 +1029,7 @@ async function* bridgeStreamBlocks(
 
     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!));
+      yield* emit(state.failLegFrames(decision.code!, decision.message!));
       return;
     }
     if (decision.kind === "end") {

structure/runtime.md

@@ -297,6 +297,16 @@ upstream terminal is `response.failed` or `response.incomplete` runs no search a
 any cell it opened rather than leaving it in progress. Assistant text is not treated as a search
 instruction.
 
+`src/web-search/passthrough-bridge.ts` withholds at most 1,000 client-tool events and
+8,388,608 UTF-16 code units of their SSE data payloads per leg; this is not a byte or total-heap
+measurement. The first over-budget event fails the leg before releasing any held tool call.
+Read failures and exhausted continuation budgets use the same cleanup: discard held calls and
+close every search cell opened by the current leg as failed before one failed terminal and DONE.
+Successful release serializes held events lazily rather than building another full frame array;
+release, discard, and the next leg reset the held payload counter and identity sets.
+`tests/web-search/web-search-progress-stream.test.ts` covers both bounds, identity-only deltas,
+upstream cancellation, cell closure, the exact event boundary, and mixed terminal controls.
+
 The bridge backend and the global `webSearchSidecar` block are configured independently, so the
 sidecar's `model` applies to a bridge search only when `resolveSidecarBackend(webSearchSidecar.backend)`
 equals that bridge backend; otherwise the bridge runs the backend's own default. An unset global

tests/web-search/web-search-progress-stream.test.ts

@@ -5,6 +5,10 @@ import {
   RoutedModelInactivityError,
   WebSearchStreamProtocolError,
 } from "../../src/web-search/progress-stream";
+import {
+  createPassthroughWebSearchBridgeStream,
+  WEB_SEARCH_BRIDGE_ERROR_CODE,
+} from "../../src/web-search/passthrough-bridge";
 import type { AdapterEvent } from "../../src/types";
 
 type ParseStream = ProviderAdapter["parseStream"];
@@ -170,12 +174,12 @@ describe("web-search streamed-body progress collector", () => {
 
   test("semantic delivery is ordered and acknowledged one event at a time", async () => {
     const marks: string[] = [];
-    const parser: ParseStream = async function* () {
-      yield { type: "text_delta", text: "a" };
+    const parser = async function* () {
+      yield { type: "text_delta", text: "a" } as AdapterEvent;
       marks.push("requested-second");
-      yield { type: "text_delta", text: "b" };
+      yield { type: "text_delta", text: "b" } as AdapterEvent;
       marks.push("requested-done");
-      yield { type: "done" };
+      yield { type: "done" } as AdapterEvent;
     };
     const iterator = parseStreamWithProgress(new Response(chunkStream([])), parser, { inactivityTimeoutMs: 200 });
     expect(await iterator.next()).toEqual({ done: false, value: { type: "text_delta", text: "a" } });
@@ -431,3 +435,178 @@ describe("web-search streamed-body progress collector", () => {
     }
   });
 });
+
+describe("web-search passthrough withheld-event stream lifecycle", () => {
+  type Payload = Record<string, unknown>;
+  type Event = {
+    type: string;
+    sequence_number: number;
+    output_index?: number;
+    item?: { type: string; id: string; status?: string; arguments?: string };
+    delta?: string;
+    response?: { error?: { code: string; message: string }; output?: unknown[] };
+  };
+
+  function* legEvents(
+    deltas: number,
+    delta = "x",
+    searches = 0,
+    itemIdOnly = false,
+    terminal = "response.completed",
+  ): Generator<Payload> {
+    for (let index = 0; index < searches; index++) {
+      yield {
+        type: "response.output_item.added", output_index: index,
+        item: {
+          type: "function_call", id: "search-" + index, call_id: "search-call-" + index,
+          name: "web_search", arguments: '{"query":"test"}',
+        },
+      };
+    }
+    const item = { type: "function_call", id: "client-tool", call_id: "client-call", name: "exec", arguments: "" };
+    yield { type: "response.output_item.added", output_index: 7, item };
+    const identity = itemIdOnly ? { item_id: item.id } : { output_index: 7 };
+    for (let index = 0; index < deltas; index++) {
+      yield { type: "response.function_call_arguments.delta", ...identity, delta };
+    }
+    const argumentsText = delta.repeat(deltas);
+    yield { type: "response.function_call_arguments.done", ...identity, arguments: argumentsText };
+    yield { type: "response.output_item.done", output_index: 7, item: { ...item, arguments: argumentsText } };
+    yield { type: terminal, response: { output: [{ ...item, arguments: argumentsText }] } };
+  }
+
+  async function runLeg(events: Iterable<Payload>) {
+    const iterator = events[Symbol.iterator]();
+    const probe = { reads: 0, cancelled: false, executions: 0, sends: 0 };
+    // One frame per pull: a cumulative-limit test must not trip the unrelated single-SSE bound.
+    const firstLeg = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        probe.reads++;
+        const next = iterator.next();
+        if (next.done) controller.close();
+        else controller.enqueue(bytes("data: " + JSON.stringify(next.value) + "\n\n"));
+      },
+      cancel() {
+        probe.cancelled = true;
+        iterator.return?.();
+      },
+    }, { highWaterMark: 0 });
+    const body = createPassthroughWebSearchBridgeStream({
+      plan: { backend: "ollama", endpoint: "https://example.com/search", maxSearches: 3, timeoutMs: 1_000 },
+      firstLeg,
+      requestBody: '{"input":[],"stream":true}',
+      execute: async () => {
+        probe.executions++;
+        return { text: "result", sources: [] };
+      },
+      send: async () => {
+        probe.sends++;
+        throw new Error("a mixed or failed test leg must not continue");
+      },
+    });
+    const wire = await new Response(body).text();
+    const output: Event[] = wire.split("\n")
+      .filter(line => line.startsWith("data: ") && line !== "data: [DONE]")
+      .map(line => JSON.parse(line.slice(6)) as Event);
+    return { wire, output, probe };
+  }
+
+  function expectFailedClosed(result: Awaited<ReturnType<typeof runLeg>>): void {
+    const failures = result.output.filter(event => event.type === "response.failed");
+    expect(failures).toHaveLength(1);
+    expect(failures[0]!.response?.error?.code).toBe(WEB_SEARCH_BRIDGE_ERROR_CODE);
+    expect(result.wire.includes('"name":"exec"')).toBe(false);
+    expect(result.output.some(event => event.type.startsWith("response.function_call_arguments."))).toBe(false);
+    expect(result.probe.executions).toBe(0);
+    expect(result.probe.sends).toBe(0);
+    expect(result.wire.split("data: [DONE]").length - 1).toBe(1);
+    expect(result.wire.endsWith("data: [DONE]\n\n")).toBe(true);
+    expect(result.output.map(event => event.sequence_number)).toEqual(result.output.map((_, index) => index));
+  }
+
+  test.each([false, true])("bounds tiny delta events matched by item id only: %s", async itemIdOnly => {
+    const result = await runLeg(legEvents(1_000, "x", 0, itemIdOnly));
+    expectFailedClosed(result);
+    expect(result.output.at(-1)?.response?.error?.message).toContain("client tool events exceeded");
+    expect(result.probe.reads).toBe(1_001);
+    expect(result.probe.cancelled).toBe(true);
+  });
+
+  test("bounds repeated client-call added events as well as deltas", async () => {
+    function* additions(): Generator<Payload> {
+      for (let index = 0; index < 1_002; index++) {
+        yield {
+          type: "response.output_item.added", output_index: index,
+          item: { type: "function_call", id: "tool-" + index, call_id: "call-" + index, name: "exec", arguments: "" },
+        };
+      }
+    }
+    const result = await runLeg(additions());
+    expectFailedClosed(result);
+    expect(result.probe.reads).toBe(1_001);
+    expect(result.probe.cancelled).toBe(true);
+  });
+
+  test("bounds cumulative payload characters while individual frames and event count remain small", async () => {
+    const result = await runLeg(legEvents(140, "x".repeat(64 * 1024), 0, true));
+    expectFailedClosed(result);
+    expect(result.output.at(-1)?.response?.error?.message).toContain("client tool events exceeded");
+    expect(result.probe.reads).toBeLessThan(140);
+    expect(result.probe.cancelled).toBe(true);
+  });
+
+  test("closes every opened search before failing a held-event overflow", async () => {
+    const result = await runLeg(legEvents(1_000, "x", 2));
+    expectFailedClosed(result);
+    const opened = result.output.filter(event => event.type === "response.output_item.added");
+    const closed = result.output.filter(event => event.type === "response.output_item.done");
+    expect(opened).toHaveLength(2);
+    expect(closed).toHaveLength(2);
+    expect(closed.map(event => [event.item?.id, event.output_index])).toEqual(
+      opened.map(event => [event.item?.id, event.output_index]),
+    );
+    expect(closed.map(event => event.item?.status)).toEqual(["failed", "failed"]);
+    expect(result.output.slice(-3).map(event => event.type)).toEqual([
+      "response.output_item.done", "response.output_item.done", "response.failed",
+    ]);
+    expect(result.probe.cancelled).toBe(true);
+  });
+
+  test("also closes opened searches when reading the upstream leg throws", async () => {
+    function* broken(): Generator<Payload> {
+      yield* Array.from(legEvents(0, "", 2)).slice(0, 3);
+      throw new Error("synthetic read failure");
+    }
+    const result = await runLeg(broken());
+    expectFailedClosed(result);
+    const closed = result.output.filter(event => event.type === "response.output_item.done");
+    expect(closed.map(event => event.item?.status)).toEqual(["failed", "failed"]);
+    expect(result.output.at(-1)?.response?.error?.message).toContain("synthetic read failure");
+  });
+
+  test("releases exactly 1000 held events without loss and preserves remapped order", async () => {
+    // added + 997 deltas + arguments.done + item.done = exactly 1000 withheld events.
+    const result = await runLeg(legEvents(997, "x", 1));
+    expect(result.output.some(event => event.type === "response.failed")).toBe(false);
+    const deltas = result.output.filter(event => event.type === "response.function_call_arguments.delta");
+    expect(deltas).toHaveLength(997);
+    expect(deltas.map(event => event.delta).join("")).toBe("x".repeat(997));
+    expect(deltas.every(event => event.output_index === 1)).toBe(true);
+    const toolDone = result.output.find(event => event.type === "response.output_item.done" && event.item?.type === "function_call");
+    expect(toolDone?.item?.arguments).toBe("x".repeat(997));
+    expect(result.output.at(-1)?.type).toBe("response.completed");
+    expect(result.output.at(-1)?.response?.output).toHaveLength(2);
+    expect(result.probe.executions).toBe(1);
+    expect(result.probe.sends).toBe(0);
+    expect(result.output.map(event => event.sequence_number)).toEqual(result.output.map((_, index) => index));
+  });
+
+  test.each(["response.failed", "response.incomplete"])("preserves mixed-leg terminal handling for %s", async terminal => {
+    const result = await runLeg(legEvents(2, "x", 1, false, terminal));
+    expect(result.output.at(-1)?.type).toBe(terminal);
+    expect(result.probe.executions).toBe(0);
+    expect(result.probe.sends).toBe(0);
+    expect(result.wire.includes('"name":"exec"')).toBe(terminal === "response.incomplete");
+    expect(result.output.find(event => event.item?.type === "web_search_call" && event.type === "response.output_item.done")?.item?.status).toBe("failed");
+  });
+});