fix(session-export): wait for response pull capacity

The ZIP loop checked desiredSize only after a push and responded to an overfull queue with setTimeout(0). A timer turn does not mean the consumer drained anything, so a slow or disconnected client still allowed the producer to enqueue the complete compressed archive while later artifact and attachment reads ran eagerly.

Give the ReadableStream a 64 KiB byte queuing strategy and block the single producer on a pull-released capacity gate whenever desiredSize is non-positive. Cancellation wakes that gate through the existing producer signal; synchronous fflate output is therefore bounded to the queue high-water mark plus one input push. A regression test exhausts timer turns without consuming and proves the next media entry remains unread until response pulling begins, and the bilingual contracts now describe the real bound.

packages/host/apiproxy/src/session-export.ts

@@ -13,10 +13,9 @@
  * the active compressor.
  * Compression runs on the host with fflate's streaming Zip API, so the archive
  * bytes are produced incrementally and the host never holds the whole archive
- * in one buffer; production yields to the consumer whenever the response queue
- * fills past its high-water mark, so a slow consumer bounds the accumulation
- * instead of piling up the whole archive (fflate's callback is synchronous —
- * this drain point is the only backpressure available).
+ * in one buffer; production waits for consumer pull whenever the response queue
+ * reaches its byte high-water mark, so a slow consumer bounds accumulation to
+ * the configured queue plus one synchronous fflate push.
  * @module
  */
 
@@ -266,30 +265,66 @@ const PUSH_CHUNK_CODE_UNITS = 1 << 16
 /** How many bytes of media one zip push carries (bounded memory; images are already size-capped). */
 const PUSH_CHUNK_BYTES = 1 << 16
 
+/** Byte capacity retained by the response stream before ZIP production waits for pull. */
+const RESPONSE_HIGH_WATER_MARK_BYTES = 1 << 16
+
+/** One producer waiter released only when ReadableStream pull restores capacity. */
+class ResponseCapacityGate {
+  private releasePending: (() => void) | undefined
+
+  /**
+   * Wait until the response queue has positive byte capacity or cancellation wins.
+   * @param controller - response controller whose desired size owns capacity.
+   * @param signal - combined request/consumer cancellation.
+   */
+  async wait(
+    controller: ReadableStreamDefaultController<Uint8Array>,
+    signal: AbortSignal,
+  ): Promise<void> {
+    signal.throwIfAborted()
+    if (controller.desiredSize === null || controller.desiredSize > 0) return
+    await new Promise<void>((resolve) => {
+      const release = (): void => {
+        this.releasePending = undefined
+        signal.removeEventListener('abort', release)
+        resolve()
+      }
+      this.releasePending = release
+      signal.addEventListener('abort', release, { once: true })
+    })
+    signal.throwIfAborted()
+  }
+
+  /** Release the current producer waiter after a consumer pull. */
+  pulled(): void {
+    this.releasePending?.()
+  }
+}
+
 /**
  * Push one media object's bytes into a deflate stream in bounded chunks,
- * yielding to a slow consumer between chunks like the artifact path does.
+ * waiting for consumer capacity between chunks like the artifact path does.
  * @param deflate - the zip entry's deflate stream.
  * @param data - the stored image bytes.
- * @param signal - optional cancellation; throws when aborted.
+ * @param controller - response queue controller.
+ * @param capacity - pull-driven response-capacity gate.
+ * @param signal - cancellation; throws when aborted.
  */
 async function pushBinaryChunks(
   deflate: ZipDeflate,
   data: Uint8Array,
   controller: ReadableStreamDefaultController<Uint8Array>,
-  signal?: AbortSignal,
+  capacity: ResponseCapacityGate,
+  signal: AbortSignal,
 ): Promise<void> {
   let offset = 0
   do {
-    signal?.throwIfAborted()
+    signal.throwIfAborted()
     const end = Math.min(offset + PUSH_CHUNK_BYTES, data.byteLength)
     const finalChunk = end >= data.byteLength
     deflate.push(data.subarray(offset, end), finalChunk)
     offset = end
-    /* v8 ignore next 2 -- only fires when a slow consumer leaves the queue over-full */
-    if (controller.desiredSize !== null && controller.desiredSize < 0) {
-      await new Promise(resolve => setTimeout(resolve, 0))
-    }
+    await capacity.wait(controller, signal)
   } while (offset < data.byteLength)
 }
 
