diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 7146188bc2..b2a011ec2e 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -13,6 +13,10 @@ import { ensureStreamReadiness } from "../utils/streamReadiness.ts"; import { createStreamController, pipeWithDisconnect } from "../utils/streamHandler.ts"; import { createSseHeartbeatTransform, shapeForClientFormat } from "../utils/sseHeartbeat.ts"; import { addBufferToUsage, filterUsageForFormat, estimateUsage } from "../utils/usageTracking.ts"; +import { + stripStaleEncodingHeaders, + filterUpstreamResponseHeaderEntries, +} from "../utils/upstreamResponseHeaders.ts"; import { refreshWithRetry } from "../services/tokenRefresh.ts"; import { createRequestLogger } from "../utils/requestLogger.ts"; import { getModelTargetFormat, PROVIDER_ID_TO_ALIAS } from "../config/providerModels.ts"; @@ -3208,7 +3212,10 @@ export async function handleChatCore({ // Non-stream: release semaphore immediately after reading full response body. const status = rawResult.response.status; const statusText = rawResult.response.statusText; - const headers = new Headers(rawResult.response.headers); + // Strip content-encoding/length/transfer-encoding: fetch() already + // decompressed the body and we are about to repack it via new Response() + // below, so forwarding the upstream encoding/length is misleading. + const headers = stripStaleEncodingHeaders(rawResult.response.headers); const contentType = (headers.get("content-type") || "").toLowerCase(); const payload = await readNonStreamingResponseBody( rawResult.response, @@ -4134,7 +4141,10 @@ export async function handleChatCore({ try { const firstChoice = translatedResponse?.choices?.[0]; const msg = firstChoice?.message; - cacheReasoningFromAssistantMessage(msg, provider, model, { requestId: skillRequestId, messageIndex: 0 }); + cacheReasoningFromAssistantMessage(msg, provider, model, { + requestId: skillRequestId, + messageIndex: 0, + }); } catch { // Cache capture is non-critical — never block the response } @@ -4364,13 +4374,14 @@ export async function handleChatCore({ await onRequestSuccess(); } + // Strip content-type (we override with text/event-stream below) and the + // stale encoding/length headers — fetch() decompressed the upstream body and + // we re-stream it through our own transforms, so forwarding the upstream + // content-encoding (e.g. "gzip") makes openai-compatible clients try to + // gunzip plain text and fail with ZlibError ("incorrect header check"). const responseHeaders: Record = { ...Object.fromEntries( - (() => { - const arr: [string, string][] = []; - providerResponse.headers.forEach((v, k) => arr.push([k, v])); - return arr; - })().filter(([k]) => k.toLowerCase() !== "content-type") + filterUpstreamResponseHeaderEntries(providerResponse.headers.entries(), ["content-type"]) ), "Content-Type": "text/event-stream", "Cache-Control": "no-cache, no-transform", @@ -4421,7 +4432,10 @@ export async function handleChatCore({ const body = streamResponseBody as Record; const choices = body.choices as { message?: Record }[] | undefined; const msg = choices?.[0]?.message; - cacheReasoningFromAssistantMessage(msg, provider, model, { requestId: skillRequestId, messageIndex: 0 }); + cacheReasoningFromAssistantMessage(msg, provider, model, { + requestId: skillRequestId, + messageIndex: 0, + }); } catch { // Cache capture is non-critical — never block the stream } diff --git a/open-sse/utils/upstreamResponseHeaders.ts b/open-sse/utils/upstreamResponseHeaders.ts new file mode 100644 index 0000000000..834d354f18 --- /dev/null +++ b/open-sse/utils/upstreamResponseHeaders.ts @@ -0,0 +1,47 @@ +/** + * Header-strip helper for upstream provider responses. + * + * `fetch()` always decompresses the upstream body before exposing it via + * `.text()` or the stream reader, so forwarding the upstream `content-encoding` + * to the downstream client (e.g. `gzip`) makes the client attempt to gunzip + * plain text and fail with `ZlibError: incorrect header check`. + * + * Similarly, `content-length` becomes stale once we transform or repack the + * response stream, and `transfer-encoding` is managed by the runtime + * (Next.js / Node), not us. + */ + +const STRIP_HEADER_NAMES: ReadonlySet = new Set([ + "content-encoding", + "content-length", + "transfer-encoding", +]); + +/** + * Return a new `Headers` instance with stale encoding/length headers removed. + * Does not mutate the input. + */ +export function stripStaleEncodingHeaders(input: Headers): Headers { + const out = new Headers(input); + for (const name of STRIP_HEADER_NAMES) out.delete(name); + return out; +} + +/** + * Return a new entries array with stale encoding/length headers removed and + * (optionally) additional header names removed. Case-insensitive. + */ +export function filterUpstreamResponseHeaderEntries( + entries: Iterable<[string, string]>, + extraToStrip: ReadonlyArray = [] +): Array<[string, string]> { + const drop = new Set(STRIP_HEADER_NAMES); + for (const h of extraToStrip) drop.add(h.toLowerCase()); + const result: Array<[string, string]> = []; + for (const [k, v] of entries) { + if (!drop.has(k.toLowerCase())) result.push([k, v]); + } + return result; +} + +export const STRIP_UPSTREAM_HEADER_NAMES: ReadonlySet = STRIP_HEADER_NAMES; diff --git a/tests/unit/upstream-response-headers-strip.test.ts b/tests/unit/upstream-response-headers-strip.test.ts new file mode 100644 index 0000000000..fe7c6f9edf --- /dev/null +++ b/tests/unit/upstream-response-headers-strip.test.ts @@ -0,0 +1,112 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { + stripStaleEncodingHeaders, + filterUpstreamResponseHeaderEntries, + STRIP_UPSTREAM_HEADER_NAMES, +} from "../../open-sse/utils/upstreamResponseHeaders.ts"; + +test("stripStaleEncodingHeaders: removes content-encoding/length/transfer-encoding (lowercase)", () => { + const input = new Headers({ + "content-encoding": "gzip", + "content-length": "1234", + "transfer-encoding": "chunked", + "content-type": "application/json", + "x-request-id": "abc", + }); + const out = stripStaleEncodingHeaders(input); + assert.strictEqual(out.get("content-encoding"), null); + assert.strictEqual(out.get("content-length"), null); + assert.strictEqual(out.get("transfer-encoding"), null); + assert.strictEqual(out.get("content-type"), "application/json"); + assert.strictEqual(out.get("x-request-id"), "abc"); +}); + +test("stripStaleEncodingHeaders: removes mixed-case header names (case-insensitive)", () => { + const input = new Headers({ + "Content-Encoding": "gzip", + "Content-Length": "1234", + "Transfer-Encoding": "chunked", + "Content-Type": "text/plain", + }); + const out = stripStaleEncodingHeaders(input); + assert.strictEqual(out.get("content-encoding"), null); + assert.strictEqual(out.get("content-length"), null); + assert.strictEqual(out.get("transfer-encoding"), null); + assert.strictEqual(out.get("content-type"), "text/plain"); +}); + +test("stripStaleEncodingHeaders: does not mutate the input Headers", () => { + const input = new Headers({ + "content-encoding": "gzip", + "content-type": "application/json", + }); + const out = stripStaleEncodingHeaders(input); + assert.strictEqual(input.get("content-encoding"), "gzip"); + assert.strictEqual(out.get("content-encoding"), null); +}); + +test("stripStaleEncodingHeaders: empty input returns empty Headers", () => { + const out = stripStaleEncodingHeaders(new Headers()); + // Iterate to confirm no entries. + const entries: Array<[string, string]> = []; + out.forEach((v, k) => entries.push([k, v])); + assert.deepEqual(entries, []); +}); + +test("filterUpstreamResponseHeaderEntries: strips default header set", () => { + const entries: Array<[string, string]> = [ + ["content-encoding", "gzip"], + ["content-length", "1234"], + ["transfer-encoding", "chunked"], + ["content-type", "application/json"], + ["x-request-id", "abc"], + ]; + const out = filterUpstreamResponseHeaderEntries(entries); + assert.deepEqual(out, [ + ["content-type", "application/json"], + ["x-request-id", "abc"], + ]); +}); + +test("filterUpstreamResponseHeaderEntries: extraToStrip is case-insensitive", () => { + const entries: Array<[string, string]> = [ + ["Content-Type", "text/event-stream"], + ["X-Request-Id", "abc"], + ]; + const out = filterUpstreamResponseHeaderEntries(entries, ["CONTENT-TYPE"]); + assert.deepEqual(out, [["X-Request-Id", "abc"]]); +}); + +test("filterUpstreamResponseHeaderEntries: empty extraToStrip preserves non-default headers", () => { + const entries: Array<[string, string]> = [ + ["content-type", "application/json"], + ["x-custom", "value"], + ]; + const out = filterUpstreamResponseHeaderEntries(entries, []); + assert.deepEqual(out, [ + ["content-type", "application/json"], + ["x-custom", "value"], + ]); +}); + +test("filterUpstreamResponseHeaderEntries: empty input returns empty array", () => { + const out = filterUpstreamResponseHeaderEntries([]); + assert.deepEqual(out, []); +}); + +test("filterUpstreamResponseHeaderEntries: handles mixed-case default header names", () => { + const entries: Array<[string, string]> = [ + ["Content-Encoding", "gzip"], + ["X-Custom", "v"], + ]; + const out = filterUpstreamResponseHeaderEntries(entries); + assert.deepEqual(out, [["X-Custom", "v"]]); +}); + +test("STRIP_UPSTREAM_HEADER_NAMES: contains expected three lowercase names", () => { + assert.strictEqual(STRIP_UPSTREAM_HEADER_NAMES.size, 3); + assert.ok(STRIP_UPSTREAM_HEADER_NAMES.has("content-encoding")); + assert.ok(STRIP_UPSTREAM_HEADER_NAMES.has("content-length")); + assert.ok(STRIP_UPSTREAM_HEADER_NAMES.has("transfer-encoding")); +});