From 30adf56babc90d469312c215080b7974c3ebe5a8 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Mon, 10 Aug 2026 03:43:39 +0900 Subject: [PATCH] fix(sidecars): bound streamed response bytes --- src/vision/describe.ts | 4 +- src/web-search/executor.ts | 4 +- src/web-search/parse.ts | 26 +++++++- tests/web-search-parse.test.ts | 105 ++++++++++++++++++++++++++++++++- 4 files changed, 131 insertions(+), 8 deletions(-) diff --git a/src/vision/describe.ts b/src/vision/describe.ts index a7ba80a7e..f51eb8c61 100644 --- a/src/vision/describe.ts +++ b/src/vision/describe.ts @@ -80,8 +80,8 @@ export async function describeImage( "what's relevant to the user's request. Output only the description.", input: [{ type: "message", role: "user", content }], reasoning: { effort: settings.reasoning }, - // The ChatGPT (codex) backend rejects `max_output_tokens` ("Unsupported parameter"); the - // description is clamped downstream (DESC_MAX_CHARS) instead. + // The ChatGPT (codex) backend rejects `max_output_tokens` ("Unsupported parameter"); the shared + // SSE parser bounds raw response bytes before DESC_MAX_CHARS applies its display clamp. store: false, stream: true, }; diff --git a/src/web-search/executor.ts b/src/web-search/executor.ts index 290e1883b..1672d6176 100644 --- a/src/web-search/executor.ts +++ b/src/web-search/executor.ts @@ -62,8 +62,8 @@ export async function runWebSearch( tool_choice: "auto", reasoning: { effort: settings.reasoning }, // NOTE: the ChatGPT (codex) backend rejects `max_output_tokens` ("Unsupported parameter") and - // requires `store: false` — keep this body minimal. Answer length is capped downstream - // (format-result clamps the injected tool_result), so no upstream cap is needed. + // requires `store: false` — keep this body minimal. The shared SSE parser bounds raw response + // bytes before format-result applies its smaller display clamp. store: false, stream: true, }; diff --git a/src/web-search/parse.ts b/src/web-search/parse.ts index c4f88fdcd..181751e19 100644 --- a/src/web-search/parse.ts +++ b/src/web-search/parse.ts @@ -29,6 +29,10 @@ interface OutputItem { content?: OutputTextBlock[]; } +// ChatGPT's Codex backend does not accept `max_output_tokens` on sidecar requests. Bound the raw +// streamed response here, before decoded text and authoritative/delta copies can accumulate. +export const MAX_SIDECAR_RESPONSE_BYTES = 64 * 1024; + /** Push a `url_citation` annotation as a source, de-duplicated by URL. */ function collectAnnotation(ann: AnnotationLike | undefined, sources: WebSearchSource[], seen: Set): void { if (!ann || ann.type !== "url_citation" || typeof ann.url !== "string" || seen.has(ann.url)) return; @@ -175,6 +179,15 @@ function fromOutputArray(output: OutputItem[], seen: Set): WebSearchResu return { text, sources }; } +function cancelReaderWithoutWaiting( + reader: ReadableStreamDefaultReader, + reason: string, +): void { + try { + void reader.cancel(reason).catch(() => undefined); + } catch { /* best-effort body teardown */ } +} + /** * Parse the sidecar's streamed Responses SSE into a final answer + sources. Tolerant of the full set of * Responses streaming events: prefers the authoritative `response.completed` output[], then the @@ -188,6 +201,7 @@ export async function parseSidecarSSE(response: Response): Promise(); // Holder object — fields are mutated inside the closure, so they can't live as narrowed locals. const acc: { @@ -242,13 +256,23 @@ export async function parseSidecarSSE(response: Response): Promise= MAX_SIDECAR_RESPONSE_BYTES) { + // Preserve complete events accepted up to the cap, but discard any unterminated line and + // TextDecoder carry. Do not let a rejecting/hung cancel turn bounded partial output into + // an error or keep this parser waiting on upstream teardown. + cancelReaderWithoutWaiting(reader, "sidecar response byte limit reached"); + break; + } } } finally { reader.releaseLock(); diff --git a/tests/web-search-parse.test.ts b/tests/web-search-parse.test.ts index 588c1e483..13930286c 100644 --- a/tests/web-search-parse.test.ts +++ b/tests/web-search-parse.test.ts @@ -1,11 +1,21 @@ import { describe, expect, test } from "bun:test"; -import { parseSidecarSSE } from "../src/web-search/parse"; +import { MAX_SIDECAR_RESPONSE_BYTES, parseSidecarSSE } from "../src/web-search/parse"; function sse(events: { type: string; [k: string]: unknown }[]): Response { const body = events.map(e => `event: ${e.type}\ndata: ${JSON.stringify(e)}\n\n`).join(""); return new Response(body, { headers: { "Content-Type": "text/event-stream" } }); } +function joinBytes(...chunks: Uint8Array[]): Uint8Array { + const out = new Uint8Array(chunks.reduce((sum, chunk) => sum + chunk.byteLength, 0)); + let offset = 0; + for (const chunk of chunks) { + out.set(chunk, offset); + offset += chunk.byteLength; + } + return out; +} + /** Parse one authoritative completed-text event through the production SSE path. */ async function parseCompletedText(text: string) { return parseSidecarSSE(sse([ @@ -14,6 +24,95 @@ async function parseCompletedText(text: string) { } describe("parseSidecarSSE trailing Sources block", () => { + test("accepts exactly the byte cap, keeps complete events, and cancels without another read", async () => { + const encoder = new TextEncoder(); + const event = encoder.encode(`data:${JSON.stringify({ + type: "response.output_text.delta", + delta: "A", + })}\n\n`); + const paddingBytes = MAX_SIDECAR_RESPONSE_BYTES - event.byteLength; + const padding = encoder.encode(`:${"x".repeat(paddingBytes - 2)}\n`); + const bytes = joinBytes(event, padding); + expect(bytes.byteLength).toBe(MAX_SIDECAR_RESPONSE_BYTES); + + const chunks = [bytes.subarray(0, 12_345), bytes.subarray(12_345)]; + let reads = 0; + let cancels = 0; + const body = new ReadableStream({ + pull(controller) { + reads += 1; + const chunk = chunks.shift(); + if (!chunk) throw new Error("parser read past the sidecar byte cap"); + controller.enqueue(chunk); + }, + cancel() { cancels += 1; }, + }); + + const out = await parseSidecarSSE(new Response(body)); + expect(out.text).toBe("A"); + expect(reads).toBe(2); + expect(cancels).toBe(1); + }); + + test("keeps complete no-space data events and drops a partial event at the cap", async () => { + const encoder = new TextEncoder(); + const complete = encoder.encode(`data:${JSON.stringify({ + type: "response.output_text.delta", + delta: "A", + })}\n\n`); + const partial = encoder.encode('data:{"type":"response.output_text.delta","delta":"B'); + const oversized = new Uint8Array(MAX_SIDECAR_RESPONSE_BYTES + 32); + oversized.set(complete); + oversized.set(partial, complete.byteLength); + oversized.fill(0x78, complete.byteLength + partial.byteLength); + let cancels = 0; + const body = new ReadableStream({ + start(controller) { controller.enqueue(oversized); }, + cancel() { cancels += 1; }, + }); + + const out = await parseSidecarSSE(new Response(body)); + expect(out.text).toBe("A"); + expect(cancels).toBe(1); + }); + + test("does not flush a truncated multibyte sequence at the cap", async () => { + const encoder = new TextEncoder(); + const complete = encoder.encode(`data:${JSON.stringify({ + type: "response.output_text.delta", + delta: "A", + })}\n\n`); + const partialPrefix = encoder.encode('data:{"type":"response.output_text.delta","delta":"'); + const filler = new Uint8Array( + MAX_SIDECAR_RESPONSE_BYTES - complete.byteLength - partialPrefix.byteLength - 1, + ).fill(0x78); + const oversized = joinBytes(complete, partialPrefix, filler, encoder.encode("😀\"}\n\n")); + + const body = new ReadableStream({ + start(controller) { controller.enqueue(oversized); }, + }); + const out = await parseSidecarSSE(new Response(body)); + expect(out.text).toBe("A"); + expect(out.text).not.toContain("�"); + }); + + test("returns bounded partial output when body cancellation rejects", async () => { + const encoder = new TextEncoder(); + const event = encoder.encode(`data:${JSON.stringify({ + type: "response.output_text.delta", + delta: "A", + })}\n\n`); + const oversized = new Uint8Array(MAX_SIDECAR_RESPONSE_BYTES + 1).fill(0x78); + oversized.set(event); + const body = new ReadableStream({ + start(controller) { controller.enqueue(oversized); }, + cancel() { return Promise.reject(new Error("cancel failed")); }, + }); + + const out = await parseSidecarSSE(new Response(body)); + expect(out.text).toBe("A"); + }); + test("extracts sources from a markdown Sources block when annotations are empty", async () => { const text = "Node 24.18.0 is the latest LTS.\n\nSources:\n" + "- Node.js Download page: https://nodejs.org/en/download/current\n" + @@ -129,14 +228,14 @@ describe("parseSidecarSSE trailing Sources block", () => { }); test("handles a long valid Sources header with two star runs", async () => { - const header = `Sources${"*".repeat(50_000)} : ${"*".repeat(49_999)}`; + const header = `Sources${"*".repeat(20_000)} : ${"*".repeat(19_999)}`; const out = await parseCompletedText(`Answer.\n\n${header}\n- https://x.test/source`); expect(out.text).toBe("Answer."); expect(out.sources).toEqual([{ url: "https://x.test/source" }]); }); test("rejects a long near-miss with a third separated star run", async () => { - const header = `Sources${"*".repeat(33_333)} ${"*".repeat(33_333)} ${"*".repeat(33_334)}`; + const header = `Sources${"*".repeat(13_333)} ${"*".repeat(13_333)} ${"*".repeat(13_334)}`; const text = `Answer.\n\n${header}\n- https://x.test/source`; const out = await parseCompletedText(text); expect(out.text).toBe(text);