fix(logs): bound response inspection without losing SSE finality

docs-site/astro.config.mjs

@@ -86,6 +86,7 @@ export default defineConfig({
           translations: { fr: "Guides", ko: "가이드", "zh-CN": "指南", "zh-TW": "指南", ru: "Руководства", ja: "ガイド", tr: "Kılavuzlar" },
           items: [
             { label: "Remote Hub Deployment", translations: { fr: "Déploiement Remote Hub", ko: "Remote Hub 배포", "zh-CN": "Remote Hub 部署", "zh-TW": "Remote Hub 部署", ru: "Развёртывание Remote Hub", ja: "Remote Hub のデプロイ", tr: "Remote Hub Dağıtımı" }, slug: "guides/remote-hub" },
+            { label: "Response Inspection", slug: "guides/response-inspection" },
             { label: "Remote Workspace", translations: { fr: "Espace de travail distant", ko: "원격 워크스페이스", "zh-CN": "远程工作区", "zh-TW": "遠端工作區", ru: "Удалённая рабочая область", ja: "リモートワークスペース", tr: "Uzak Çalışma Alanı" }, slug: "guides/remote-workspace" },
             { label: "Providers", translations: { fr: "Fournisseurs", ko: "프로바이더", "zh-CN": "提供商", "zh-TW": "供應商", ru: "Провайдеры", ja: "プロバイダー", tr: "Sağlayıcılar" }, slug: "guides/providers" },
             { label: "Factory Droid Bridge", translations: { fr: "Pont Factory Droid", ko: "Factory Droid 브리지" }, slug: "guides/factory-droid" },

docs-site/src/content/docs/guides/response-inspection.md

@@ -0,0 +1,36 @@
+---
+title: Response inspection and large responses
+description: How bounded diagnostic retention and streaming inspection interact with response delivery.
+---
+
+OpenCodex keeps response diagnostics bounded without making the logging limit a
+limit on the bytes delivered to your client. Other provider, request and transport
+limits still apply independently.
+
+## JSON and ordinary error responses
+
+JSON inspection retains at most 32 MiB of source bytes. If the body exceeds that
+allowance, logging drops its retained copy and continues forwarding the original
+response. It does not parse a truncated prefix as authoritative usage or model
+metadata. Usage already supplied by another trusted path is preserved; missing
+usage is not replaced with an invented zero. Ordinary non-JSON error diagnostics
+retain only the first 8 KiB and pass through the existing redaction logic.
+
+The client receives chunks as it reads them rather than waiting for diagnostic
+inspection of the whole body. A read failure is recorded as 502 and cancellation
+as 499 in request history; these diagnostic outcomes do not rewrite HTTP headers
+that have already been sent. Logging is finalized once.
+
+## Streaming responses
+
+Native SSE inspection pauses when it runs too far ahead of client consumption.
+The allowance is 32 MiB plus source-chunk/native-prefetch overhead, not a total
+response-size limit or a cap on all process memory. A longer response is still
+inspected through its actual completion event, including terminal usage and
+continuation state.
+
+After the client disconnects, the existing bounded drain can still observe a late
+completion for up to 15 seconds or 32 MiB of additional inspection. A forced
+shutdown is different: it discards uncompleted candidates rather than recording
+them as a completed response. Existing transport selection and WebSocket memory
+bounds are unchanged. No new configuration setting is required.

scripts/test-layout/layout.json

@@ -1466,7 +1466,9 @@
     "execution-budget-permits.test.ts": "lib",
     "spend-instrumentation-log.test.ts": "server",
     "codex-pool-refresh-backoff.test.ts": "codex-integration",
-    "responses-account-change-scrub.test.ts": "responses"
+    "responses-account-change-scrub.test.ts": "responses",
+    "response-log-inspection.test.ts": "server",
+    "request-log-nonstream.test.ts": "usage"
   },
   "migrated": [
     "adapters",

src/server/inspection-tee.ts

@@ -0,0 +1,107 @@
+/** Read-ahead allowance, not a lifetime limit on SSE inspection or delivery. */
+export const MAX_INSPECTION_READ_AHEAD_BYTES = 32 * 1024 * 1024;
+
+export interface InspectionTeeOptions {
+  /** Release pacing so the inspection owner's existing bounded drain can run. */
+  clientGoneSignal?: AbortSignal;
+  /** Internal test seam; callers must not take this value from upstream data. */
+  maxReadAheadBytes?: number;
+}
+
+/**
+ * Keep native tee cancellation semantics while pacing the inspection branch.
+ *
+ * Inspection may lead raw client consumption by the allowance plus one source
+ * chunk (and native tee prefetch). It never stops merely because the whole turn
+ * crossed that size: terminal, usage and continuation observers retain ownership.
+ * Count raw client bytes BEFORE rewrites, which may shrink, drop or expand them.
+ * Push sources still need their own producer-side bound; this is not an RSS cap.
+ */
+export function teeWithBoundedInspection(
+  source: ReadableStream<Uint8Array>,
+  options: InspectionTeeOptions = {},
+): [ReadableStream<Uint8Array>, ReadableStream<Uint8Array>] {
+  const limit = options.maxReadAheadBytes ?? MAX_INSPECTION_READ_AHEAD_BYTES;
+  if (!Number.isSafeInteger(limit) || limit <= 0) {
+    throw new RangeError("Inspection read-ahead limit must be a positive safe integer");
+  }
+  const [client, inspection] = source.tee();
+  let leadBytes = 0;
+  let pacingReleased = false;
+  let credit: Promise<void> | undefined;
+  let wake: (() => void) | undefined;
+
+  const wakeReader = () => {
+    const resolve = wake;
+    wake = undefined;
+    credit = undefined;
+    resolve?.();
+  };
+  const releasePacing = () => {
+    pacingReleased = true;
+    wakeReader();
+  };
+  const signal = options.clientGoneSignal;
+  if (signal?.aborted) releasePacing();
+  else signal?.addEventListener("abort", releasePacing, { once: true });
+
+  const wrap = (
+    body: ReadableStream<Uint8Array>,
+    isClient: boolean,
+  ): ReadableStream<Uint8Array> => {
+    const reader = body.getReader();
+    let ended = false;
+    // An upstream error must wake an inspector waiting for credit, not remain
+    // hidden until the client happens to issue another read. One observer per
+    // reader, not a permanent Promise.race reaction attached on every chunk.
+    void reader.closed.catch(releasePacing);
+    const releaseLock = () => {
+      try { reader.releaseLock(); } catch { /* a pending read owns the lock */ }
+    };
+    const finish = () => {
+      ended = true;
+      releasePacing();
+      if (!isClient) signal?.removeEventListener("abort", releasePacing);
+    };
+    return new ReadableStream<Uint8Array>({
+      async pull(controller) {
+        if (ended) return;
+        if (!isClient && !pacingReleased && leadBytes >= limit) {
+          credit ??= new Promise<void>(resolve => { wake = resolve; });
+          await credit;
+          if (ended) return;
+        }
+        try {
+          const result = await reader.read();
+          if (ended) return;
+          if (result.done) {
+            finish();
+            releaseLock();
+            controller.close();
+            return;
+          }
+          if (isClient) {
+            leadBytes -= result.value.byteLength;
+            if (leadBytes < limit) wakeReader();
+          } else {
+            leadBytes += result.value.byteLength;
+          }
+          controller.enqueue(result.value);
+        } catch (error) {
+          if (ended) return;
+          finish();
+          releaseLock();
+          controller.error(error);
+        }
+      },
+      cancel(reason) {
+        finish();
+        // Awaiting one tee branch's cancellation waits for the sibling. The
+        // owner must be free to finish cleanup and cancel/read that sibling.
+        void reader.cancel(reason).catch(() => undefined);
+        releaseLock();
+      },
+    }, { highWaterMark: 0 });
+  };
+  return [wrap(client, true), wrap(inspection, false)];
+}

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 { createBoundedResponseLogBody } from "./response-log-body";
 
 const nativePassthroughSseResponses = new WeakSet<Response>();
 const eagerRelaySseResponses = new WeakSet<Response>();
@@ -718,25 +719,16 @@ 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) {
-          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 */ }
-          }
+      const body = createBoundedResponseLogBody(response.body, {
+        json: contentType.includes("application/json"),
+        inspect: text => inspectResponseLogJson(logCtx, text),
+        finalize: reason => {
+          // Preserve wire status; request history follows the adjacent SSE
+          // convention for a client cancellation or upstream read failure.
+          const status = reason === "cancel" ? 499 : reason === "read_error" ? 502 : response.status;
+          addFinalRequestLog(requestId, start, logCtx, status, {
+            closeReason: reason === "cancel" ? "client_cancel" : "non_stream",
+          }, addLog);
         },
       });
       return new Response(body, {
@@ -1338,6 +1330,9 @@ function startBoundedInspectionPump(options: InspectionPumpOptions): void {
     try {
       for (;;) {
         const { done, value } = await reader.read();
+        // Hard cancellation settles a pending read as EOF. Do not flush a
+        // partial terminal after the owner already finalized cancellation.
+        if (cancelled) break;
         if (clientGoneSignal?.aborted) markClientGone();
         if (drainStopped) {
           // stopDrain() cancelled the reader; the settled read is the wake-up.

src/server/response-log-body.ts

@@ -0,0 +1,153 @@
+/** Bounds diagnostic retention, never the bytes delivered to the caller. */
+export const MAX_RESPONSE_LOG_INSPECTION_BYTES = 32 * 1024 * 1024;
+export const MAX_NON_JSON_ERROR_INSPECTION_BYTES = 8 * 1024;
+const INSPECTION_BLOCK_BYTES = 64 * 1024;
+
+export type ResponseLogBodyEnd = "eof" | "read_error" | "cancel";
+
+export interface ResponseLogBodyOptions {
+  json: boolean;
+  inspect: (text: string) => void;
+  finalize: (reason: ResponseLogBodyEnd) => void;
+  /** Internal test seam; this is not a provider-controlled limit. */
+  maxInspectionBytes?: number;
+}
+
+/** Fixed-size blocks bound both retained bytes and per-chunk bookkeeping. */
+class ResponseLogInspection {
+  private blocks: Uint8Array[] = [];
+  private bytes = 0;
+  private overflowed = false;
+
+  constructor(private readonly json: boolean, private readonly limit: number) {}
+
+  append(chunk: Uint8Array): void {
+    if (this.overflowed || chunk.byteLength === 0) return;
+    if (this.json && chunk.byteLength > this.limit - this.bytes) {
+      // A JSON prefix is not an authoritative response. Drop it immediately,
+      // without either truncating the delivery stream or retaining later bytes.
+      this.overflowed = true;
+      this.dispose();
+      return;
+    }
+    let remaining = Math.min(chunk.byteLength, this.limit - this.bytes);
+    let offset = 0;
+    while (remaining > 0) {
+      const blockOffset = this.bytes % INSPECTION_BLOCK_BYTES;
+      if (blockOffset === 0) {
+        this.blocks.push(new Uint8Array(Math.min(INSPECTION_BLOCK_BYTES, this.limit - this.bytes)));
+      }
+      const block = this.blocks[this.blocks.length - 1]!;
+      const length = Math.min(remaining, block.byteLength - blockOffset);
+      block.set(chunk.subarray(offset, offset + length), blockOffset);
+      this.bytes += length;
+      offset += length;
+      remaining -= length;
+    }
+  }
+
+  text(reason: ResponseLogBodyEnd): string | undefined {
+    // Even a syntactically valid JSON prefix must not update usage/model
+    // metadata when the transport did not reach EOF.
+    if (this.json && (reason !== "eof" || this.overflowed)) return undefined;
+    const combined = new Uint8Array(this.bytes);
+    let offset = 0;
+    for (const block of this.blocks) {
+      const length = Math.min(block.byteLength, this.bytes - offset);
+      combined.set(block.subarray(0, length), offset);
+      offset += length;
+    }
+    return new TextDecoder().decode(combined);
+  }
+
+  dispose(): void {
+    this.blocks.length = 0;
+    this.bytes = 0;
+  }
+}
+
+/** Optional diagnostics must not change the response's transport outcome. */
+function bestEffort(callback: () => void): void {
+  try {
+    callback();
+  } catch {
+    return;
+  }
+}
+
+/**
+ * Forward one upstream read per downstream pull, with bounded side inspection.
+ * EOF, read failure and cancellation each finalize at most once. Cancellation
+ * does not await the upstream cancel promise: one branch of a tee can otherwise
+ * wait for a sibling that the same caller intends to consume or cancel later.
+ */
+export function createBoundedResponseLogBody(
+  body: ReadableStream<Uint8Array>,
+  options: ResponseLogBodyOptions,
+): ReadableStream<Uint8Array> {
+  const limit = options.maxInspectionBytes ?? (options.json
+    ? 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 inspection = new ResponseLogInspection(options.json, limit);
+  const reader = body.getReader();
+  let ended = false;
+  let inspectionFailed = false;
+
+  const release = () => bestEffort(() => reader.releaseLock());
+  const finish = (reason: ResponseLogBodyEnd) => {
+    if (ended) return;
+    ended = true; // Set before callbacks or a pending read resumes.
+    try {
+      if (!inspectionFailed) {
+        bestEffort(() => {
+          const text = inspection.text(reason);
+          if (text !== undefined) options.inspect(text);
+        });
+      }
+    } finally {
+      inspection.dispose();
+      bestEffort(() => options.finalize(reason));
+    }
+  };
+
+  return new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      if (ended) return;
+      let result: Awaited<ReturnType<typeof reader.read>>;
+      try {
+        result = await reader.read();
+      } catch (error) {
+        if (ended) return; // Cancellation owns its pending-read settlement.
+        finish("read_error");
+        release();
+        controller.error(error);
+        return;
+      }
+      if (ended) return;
+      if (result.done) {
+        finish("eof");
+        release();
+        controller.close();
+        return;
+      }
+      if (!inspectionFailed) {
+        try {
+          inspection.append(result.value);
+        } catch {
+          inspectionFailed = true;
+          inspection.dispose();
+        }
+      }
+      // Do not decode/re-encode transport bytes, including malformed UTF-8.
+      controller.enqueue(result.value);
+    },
+    cancel(reason) {
+      finish("cancel");
+      bestEffort(() => { void reader.cancel(reason).catch(() => undefined); });
+      release();
+    },
+  }, { highWaterMark: 0 });
+}