diff --git a/src/web-search/passthrough-bridge.ts b/src/web-search/passthrough-bridge.ts index be03084fb7..002470a558 100644 --- a/src/web-search/passthrough-bridge.ts +++ b/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(); private heldItemIds = new Set(); private terminalPayload: Record | undefined; @@ -497,9 +502,7 @@ 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; } @@ -507,6 +510,15 @@ class BridgeStreamState { return this.heldCalls.length > 0; } + private holdCall(payload: Record, 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 = { ...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 { + try { + 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); + 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 { + 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 { + const emit = function* (blocks: Iterable): Generator { 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") { diff --git a/structure/runtime.md b/structure/runtime.md index 37cee672c1..0ced4d8b61 100644 --- a/structure/runtime.md +++ b/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 diff --git a/tests/web-search/web-search-progress-stream.test.ts b/tests/web-search/web-search-progress-stream.test.ts index d36bb42594..e97da2a5c4 100644 --- a/tests/web-search/web-search-progress-stream.test.ts +++ b/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; + 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 { + 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) { + 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({ + 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>): 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 { + 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 { + 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"); + }); +});