fix(sse): handle non-SSE JSON upstream body on streaming path + SSE-wrap cache hits (#3089, #2952) (#3108)

#3089: reasoning openai-compatible upstreams that ignore stream:true and return application/json produced STREAM_EARLY_EOF because readiness only scans SSE data: frames. chatCore now detects a non-SSE JSON upstream body on the streaming path and synthesizes an equivalent OpenAI SSE stream (new synthesizeOpenAiSseFromJson util), preserving content + reasoning_content. #2952: semantic-cache hits returned application/json regardless of stream flag, so streaming clients lost reasoning_content; stream requests now SSE-wrap the cached completion via the same helper. Unit tests for the converter (4).

CHANGELOG.md

@@ -22,6 +22,8 @@ _Development cycle in progress — entries are added as work merges into `releas
 
 ### 🔧 Bug Fixes
 
+- **sse:** stop 502'ing streaming requests when a "reasoning" openai-compatible upstream ignores `stream:true` and returns a complete `application/json` body — the streaming readiness check only recognized SSE `data:` frames, so such a JSON body (even with valid `content`/`reasoning_content`) produced a spurious `STREAM_EARLY_EOF`. OmniRoute now detects a non-SSE JSON upstream body on the streaming path and synthesizes an equivalent OpenAI SSE stream (`synthesizeOpenAiSseFromJson`), preserving content + reasoning_content. ([#3089](https://github.com/diegosouzapw/OmniRoute/issues/3089))
+- **cache:** serve semantic-cache hits as SSE for streaming clients — a cache hit returned `application/json` regardless of the `stream` flag, so OpenAI-compatible streaming clients lost `reasoning_content` (and got a non-stream body) on cached responses. Stream requests now SSE-wrap the cached completion. ([#2952](https://github.com/diegosouzapw/OmniRoute/issues/2952))
 - **i18n:** fill the missing Chinese (zh-CN) and Russian (ru) UI translations — both locales were missing 9 entire sections (`quotaPlans`, `activity`, `agentBridge`, `trafficInspector`, `cliCommon`, `cliCode`, `cliAgents`, `acpAgents`, `agentSkills`, ~823 keys each) added after the last translation sweep, so those buttons/labels rendered in English. Both catalogs are now at full key parity with `en.json` (8025 keys). ([#3026](https://github.com/diegosouzapw/OmniRoute/issues/3026), [#3067](https://github.com/diegosouzapw/OmniRoute/issues/3067))
 - **dashboard:** fix "Ambiguous model" error in the provider Playground for vendor-namespaced models — the Playground only prefixed models without a `/`, so ids like `moonshotai/kimi-k2.6` or `nvidia/zyphra/zamba2-7b-instruct` (NVIDIA NIM) were sent bare and rejected when the same id exists under multiple providers. The Playground now always qualifies the selected model with its `providerId/` prefix (without double-prefixing). ([#3050](https://github.com/diegosouzapw/OmniRoute/issues/3050))
 - **db:** stop accepting duplicate API keys for the same provider — `createProviderConnection` now dedups by the decrypted key value (not just by name), so re-adding the same key under a different/blank name updates the existing connection instead of inserting a second row. Whitespace-only differences also dedup. ([#3023](https://github.com/diegosouzapw/OmniRoute/issues/3023))

open-sse/handlers/chatCore.ts

@@ -13,6 +13,7 @@ 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 { createSseHeartbeatTransform, shapeForClientFormat } from "../utils/sseHeartbeat.ts";
@@ -2290,20 +2291,28 @@ export async function handleChatCore({
         cacheSource: "semantic",
       });
       trackPendingRequest(model, provider, connectionId, false);
+      // #2952 — when the client requested a stream, serve the cached completion
+      // as an SSE stream (not a raw JSON body) so content + reasoning_content
+      // arrive in the streaming shape the client expects. Cache hits previously
+      // returned application/json regardless of the stream flag, which made
+      // OpenAI-compatible streaming clients lose reasoning_content. Non-OpenAI
+      // shapes (no `choices`) yield "" and fall back to the JSON body unchanged.
+      const cachedSse = stream ? synthesizeOpenAiSseFromJson(JSON.stringify(cached)) : "";
+      const cacheHitMetaHeaders = buildOmniRouteResponseMetaHeaders({
+        provider,
+        model,
+        cacheHit: true,
+        latencyMs: Date.now() - startTime,
+        usage: cachedUsage,
+        costUsd: cachedCost,
+      });
       return {
         success: true,
-        response: new Response(JSON.stringify(cached), {
+        response: new Response(cachedSse || JSON.stringify(cached), {
           headers: {
-            "Content-Type": "application/json",
+            "Content-Type": cachedSse ? "text/event-stream" : "application/json",
             [OMNIROUTE_RESPONSE_HEADERS.cache]: "HIT",
-            ...buildOmniRouteResponseMetaHeaders({
-              provider,
-              model,
-              cacheHit: true,
-              latencyMs: Date.now() - startTime,
-              usage: cachedUsage,
-              costUsd: cachedCost,
-            }),
+            ...cacheHitMetaHeaders,
           },
         }),
       };
@@ -5481,6 +5490,47 @@ export async function handleChatCore({
   }
 
   // Streaming response
+  // #3089 — some "reasoning" openai-compatible upstreams ignore a stream:true
+  // request and return a complete application/json chat-completion body instead
+  // of an SSE stream. The readiness check below only recognizes SSE `data:`
+  // frames, so that body produced a spurious STREAM_EARLY_EOF / HTTP 502 even
+  // 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,
+        });
+      }
+    }
+  }
   const streamReadinessPolicy = resolveStreamReadinessTimeout({
     baseTimeoutMs: STREAM_READINESS_TIMEOUT_MS,
     provider,

open-sse/utils/jsonToSse.ts

@@ -0,0 +1,80 @@
+/**
+ * #3089 — Convert a complete OpenAI-style chat-completion JSON body into an
+ * equivalent OpenAI SSE (`chat.completion.chunk`) stream.
+ *
+ * Some "reasoning" openai-compatible upstreams ignore a `stream: true` request
+ * and reply with a single `application/json` chat-completion body instead of an
+ * SSE stream. OmniRoute's streaming readiness check only recognizes SSE `data:`
+ * frames, so such a body produced a spurious `STREAM_EARLY_EOF` / HTTP 502 even
+ * though it carried valid `content` / `reasoning_content`. Synthesizing an SSE
+ * stream from that JSON lets the normal streaming pipeline (and the client) get
+ * a valid stream that preserves both `content` and `reasoning_content`.
+ *
+ * Returns "" when the text is not a parseable chat-completion object with at
+ * least one choice — callers then fall back to the original (error) handling.
+ */
+
+type JsonRecord = Record<string, unknown>;
+
+function isRecord(value: unknown): value is JsonRecord {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function sseEvent(payload: JsonRecord): string {
+  return `data: ${JSON.stringify(payload)}\n\n`;
+}
+
+export function synthesizeOpenAiSseFromJson(jsonText: string): string {
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(jsonText);
+  } catch {
+    return "";
+  }
+  if (!isRecord(parsed)) return "";
+
+  const choices = parsed.choices;
+  if (!Array.isArray(choices) || choices.length === 0) return "";
+
+  const id = typeof parsed.id === "string" && parsed.id ? parsed.id : "chatcmpl-omniroute-sse";
+  const created = typeof parsed.created === "number" ? parsed.created : 0;
+  const model = typeof parsed.model === "string" ? parsed.model : "";
+  const base = { id, object: "chat.completion.chunk", created, model };
+
+  let out = "";
+  let emittedAny = false;
+
+  choices.forEach((choice, fallbackIndex) => {
+    if (!isRecord(choice)) return;
+    const index = typeof choice.index === "number" ? choice.index : fallbackIndex;
+    const message = isRecord(choice.message) ? choice.message : {};
+
+    // First chunk carries role + whatever the message produced (content,
+    // reasoning_content, tool_calls). Putting them in one delta is valid and
+    // keeps downstream translation simple.
+    const delta: JsonRecord = { role: typeof message.role === "string" ? message.role : "assistant" };
+    if (typeof message.content === "string" && message.content.length > 0) {
+      delta.content = message.content;
+    }
+    if (typeof message.reasoning_content === "string" && message.reasoning_content.length > 0) {
+      delta.reasoning_content = message.reasoning_content;
+    }
+    if (Array.isArray(message.tool_calls) && message.tool_calls.length > 0) {
+      delta.tool_calls = message.tool_calls;
+    }
+
+    out += sseEvent({ ...base, choices: [{ index, delta, finish_reason: null }] });
+
+    const finishReason =
+      typeof choice.finish_reason === "string" && choice.finish_reason ? choice.finish_reason : "stop";
+    const finalChoice: JsonRecord = { index, delta: {}, finish_reason: finishReason };
+    const finalChunk: JsonRecord = { ...base, choices: [finalChoice] };
+    if (isRecord(parsed.usage)) finalChunk.usage = parsed.usage;
+    out += sseEvent(finalChunk);
+    emittedAny = true;
+  });
+
+  if (!emittedAny) return "";
+  out += "data: [DONE]\n\n";
+  return out;
+}

tests/unit/json-to-sse-3089.test.ts

@@ -0,0 +1,85 @@
+import { describe, test } from "node:test";
+import assert from "node:assert/strict";
+import { synthesizeOpenAiSseFromJson } from "../../open-sse/utils/jsonToSse.ts";
+
+function parseDataChunks(sse: string) {
+  return sse
+    .split(/\n\n/)
+    .map((b) => b.trim())
+    .filter((b) => b.startsWith("data:"))
+    .map((b) => b.slice(5).trim());
+}
+
+describe("synthesizeOpenAiSseFromJson (#3089)", () => {
+  test("converts a reasoning chat-completion JSON to SSE preserving content + reasoning_content", () => {
+    const body = JSON.stringify({
+      id: "mock-1",
+      object: "chat.completion",
+      created: 123,
+      model: "mock-reasoner",
+      choices: [
+        {
+          index: 0,
+          message: {
+            role: "assistant",
+            reasoning_content: "thinking...",
+            content: "HI there",
+          },
+          finish_reason: "stop",
+        },
+      ],
+      usage: { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 },
+    });
+
+    const sse = synthesizeOpenAiSseFromJson(body);
+    const chunks = parseDataChunks(sse);
+
+    assert.ok(sse.startsWith("data: "), "must be SSE");
+    assert.equal(chunks[chunks.length - 1], "[DONE]", "must terminate with [DONE]");
+
+    const first = JSON.parse(chunks[0]);
+    assert.equal(first.object, "chat.completion.chunk");
+    assert.equal(first.model, "mock-reasoner");
+    assert.equal(first.choices[0].delta.role, "assistant");
+    assert.equal(first.choices[0].delta.content, "HI there");
+    assert.equal(first.choices[0].delta.reasoning_content, "thinking...");
+
+    const finishChunk = JSON.parse(chunks[1]);
+    assert.equal(finishChunk.choices[0].finish_reason, "stop");
+    assert.deepEqual(finishChunk.usage, { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 });
+  });
+
+  test("content-only completion converts without a reasoning_content delta", () => {
+    const sse = synthesizeOpenAiSseFromJson(
+      JSON.stringify({ choices: [{ message: { role: "assistant", content: "ok" } }] })
+    );
+    const first = JSON.parse(parseDataChunks(sse)[0]);
+    assert.equal(first.choices[0].delta.content, "ok");
+    assert.equal("reasoning_content" in first.choices[0].delta, false);
+  });
+
+  test("forwards tool_calls in the delta", () => {
+    const sse = synthesizeOpenAiSseFromJson(
+      JSON.stringify({
+        choices: [
+          {
+            message: {
+              role: "assistant",
+              tool_calls: [{ id: "t1", type: "function", function: { name: "f", arguments: "{}" } }],
+            },
+            finish_reason: "tool_calls",
+          },
+        ],
+      })
+    );
+    const first = JSON.parse(parseDataChunks(sse)[0]);
+    assert.equal(first.choices[0].delta.tool_calls[0].id, "t1");
+  });
+
+  test("returns empty string for non-completion JSON / invalid JSON", () => {
+    assert.equal(synthesizeOpenAiSseFromJson('{"error":{"message":"x"}}'), "");
+    assert.equal(synthesizeOpenAiSseFromJson("{not json"), "");
+    assert.equal(synthesizeOpenAiSseFromJson('{"choices":[]}'), "");
+    assert.equal(synthesizeOpenAiSseFromJson("[]"), "");
+  });
+});