diff --git a/CHANGELOG.md b/CHANGELOG.md index ec89baf291..6786ab4b49 100644 --- a/CHANGELOG.md +++ b/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)) diff --git a/open-sse/handlers/chatCore.ts b/open-sse/handlers/chatCore.ts index 96d25e4cf6..143daf9500 100644 --- a/open-sse/handlers/chatCore.ts +++ b/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(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, diff --git a/open-sse/utils/jsonToSse.ts b/open-sse/utils/jsonToSse.ts new file mode 100644 index 0000000000..a9a2f267c9 --- /dev/null +++ b/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; + +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; +} diff --git a/tests/unit/json-to-sse-3089.test.ts b/tests/unit/json-to-sse-3089.test.ts new file mode 100644 index 0000000000..5e42ba4825 --- /dev/null +++ b/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("[]"), ""); + }); +});