refactor(chatCore): extrai maybeConvertJsonBodyToSse (#3089 JSON→SSE streaming, #3501) (#4833)

Integrated into release/v3.8.36 (#3501 chatCore extraction stack 10/13)

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<string>(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,

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<Response> {
+  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<string>(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,
+  });
+}

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 <T>(p: Promise<T>) => p,
+    synthesizeOpenAiSseFromJson: synth,
+  } as Parameters<typeof maybeConvertJsonBodyToSse>[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");
+});