fix(logs): bound JSON and error response inspection

src/server/relay.ts

@@ -26,6 +26,7 @@ import {
   MAX_CLIENT_SSE_FRAME_BYTES,
 } from "./sse-frame-buffer";
 import { replaceSseDataPayload } from "./sse-payload-rewrite";
+import { relayResponseLogBody } from "./response-log-body";
 
 const nativePassthroughSseResponses = new WeakSet<Response>();
 const eagerRelaySseResponses = new WeakSet<Response>();
@@ -718,24 +719,21 @@ export function responseWithDeferredRequestLog(
   }
   if (!response.body || !contentType.includes("text/event-stream")) {
     if (response.body && (contentType.includes("application/json") || response.status >= 400)) {
-      const finalizeJsonLog = async () => {
-        const text = await response.text();
-        // Non-JSON error bodies: inspect/log only a bounded prefix (the stored
-        // upstreamError is 500 chars anyway); the FULL text is still forwarded to the
-        // client below, unchanged. JSON bodies keep full inspection (usage parsing).
-        const isJson = contentType.includes("application/json");
-        inspectResponseLogJson(logCtx, isJson ? text : text.slice(0, 8192));
-        addFinalRequestLog(requestId, start, logCtx, response.status, { closeReason: "non_stream" }, addLog);
-        return text;
-      };
-      const body = new ReadableStream<Uint8Array>({
-        async start(controller) {
+      const body = relayResponseLogBody(response.body, {
+        isJson: contentType.includes("application/json"),
+        onFinalize(outcome, text) {
           try {
-            controller.enqueue(new TextEncoder().encode(await finalizeJsonLog()));
-            controller.close();
-          } catch (err) {
-            addFinalRequestLog(requestId, start, logCtx, 502, { closeReason: "non_stream" }, addLog);
-            try { controller.error(err); } catch { /* already torn down */ }
+            // Only complete, within-budget JSON reaches the existing usage parser.
+            // Error/cancel prefixes must not masquerade as reported JSON usage.
+            if (text !== undefined) inspectResponseLogJson(logCtx, text);
+          } finally {
+            if (outcome === "error" && logCtx.activeAttempt) {
+              logCtx.activeAttempt.streamAborted = true;
+            }
+            const status = outcome === "cancel" ? 499 : outcome === "error" ? 502 : response.status;
+            addFinalRequestLog(requestId, start, logCtx, status, {
+              closeReason: outcome === "cancel" ? "client_cancel" : "non_stream",
+            }, addLog);
           }
         },
       });

src/server/response-log-body.ts

@@ -0,0 +1,134 @@
+export const MAX_RESPONSE_LOG_INSPECTION_BYTES = 32 * 1024 * 1024;
+export const MAX_NON_JSON_ERROR_INSPECTION_BYTES = 8 * 1024;
+
+export type ResponseLogBodyOutcome = "eof" | "error" | "cancel";
+
+export type ResponseLogBodyOptions = {
+  isJson: boolean;
+  onFinalize(outcome: ResponseLogBodyOutcome, text: string | undefined): void;
+  /** Test seam; production uses the JSON / non-JSON byte ceilings above. */
+  inspectionLimit?: number;
+};
+
+/**
+ * Forward original bytes on demand, retaining only a bounded log-inspection copy.
+ * JSON is inspected only after clean EOF and only when the complete body fits.
+ * Other error bodies keep a diagnostic prefix, including on error/cancellation.
+ * This is not an SSE observer and never owns a second, eagerly drained tee branch.
+ */
+export function relayResponseLogBody(
+  source: ReadableStream<Uint8Array>,
+  options: ResponseLogBodyOptions,
+): ReadableStream<Uint8Array> {
+  const limit = options.inspectionLimit ?? (options.isJson
+    ? MAX_RESPONSE_LOG_INSPECTION_BYTES
+    : MAX_NON_JSON_ERROR_INSPECTION_BYTES);
+  if (!Number.isSafeInteger(limit) || limit < 0) {
+    throw new RangeError("Response log inspection limit must be a non-negative safe integer");
+  }
+  const reader = source.getReader();
+  let retained: Uint8Array = new Uint8Array(0);
+  let retainedBytes = 0;
+  let inspectionUnavailable = false;
+  let finalized = false;
+  let released = false;
+
+  const discard = () => {
+    retained = new Uint8Array(0);
+    retainedBytes = 0;
+  };
+  const release = () => {
+    if (released) return;
+    try {
+      reader.releaseLock();
+      released = true;
+    } catch {
+      // Some runtimes defer settling a cancelled read; its continuation retries.
+      return;
+    }
+  };
+  const retain = (chunk: Uint8Array) => {
+    if (inspectionUnavailable || chunk.byteLength === 0) return;
+    const remaining = limit - retainedBytes;
+    if (options.isJson && chunk.byteLength > remaining) {
+      inspectionUnavailable = true;
+      discard();
+      return;
+    }
+    const count = Math.min(chunk.byteLength, remaining);
+    if (count === 0) return;
+    const required = retainedBytes + count;
+    if (required > retained.byteLength) {
+      // One geometrically grown allocation bounds both bytes and object count;
+      // retaining a separate slice per tiny chunk would still grow metadata.
+      const capacity = Math.min(limit, Math.max(required, retained.byteLength * 2, 4096));
+      const grown = new Uint8Array(capacity);
+      grown.set(retained.subarray(0, retainedBytes));
+      retained = grown;
+    }
+    retained.set(chunk.subarray(0, count), retainedBytes);
+    retainedBytes = required;
+  };
+  const finalize = (outcome: ResponseLogBodyOutcome) => {
+    if (finalized) return;
+    finalized = true;
+    const inspect = !inspectionUnavailable && (!options.isJson || outcome === "eof");
+    const bytes = retained.subarray(0, retainedBytes);
+    discard();
+    let text: string | undefined;
+    try {
+      if (inspect) text = new TextDecoder().decode(bytes);
+    } catch {
+      // A diagnostic decoding failure must still finalize the request log.
+      text = undefined;
+    }
+    try {
+      options.onFinalize(outcome, text);
+    } catch {
+      // Optional logging must neither corrupt delivery nor prevent cancellation.
+      return;
+    }
+  };
+
+  return new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      try {
+        const { done, value } = await reader.read();
+        if (finalized) {
+          release();
+          return;
+        }
+        if (done) {
+          finalize("eof");
+          release();
+          controller.close();
+          return;
+        }
+        try {
+          retain(value);
+        } catch {
+          // Inspection allocation is not a reason to drop client bytes.
+          inspectionUnavailable = true;
+          discard();
+        }
+        controller.enqueue(value);
+      } catch (error) {
+        if (finalized) {
+          release();
+          return;
+        }
+        finalize("error");
+        release();
+        controller.error(error);
+      }
+    },
+    cancel(reason) {
+      if (finalized) return;
+      finalize("cancel");
+      // Never await tee cancellation: it can wait for the other branch's EOF.
+      // Cancelling settles our pending read; no retained inspection bytes remain.
+      void reader.cancel(reason).catch(() => undefined);
+      release();
+    },
+  }, { highWaterMark: 0 });
+}

structure/transports/byte-accounting.md

@@ -39,3 +39,26 @@ These optimizations do not add request queues, retry policies, or RSS-based admi
 
 Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate.
 Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged.
+
+## Non-stream response-log inspection
+
+`src/server/relay.ts` delegates JSON and non-JSON error-body delivery to
+`src/server/response-log-body.ts`. A single pull-driven stream forwards the original bytes;
+logging neither waits for the whole body before delivery nor eagerly drains a second tee branch.
+The inspection copy retains at most 32 MiB for JSON, or an 8 KiB byte prefix for other errors.
+One geometrically grown allocation also bounds retained chunk metadata. These are per-response
+inspection-copy limits, not a process-memory ceiling or an output-size limit; decoding and parsing
+can allocate additional bounded objects.
+
+JSON is inspected only after clean EOF when the entire body fits. Crossing its budget discards
+its inspection copy and skips parsing without truncating delivery, changing the HTTP status,
+or inventing usage. Error and cancellation paths never parse a partial JSON document. Other
+error bodies retain their bounded diagnostic prefix. The existing request-log parser and final
+log writer remain responsible for redaction, usage provenance, and attempt accounting.
+
+EOF logs the original status, a failed body read logs 502, and downstream cancellation logs 499
+with `client_cancel`, exactly once. The same cancellation reason reaches the source reader;
+reader cancellation is not awaited because a tee sibling can remain open. This limit does not
+apply to SSE turn length: existing frame/output-item budgets and post-disconnect drain ownership
+remain unchanged. `tests/server/consume-for-inspection-cancel.test.ts` registers the shared
+body-lifecycle cases and covers the integration and a late SSE terminal beyond the JSON budget.

tests/helpers/response-log-body-cases.ts

@@ -0,0 +1,146 @@
+import assert from 'node:assert/strict';
+import { relayResponseLogBody, MAX_RESPONSE_LOG_INSPECTION_BYTES, MAX_NON_JSON_ERROR_INSPECTION_BYTES } from '../../src/server/response-log-body';
+const enc = new TextEncoder();
+const tick = () => new Promise<void>(r => setTimeout(r, 0));
+function controlled() {
+  let controller!: ReadableStreamDefaultController<Uint8Array>;
+  const reasons: unknown[] = [];
+  const stream = new ReadableStream<Uint8Array>({ start(c) { controller = c; }, cancel(r) { reasons.push(r); } }, { highWaterMark: 0 });
+  return { stream, reasons, push: (v: string | Uint8Array) => controller.enqueue(typeof v === 'string' ? enc.encode(v) : v), close: () => controller.close(), error: (e: unknown) => controller.error(e) };
+}
+function wrapped(source: ReadableStream<Uint8Array>, isJson: boolean, inspectionLimit?: number) {
+  const calls: [string, string | undefined][] = [];
+  return { body: relayResponseLogBody(source, { isJson, inspectionLimit, onFinalize: (...v) => calls.push(v) }), calls };
+}
+export function registerResponseLogBodyCases(test: (name: string, run: () => Promise<void>) => unknown): void {
+  for (const isJson of [false, true]) {
+    test(`first chunk before EOF; no early log (${isJson})`, async () => {
+      const s = controlled(), w = wrapped(s.stream, isJson);
+      const reader = w.body.getReader(), first = reader.read();
+      s.push('{}');
+      assert.deepEqual(await first, { value: enc.encode('{}'), done: false });
+      assert.equal(w.calls.length, 0);
+      s.close();
+      assert.equal((await reader.read()).done, true);
+      assert.deepEqual(w.calls, [['eof', '{}']]);
+      assert.equal(s.stream.locked, false);
+    });
+    test(`slow readers do not cause prefetch (${isJson})`, async () => {
+      let pulls = 0;
+      const source = new ReadableStream({ pull(c) { ++pulls; c.enqueue(enc.encode('x')); } }, { highWaterMark: 0 });
+      const w = wrapped(source, isJson), r = w.body.getReader();
+      await tick(); assert.equal(pulls, 0);
+      await r.read(); await tick(); assert.equal(pulls, 1);
+      await r.cancel('stop'); assert.equal(w.calls.length, 1);
+    });
+    test(`pending cancellation finalizes once and propagates reason (${isJson})`, async () => {
+      const s = controlled(), w = wrapped(s.stream, isJson), r = w.body.getReader();
+      const pending = r.read(); await tick();
+      const reason = new Error('client stop');
+      await r.cancel(reason); await pending; await tick();
+      assert.deepEqual(s.reasons, [reason]);
+      assert.deepEqual(w.calls, [['cancel', isJson ? undefined : '']]);
+      assert.equal(s.stream.locked, false);
+    });
+    test(`source error preserves exception and prefix rules (${isJson})`, async () => {
+      const s = controlled(), w = wrapped(s.stream, isJson), r = w.body.getReader();
+      const first = r.read(); s.push('{}'); await first;
+      const failure = new Error('upstream reset');
+      const pending = r.read(); s.error(failure);
+      await assert.rejects(pending, (e: unknown) => e === failure);
+      assert.deepEqual(w.calls, [['error', isJson ? undefined : '{}']]);
+      assert.equal(s.stream.locked, false);
+    });
+  }
+  test('binary and split UTF-8 forwarding is byte exact', async () => {
+    const chunks = [new Uint8Array([0xff, 0, 0xe3]), new Uint8Array([0x81, 0x82, 0xfe])];
+    const source = new ReadableStream<Uint8Array>({ start(c) { chunks.forEach(v => c.enqueue(v)); c.close(); } });
+    const w = wrapped(source, false);
+    assert.deepEqual(new Uint8Array(await new Response(w.body).arrayBuffer()), new Uint8Array(chunks.flatMap(v => [...v])));
+    assert.equal(w.calls.length, 1);
+  });
+  for (const [size, chunkSize, inspect] of [[8, 8, true], [9, 9, false], [9, 3, false], [8, 1, true]] as const) {
+    test(`JSON cap: ${size} bytes in ${chunkSize}-byte chunks`, async () => {
+      let sent = 0;
+      const source = new ReadableStream({ pull(c) { if (sent === size) return c.close(); const n = Math.min(chunkSize, size-sent); sent += n; c.enqueue(enc.encode('x'.repeat(n))); } });
+      const w = wrapped(source, true, 8);
+      assert.equal(await new Response(w.body).text(), 'x'.repeat(size));
+      assert.deepEqual(w.calls, [['eof', inspect ? 'x'.repeat(size) : undefined]]);
+    });
+  }
+  test('actual oversized JSON budget still forwards every byte', async () => {
+    const size = MAX_RESPONSE_LOG_INSPECTION_BYTES + 1;
+    let sent = 0;
+    const block = new Uint8Array(65536).fill(120);
+    const source = new ReadableStream({ pull(c) { if (sent === size) return c.close(); const n = Math.min(block.length, size-sent); sent += n; c.enqueue(block.subarray(0,n)); } });
+    const w = wrapped(source, true), r = w.body.getReader();
+    let got = 0;
+    for (;;) { const { done, value } = await r.read(); if (done) break; got += value.length; assert.equal(value[0], 120); }
+    assert.equal(got, size);
+    assert.deepEqual(w.calls, [['eof', undefined]]);
+  });
+  test('non-JSON prefix is byte bounded, not character bounded', async () => {
+    const text = '한'.repeat(10000);
+    const w = wrapped(new Response(text).body!, false);
+    assert.equal(await new Response(w.body).text(), text);
+    assert.deepEqual(w.calls, [['eof', new TextDecoder().decode(enc.encode(text).subarray(0, MAX_NON_JSON_ERROR_INSPECTION_BYTES))]]);
+  });
+  test('cancel before any read never pulls', async () => {
+    let pulls = 0; let reason: unknown;
+    const s = new ReadableStream({ pull() { ++pulls; }, cancel(r) { reason = r; } }, { highWaterMark: 0 });
+    const w = wrapped(s, true);
+    await w.body.cancel('before');
+    assert.equal(pulls, 0); assert.equal(reason, 'before'); assert.equal(s.locked, false);
+    assert.deepEqual(w.calls, [['cancel', undefined]]);
+  });
+  test('valid JSON prefix on cancellation is not treated as a complete response', async () => {
+    const s = controlled(), w = wrapped(s.stream, true), r = w.body.getReader();
+    const pending = r.read(); s.push('{"usage":{"total_tokens":900}}'); await pending;
+    await r.cancel(); assert.deepEqual(w.calls, [['cancel', undefined]]);
+  });
+  for (const outcome of ['eof', 'cancel', 'error']) {
+    test(`throwing logger does not disrupt ${outcome}`, async () => {
+      const s = controlled(); let calls = 0;
+      const body = relayResponseLogBody(s.stream, { isJson: false, onFinalize() { ++calls; throw new Error('logging failed'); } });
+      const r = body.getReader(); const pending = r.read(); s.push('ok'); await pending;
+      if (outcome === 'eof') { s.close(); assert.equal((await r.read()).done, true); }
+      if (outcome === 'cancel') { await r.cancel('gone'); assert.deepEqual(s.reasons, ['gone']); }
+      if (outcome === 'error') { const e = new Error('reset'); s.error(e); await assert.rejects(r.read(), (v: unknown) => v === e); }
+      assert.equal(calls, 1); assert.equal(s.stream.locked, false);
+    });
+  }
+  test('cancel rejection does not stall teardown', async () => {
+    const source = new ReadableStream({ cancel() { return Promise.reject(new Error('cancel failed')); } });
+    const w = wrapped(source, false); await w.body.cancel(); await tick();
+    assert.equal(w.calls.length, 1); assert.equal(source.locked, false);
+  });
+  test('tee cancellation does not wait for sibling; sibling remains intact', async () => {
+    const s = controlled(), [left, right] = s.stream.tee();
+    const w = wrapped(left, false), reader = w.body.getReader(), sibling = right.getReader();
+    const a = reader.read(), b = sibling.read(); s.push('first'); await a; await b;
+    await reader.cancel('left stopped');
+    assert.equal(w.calls.length, 1); assert.equal(s.reasons.length, 0);
+    const next = sibling.read(); s.push('second'); assert.equal(new TextDecoder().decode((await next).value), 'second');
+    s.close(); assert.equal((await sibling.read()).done, true);
+  });
+  test('cancel wins a racing source error without duplicate finalization', async () => {
+    const s = controlled(), w = wrapped(s.stream, true), r = w.body.getReader();
+    const pending = r.read(); await tick();
+    s.error(new Error('reset'));
+    await r.cancel('stop'); await pending; await tick();
+    assert.deepEqual(w.calls, [['cancel', undefined]]);
+  });
+  test('retained inspection is a copy, not a view of forwarded storage', async () => {
+    const s = controlled(), w = wrapped(s.stream, false), r = w.body.getReader();
+    const buffer = enc.encode('original'); const read = r.read(); s.push(buffer); await read; buffer.fill(120);
+    s.close(); await r.read(); assert.deepEqual(w.calls, [['eof', 'original']]);
+  });
+  test('empty JSON at zero cap and invalid limits', async () => {
+    const w = wrapped(new Response('').body!, true, 0); await new Response(w.body).text(); assert.deepEqual(w.calls, [['eof', '']]);
+    for (const limit of [-1, NaN, Infinity, 1.5]) {
+      const source = controlled().stream;
+      assert.throws(() => wrapped(source, true, limit), RangeError); assert.equal(source.locked, false);
+    }
+  });
+
+}