fix: fold the wp6 re-audit round (house bounded-body primitive, exact UTF-8 cap)

Re-audit findings on 9794e24:

- readBoundedResponseText awaited reader.cancel(), so a broken stream whose
  cancellation never settles would hang the overflow path instead of returning
  the documented 502. The custom helper is replaced by the house primitive
  readBoundedResponseBody, which cancels fire-and-forget with synchronous-throw
  protection and adds total (180s) and inactivity (30s) transfer deadlines on
  top of the byte ceiling; oversize and stalls both fail closed, and a partial
  body is never parsed.
- bounded-body.ts gains a maxBytes option (default unchanged at 64 KiB) and
  accumulates into a geometrically growing single buffer, so per-chunk metadata
  cannot amplify beyond the payload budget on large ceilings.
- the repair byte cap measured UTF-16 code units; astral text could enter the
  parse attempts at 4x the intended bytes. utf8BytesExceed measures the exact
  UTF-8 length with early exit and no allocation; regression test covers a
  600k-code-unit, 1.2 MB input that a length check would have admitted.

Author: lidge-jun

src/adapters/anthropic.ts

@@ -288,6 +288,26 @@ function usableToolUseId(id: unknown): string {
  */
 const MAX_REPAIRABLE_TOOL_ARGUMENT_BYTES = 1024 * 1024;
 
+/**
+ * Whether `input` encodes to more than `max` UTF-8 bytes, with an early exit so the check
+ * itself never allocates a copy of a hostile string. `string.length` counts UTF-16 code
+ * units, which undercounts astral text by 2x against a byte budget.
+ */
+function utf8BytesExceed(input: string, max: number): boolean {
+  let bytes = 0;
+  for (let i = 0; i < input.length; i++) {
+    const code = input.charCodeAt(i);
+    if (code < 0x80) bytes += 1;
+    else if (code < 0x800) bytes += 2;
+    else if (code >= 0xd800 && code <= 0xdbff && i + 1 < input.length) {
+      bytes += 4;
+      i++;
+    } else bytes += 3; // lone surrogates encode as U+FFFD (3 bytes)
+    if (bytes > max) return true;
+  }
+  return false;
+}
+
 function lastValidJsonObject(input: string, maxCandidates: number): string | undefined {
   const tryParseObject = (candidate: string): string | undefined => {
     try {
@@ -323,7 +343,7 @@ function toolUseArguments(input: unknown, lenient = false): string {
       JSON.parse(trimmed);
       return trimmed;
     } catch {
-      if (lenient && trimmed.length <= MAX_REPAIRABLE_TOOL_ARGUMENT_BYTES) {
+      if (lenient && !utf8BytesExceed(trimmed, MAX_REPAIRABLE_TOOL_ARGUMENT_BYTES)) {
         const repaired = lastValidJsonObject(trimmed, 32);
         if (repaired !== undefined) return repaired;
       }

src/lib/bounded-body.ts

@@ -7,6 +7,12 @@ export const BOUNDED_BODY_TIMEOUT_MS = 5_000;
 export interface BoundedBodyOptions {
 	/** Abort the read with this signal. Its reason is rethrown by identity. */
 	signal?: AbortSignal;
+	/**
+	 * Byte ceiling for retained body data. Defaults to BOUNDED_BODY_MAX_BYTES (64 KiB),
+	 * which suits error bodies; callers materializing whole success payloads (e.g. a
+	 * non-streaming upstream JSON completion) pass a larger explicit budget.
+	 */
+	maxBytes?: number;
 	/** Total wall-clock deadline. Exposed for focused tests. */
 	totalTimeoutMs?: number;
 	/** Deadline between non-empty raw chunks. Exposed for focused tests. */
@@ -93,7 +99,11 @@ export async function readBoundedResponseBody(
 	}
 
 	const reader = body.getReader();
-	const chunks: Uint8Array[] = [];
+	const maxBytes = options.maxBytes ?? BOUNDED_BODY_MAX_BYTES;
+	// Geometrically growing single buffer: per-chunk arrays would retain one object per
+	// transport chunk, which a hostile peer could inflate into metadata amplification far
+	// beyond the payload ceiling on large budgets.
+	let retained = new Uint8Array(Math.min(maxBytes, 64 * 1024));
 	let retainedBytes = 0;
 	let mustCancel = false;
 	let cancelReason: unknown;
@@ -134,7 +144,7 @@ export async function readBoundedResponseBody(
 					"TimeoutError",
 				);
 				return {
-					text: decodeUtf8(chunks),
+					text: decodeUtf8([retained.subarray(0, retainedBytes)]),
 					truncated: true,
 					timedOut: true,
 					totalTimedOut: outcome === TOTAL_TIMEOUT,
@@ -147,7 +157,7 @@ export async function readBoundedResponseBody(
 			const { value, done } = outcome as ReadableStreamReadResult<Uint8Array>;
 			if (done) {
 				return {
-					text: decodeUtf8(chunks),
+					text: decodeUtf8([retained.subarray(0, retainedBytes)]),
 					truncated: false,
 					timedOut: false,
 					totalTimedOut: false,
@@ -165,10 +175,10 @@ export async function readBoundedResponseBody(
 				INACTIVITY_TIMEOUT,
 			);
 
-			if (value.byteLength > BOUNDED_BODY_MAX_BYTES - retainedBytes) {
+			if (value.byteLength > maxBytes - retainedBytes) {
 				mustCancel = true;
 				cancelReason = new DOMException("Error body size limit reached", "QuotaExceededError");
-				chunks.length = 0;
+				retained = new Uint8Array(0);
 				retainedBytes = 0;
 				return {
 					text: "",
@@ -181,7 +191,14 @@ export async function readBoundedResponseBody(
 				};
 			}
 
-			chunks.push(value);
+			if (retainedBytes + value.byteLength > retained.length) {
+				const grown = new Uint8Array(
+					Math.min(maxBytes, Math.max(retained.length * 2, retainedBytes + value.byteLength)),
+				);
+				grown.set(retained.subarray(0, retainedBytes));
+				retained = grown;
+			}
+			retained.set(value, retainedBytes);
 			retainedBytes += value.byteLength;
 		}
 	} catch (error) {

src/server/relay.ts

@@ -20,48 +20,6 @@ export const MAX_INSPECTION_SSE_FRAME_BYTES = 4 * 1024 * 1024;
 export const MAX_COMPLETED_OUTPUT_ITEMS = 256;
 export const MAX_COMPLETED_OUTPUT_ITEM_SOURCE_BYTES = 8 * 1024 * 1024;
 export const MAX_TAIL_ERROR_MESSAGE_CHARS = 512;
-// Whole-body ceiling for a non-streaming upstream JSON response. The caller materializes
-// the body for logging and (on the WebSocket bridge) reframing, so an unbounded `.text()`
-// read would let a hostile or broken upstream grow proxy memory without limit. 32 MiB
-// matches the continuation snapshot read bound and is far above any legitimate
-// non-streaming completion (including base64 image payloads).
-export const MAX_UPSTREAM_JSON_BODY_BYTES = 32 * 1024 * 1024;
-
-/**
- * Read an entire upstream body as text with a hard byte ceiling. Returns `truncated: true`
- * (with the body already cancelled) when the body exceeds `maxBytes`; callers must treat
- * truncation as an upstream failure, never parse the partial text. A null body reads as "".
- */
-export async function readBoundedResponseText(
-  body: ReadableStream<Uint8Array> | null,
-  maxBytes: number = MAX_UPSTREAM_JSON_BODY_BYTES,
-): Promise<{ text: string; truncated: boolean }> {
-  if (!body) return { text: "", truncated: false };
-  const reader = body.getReader();
-  const chunks: Uint8Array[] = [];
-  let total = 0;
-  try {
-    for (;;) {
-      const { done, value } = await reader.read();
-      if (done) break;
-      total += value.byteLength;
-      if (total > maxBytes) {
-        await reader.cancel("upstream body exceeded the safe byte limit").catch(() => {});
-        return { text: "", truncated: true };
-      }
-      chunks.push(value);
-    }
-  } finally {
-    reader.releaseLock();
-  }
-  const merged = new Uint8Array(total);
-  let offset = 0;
-  for (const chunk of chunks) {
-    merged.set(chunk, offset);
-    offset += chunk.byteLength;
-  }
-  return { text: new TextDecoder().decode(merged), truncated: false };
-}
 
 export type InspectionCounters = {
   frameBufferHighWaterBytes: number;

src/server/responses/core.ts

@@ -139,7 +139,6 @@ import {
   isNativePassthroughSseResponse,
   markEagerRelaySseResponse,
   markNativePassthroughSseResponse,
-  readBoundedResponseText,
   relaySseWithFailedTail,
   relayWithAbort,
   sanitizePassthroughHeaders,
@@ -688,6 +687,15 @@ export function buildComboChildHeaders(parentHeaders: HeadersInit): Headers {
 const UNREADABLE_ENCRYPTED_AGENT_TASK_MESSAGE =
   "Routed V2 worker task is encrypted for the native ChatGPT backend and cannot be read by the selected provider. Use plaintext V2 agent-message delivery or select a native ChatGPT model.";
 
+// Whole-body policy for non-streaming upstream JSON responses (see the application/json
+// branch of the passthrough return path). 32 MiB matches the continuation snapshot read
+// bound and is far above any legitimate non-streaming completion, including base64 image
+// payloads. The stall deadlines only govern the body transfer — generation time before
+// the response headers is untouched.
+const MAX_UPSTREAM_JSON_BODY_BYTES = 32 * 1024 * 1024;
+const UPSTREAM_JSON_BODY_TOTAL_TIMEOUT_MS = 180_000;
+const UPSTREAM_JSON_BODY_INACTIVITY_TIMEOUT_MS = 30_000;
+
 function unreadableEncryptedAgentTaskResponse(): Response {
   return new Response(
     JSON.stringify({
@@ -1969,10 +1977,18 @@ async function handleResponsesInner(
       // so an unbounded .text() would let a hostile or stuck upstream grow proxy memory
       // without limit. This path is no longer rare — WebSocket turns for models whose
       // streaming terminal event is unreliable are deliberately answered with bounded JSON.
-      const bounded = await readBoundedResponseText(upstreamResponse.body);
-      if (bounded.truncated) {
+      // Oversize and stall deadlines both fail closed; a partial body is never parsed.
+      const bounded = await readBoundedResponseBody(upstreamResponse, {
+        maxBytes: MAX_UPSTREAM_JSON_BODY_BYTES,
+        totalTimeoutMs: UPSTREAM_JSON_BODY_TOTAL_TIMEOUT_MS,
+        inactivityTimeoutMs: UPSTREAM_JSON_BODY_INACTIVITY_TIMEOUT_MS,
+      });
+      if (bounded.oversized) {
         return formatErrorResponse(502, "upstream_error", "upstream JSON response exceeded the safe body limit");
       }
+      if (bounded.truncated) {
+        return formatErrorResponse(502, "upstream_error", "upstream JSON response stalled before completing");
+      }
       const text = bounded.text;
       inspectResponseLogJson(logCtx, text);
       if (rememberPassthroughResponse) {

tests/anthropic-eof-tolerance.test.ts

@@ -178,4 +178,16 @@ describe("AgentRouter Anthropic EOF tolerance (#658)", () => {
     const events = await createAnthropicAdapter(tolerant).parseResponse(new Response(payload));
     expect(events).toContainEqual({ type: "tool_call_delta", arguments: "{}" });
   });
+
+  test("the repair cap measures UTF-8 bytes, not UTF-16 code units", async () => {
+    // 600k 2-byte characters: 1.2 MB on the wire but only 600k code units, which a
+    // `string.length` check would wrongly admit into the parse attempts.
+    const oversized = `{${"é".repeat(600_000)}`;
+    const payload = JSON.stringify({
+      content: [{ type: "tool_use", id: "toolu_1", name: "get_weather", input: oversized }],
+    });
+
+    const events = await createAnthropicAdapter(tolerant).parseResponse(new Response(payload));
+    expect(events).toContainEqual({ type: "tool_call_delta", arguments: "{}" });
+  });
 });