From b1453a38385a52eece45112799202f6b42da71fd Mon Sep 17 00:00:00 2001 From: JUN Date: Sun, 13 Sep 2026 15:58:40 +0900 Subject: [PATCH] fix(images): bound bridge iteration buffering [skip ci] Carry #4388 from 9518b528166122e775ef4246e81dff4d1b903064 onto the lane B chain above the #4381 carry. A hidden image/video iteration collected adapter events into an unbounded array. Both the runTurn emit path and ordinary stream collection kept every non-heartbeat event until the scanner ran, so an oversized tool argument or a long text stream grew one iteration without limit even though src/lib/translator-budget.ts already defines the turn and call-argument ceilings. Each iteration now owns an independent budget: 32 MiB of serialized retained events including array framing, and 2 MiB of UTF-8 arguments per open call, with surrogate pairs that straddle two deltas counted once. Overflow aborts the producer and surfaces translation_buffer_limit instead of a truncated success. Folded in the Codex review finding left on the source head, which was not addressed there. createAdapterEventQueue merges adjacent text and thinking deltas into chunks of up to 64 KiB while no reader is waiting, so a synchronous producer's token-granular deltas survive as a handful of strings. Charging each pre-merge envelope billed about 31 bytes for every one-character delta, so roughly 1 MiB of retained output crossed the 32 MiB turn limit and aborted a healthy turn. push now reports whether it merged the event into its buffered tail, and a merged delta is charged only for the payload it appended. The return value is additive: every other caller passes push as a void callback and is unaffected. That accounting is deliberately conservative in one place. JSON escaping is per character, so a merged delta costs its quoted form minus the quotes; a surrogate pair split across two deltas over-counts by eight bytes and never under-counts. Verification on this carry: bun test tests/images/loop.test.ts tests/adapters/run-turn-queue.test.ts (74 pass, 0 fail), bun run typecheck, bun run structure:check, bun run privacy:scan. The new coalescing regression was driven red against the unfixed emit path first, where it reproduced the spurious translation_buffer_limit response. Local full suite: NOT RUN. Hosted CI on the lane tip is the suite proof. Co-authored-by: luvs01 <27862058+luvs01@users.noreply.github.com> --- src/adapters/run-turn-queue.ts | 19 +- src/images/loop.ts | 155 ++++++++++++---- structure/runtime.md | 3 + structure/transports/inventory.md | 18 ++ tests/images/loop.test.ts | 288 +++++++++++++++++++++++++++++- 5 files changed, 438 insertions(+), 45 deletions(-) diff --git a/src/adapters/run-turn-queue.ts b/src/adapters/run-turn-queue.ts index 4b63e387de..253407baf2 100644 --- a/src/adapters/run-turn-queue.ts +++ b/src/adapters/run-turn-queue.ts @@ -12,7 +12,13 @@ export const PREFLIGHT_HEARTBEAT_RETAIN_LIMIT = 16; export const COALESCE_MAX_CHUNK_LENGTH = 64 * 1024; export interface AdapterEventQueue { - push(event: AdapterEvent): void; + /** + * Returns true when the event was merged into the buffered tail instead of + * becoming its own retained item. A caller that charges a memory budget for + * what the queue holds needs that distinction: a merged delta costs only its + * appended payload, while a new item costs a whole serialized event. + */ + push(event: AdapterEvent): boolean; close(): void; stream(): AsyncIterable; collect(): Promise; @@ -98,21 +104,22 @@ export function createAdapterEventQueue(opts?: { return false; }; - const push = (event: AdapterEvent): void => { - if (closed) return; + const push = (event: AdapterEvent): boolean => { + if (closed) return false; const reader = readers.shift(); if (reader) { reader({ done: false, value: event }); - return; + return false; } - if (coalesceIntoTail(event)) return; + if (coalesceIntoTail(event)) return true; if (queued.length >= maxBacklog) { opts?.onBacklogExceeded?.(); queued.push({ type: "error", message: "consumer stalled: adapter event backlog exceeded — turn aborted" }); close(); - return; + return false; } queued.push(event); + return false; }; const close = (): void => { diff --git a/src/images/loop.ts b/src/images/loop.ts index e33ae02b41..6f9eacad1f 100644 --- a/src/images/loop.ts +++ b/src/images/loop.ts @@ -24,7 +24,9 @@ import { readBoundedResponseBody } from "../lib/bounded-body"; import { applyUpstreamRecoveryInit, fetchWithResetRetry, prepareSameTarget429Wait } from "../lib/upstream-retry"; import { rateLimitRetryDelayMs } from "../providers/key-failover"; import { + createTranslatorBudget, isTranslatorBudgetExceededError, + TRANSLATOR_MAX_CALL_ARGUMENT_BYTES, TRANSLATOR_MAX_TURN_BYTES, TranslatorBudgetExceededError, } from "../lib/translator-budget"; @@ -101,6 +103,57 @@ interface ImageCall { providerMetadata?: OcxProviderOpaqueToolCallMetadata; } +/** Independent retention owner: adapter leases and final SSE buffers have separate lifetimes. */ +function createIterationEventBudget() { + const budget = createTranslatorBudget(); + let firstEvent = true; + let argumentBytes: number | undefined; + let trailingHighSurrogate = false; + return { + retain(event: AdapterEvent, coalescedIntoTail = false): void { + // Heartbeats are never retained or passed to scanEventsForImageCall. + if (event.type === "heartbeat") return; + if (event.type === "tool_call_start") { + argumentBytes = 0; + trailingHighSurrogate = false; + } else if (event.type === "tool_call_delta" && argumentBytes !== undefined) { + const chunk = event.arguments; + argumentBytes += Buffer.byteLength(chunk); + // A surrogate pair may straddle adapter deltas; count the concatenated UTF-8 string. + if (trailingHighSurrogate && /^[\uDC00-\uDFFF]/.test(chunk)) argumentBytes -= 2; + if (chunk.length > 0) trailingHighSurrogate = /[\uD800-\uDBFF]$/.test(chunk); + if (argumentBytes > TRANSLATOR_MAX_CALL_ARGUMENT_BYTES) { + throw new TranslatorBudgetExceededError("tool_args", TRANSLATOR_MAX_CALL_ARGUMENT_BYTES); + } + } else { + argumentBytes = undefined; + trailingHighSurrogate = false; + } + // A delta the queue merged into its buffered tail leaves one object behind, not two, + // so it costs the appended payload rather than another envelope. Charging the whole + // event here would bill over 32 MiB for the ~1 MiB that a million one-character text + // deltas actually retain, and abort a turn far below the documented limit. + const appended = coalescedIntoTail + ? event.type === "text_delta" ? event.text : event.type === "thinking_delta" ? event.thinking : undefined + : undefined; + if (appended !== undefined) { + // JSON escaping is per character, so the tail grows by the quoted form minus its + // quotes. A surrogate pair split across two deltas is the one over-count, by eight + // bytes, which bounds memory conservatively and never under-charges. + budget.chargeRetained(Buffer.byteLength(JSON.stringify(appended)) - 2, { + kind: "retained_collectors", + }); + return; + } + budget.chargeRetained(Buffer.byteLength(JSON.stringify(event)) + (firstEvent ? 2 : 1), { + kind: "retained_collectors", + }); + firstEvent = false; + }, + dispose: () => budget.dispose(), + }; +} + /** * Split an iteration's adapter events into (a) the image-generation tool calls to intercept and * (b) the events to pass through to Codex. An image tool-call's own start/delta/end events are @@ -370,6 +423,8 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise; @@ -397,29 +452,31 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise internalAbort.abort("runTurn backlog exceeded"), }); - // Attempt telemetry must fire at dispatch time (parity with fetchOnce), not after collect. - deps.onAttemptSend?.(); - void adapter - .runTurn( - iterParsed, - { - headers: deps.forwardHeaders ? new Headers(deps.forwardHeaders) : new Headers(), - abortSignal: signal, - translatorBudget, - }, - queue.push, - ) - .then(() => queue.close()) - .catch(err => { - queue.push({ type: "error", message: err instanceof Error ? err.message : String(err) }); - queue.close(); - }); + const iterationBudget = createIterationEventBudget(); + let accepting = true; + let collectionError: unknown; + const closeOnAbort = (): void => { accepting = false; queue.close(); }; + signal.addEventListener("abort", closeOnAbort, { once: true }); + const emit = (event: AdapterEvent): void => { + if (!accepting || signal.aborted) return; + try { + // Check at emission, before even a synchronous producer can fill the queue. push + // reports whether it merged this delta into the buffered tail, which is what the + // iteration actually retains once the consumer drains it. + iterationBudget.retain(event, queue.push(event)); + } catch (error) { + collectionError = error; + closeOnAbort(); + internalAbort.abort(error); + throw error; + } + }; // Bound collect with a real *idle* deadline that resets on each emitted event. // A fixed wall-clock race would abort legitimate long Cursor turns that keep @@ -439,14 +496,34 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise { + if (accepting) { + collectionError = err; + if (isTranslatorBudgetExceededError(err)) internalAbort.abort(err); + } + closeOnAbort(); + }); idle.reset(); for await (const event of queue.stream()) { if (timedOut) break; idle.reset(); - events.push(event); + if (event.type !== "heartbeat") events.push(event); } } finally { + accepting = false; idle.cancel(); + signal.removeEventListener("abort", closeOnAbort); + iterationBudget.dispose(); + } + if (collectionError) { + if (isTranslatorBudgetExceededError(collectionError)) throw collectionError; + throw new LoopError(502, collectionError instanceof Error ? collectionError.message : String(collectionError)); } if (timedOut) { throw new LoopError(504, `runTurn inactivity timeout after ${stallTimeoutMs}ms during image-bridge`); @@ -466,14 +543,9 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise { - const events: AdapterEvent[] = []; + const events: AdapterEvent[] = prepared.collectedEvents ?? []; + const iterationBudget = prepared.collectedEvents ? undefined : createIterationEventBudget(); try { - const parse = prepared.responseAdapter.parseStream.bind(prepared.responseAdapter); - for await (const event of parseStreamWithProgress(prepared.response, parse, { - signal, - inactivityTimeoutMs: stallTimeoutMs, - translatorBudget, - })) { - if (event.type === "heartbeat") yield event; - else events.push(event); + if (iterationBudget) { + const parse = prepared.responseAdapter.parseStream.bind(prepared.responseAdapter); + for await (const event of parseStreamWithProgress(prepared.response, parse, { + signal, + inactivityTimeoutMs: stallTimeoutMs, + translatorBudget, + })) { + if (event.type === "heartbeat") yield event; + else { + iterationBudget.retain(event); + events.push(event); + } + } } } catch (error) { - if (isTranslatorBudgetExceededError(error)) throw error; + if (isTranslatorBudgetExceededError(error)) { + internalAbort.abort(error); + throw error; + } if (signal.aborted) throw new LoopError(499, "client closed request during image-bridge"); if (error instanceof RoutedModelInactivityError) throw new LoopError(504, error.message); if (error instanceof WebSearchStreamProtocolError) throw new LoopError(502, error.message); throw new LoopError(502, `Provider stream error: ${error instanceof Error ? error.message : String(error)}`); + } finally { + iterationBudget?.dispose(); } const terminalIndexes = events.flatMap((event, index) => diff --git a/structure/runtime.md b/structure/runtime.md index 35b1e1f31a..c03c14d914 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -173,6 +173,9 @@ The server exposes `POST /api/stop` which restores native Codex config, stops an Adapter output must stay in internal `AdapterEvent` form until `bridge.ts` converts it back to Responses SSE or WebSocket frames. +The image/video loop bounds each hidden iteration before replay or fulfillment; see +[media iteration retention](transports/inventory.md#media-iteration-retention). + Live model discovery is bounded and registry-driven through `src/providers/model-discovery.ts`. Custom providers keep the conventional `${baseUrl}/models` request; canonical presets may select a trusted URL/path/query and declarative eligibility filter without persisting that policy into user diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index b2ba773ba1..ff966bb1bc 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -38,6 +38,24 @@ Native Composer/MCP behavior and text-only historical replay remain unchanged. The shared Responses path follows the [bounded multipart recovery contract](../subagents.md#multipart-encrypted-task-recovery); credential admission and retry policy remain unchanged. +## Media iteration retention + +`src/images/loop.ts` admits at most 32 MiB of serialized non-heartbeat adapter events per +iteration, including array framing, and 2 MiB of UTF-8 arguments per current tool call. Both +`runTurn` emission and ordinary stream collection enforce the shared translator limits before +retaining another event. Overflow aborts the producer and surfaces `translation_buffer_limit`. +The iteration budget is separate from adapter leases and final response buffers; collected +`runTurn` events reach the scanner directly without a second charge. Heartbeats do not reset +argument accounting, and each new iteration receives a fresh retention budget. The charge follows +what `src/adapters/run-turn-queue.ts` keeps: `push` reports whether it merged a text or thinking +delta into its buffered tail, and a merged delta costs only its appended payload. Billing every +pre-merge envelope would abort a turn on roughly a thirtieth of the documented limit whenever a +producer streams token-granular deltas ahead of its consumer. These bounds do +not cap process RSS or the conversation messages accumulated across completed media iterations. +`tests/images/loop.test.ts` covers early producer cancellation, byte boundaries, iteration reset, +opaque metadata, normal tool passthrough, coalesced-tail accounting, and consumer cancellation on +both execution paths. + ## Provider diagnostic outbound safety Provider connection tests and live model discovery share the GET-only provider outbound wrapper. diff --git a/tests/images/loop.test.ts b/tests/images/loop.test.ts index 4b9d6423b7..9dbddf1a68 100644 --- a/tests/images/loop.test.ts +++ b/tests/images/loop.test.ts @@ -7,6 +7,12 @@ import type { AdapterEvent, OcxParsedRequest } from "../../src/types"; import type { ImageBridgePlan, ImageCallResult } from "../../src/images/types"; import type { ImageBridgeDeps } from "../../src/images/loop"; import { createTestTranslatorBudget } from "../helpers/translator-budget"; +import { parseStreamWithProgress, type ParseStreamWithProgressOptions } from "../../src/web-search/progress-stream"; +import { TRANSLATOR_MAX_CALL_ARGUMENT_BYTES, TRANSLATOR_MAX_TURN_BYTES, translatorLiveBudgetCountForTests } from "../../src/lib/translator-budget"; + +const realParseStreamWithProgress = parseStreamWithProgress; +let useRealProgressStream = false; +let fulfillCallCount = 0; const PREV_HOME = process.env.OPENCODEX_HOME; let runWithImageBridgeProduction: typeof import("../../src/images/loop")["runWithImageBridge"]; @@ -23,14 +29,15 @@ beforeAll(async () => { process.env.OPENCODEX_HOME = join(tmpdir(), "ocx-test-" + randomUUID()); mock.restore(); mock.module("../../src/web-search/progress-stream", () => ({ - parseStreamWithProgress: async function* (_resp: Response, parse: (r: Response) => AsyncGenerator, _opts: unknown) { - for await (const e of parse(_resp)) yield e; + parseStreamWithProgress: async function* (_resp: Response, parse: ProviderAdapter["parseStream"], opts: ParseStreamWithProgressOptions) { + if (useRealProgressStream) yield* realParseStreamWithProgress(_resp, parse, opts); + else for await (const e of parse(_resp, opts.translatorBudget)) yield e; }, RoutedModelInactivityError: class extends Error { readonly timeoutMs = 0; }, WebSearchStreamProtocolError: class extends Error { /* */ }, })); mock.module("../../src/images/fulfill", () => ({ - fulfillImageCall: async (): Promise => fulfillResult, + fulfillImageCall: async (): Promise => { fulfillCallCount++; return fulfillResult; }, })); ({ runWithImageBridge: runWithImageBridgeProduction, @@ -62,11 +69,195 @@ const defaultFulfillResult: ImageCallResult = { files: ["/test/img.png"], count: 1, markdown: "![image](/test/img.png)", }; beforeEach(() => { + useRealProgressStream = false; + fulfillCallCount = 0; fulfillResult = { ...defaultFulfillResult, files: [...defaultFulfillResult.files] }; buildRequestCalls = 0; streamQueue = []; }); +describe.each(["runTurn", "parseStream"] as const)("image-loop collection bounds — %s", mode => { + beforeEach(() => { useRealProgressStream = true; }); + + function streamingAdapter(events: () => Generator) { + const state = { produced: 0, terminalProduced: false, closed: false, cancelled: false, signal: undefined as AbortSignal | undefined, requests: [] as OcxParsedRequest[] }; + async function* source(): AsyncGenerator { + try { + for (const event of events()) { + if (state.signal?.aborted) return; + state.produced++; + if (event.type === "done") state.terminalProduced = true; + yield event; + // Keep queue backlog small: the regression is cumulative iteration retention. + await Bun.sleep(1); + } + } finally { state.closed = true; } + } + const adapter: ProviderAdapter = { + name: "bounded-media-fixture", + buildRequest: async (_parsed, incoming) => { + state.signal = incoming.abortSignal; + state.requests.push(_parsed); + return { url: "https://example.invalid/model", method: "POST", headers: {}, body: "{}" }; + }, + fetchResponse: async () => new Response(new ReadableStream({ + cancel() { state.cancelled = true; }, + })), + parseStream: source, + ...(mode === "runTurn" ? { + runTurn: async (_parsed: OcxParsedRequest, incoming: IncomingMeta, emit: (event: AdapterEvent) => void) => { + state.signal = incoming.abortSignal; + state.requests.push(_parsed); + for await (const event of source()) emit(event); + }, + } : {}), + }; + return { adapter, state }; + } + + test("aborts retained-event overflow before the producer reaches its terminal", async () => { + const { adapter, state } = streamingAdapter(function* () { + const text = "x".repeat(1024 * 1024); + for (let i = 0; i < 40; i++) yield { type: "text_delta", text }; + yield { type: "done" }; + }); + const response = await runWithImageBridge({ parsed: makeParsed(), adapter, plan }); + const sse = await response.text(); + await Bun.sleep(5); + expect(state.terminalProduced).toBe(false); + expect(state.produced).toBeLessThan(40); + expect(state.signal?.aborted).toBe(true); + expect(state.closed).toBe(true); + if (mode === "parseStream") expect(state.cancelled).toBe(true); + expect(sse).toContain('"code":"translation_buffer_limit"'); + expect(sse).not.toContain("event: response.completed"); + expect(fulfillCallCount).toBe(0); + }); + + test("aborts cumulative UTF-8 arguments before media fulfillment or terminal", async () => { + const { adapter, state } = streamingAdapter(function* () { + yield { type: "tool_call_start", id: "oversize", name: "image_gen" }; + const argumentsChunk = "한".repeat(Math.floor(TRANSLATOR_MAX_CALL_ARGUMENT_BYTES / 6)); + for (let i = 0; i < 4; i++) { + yield { type: "tool_call_delta", arguments: argumentsChunk }; + yield { type: "heartbeat" }; + } + yield { type: "tool_call_end" }; + yield { type: "done" }; + }); + const response = await runWithImageBridge({ parsed: makeParsed(), adapter, plan }); + const sse = await response.text(); + await Bun.sleep(5); + expect(state.terminalProduced).toBe(false); + expect(state.produced).toBeLessThan(9); + expect(state.signal?.aborted).toBe(true); + expect(state.closed).toBe(true); + if (mode === "parseStream") expect(state.cancelled).toBe(true); + expect(sse).toContain('"code":"translation_buffer_limit"'); + expect(sse).not.toContain("event: response.completed"); + expect(fulfillCallCount).toBe(0); + expect(translatorLiveBudgetCountForTests()).toBe(1); // Only the caller-owned budget remains. + }); + + test.each([0, 1])("retained JSON array boundary plus %i byte", async extra => { + const first: AdapterEvent[] = [{ type: "text_delta", text: "" }, ...imageCallEvents]; + const overhead = Buffer.byteLength(JSON.stringify(first)); + first[0] = { type: "text_delta", text: "x".repeat(TRANSLATOR_MAX_TURN_BYTES - overhead + extra) }; + let iteration = 0; + const { adapter } = streamingAdapter(function* () { + if (iteration++ === 0) yield* first; + else { yield { type: "text_delta", text: "finished" }; yield { type: "done" }; } + }); + const response = await runWithImageBridge({ parsed: makeParsed(), adapter, plan }); + const sse = await response.text(); + expect(sse.includes('"code":"translation_buffer_limit"')).toBe(extra === 1); + expect(sse.includes("event: response.completed")).toBe(extra === 0); + expect(fulfillCallCount).toBe(extra === 0 ? 1 : 0); + }); + + test("resets the retained-event budget between media iterations", async () => { + let iteration = 0; + const { adapter } = streamingAdapter(function* () { + if (iteration++ < 2) { + yield { type: "text_delta", text: "x".repeat(18 * 1024 * 1024) }; + yield* imageCallEvents; + } else { yield { type: "text_delta", text: "finished" }; yield { type: "done" }; } + }); + const response = await runWithImageBridge({ parsed: makeParsed(), adapter, plan }); + const sse = await response.text(); + expect(sse).toContain("event: response.completed"); + expect(sse).not.toContain("translation_buffer_limit"); + expect(fulfillCallCount).toBe(2); + }); + + test("accepts exact UTF-8 argument limits per call and preserves opaque metadata", async () => { + let iteration = 0; + const signatures = ["first-synthetic-signature", "second-synthetic-signature"]; + const { adapter, state } = streamingAdapter(function* () { + if (iteration++ > 0) { yield { type: "done" }; return; } + for (const signature of signatures) { + yield { type: "tool_call_start", id: signature, name: "image_gen", providerMetadata: { google: { thoughtSignature: signature } } }; + const prefix = '{"prompt":"'; + const suffix = '"}'; + yield { type: "tool_call_delta", arguments: prefix + "x".repeat(TRANSLATOR_MAX_CALL_ARGUMENT_BYTES - prefix.length - suffix.length - 4) }; + yield { type: "tool_call_delta", arguments: "\uD83D" }; + yield { type: "heartbeat" }; + yield { type: "tool_call_delta", arguments: "\uDE00" + suffix }; + yield { type: "tool_call_end" }; + } + yield { type: "done" }; + }); + const response = await runWithImageBridge({ parsed: makeParsed(), adapter, plan }); + const sse = await response.text(); + expect(sse).toContain("event: response.completed"); + expect(sse).not.toContain("translation_buffer_limit"); + expect(fulfillCallCount).toBe(2); + const assistant = state.requests[1]?.context.messages.find(message => message.role === "assistant"); + const calls = assistant?.role === "assistant" ? assistant.content.filter(part => part.type === "toolCall") : []; + expect(calls.map(call => call.providerMetadata?.google?.thoughtSignature)).toEqual(signatures); + expect(calls.map(call => Buffer.byteLength(JSON.stringify(call.arguments)))).toEqual([TRANSLATOR_MAX_CALL_ARGUMENT_BYTES, TRANSLATOR_MAX_CALL_ARGUMENT_BYTES]); + }); + + test("passes normal real tool calls through without media fulfillment", async () => { + const { adapter } = streamingAdapter(function* () { + yield { type: "tool_call_start", id: "real", name: "read_file", providerMetadata: { google: { thoughtSignature: "real-call-signature" } } }; + yield { type: "tool_call_delta", arguments: '{"path":"example.txt"}' }; + yield { type: "tool_call_end" }; + yield { type: "done" }; + }); + const response = await runWithImageBridge({ parsed: makeParsed(), adapter, plan }); + const sse = await response.text(); + expect(sse).toContain("event: response.completed"); + expect(sse).toContain('"name":"read_file"'); + expect(sse).toContain('"thought_signature":"real-call-signature"'); + expect(fulfillCallCount).toBe(0); + }); + + test("consumer cancellation aborts and releases an active collector", async () => { + const { adapter, state } = streamingAdapter(function* () { + for (let i = 0; i < 100; i++) yield { type: "text_delta", text: "pending" }; + yield { type: "done" }; + }); + const response = await runWithImageBridge({ parsed: makeParsed(), adapter, plan }); + const reader = response.body!.getReader(); + const draining = (async () => { while (!(await reader.read()).done) { /* keep demanding SSE */ } })(); + try { + for (let i = 0; i < 20 && state.produced === 0; i++) await Bun.sleep(1); + expect(state.produced).toBeGreaterThan(0); + } finally { + await reader.cancel("synthetic consumer closed"); + await draining; + } + await Bun.sleep(5); + expect(state.signal?.aborted).toBe(true); + expect(state.closed).toBe(true); + expect(state.terminalProduced).toBe(false); + if (mode === "parseStream") expect(state.cancelled).toBe(true); + expect(fulfillCallCount).toBe(0); + expect(translatorLiveBudgetCountForTests()).toBe(1); + }); +}); + const mockAdapter: ProviderAdapter = { name: "test", buildRequest: async () => { buildRequestCalls++; return { url: "https://test/v1/chat", method: "POST", headers: {}, body: "{}" }; }, @@ -731,6 +922,97 @@ describe("runWithImageBridge", () => { // --------------------------------------------------------------------------- describe("runWithImageBridge — runTurn adapter", () => { + test("charges the queue's coalesced tail, not each delta it discarded", async () => { + // createAdapterEventQueue merges adjacent text deltas into chunks while no reader is + // scheduled, so a synchronous producer's one-character deltas survive as a handful of + // strings. Charging each original event's envelope instead billed ~31 bytes apiece and + // tripped the 32 MiB turn limit on roughly 1 MiB of retained output. + const deltas = 1_200_000; + const response = await runWithImageBridge({ + parsed: makeParsed(), plan, + adapter: { + ...mockAdapter, + runTurn: async (_parsed, _incoming, emit) => { + for (let i = 0; i < deltas; i++) emit({ type: "text_delta", text: "x" }); + emit({ type: "done" }); + }, + }, + }); + const sse = await response.text(); + expect(Buffer.byteLength(JSON.stringify({ type: "text_delta", text: "x" })) * deltas) + .toBeGreaterThan(TRANSLATOR_MAX_TURN_BYTES); + expect(sse).not.toContain("translation_buffer_limit"); + expect(sse).toContain("event: response.completed"); + }); + + test("queue backlog overflow keeps its upstream error instead of becoming client cancellation", async () => { + const response = await runWithImageBridge({ + parsed: makeParsed(), plan, + adapter: { + ...mockAdapter, + runTurn: async (_parsed, _incoming, emit) => { + for (let i = 0; i < 1100; i++) emit({ type: "tool_call_start", id: `call_${i}`, name: "read_file" }); + emit({ type: "done" }); + }, + }, + }); + const sse = await response.text(); + expect(sse).toContain("adapter event backlog exceeded"); + expect(sse).not.toContain("client closed request"); + expect(sse).not.toContain("event: response.completed"); + }); + + test("a completed batch is not fulfilled after its turn signal aborts", async () => { + const abort = new AbortController(); + const adapter: ProviderAdapter = { + ...mockAdapter, + runTurn: async (_parsed, _incoming, emit) => { + for (const event of imageCallEvents) emit(event); + abort.abort("synthetic cancelled turn"); + }, + }; + const response = await runWithImageBridge({ parsed: makeParsed(), adapter, plan, abortSignal: abort.signal }); + const sse = await response.text(); + expect(sse).not.toContain("event: response.completed"); + expect(fulfillCallCount).toBe(0); + }); + + test("emits after runTurn settles cannot recharge its collection", async () => { + let lateEmit!: (event: AdapterEvent) => void; + let resolveRun!: () => void; + let incomingSignal: AbortSignal | undefined; + const finished = new Promise(resolve => { resolveRun = resolve; }); + const adapter: ProviderAdapter = { + ...mockAdapter, + runTurn: (_parsed, incoming, emit) => { + incomingSignal = incoming.abortSignal; + lateEmit = emit; + // Alternate event types so the queue still has a batch to drain after producer settlement. + for (let i = 0; i < 32; i++) { + emit({ type: "text_delta", text: "finished" }); + emit({ type: "thinking_delta", thinking: "synthetic thought" }); + } + emit({ type: "done" }); + resolveRun(); + return finished; + }, + }; + const response = await runWithImageBridge({ parsed: makeParsed(), adapter, plan }); + const reading = response.text(); + await finished; + // Queue the emit after the bridge's promise completion handler, while the batch can still drain. + await Promise.resolve(); + const abortedBeforeLateEmit = incomingSignal?.aborted; + expect(abortedBeforeLateEmit).toBe(false); + expect(() => lateEmit({ type: "tool_call_start", id: "late", name: "image_gen" })).not.toThrow(); + expect(() => lateEmit({ type: "tool_call_delta", arguments: "x".repeat(TRANSLATOR_MAX_CALL_ARGUMENT_BYTES + 1) })).not.toThrow(); + expect(incomingSignal?.aborted).toBe(abortedBeforeLateEmit); + const sse = await reading; + expect(sse).toContain("event: response.completed"); + expect(sse).not.toContain("translation_buffer_limit"); + expect(fulfillCallCount).toBe(0); + }); + let runTurnEventQueue: AdapterEvent[][] = []; const runTurnAdapter: ProviderAdapter = { ...mockAdapter,