diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 3f96a6c699..d4caf28454 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -10,6 +10,7 @@ import { buildPostCallGuardrailContext } from "./chatCore/postCallGuardrailConte import { storeSemanticCacheResponse } from "./chatCore/semanticCacheStore.ts"; import { buildNonStreamingResponseHeaders } from "./chatCore/nonStreamingResponseHeaders.ts"; import { maybeConvertJsonBodyToSse } from "./chatCore/jsonBodyToSse.ts"; +import { assembleStreamingResponseHeaders } from "./chatCore/streamingResponseHeaders.ts"; import { sanitizeChatRequestBody } from "./chatCore/sanitization.ts"; import { getHeaderValueCaseInsensitive, @@ -3709,20 +3710,13 @@ export async function handleChatCore({ await onRequestSuccess(); } - const responseHeaders: Record = { - ...buildStreamingResponseHeaders(providerResponse.headers, { - provider, - model, - cacheHit: false, - latencyMs: 0, - usage: null, - costUsd: 0, - }), - "x-omniroute-request-id": pendingRequestId, - }; - if (compressionResponseMeta) { - responseHeaders[OMNIROUTE_RESPONSE_HEADERS.compression] = compressionResponseMeta; - } + const responseHeaders = assembleStreamingResponseHeaders({ + providerHeaders: providerResponse.headers, + provider, + model, + pendingRequestId, + compressionResponseMeta, + }); // Create transform stream with logger for streaming response let transformStream; diff --git a/open-sse/handlers/chatCore/streamingResponseHeaders.ts b/open-sse/handlers/chatCore/streamingResponseHeaders.ts new file mode 100644 index 0000000000..a978a0a212 --- /dev/null +++ b/open-sse/handlers/chatCore/streamingResponseHeaders.ts @@ -0,0 +1,39 @@ +/** + * chatCore streaming response headers (Quality Gate v2 / Fase 9 — chatCore god-file decomposition, + * #3501). + * + * Extracted from handleChatCore's streaming success path: assemble the streaming response header map + * — the upstream-derived streaming headers (via buildStreamingResponseHeaders, with zeroed + * latency/usage/cost since those are not yet known at stream start), the per-request id, and the + * optional compression header. Pure builder (returns a fresh map). Behaviour is byte-identical to + * the previous inline block. + */ +import { OMNIROUTE_RESPONSE_HEADERS } from "@/shared/constants/headers"; +import { buildStreamingResponseHeaders as defaultBuildStreaming } from "./responseHeaders.ts"; + +export function assembleStreamingResponseHeaders( + args: { + providerHeaders: Headers; + provider: string | null | undefined; + model: string | null | undefined; + pendingRequestId: string; + compressionResponseMeta?: string | null | undefined; + }, + buildStreamingResponseHeaders: typeof defaultBuildStreaming = defaultBuildStreaming +): Record { + const responseHeaders: Record = { + ...buildStreamingResponseHeaders(args.providerHeaders, { + provider: args.provider, + model: args.model, + cacheHit: false, + latencyMs: 0, + usage: null, + costUsd: 0, + }), + "x-omniroute-request-id": args.pendingRequestId, + }; + if (args.compressionResponseMeta) { + responseHeaders[OMNIROUTE_RESPONSE_HEADERS.compression] = args.compressionResponseMeta; + } + return responseHeaders; +} diff --git a/tests/unit/chatcore-streaming-response-headers.test.ts b/tests/unit/chatcore-streaming-response-headers.test.ts new file mode 100644 index 0000000000..8b714ca4eb --- /dev/null +++ b/tests/unit/chatcore-streaming-response-headers.test.ts @@ -0,0 +1,65 @@ +// Characterization of assembleStreamingResponseHeaders — the streaming response header builder +// extracted from handleChatCore's streaming success path (chatCore god-file decomposition, #3501). +// buildStreamingResponseHeaders is injected so the merge of upstream headers + request-id + the +// optional compression header is observable. Locks: zeroed latency/usage/cost at stream start, the +// x-omniroute-request-id, and the compression header only when meta is present. +import { test } from "node:test"; +import assert from "node:assert/strict"; + +const { assembleStreamingResponseHeaders } = await import( + "../../open-sse/handlers/chatCore/streamingResponseHeaders.ts" +); + +function makeBuild() { + const calls: Array<{ headers: unknown; meta: Record }> = []; + const build = (headers: unknown, meta: Record) => { + calls.push({ headers, meta }); + return { "x-upstream": "kept" }; + }; + return { build: build as Parameters[1], calls }; +} + +function baseArgs(overrides: Record = {}) { + return { + providerHeaders: new Headers({ "content-type": "text/event-stream" }), + provider: "openai", + model: "gpt-x", + pendingRequestId: "preq-1", + compressionResponseMeta: undefined, + ...overrides, + } as Parameters[0]; +} + +test("merges upstream headers and sets x-omniroute-request-id", () => { + const { build } = makeBuild(); + const h = assembleStreamingResponseHeaders(baseArgs(), build); + assert.equal(h["x-upstream"], "kept"); + assert.equal(h["x-omniroute-request-id"], "preq-1"); +}); + +test("buildStreamingResponseHeaders receives zeroed latency/usage/cost and cacheHit false", () => { + const { build, calls } = makeBuild(); + assembleStreamingResponseHeaders(baseArgs(), build); + assert.equal(calls.length, 1); + assert.equal(calls[0].meta.cacheHit, false); + assert.equal(calls[0].meta.latencyMs, 0); + assert.equal(calls[0].meta.usage, null); + assert.equal(calls[0].meta.costUsd, 0); + assert.equal(calls[0].meta.provider, "openai"); + assert.equal(calls[0].meta.model, "gpt-x"); +}); + +test("no compression meta → no compression header", () => { + const { build } = makeBuild(); + const h = assembleStreamingResponseHeaders(baseArgs({ compressionResponseMeta: undefined }), build); + assert.ok(!Object.values(h).includes("engine:z")); +}); + +test("compression meta present → compression header set", () => { + const { build } = makeBuild(); + const h = assembleStreamingResponseHeaders( + baseArgs({ compressionResponseMeta: "engine:z; source=routing" }), + build + ); + assert.ok(Object.values(h).includes("engine:z; source=routing")); +});