fix(images): bound bridge iteration buffering

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<AdapterEvent>;
   collect(): Promise<AdapterEvent[]>;
@@ -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 => {

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<Respons
   interface IterationResponse {
     response: Response;
     responseAdapter: ProviderAdapter;
+    /** runTurn events are already bounded; consume them without replaying or charging twice. */
+    collectedEvents?: AdapterEvent[];
   }
   type IterationSplit = ReturnType<typeof scanEventsForImageCall>;
 
@@ -397,29 +452,31 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise<Respons
 
     // runTurn adapters (Cursor) own all upstream communication via an emit callback. They don't
     // expose buildRequest/fetchResponse/parseStream to the bridge, so collect their events through
-    // an AdapterEventQueue and wrap them in a pseudo-response whose parseStream replays them.
+    // an AdapterEventQueue and pass the bounded collection directly to the common scanner.
     if (adapter.runTurn) {
       await deps.waitForRequestSlot?.(signal);
       const queue = createAdapterEventQueue({
         onBacklogExceeded: () => 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<Respons
       });
       const events: AdapterEvent[] = [];
       try {
+        // Attempt telemetry must fire at dispatch time, not after collection.
+        deps.onAttemptSend?.();
+        void adapter.runTurn(iterParsed, {
+          headers: deps.forwardHeaders ? new Headers(deps.forwardHeaders) : new Headers(),
+          abortSignal: signal,
+          translatorBudget,
+        }, emit).then(closeOnAbort).catch(err => {
+          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<Respons
         }
         throw new LoopError(502, errorEvent.message);
       }
+      if (signal.aborted) throw new LoopError(499, "client closed request during image-bridge");
 
-      const wrappedAdapter: ProviderAdapter = {
-        ...adapter,
-        async *parseStream() {
-          for (const e of events) yield e;
-        },
-      };
-      return { response: new Response(new Uint8Array(0), { status: 200 }), responseAdapter: wrappedAdapter };
+      return { response: new Response(new Uint8Array(0), { status: 200 }), responseAdapter: adapter, collectedEvents: events };
     }
 
     let headerDeadline = clearableDeadline(connectTimeoutMs, signal);
@@ -643,23 +715,34 @@ export async function runWithImageBridge(deps: ImageBridgeDeps): Promise<Respons
   // Consume and validate one successful response body. Only invisible heartbeat events escape while
   // semantic output remains buffered for safe scanning.
   const consumeIterationEvents = async function* (prepared: IterationResponse): AsyncGenerator<AdapterEvent, IterationSplit> {
-    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) =>

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

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.