diff --git a/src/server/relay.ts b/src/server/relay.ts index 3b5aae2366..778f5658d3 100644 --- a/src/server/relay.ts +++ b/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({ + 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"; } diff --git a/tests/request-log.test.ts b/tests/request-log.test.ts index 50042794d7..7d067052d6 100644 --- a/tests/request-log.test.ts +++ b/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({ + 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({