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,