fix(relay): stream non-JSON upstream error bodies safely

src/server/relay.ts

@@ -20,6 +20,7 @@ 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;
+const MAX_NON_JSON_ERROR_INSPECTION_BYTES = 8192;
 
 export type InspectionCounters = {
   frameBufferHighWaterBytes: number;
@@ -427,14 +428,10 @@ export function responseWithDeferredRequestLog(
     return response;
   }
   if (!response.body || !contentType.includes("text/event-stream")) {
-    if (response.body && (contentType.includes("application/json") || response.status >= 400)) {
+    if (response.body && contentType.includes("application/json")) {
       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));
+        inspectResponseLogJson(logCtx, text);
         addFinalRequestLog(requestId, start, logCtx, response.status, { closeReason: "non_stream" }, addLog);
         return text;
       };
@@ -455,6 +452,55 @@ export function responseWithDeferredRequestLog(
         headers: response.headers,
       });
     }
+    if (response.body && response.status >= 400) {
+      const reader = response.body.getReader();
+      const inspected: Uint8Array[] = [];
+      let inspectedBytes = 0;
+      let logged = false;
+      const finalize = (status = response.status) => {
+        if (logged) return;
+        logged = true;
+        const prefix = new Uint8Array(inspectedBytes);
+        let offset = 0;
+        for (const chunk of inspected) {
+          prefix.set(chunk, offset);
+          offset += chunk.byteLength;
+        }
+        inspectResponseLogJson(logCtx, new TextDecoder().decode(prefix));
+        addFinalRequestLog(requestId, start, logCtx, status, { closeReason: "non_stream" }, addLog);
+      };
+      const body = new ReadableStream<Uint8Array>({
+        async pull(controller) {
+          try {
+            const { done, value } = await reader.read();
+            if (done) {
+              finalize();
+              controller.close();
+              return;
+            }
+            const remaining = MAX_NON_JSON_ERROR_INSPECTION_BYTES - inspectedBytes;
+            if (remaining > 0) {
+              const prefix = value.byteLength <= remaining ? value.slice() : value.slice(0, remaining);
+              inspected.push(prefix);
+              inspectedBytes += prefix.byteLength;
+            }
+            controller.enqueue(value);
+          } catch (err) {
+            finalize(502);
+            try { controller.error(err); } catch { /* already torn down */ }
+          }
+        },
+        cancel(reason) {
+          finalize();
+          reader.cancel(reason).catch(() => {});
+        },
+      });
+      return new Response(body, {
+        status: response.status,
+        statusText: response.statusText,
+        headers: response.headers,
+      });
+    }
     if (isUsageDebugEnabled() && logCtx.usageDebugBodyKind === undefined) {
       logCtx.usageDebugBodyKind = response.body ? "other" : "none";
     }

tests/request-log.test.ts

@@ -676,6 +676,50 @@ describe("request log metadata", () => {
     });
   });
 
+  test("non-JSON error logging streams without waiting for the upstream body to end", async () => {
+    const entries: RequestLogEntry[] = [];
+    const encoder = new TextEncoder();
+    const upstream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(encoder.encode("upstream failed: first chunk"));
+      },
+    });
+    const response = responseWithDeferredRequestLog(
+      new Response(upstream, { status: 502, headers: { "content-type": "text/plain" } }),
+      "ocx-test-streaming-error",
+      Date.now(),
+      { model: "gpt-5.5", provider: "custom" },
+      entry => entries.push(entry),
+    );
+
+    const reader = response.body!.getReader();
+    const first = await Promise.race([
+      reader.read(),
+      Bun.sleep(250).then(() => { throw new Error("error response was buffered"); }),
+    ]);
+    expect(new TextDecoder().decode(first.value)).toBe("upstream failed: first chunk");
+    expect(entries).toHaveLength(0);
+    await reader.cancel();
+    expect(entries[0]?.upstreamError).toBe("upstream failed: first chunk");
+  });
+
+  test("non-JSON error logging inspects only a bounded prefix while preserving the body", async () => {
+    const entries: RequestLogEntry[] = [];
+    const body = `provider error: ${"x".repeat(20_000)}`;
+    const response = responseWithDeferredRequestLog(
+      new Response(body, { status: 500, headers: { "content-type": "text/plain" } }),
+      "ocx-test-bounded-error",
+      Date.now(),
+      { model: "gpt-5.5", provider: "custom" },
+      entry => entries.push(entry),
+    );
+
+    expect(await response.text()).toBe(body);
+    expect(entries).toHaveLength(1);
+    expect(entries[0]?.upstreamError).toHaveLength(500);
+    expect(entries[0]?.upstreamError).toStartWith("provider error: ");
+  });
+
   test("deferred SSE logging captures terminal reported usage", async () => {
     const entries: RequestLogEntry[] = [];
     const body = new ReadableStream<Uint8Array>({