@@ -299,19 +334,22 @@ async function pushBinaryChunks(
  * re-encodes as U+FFFD and would silently corrupt the exported artifact).
  * @param deflate - the zip entry's deflate stream.
  * @param content - the artifact text verbatim.
- * @param signal - optional cancellation; throws when aborted.
+ * @param controller - response queue controller.
+ * @param capacity - pull-driven response-capacity gate.
+ * @param signal - cancellation; throws when aborted.
  */
 async function pushArtifactChunks(
   deflate: ZipDeflate,
   content: string,
   controller: ReadableStreamDefaultController<Uint8Array>,
-  signal?: AbortSignal,
+  capacity: ResponseCapacityGate,
+  signal: AbortSignal,
 ): Promise<void> {
   const encoder = new TextEncoder()
   let offset = 0
   let finalChunk: boolean
   do {
-    signal?.throwIfAborted()
+    signal.throwIfAborted()
     let end = Math.min(offset + PUSH_CHUNK_CODE_UNITS, content.length)
     if (end < content.length && end - offset > 1) {
       // Back off one code unit when the boundary lands inside a surrogate
@@ -322,10 +360,7 @@ async function pushArtifactChunks(
     finalChunk = end >= content.length
     deflate.push(encoder.encode(content.slice(offset, end)), finalChunk)
     offset = end
-    /* v8 ignore next 2 -- only fires when a slow consumer leaves the queue over-full */
-    if (controller.desiredSize !== null && controller.desiredSize < 0) {
-      await new Promise(resolve => setTimeout(resolve, 0))
-    }
+    await capacity.wait(controller, signal)
   } while (!finalChunk)
 }
 
@@ -354,6 +389,7 @@ export function streamSessionLogZip(
   const producerSignal = AbortSignal.any([signal, consumerAbort.signal])
   let zip: Zip | undefined
   let zipTerminated = false
+  const capacity = new ResponseCapacityGate()
   const terminateZip = (): void => {
     if (zip === undefined || zipTerminated) return
     zipTerminated = true
@@ -362,9 +398,9 @@ export function streamSessionLogZip(
   return new ReadableStream<Uint8Array>({
     start(controller) {
       // fflate invokes the callback synchronously per compressed chunk, so a
-      // single push can enqueue ahead of a slow consumer; pushArtifactChunks
-      // yields between chunks once the queue is over-full, bounding the
-      // accumulation to the queue high-water mark plus one push.
+      // single push can enqueue ahead of a slow consumer; the capacity gate
+      // waits for pull between pushes once the byte queue is full, bounding
+      // accumulation to the queue high-water mark plus one synchronous push.
       const archive = new Zip((error, data, final) => {
         /* v8 ignore next 3 -- fflate reports only internal zip failures, unreachable for valid inputs */
         if (error) {
@@ -382,9 +418,9 @@ export function streamSessionLogZip(
             const deflate = new ZipDeflate(entry.path, { level: 6 })
             archive.add(deflate)
             if ('content' in entry) {
-              await pushArtifactChunks(deflate, entry.content, controller, producerSignal)
+              await pushArtifactChunks(deflate, entry.content, controller, capacity, producerSignal)
             } else {
-              await pushBinaryChunks(deflate, entry.data, controller, producerSignal)
+              await pushBinaryChunks(deflate, entry.data, controller, capacity, producerSignal)
             }
           }
           archive.end()
@@ -397,11 +433,17 @@ export function streamSessionLogZip(
         }
       })()
     },
+    pull() {
+      capacity.pulled()
+    },
     cancel(reason) {
       consumerAbort.abort(
         reason instanceof Error ? reason : new Error('session log export stream cancelled'),
       )
       terminateZip()
     },
+  }, {
+    highWaterMark: RESPONSE_HIGH_WATER_MARK_BYTES,
+    size: chunk => chunk.byteLength,
   })
 }