From c69bd6a4ebdc92aa2abf0e9b40956cdd41940968 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Tue, 23 Jun 2026 20:02:42 -0300 Subject: [PATCH] =?UTF-8?q?refactor(chatCore):=20extrai=20maybeConvertJson?= =?UTF-8?q?BodyToSse=20(#3089=20JSON=E2=86=92SSE=20streaming,=20#3501)=20(?= =?UTF-8?q?#4833)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Integrated into release/v3.8.36 (#3501 chatCore extraction stack 10/13) --- open-sse/handlers/chatCore.ts | 37 +--------- open-sse/handlers/chatCore/jsonBodyToSse.ts | 68 ++++++++++++++++++ tests/unit/chatcore-json-body-to-sse.test.ts | 73 ++++++++++++++++++++ 3 files changed, 143 insertions(+), 35 deletions(-) create mode 100644 open-sse/handlers/chatCore/jsonBodyToSse.ts create mode 100644 tests/unit/chatcore-json-body-to-sse.test.ts diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 6fd5168e0e..3f96a6c699 100644 --- a/open-sse/handlers/chatCore.ts +++ b/open-sse/handlers/chatCore.ts @@ -9,6 +9,7 @@ import { applyClientUsageBuffer } from "./chatCore/clientUsageBuffer.ts"; import { buildPostCallGuardrailContext } from "./chatCore/postCallGuardrailContext.ts"; import { storeSemanticCacheResponse } from "./chatCore/semanticCacheStore.ts"; import { buildNonStreamingResponseHeaders } from "./chatCore/nonStreamingResponseHeaders.ts"; +import { maybeConvertJsonBodyToSse } from "./chatCore/jsonBodyToSse.ts"; import { sanitizeChatRequestBody } from "./chatCore/sanitization.ts"; import { getHeaderValueCaseInsensitive, @@ -67,7 +68,6 @@ import { withBodyTimeout, } from "../utils/stream.ts"; import { ensureStreamReadiness } from "../utils/streamReadiness.ts"; -import { synthesizeOpenAiSseFromJson } from "../utils/jsonToSse.ts"; import { resolveStreamReadinessTimeout } from "../utils/streamReadinessPolicy.ts"; import { createStreamController, pipeWithDisconnect } from "../utils/streamHandler.ts"; import * as streamFailure from "../utils/streamFailureFinalization.ts"; @@ -3646,40 +3646,7 @@ export async function handleChatCore({ // though it carried valid content/reasoning_content. Detect a JSON (non-SSE) // upstream body and synthesize an equivalent OpenAI SSE stream so the // streaming pipeline (and the client) get a valid stream. - { - const upstreamContentType = (providerResponse.headers.get("content-type") || "").toLowerCase(); - const isNonSseJsonBody = - !!providerResponse.body && - upstreamContentType.includes("application/json") && - !upstreamContentType.includes("text/event-stream") && - !upstreamContentType.includes("application/x-ndjson"); - if (isNonSseJsonBody) { - const jsonText = await withBodyTimeout(providerResponse.text()); - const synthesizedSse = synthesizeOpenAiSseFromJson(jsonText); - const rebuiltHeaders = new Headers(providerResponse.headers); - rebuiltHeaders.delete("content-length"); - if (synthesizedSse) { - log?.debug?.( - "STREAM", - `Upstream returned application/json on a streaming request — converting to SSE (${provider}/${model})` - ); - rebuiltHeaders.set("content-type", "text/event-stream"); - providerResponse = new Response(synthesizedSse, { - status: providerResponse.status, - statusText: providerResponse.statusText, - headers: rebuiltHeaders, - }); - } else { - // Not a convertible chat-completion JSON — rebuild the consumed body so - // the existing readiness/error path still runs unchanged. - providerResponse = new Response(jsonText, { - status: providerResponse.status, - statusText: providerResponse.statusText, - headers: rebuiltHeaders, - }); - } - } - } + providerResponse = await maybeConvertJsonBodyToSse(providerResponse, { log, provider, model }); const streamReadinessPolicy = resolveStreamReadinessTimeout({ baseTimeoutMs: STREAM_READINESS_TIMEOUT_MS, provider, diff --git a/open-sse/handlers/chatCore/jsonBodyToSse.ts b/open-sse/handlers/chatCore/jsonBodyToSse.ts new file mode 100644 index 0000000000..fa4db8867d --- /dev/null +++ b/open-sse/handlers/chatCore/jsonBodyToSse.ts @@ -0,0 +1,68 @@ +/** + * chatCore non-SSE JSON → SSE conversion (Quality Gate v2 / Fase 9 — chatCore god-file + * decomposition, #3501). + * + * Extracted from handleChatCore's streaming entry (#3089): some "reasoning" openai-compatible + * upstreams ignore `stream:true` and return a complete application/json chat-completion body + * instead of an SSE stream. The readiness check only recognizes SSE `data:` frames, so that body + * produced a spurious STREAM_EARLY_EOF / HTTP 502. Detect a JSON (non-SSE) upstream body and + * synthesize an equivalent OpenAI SSE stream so the streaming pipeline gets a valid stream. + * + * Returns the (possibly rebuilt) provider response — unchanged when the body is not a non-SSE JSON + * body, an SSE stream when convertible, or a rebuilt-with-consumed-body response otherwise (so the + * existing readiness/error path still runs unchanged). Behaviour is byte-identical to the previous + * inline block. + */ +import { withBodyTimeout as defaultWithBodyTimeout } from "../../utils/stream.ts"; +import { synthesizeOpenAiSseFromJson as defaultSynthesize } from "../../utils/jsonToSse.ts"; + +type LoggerLike = { debug?: (...args: unknown[]) => void } | null | undefined; + +export interface JsonBodyToSseDeps { + withBodyTimeout: typeof defaultWithBodyTimeout; + synthesizeOpenAiSseFromJson: typeof defaultSynthesize; +} + +const DEFAULT_DEPS: JsonBodyToSseDeps = { + withBodyTimeout: defaultWithBodyTimeout, + synthesizeOpenAiSseFromJson: defaultSynthesize, +}; + +export async function maybeConvertJsonBodyToSse( + providerResponse: Response, + ctx: { log?: LoggerLike; provider: string | null | undefined; model: string | null | undefined }, + deps: JsonBodyToSseDeps = DEFAULT_DEPS +): Promise { + const upstreamContentType = (providerResponse.headers.get("content-type") || "").toLowerCase(); + const isNonSseJsonBody = + !!providerResponse.body && + upstreamContentType.includes("application/json") && + !upstreamContentType.includes("text/event-stream") && + !upstreamContentType.includes("application/x-ndjson"); + if (!isNonSseJsonBody) { + return providerResponse; + } + const jsonText = await deps.withBodyTimeout(providerResponse.text()); + const synthesizedSse = deps.synthesizeOpenAiSseFromJson(jsonText); + const rebuiltHeaders = new Headers(providerResponse.headers); + rebuiltHeaders.delete("content-length"); + if (synthesizedSse) { + ctx.log?.debug?.( + "STREAM", + `Upstream returned application/json on a streaming request — converting to SSE (${ctx.provider}/${ctx.model})` + ); + rebuiltHeaders.set("content-type", "text/event-stream"); + return new Response(synthesizedSse, { + status: providerResponse.status, + statusText: providerResponse.statusText, + headers: rebuiltHeaders, + }); + } + // Not a convertible chat-completion JSON — rebuild the consumed body so the existing + // readiness/error path still runs unchanged. + return new Response(jsonText, { + status: providerResponse.status, + statusText: providerResponse.statusText, + headers: rebuiltHeaders, + }); +} diff --git a/tests/unit/chatcore-json-body-to-sse.test.ts b/tests/unit/chatcore-json-body-to-sse.test.ts new file mode 100644 index 0000000000..ed778e5fec --- /dev/null +++ b/tests/unit/chatcore-json-body-to-sse.test.ts @@ -0,0 +1,73 @@ +// Characterization of maybeConvertJsonBodyToSse — the #3089 non-SSE JSON → SSE conversion +// extracted from handleChatCore's streaming entry (chatCore god-file decomposition, #3501). Real +// Response objects + injected deps make the content-type gate, the synthesize-or-rebuild branch, +// and the header rewrites observable. Locks: pass-through for SSE/ndjson/no-body, conversion to +// text/event-stream when synthesizable, and rebuild-with-consumed-body otherwise. +import { test } from "node:test"; +import assert from "node:assert/strict"; + +const { maybeConvertJsonBodyToSse } = await import( + "../../open-sse/handlers/chatCore/jsonBodyToSse.ts" +); + +const ctx = { log: undefined, provider: "openai", model: "gpt-x" }; + +function makeDeps(synth: (s: string) => string | null) { + return { + withBodyTimeout: async (p: Promise) => p, + synthesizeOpenAiSseFromJson: synth, + } as Parameters[2]; +} + +test("already text/event-stream → returned unchanged (same reference)", async () => { + const resp = new Response("data: {}\n\n", { + headers: { "content-type": "text/event-stream" }, + }); + const out = await maybeConvertJsonBodyToSse(resp, ctx, makeDeps(() => "X")); + assert.equal(out, resp); +}); + +test("application/x-ndjson → returned unchanged", async () => { + const resp = new Response('{"a":1}\n', { headers: { "content-type": "application/x-ndjson" } }); + const out = await maybeConvertJsonBodyToSse(resp, ctx, makeDeps(() => "X")); + assert.equal(out, resp); +}); + +test("no body → returned unchanged", async () => { + const resp = new Response(null, { headers: { "content-type": "application/json" } }); + const out = await maybeConvertJsonBodyToSse(resp, ctx, makeDeps(() => "X")); + assert.equal(out, resp); +}); + +test("application/json + synthesizable → SSE response with text/event-stream, no content-length", async () => { + const resp = new Response('{"choices":[]}', { + headers: { "content-type": "application/json", "content-length": "14" }, + }); + const out = await maybeConvertJsonBodyToSse(resp, ctx, makeDeps(() => "data: synth\n\n")); + assert.notEqual(out, resp); + assert.equal(out.headers.get("content-type"), "text/event-stream"); + assert.equal(out.headers.get("content-length"), null); + assert.equal(await out.text(), "data: synth\n\n"); +}); + +test("application/json + not synthesizable → rebuilt with consumed body, content-type unchanged", async () => { + const resp = new Response('{"not":"chat"}', { + headers: { "content-type": "application/json" }, + }); + const out = await maybeConvertJsonBodyToSse(resp, ctx, makeDeps(() => null)); + assert.notEqual(out, resp); + // not converted → content-type stays application/json (rebuilt headers) + assert.equal(out.headers.get("content-type"), "application/json"); + assert.equal(await out.text(), '{"not":"chat"}'); +}); + +test("status and statusText are preserved on conversion", async () => { + const resp = new Response('{"choices":[]}', { + status: 201, + statusText: "Created", + headers: { "content-type": "application/json" }, + }); + const out = await maybeConvertJsonBodyToSse(resp, ctx, makeDeps(() => "data: x\n\n")); + assert.equal(out.status, 201); + assert.equal(out.statusText, "Created"); +});