fix: fold the wp6 reviewer round (bounded upstream JSON body, backward-scan repair, strict EOF predicates)

Review findings on the stacked contributor fixes, all red-green verified:

- core.ts upstream JSON branch read the whole body with an unbounded .text();
  every non-streaming upstream — now including WebSocket turns deliberately
  answered with bounded JSON — could grow proxy memory without limit. The read
  goes through relay.readBoundedResponseText (32 MiB ceiling, body cancelled on
  overflow) and fails closed with a 502 instead of parsing a partial body.
- anthropic lastValidJsonObject collected every brace offset into two arrays
  before its candidate cap, so brace-dense hostile input cost O(n) index storage.
  It now scans backwards from the end with lastIndexOf and never materializes an
  index; inputs above 1 MiB are not repaired at all.
- an empty text_delta marked sawVisibleText, letting a cut-off tolerant stream
  complete as a successful empty answer; only non-empty text authorizes tolerant
  completion now.
- a translator-budget overflow could be followed by tool_call_end/done when the
  generator was fully drained; the budget error now returns immediately as the
  single terminal event.

Author: lidge-jun

src/adapters/anthropic.ts

@@ -281,30 +281,36 @@ function usableToolUseId(id: unknown): string {
  * Bound repair for a malformed tool-arguments string under the compatibility profile (#658):
  * a gateway such as AgentRouter can concatenate JSON objects (`{}{"value":42}`). Find the
  * last parseable JSON object by scanning suffixes from each object-open brace and prefixes
- * ending at each object-close brace, bounded so hostile input cannot cost unbounded time.
+ * ending at each object-close brace. Both scans walk backwards from the end trying at most
+ * `maxCandidates` positions, so no offset index is ever materialized: a brace-dense hostile
+ * input costs at most 2 × maxCandidates bounded JSON.parse attempts and no extra storage.
+ * Inputs above MAX_REPAIRABLE_TOOL_ARGUMENT_BYTES are not repaired at all.
  */
+const MAX_REPAIRABLE_TOOL_ARGUMENT_BYTES = 1024 * 1024;
+
 function lastValidJsonObject(input: string, maxCandidates: number): string | undefined {
-  const opens: number[] = [];
-  const closes: number[] = [];
-  for (let i = 0; i < input.length; i++) {
-    if (input[i] === "{") opens.push(i);
-    else if (input[i] === "}") closes.push(i);
-  }
-  let tried = 0;
-  for (let i = opens.length - 1; i >= 0 && tried < maxCandidates; i--, tried++) {
-    const candidate = input.slice(opens[i]);
+  const tryParseObject = (candidate: string): string | undefined => {
     try {
       const parsed = JSON.parse(candidate) as unknown;
       if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) return candidate;
     } catch { /* keep scanning */ }
+    return undefined;
+  };
+  let scanFrom = input.length - 1;
+  for (let tried = 0; tried < maxCandidates && scanFrom >= 0; tried++) {
+    const open = input.lastIndexOf("{", scanFrom);
+    if (open === -1) break;
+    const repaired = tryParseObject(input.slice(open));
+    if (repaired !== undefined) return repaired;
+    scanFrom = open - 1;
   }
-  tried = 0;
-  for (let i = closes.length - 1; i >= 0 && tried < maxCandidates; i--, tried++) {
-    const candidate = input.slice(0, closes[i] + 1);
-    try {
-      const parsed = JSON.parse(candidate) as unknown;
-      if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) return candidate;
-    } catch { /* keep scanning */ }
+  scanFrom = input.length - 1;
+  for (let tried = 0; tried < maxCandidates && scanFrom >= 0; tried++) {
+    const close = input.lastIndexOf("}", scanFrom);
+    if (close === -1) break;
+    const repaired = tryParseObject(input.slice(0, close + 1));
+    if (repaired !== undefined) return repaired;
+    scanFrom = close - 1;
   }
   return undefined;
 }
@@ -317,7 +323,7 @@ function toolUseArguments(input: unknown, lenient = false): string {
       JSON.parse(trimmed);
       return trimmed;
     } catch {
-      if (lenient) {
+      if (lenient && trimmed.length <= MAX_REPAIRABLE_TOOL_ARGUMENT_BYTES) {
         const repaired = lastValidJsonObject(trimmed, 32);
         if (repaired !== undefined) return repaired;
       }
@@ -890,7 +896,10 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
                 const delta = data.delta as Record<string, unknown> | undefined;
                 if (!delta) break;
                 if (delta.type === "text_delta" && typeof delta.text === "string") {
-                  sawVisibleText = true;
+                  // Only non-empty text proves the upstream produced usable output; an empty
+                  // delta followed by EOF must stay a truncation error even on the tolerant
+                  // profile, or a cut-off turn would surface as a successful empty answer.
+                  if (delta.text.length > 0) sawVisibleText = true;
                   yield { type: "text_delta", text: delta.text };
                 } else if (delta.type === "thinking_delta" && typeof delta.thinking === "string") {
                   yield { type: "thinking_delta", thinking: delta.thinking };
@@ -972,6 +981,10 @@ export function createAnthropicAdapter(provider: OcxProviderConfig, cacheRetenti
           code: "translation_buffer_limit",
           message: "upstream translation buffer exceeded the safe limit",
         };
+        // The budget error IS the terminal event for this stream. Falling through to the
+        // EOF handling below could append tool_call_end/done after it, violating the
+        // one-terminal-event contract for consumers that keep draining the generator.
+        return;
       } finally {
         if (currentToolCallId) budget.closeCall(currentToolCallId);
       }

src/server/relay.ts

@@ -20,6 +20,48 @@ 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,6 +139,7 @@ import {
   isNativePassthroughSseResponse,
   markEagerRelaySseResponse,
   markNativePassthroughSseResponse,
+  readBoundedResponseText,
   relaySseWithFailedTail,
   relayWithAbort,
   sanitizePassthroughHeaders,
@@ -1963,7 +1964,16 @@ async function handleResponsesInner(
       }));
     }
     if (headers.get("content-type")?.toLowerCase().includes("application/json")) {
-      const text = await upstreamResponse.text();
+      // Bounded whole-body read: a non-streaming upstream JSON body is fully materialized
+      // here (and again by the request-log finalizer and the WebSocket bridge's reframing),
+      // 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) {
+        return formatErrorResponse(502, "upstream_error", "upstream JSON response exceeded the safe body limit");
+      }
+      const text = bounded.text;
       inspectResponseLogJson(logCtx, text);
       if (rememberPassthroughResponse) {
         try {

tests/anthropic-eof-tolerance.test.ts

@@ -2,7 +2,7 @@ import { describe, expect, test } from "bun:test";
 import { createAnthropicAdapter as createAnthropicAdapterProduction } from "../src/adapters/anthropic";
 import { FREE_PROVIDER_DIRECTORY } from "../src/providers/free-directory";
 import type { AdapterEvent, OcxProviderConfig } from "../src/types";
-import { withTestTranslatorBudget } from "./helpers/translator-budget";
+import { createTestTranslatorBudget, withTestTranslatorBudget } from "./helpers/translator-budget";
 
 /**
  * #658: AgentRouter's Anthropic-compatible endpoint can close the stream before
@@ -119,4 +119,63 @@ describe("AgentRouter Anthropic EOF tolerance (#658)", () => {
     const row = FREE_PROVIDER_DIRECTORY.find(provider => provider.id === "agentrouter");
     expect(row?.anthropicEofTolerance).toBe(true);
   });
+
+  test("an empty text delta at EOF stays a truncation error even when tolerant", async () => {
+    // Review finding: `sawVisibleText` must require non-empty text, or a cut-off turn
+    // surfaces as a successful empty answer.
+    const events = await collect(tolerant, [
+      'event: message_start\ndata: {"type":"message_start","message":{"usage":{"input_tokens":2}}}',
+      'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}',
+      'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":""}}',
+    ]);
+
+    expect(events.at(-1)).toEqual({ type: "error", message: TRUNCATION });
+    expect(events.some(event => event.type === "done")).toBe(false);
+  });
+
+  test("a budget overflow is the single terminal event on the tolerant path", async () => {
+    // Review finding: after the translation_buffer_limit error the generator must return,
+    // not fall through to EOF tolerance and append tool_call_end/done behind the error.
+    const adapter = createAnthropicAdapter(tolerant);
+    const events: AdapterEvent[] = [];
+    for await (const event of adapter.parseStream(
+      sseResponse(toolEof('{"value":42}', "toolu_1").concat([
+        `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "input_json_delta", partial_json: ',"extra":true' } })}`,
+      ])),
+      createTestTranslatorBudget({ maxCallArgumentBytes: 16 }),
+    )) {
+      events.push(event);
+    }
+
+    const terminal = events.at(-1);
+    expect(terminal?.type).toBe("error");
+    expect((terminal as { code?: string }).code).toBe("translation_buffer_limit");
+    const firstError = events.findIndex(event => event.type === "error");
+    expect(events.slice(firstError + 1).some(event => event.type === "done" || event.type === "tool_call_end")).toBe(false);
+  });
+
+  test("repair never materializes a brace index over hostile input", async () => {
+    // Review finding: the repair scan must stay backward and candidate-bounded. 4 MiB of
+    // unmatched opens would have built two O(n) offset arrays in the original helper; the
+    // result must still be the plain fallback.
+    const hostile = "{".repeat(4 * 1024 * 1024);
+    const payload = JSON.stringify({
+      content: [{ type: "tool_use", id: "toolu_1", name: "get_weather", input: hostile }],
+    });
+
+    const started = Date.now();
+    const events = await createAnthropicAdapter(tolerant).parseResponse(new Response(payload));
+    expect(Date.now() - started).toBeLessThan(10_000);
+    expect(events).toContainEqual({ type: "tool_call_delta", arguments: "{}" });
+  });
+
+  test("repair declines input above the byte cap", async () => {
+    const oversized = `{"pad":"${"x".repeat(1024 * 1024 + 8)}`; // > 1 MiB, unparseable
+    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: "{}" });
+  });
 });

tests/deepseek-inbound-wire.test.ts

@@ -135,6 +135,32 @@ describe("the inbound scope survives the handleResponses replay", () => {
   test("ordinary HTTP Responses requests keep streaming upstream", async () => {
     expect((await drive("responses")).body.stream).toBe(true);
   });
+
+  test("an oversized upstream JSON body fails closed instead of buffering without limit", async () => {
+    // Review finding: the WebSocket bounded-JSON path (and every non-streaming upstream)
+    // materializes the whole body, so the read must have a hard byte ceiling. 33 MiB is
+    // one MiB over MAX_UPSTREAM_JSON_BODY_BYTES.
+    globalThis.fetch = (async () => new Response(" ".repeat(33 * 1024 * 1024), {
+      status: 200,
+      headers: { "content-type": "application/json" },
+    })) as typeof fetch;
+    const config = { providers: { deepseek: deepseekProvider() } } as unknown as OcxConfig;
+    const response = await handleResponses(
+      new Request("http://localhost/v1/responses", {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ model: MODEL, input: "ping", stream: true }),
+      }),
+      config,
+      { model: "", provider: "" },
+      { inboundWire: "responses", inboundTransport: "websocket" },
+    );
+
+    expect(response.status).toBe(502);
+    const payload = (await response.json()) as { error?: { code?: string; message?: string } };
+    expect(payload.error?.code).toBe("upstream_server_error");
+    expect(payload.error?.message).toContain("exceeded the safe body limit");
+  });
 });
 
 /**