fix(chat): preserve completion semantics in JSON-to-SSE fallback

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.

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",

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("")), {

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

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",

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<typeof Bun.serve> | 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<Chunk[]> {
+  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);
+});