From 7362d77b74af238d2e43330b4d7901a3e5987246 Mon Sep 17 00:00:00 2001 From: Ingwannu Date: Sun, 6 Sep 2026 14:44:00 +0000 Subject: [PATCH] fix(chat): preserve completion semantics in JSON-to-SSE fallback --- .../content/docs/reference/proxy-formats.md | 7 ++ scripts/test-layout/layout.json | 1 + src/server/chat-completions.ts | 24 ++-- structure/04_transports-and-sidecars.md | 18 +++ tests/fixtures/test-layout-expected.json | 1 + .../responses/chat-json-sse-fallback.test.ts | 103 ++++++++++++++++++ 6 files changed, 146 insertions(+), 8 deletions(-) create mode 100644 tests/responses/chat-json-sse-fallback.test.ts diff --git a/docs-site/src/content/docs/reference/proxy-formats.md b/docs-site/src/content/docs/reference/proxy-formats.md index bba98afef1a..de0b6bf1743 100644 --- a/docs-site/src/content/docs/reference/proxy-formats.md +++ b/docs-site/src/content/docs/reference/proxy-formats.md @@ -251,6 +251,13 @@ Non-streaming output has `object: "chat.completion"`. Streaming output uses SSE `data: [DONE]`. Tool-call and usage information are translated back where the source events carry them. +If a streaming Chat request receives a complete JSON Responses result upstream, the proxy +synthesizes SSE from the converted completion. It preserves answer and reasoning content, +function tool calls (with a separate stream `index` for each call), usage, and the converted +`finish_reason`, including `tool_calls` and `length`. This fallback delivers the completed result +in chunks; it cannot provide token-by-token delivery before the upstream JSON response arrives. +It does not issue an additional inference request. + Because the internal execution path is Responses-based, a provider adapter can impose a narrower feature set. For example, a request feature that cannot be represented by the selected adapter is returned as an error instead of silently changing its meaning. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 0fbe7cf746b..2e2b806deea 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -270,6 +270,7 @@ "catalog-verbosity-default.test.ts": "codex-integration", "catalog-vision-sidecar-modalities.test.ts": "codex-integration", "chat-completions-endpoint.test.ts": "responses", + "chat-json-sse-fallback.test.ts": "responses", "chatgpt-device-auth.test.ts": "oauth", "chatgpt-oauth.test.ts": "oauth", "chatgpt-token-expiry.test.ts": "oauth", diff --git a/src/server/chat-completions.ts b/src/server/chat-completions.ts index d66a0df0b68..6e9e4e4c55c 100644 --- a/src/server/chat-completions.ts +++ b/src/server/chat-completions.ts @@ -455,20 +455,28 @@ async function handleChatCompletionsWithBudget( }); } - // Streaming client + JSON upstream: synthesize a minimal Chat Completions stream. + // JSON upstream changes delivery, not semantics: preserve the converted message and finish. const encoder = new TextEncoder(); const id = typeof completion.id === "string" ? completion.id : `chatcmpl-${Date.now()}`; const created = typeof completion.created === "number" ? completion.created : Math.floor(Date.now() / 1000); - const message = isRec((completion.choices as Rec[] | undefined)?.[0]) - ? ((completion.choices as Rec[])[0] as Rec).message as Rec | undefined - : undefined; - const content = message && typeof message.content === "string" ? message.content : ""; + const rawChoice = Array.isArray(completion.choices) ? completion.choices[0] : undefined; + const choice = isRec(rawChoice) ? rawChoice : {}; + const message = isRec(choice.message) ? choice.message : {}; + const delta: Rec = {}; + for (const field of ["content", "reasoning_content", "refusal"] as const) { + if (typeof message[field] === "string" && message[field].length > 0) delta[field] = message[field]; + } + if (Array.isArray(message.tool_calls) && message.tool_calls.length > 0) { + // JSON tool calls have no index; streamed clients use it to assemble each parallel call. + delta.tool_calls = message.tool_calls.filter(isRec).map((call, index) => ({ ...call, index })); + } + const finishReason = typeof choice.finish_reason === "string" ? choice.finish_reason : "stop"; const frames = [ `data: ${JSON.stringify({ id, object: "chat.completion.chunk", created, model: requestedModel, choices: [{ index: 0, delta: { role: "assistant", content: "" }, finish_reason: null }] })}\n\n`, - ...(content - ? [`data: ${JSON.stringify({ id, object: "chat.completion.chunk", created, model: requestedModel, choices: [{ index: 0, delta: { content }, finish_reason: null }] })}\n\n`] + ...(Object.keys(delta).length > 0 + ? [`data: ${JSON.stringify({ id, object: "chat.completion.chunk", created, model: requestedModel, choices: [{ index: 0, delta, finish_reason: null }] })}\n\n`] : []), - `data: ${JSON.stringify({ id, object: "chat.completion.chunk", created, model: requestedModel, choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: completion.usage })}\n\n`, + `data: ${JSON.stringify({ id, object: "chat.completion.chunk", created, model: requestedModel, choices: [{ index: 0, delta: {}, finish_reason: finishReason }], usage: completion.usage })}\n\n`, "data: [DONE]\n\n", ]; return new Response(encoder.encode(frames.join("")), { diff --git a/structure/04_transports-and-sidecars.md b/structure/04_transports-and-sidecars.md index 4716c447bb6..8fff1ffb3b0 100644 --- a/structure/04_transports-and-sidecars.md +++ b/structure/04_transports-and-sidecars.md @@ -1302,6 +1302,24 @@ messages are redacted before either JSON or SSE reaches the client. The native p request-attempt logging, reset retry, same-key 429 replay, key rotation, usage extraction, and request-signal cancellation contracts as routed Responses transport. +## Chat streaming client with a JSON upstream result + +The translated inbound path in `src/server/chat-completions.ts` may receive a complete JSON +Responses result even when the Chat client requested SSE. Its synthetic stream reuses +`responsesJsonToChatCompletion` as the semantic authority: converted text, reasoning, available +refusal content, tool calls, finish reason, and usage must survive this final delivery conversion. +Tool calls gain their array-order stream `index`; the stream retains one assistant-role frame, +one terminal choice, and one `[DONE]`. The existing response-body lifecycle owns translation-budget +release on consumption or cancellation. Actual upstream SSE and native Chat bypass this fallback. + +[Decision Log] +- 목적과 의도: Keep tool execution and incomplete-response detection working when a streaming client receives a JSON upstream result. +- 기존 구현 및 제약 조건: The existing fallback copied only text and forced `stop`, despite the JSON converter already retaining tool calls, reasoning, and incomplete status. +- 검토한 주요 대안: Duplicate Responses parsing in the emitter; perform another inference request; preserve the already-converted Chat completion. +- 선택한 방식: Copy supported converted message fields into one delta, assign tool-call stream indexes, and retain the converted finish reason. +- 다른 대안 대신 이 방식을 선택한 이유: One conversion authority prevents the streaming fallback from drifting from non-streaming semantics without changing routing or retry behavior. +- 장점, 단점 및 영향: No additional upstream request or dependency; this remains buffered delivery, not token-by-token upstream streaming. Handler regressions cover tools, reasoning, length, ordinary and empty completions, and budget release. + ## Parallel tool calls (default-on for chat providers) The openai-chat adapter buffers ALL streamed `tool_calls` deltas (keyed by `index`, falling back to diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index db2583b00b7..a1f0198e44a 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -105,6 +105,7 @@ "catalog-verbosity-default.test.ts": "codex-integration", "catalog-vision-sidecar-modalities.test.ts": "codex-integration", "chat-completions-endpoint.test.ts": "responses", + "chat-json-sse-fallback.test.ts": "responses", "chatgpt-device-auth.test.ts": "oauth", "chatgpt-oauth.test.ts": "oauth", "chatgpt-token-expiry.test.ts": "oauth", diff --git a/tests/responses/chat-json-sse-fallback.test.ts b/tests/responses/chat-json-sse-fallback.test.ts new file mode 100644 index 00000000000..37015b024d1 --- /dev/null +++ b/tests/responses/chat-json-sse-fallback.test.ts @@ -0,0 +1,103 @@ +import { afterEach, expect, test } from "bun:test"; +import { handleChatCompletions } from "../../src/server/chat-completions"; +import { translatorObservedBufferSnapshot } from "../../src/lib/translator-budget"; +import type { OcxConfig } from "../../src/types"; + +let upstream: ReturnType | undefined; +afterEach(async () => { await upstream?.stop(true); upstream = undefined; }); + +interface Chunk { + choices: Array<{ index: number; delta: { + role?: string; content?: string; reasoning_content?: string; + tool_calls?: Array<{ index: number; id: string; type: string; function: { name: string; arguments: string } }>; + }; finish_reason: string | null }>; + usage?: { prompt_tokens: number; completion_tokens: number }; +} + +async function streamFixture(output: unknown[], status = "completed", cancel = false): Promise { + const budgetBefore = translatorObservedBufferSnapshot().currentBytes; + let requests = 0; + upstream = Bun.serve({ hostname: "127.0.0.1", port: 0, async fetch(req) { + expect(new URL(req.url).pathname).toBe("/v1/responses"); + expect((await req.json() as { stream: boolean }).stream).toBe(true); + requests++; + return Response.json({ id: "resp_fixture", status, output, + ...(status === "incomplete" ? { incomplete_details: { reason: "max_output_tokens" } } : {}), + usage: { input_tokens: 11, output_tokens: 7 } }); + } }); + const config: OcxConfig = { port: 0, defaultProvider: "fixture", providers: { fixture: { + adapter: "openai-responses", baseUrl: `http://127.0.0.1:${upstream.port}/v1`, + authMode: "key", apiKey: "fixture-key", allowPrivateNetwork: true, models: ["model"], + } } }; + const response = await handleChatCompletions(new Request("http://localhost/v1/chat/completions", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "fixture/model", stream: true, messages: [{ role: "user", content: "fixture" }], + tools: [{ type: "function", function: { name: "lookup", parameters: { type: "object" } } }] }), + }), config, { model: "", provider: "" }); + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("text/event-stream"); + if (cancel) { + await response.body!.cancel("fixture cancellation"); + expect(requests).toBe(1); + expect(translatorObservedBufferSnapshot().currentBytes).toBe(budgetBefore); + return []; + } + const text = await response.text(); + expect(translatorObservedBufferSnapshot().currentBytes).toBe(budgetBefore); + expect(requests).toBe(1); + const payloads = text.split(/\r?\n/).filter(line => line.startsWith("data: ")).map(line => line.slice(6)); + expect(payloads.filter(value => value === "[DONE]")).toHaveLength(1); + expect(payloads.at(-1)).toBe("[DONE]"); + const chunks = payloads.filter(value => value !== "[DONE]").map(value => JSON.parse(value) as Chunk); + expect(chunks.flatMap(chunk => chunk.choices).filter(choice => choice.finish_reason !== null)).toHaveLength(1); + expect(chunks.at(-1)?.usage).toMatchObject({ prompt_tokens: 11, completion_tokens: 7 }); + return chunks; +} + +test.each([1, 2])("JSON-to-SSE keeps %s indexed tool calls and tool_calls finish", async count => { + const calls = Array.from({ length: count }, (_, index) => ({ type: "function_call", + call_id: `call_fixture_${index}`, name: "lookup", arguments: JSON.stringify({ index }) })); + const chunks = await streamFixture(calls); + expect(chunks.flatMap(chunk => chunk.choices.flatMap(choice => choice.delta.tool_calls ?? []))) + .toEqual(calls.map((call, index) => ({ index, id: call.call_id, type: "function", + function: { name: call.name, arguments: call.arguments } }))); + expect(chunks.at(-1)?.choices[0]?.finish_reason).toBe("tool_calls"); +}); + +test("JSON-to-SSE keeps reasoning alongside answer text", async () => { + const chunks = await streamFixture([ + { type: "reasoning", summary: [{ type: "summary_text", text: "Fixture reasoning." }] }, + { type: "message", role: "assistant", content: [{ type: "output_text", text: "Answer." }] }, + ]); + expect(chunks.flatMap(chunk => chunk.choices).map(choice => choice.delta.reasoning_content ?? "").join("")) + .toBe("Fixture reasoning."); + expect(chunks.flatMap(chunk => chunk.choices).map(choice => choice.delta.content ?? "").join("")) + .toBe("Answer."); + expect(chunks.at(-1)?.choices[0]?.finish_reason).toBe("stop"); +}); + +test("JSON-to-SSE preserves length instead of claiming a normal stop", async () => { + const chunks = await streamFixture([ + { type: "message", role: "assistant", content: [{ type: "output_text", text: "Partial answer." }] }, + ], "incomplete"); + expect(chunks.at(-1)?.choices[0]?.finish_reason).toBe("length"); +}); + +test("JSON-to-SSE preserves ordinary text and a single empty completion terminal", async () => { + const chunks = await streamFixture([ + { type: "message", role: "assistant", content: [{ type: "output_text", text: "Ordinary text." }] }, + ]); + expect(chunks.flatMap(chunk => chunk.choices).map(choice => choice.delta.content ?? "").join("")) + .toBe("Ordinary text."); + expect(chunks.at(-1)?.choices[0]?.finish_reason).toBe("stop"); +}); + +test("JSON-to-SSE empty completion still terminates once", async () => { + const chunks = await streamFixture([]); + expect(chunks).toHaveLength(2); + expect(chunks.at(-1)?.choices[0]?.finish_reason).toBe("stop"); +}); + +test("JSON-to-SSE cancellation releases the existing translation budget", async () => { + await streamFixture([{ type: "function_call", call_id: "call_cancel", name: "lookup", arguments: "{}" }], "completed", true); +});