From 45d38bc4b22cf93e877c267b8ba2ed8dce263e9e Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza <8016841+diegosouzapw@users.noreply.github.com> Date: Sun, 12 Jul 2026 10:23:33 -0300 Subject: [PATCH] fix(sse): defer response.completed until trailing usage-only chunk (#6906) (#6965) Real OpenAI-compatible upstreams with stream_options.include_usage=true send finish_reason in one chunk (usage: null) and the actual token counts in a separate, trailing usage-only chunk (choices: [], usage: {...}). Both the live translator (openai-responses.ts) and the legacy transformer (responsesTransformer.ts) fired response.completed as soon as they saw finish_reason, so the trailing usage chunk's token counts were captured into state but never emitted -- Codex CLI and other /v1/responses consumers saw response.completed with no usage field (permanent 0% context-used). Both translators now defer response.completed via an awaitingTrailingUsage state flag when finish_reason arrives without usage already captured, and complete on the next usage-only chunk (or at stream end via the existing flush fallback) instead. Extracted the duplicated events/emit boilerplate into a new openai-responses/eventEmitter.ts leaf to keep the frozen openai-responses.ts file under its file-size baseline. Fixes 3 existing tests that encoded the old chunk ordering and adds a permanent regression test (tests/unit/responses-usage-trailing-6906.ts) covering both translators. --- .../fixes/6906-responses-usage-trailing.md | 1 + open-sse/transformer/responsesTransformer.ts | 22 ++- .../translator/response/openai-responses.ts | 33 +++-- .../response/openai-responses/eventEmitter.ts | 12 ++ .../unit/responses-translation-fixes.test.ts | 5 +- .../responses-usage-trailing-6906.test.ts | 133 ++++++++++++++++++ ...-openai-responses-custom-tool-1007.test.ts | 3 + .../translator-resp-openai-responses.test.ts | 3 + 8 files changed, 196 insertions(+), 16 deletions(-) create mode 100644 changelog.d/fixes/6906-responses-usage-trailing.md create mode 100644 open-sse/translator/response/openai-responses/eventEmitter.ts create mode 100644 tests/unit/responses-usage-trailing-6906.test.ts diff --git a/changelog.d/fixes/6906-responses-usage-trailing.md b/changelog.d/fixes/6906-responses-usage-trailing.md new file mode 100644 index 0000000000..bfa9691b4b --- /dev/null +++ b/changelog.d/fixes/6906-responses-usage-trailing.md @@ -0,0 +1 @@ +- fix(sse): defer `response.completed` until a trailing usage-only chunk arrives on `/v1/responses` streams (#6906) diff --git a/open-sse/transformer/responsesTransformer.ts b/open-sse/transformer/responsesTransformer.ts index afe245a019..ab8ce9596f 100644 --- a/open-sse/transformer/responsesTransformer.ts +++ b/open-sse/transformer/responsesTransformer.ts @@ -108,6 +108,10 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv completedSent: false, usage: null, keepaliveTimer: null, + // #6906: true once a finish_reason chunk closed all output items but deferred + // response.completed — a trailing usage-only chunk (choices: [], usage: {...}) may + // still arrive for stream_options.include_usage=true upstreams. + awaitingTrailingUsage: false, }; const encoder = new TextEncoder(); @@ -403,6 +407,11 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv if (parsed.usage) { state.usage = parsed.usage; } + // #6906: trailing usage-only chunk after finish_reason already deferred + // completion — send it now with the usage just captured above. + if (state.awaitingTrailingUsage && !state.completedSent) { + sendCompleted(controller); + } continue; } @@ -630,7 +639,18 @@ export function createResponsesApiTransformStream(logger = null, keepaliveInterv for (const i in state.msgItemAdded) closeMessage(controller, i); closeReasoning(controller); for (const i in state.funcCallIds) closeToolCall(controller, i); - sendCompleted(controller); + if (state.usage) { + // Usage already captured — either it arrived in this same chunk, or an + // earlier usage-bearing chunk already populated state.usage. Either way + // there is nothing left to wait for, so complete right away. + sendCompleted(controller); + } else { + // #6906: defer response.completed — a trailing usage-only chunk may + // still arrive (stream_options.include_usage=true). The empty-choices + // branch above (or flush() at stream end, as a fallback) actually + // calls sendCompleted(). + state.awaitingTrailingUsage = true; + } } } }, diff --git a/open-sse/translator/response/openai-responses.ts b/open-sse/translator/response/openai-responses.ts index 06f6b47d06..fe95ee7599 100644 --- a/open-sse/translator/response/openai-responses.ts +++ b/open-sse/translator/response/openai-responses.ts @@ -14,6 +14,7 @@ import { normalizeUpstreamFailure, extractResponsesReasoningSummaryText, } from "./openai-responses/pureHelpers.ts"; +import { createEventEmitter } from "./openai-responses/eventEmitter.ts"; // normalizeUpstreamFailure is re-exported for external importers (tests). export { normalizeUpstreamFailure } from "./openai-responses/pureHelpers.ts"; @@ -93,16 +94,19 @@ export function openaiToOpenAIResponsesResponse(chunk, state) { } if (!chunk.choices?.length) { + // #6906: a deferred finish_reason (awaitingTrailingUsage, see below) completes here — + // the trailing usage-only chunk (choices: [], usage: {...}) is what real + // stream_options.include_usage=true upstreams send after finish_reason (see the + // "READ THIS" block in stream.ts); state.usage was already captured above. + if (state.awaitingTrailingUsage && !state.completedSent) { + const { events, emit } = createEventEmitter(state); + sendCompleted(state, emit); + return events; + } return []; } - const events = []; - const nextSeq = () => ++state.seq; - - const emit = (eventType, data) => { - data.sequence_number = nextSeq(); - events.push({ event: eventType, data }); - }; + const { events, emit } = createEventEmitter(state); const choice = chunk.choices[0]; const idx = choice.index || 0; @@ -215,7 +219,13 @@ export function openaiToOpenAIResponsesResponse(chunk, state) { for (const i in state.msgItemAdded) closeMessage(state, emit, i); closeReasoning(state, emit); for (const i in state.funcCallIds) closeToolCall(state, emit, i); - sendCompleted(state, emit); + // #6906: usage already captured (same chunk or earlier) completes now; otherwise + // defer for a trailing usage-only chunk, handled above and in flushEvents(). + if (state.usage) { + sendCompleted(state, emit); + } else { + state.awaitingTrailingUsage = true; + } } return events; @@ -589,12 +599,7 @@ function sendCompleted(state, emit) { function flushEvents(state) { if (state.completedSent) return []; - const events = []; - const nextSeq = () => ++state.seq; - const emit = (eventType, data) => { - data.sequence_number = nextSeq(); - events.push({ event: eventType, data }); - }; + const { events, emit } = createEventEmitter(state); for (const i in state.msgItemAdded) closeMessage(state, emit, i); closeReasoning(state, emit); diff --git a/open-sse/translator/response/openai-responses/eventEmitter.ts b/open-sse/translator/response/openai-responses/eventEmitter.ts new file mode 100644 index 0000000000..1db4e29e26 --- /dev/null +++ b/open-sse/translator/response/openai-responses/eventEmitter.ts @@ -0,0 +1,12 @@ +// Shared event-collector for the OpenAI Responses response translator's three emission +// sites (main per-chunk pass, the deferred-completion empty-choices branch introduced by +// #6906, and flushEvents at stream end) — avoids duplicating the events/emit boilerplate. +// Carries stream state (state.seq), so it lives outside the stateless pureHelpers.ts leaf. +export function createEventEmitter(state: { seq: number }) { + const events: Array<{ event: string; data: Record }> = []; + const emit = (eventType: string, data: Record) => { + data.sequence_number = ++state.seq; + events.push({ event: eventType, data }); + }; + return { events, emit }; +} diff --git a/tests/unit/responses-translation-fixes.test.ts b/tests/unit/responses-translation-fixes.test.ts index 61cf4b23f1..c29641c6a3 100644 --- a/tests/unit/responses-translation-fixes.test.ts +++ b/tests/unit/responses-translation-fixes.test.ts @@ -458,7 +458,10 @@ test("Chat→Responses streaming: completed event includes accumulated output", // Finish const finishChunk = { choices: [{ index: 0, delta: {}, finish_reason: "stop" }] }; - const events = openaiToOpenAIResponsesResponse(finishChunk, state); + openaiToOpenAIResponsesResponse(finishChunk, state); + // #6906: no usage was ever sent for this stream, so response.completed is deferred + // until the stream-end flush (no trailing usage-only chunk will ever arrive). + const events = openaiToOpenAIResponsesResponse(null, state); const completedEvent = events.find((e) => e.event === "response.completed"); assert.ok(completedEvent.data.response.output, "completed should have output"); assert.ok(completedEvent.data.response.output.length > 0, "output should not be empty"); diff --git a/tests/unit/responses-usage-trailing-6906.test.ts b/tests/unit/responses-usage-trailing-6906.test.ts new file mode 100644 index 0000000000..5bf66b0cb4 --- /dev/null +++ b/tests/unit/responses-usage-trailing-6906.test.ts @@ -0,0 +1,133 @@ +// Regression guard for #6906: response.completed must carry usage even when +// the upstream sends a trailing usage-only chunk AFTER the finish_reason chunk +// (the real-world OpenAI-compatible `stream_options.include_usage: true` order). +import test from "node:test"; +import assert from "node:assert/strict"; + +import { openaiToOpenAIResponsesResponse } from "../../open-sse/translator/response/openai-responses.ts"; +import { initState } from "../../open-sse/translator/index.ts"; +import { FORMATS } from "../../open-sse/translator/formats.ts"; +import { createResponsesApiTransformStream } from "../../open-sse/transformer/responsesTransformer.ts"; + +function collectLiveTranslatorEvents(chunks) { + const state = initState(FORMATS.OPENAI_RESPONSES); + const events = []; + for (const chunk of chunks) { + const result = openaiToOpenAIResponsesResponse(chunk, state); + if (result) events.push(...result); + } + return events; +} + +test("BUG #6906: live translator — response.completed carries usage when the usage-only chunk arrives AFTER finish_reason", () => { + const events = collectLiveTranslatorEvents([ + { + id: "chatcmpl-1", + model: "gpt-4.1", + choices: [{ index: 0, delta: { content: "Hello" }, finish_reason: null }], + }, + // finish_reason chunk arrives BEFORE the usage-only chunk (real-world order) + { + id: "chatcmpl-1", + model: "gpt-4.1", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + usage: null, + }, + // trailing usage-only chunk per stream_options.include_usage=true semantics + { + id: "chatcmpl-1", + model: "gpt-4.1", + choices: [], + usage: { prompt_tokens: 2249, completion_tokens: 123, total_tokens: 2372 }, + }, + ]); + + const completedEvent = events.find((event) => event.event === "response.completed"); + assert.ok(completedEvent, "response.completed event should be emitted"); + assert.deepEqual( + completedEvent.data.response.usage, + { input_tokens: 2249, output_tokens: 123, total_tokens: 2372 }, + "response.completed must carry usage even when the usage-only chunk trails finish_reason" + ); +}); + +test("BUG #6906 (flush fallback): live translator still emits response.completed on stream end when no trailing usage chunk ever arrives", () => { + const state = initState(FORMATS.OPENAI_RESPONSES); + const events = []; + const chunks = [ + { + id: "chatcmpl-2", + model: "gpt-4.1", + choices: [{ index: 0, delta: { content: "Hi" }, finish_reason: null }], + }, + { + id: "chatcmpl-2", + model: "gpt-4.1", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + }, + ]; + for (const chunk of chunks) { + const result = openaiToOpenAIResponsesResponse(chunk, state); + if (result) events.push(...result); + } + // stream end (flush) + const flushed = openaiToOpenAIResponsesResponse(null, state); + if (flushed) events.push(...flushed); + + const completedEvent = events.find((event) => event.event === "response.completed"); + assert.ok(completedEvent, "response.completed event should be emitted on flush"); +}); + +test("BUG #6906: legacy transformer — response.completed carries usage when the usage-only chunk arrives AFTER finish_reason", async () => { + const chunks = [ + { + id: "chatcmpl-3", + model: "gpt-4.1", + choices: [{ index: 0, delta: { content: "Hello" }, finish_reason: null }], + }, + { + id: "chatcmpl-3", + model: "gpt-4.1", + choices: [{ index: 0, delta: {}, finish_reason: "stop" }], + usage: null, + }, + { + id: "chatcmpl-3", + model: "gpt-4.1", + choices: [], + usage: { prompt_tokens: 55, completion_tokens: 11, total_tokens: 66 }, + }, + ]; + + const encoder = new TextEncoder(); + const decoder = new TextDecoder(); + const readable = new ReadableStream({ + start(controller) { + for (const chunk of chunks) { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`)); + } + controller.enqueue(encoder.encode("data: [DONE]\n\n")); + controller.close(); + }, + }); + + const transformed = readable.pipeThrough(createResponsesApiTransformStream(null, 60000)); + const reader = transformed.getReader(); + let full = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + full += decoder.decode(value, { stream: true }); + } + + // Events are emitted as `event: \ndata: \n\n` blocks. + const blocks = full.split("\n\n").filter((block) => block.includes("event: response.completed")); + assert.ok(blocks.length > 0, "response.completed event should be emitted"); + const dataLine = blocks[0].split("\n").find((line) => line.startsWith("data:")); + const payload = JSON.parse(dataLine.replace(/^data:\s*/, "")); + assert.deepEqual( + payload.response.usage, + { prompt_tokens: 55, completion_tokens: 11, total_tokens: 66 }, + "legacy transformer response.completed must carry usage even when the usage-only chunk trails finish_reason" + ); +}); diff --git a/tests/unit/translator-openai-responses-custom-tool-1007.test.ts b/tests/unit/translator-openai-responses-custom-tool-1007.test.ts index dc21c8c934..cfd47fe597 100644 --- a/tests/unit/translator-openai-responses-custom-tool-1007.test.ts +++ b/tests/unit/translator-openai-responses-custom-tool-1007.test.ts @@ -130,6 +130,9 @@ test("OpenAI -> Responses: apply_patch streams as custom_tool_call with raw inpu }, ], }, + // #6906: real providers may send no separate usage chunk at all — the stream-end + // flush is what finalizes response.completed in that case. + null, ]); const added = events.find((e) => e.event === "response.output_item.added"); diff --git a/tests/unit/translator-resp-openai-responses.test.ts b/tests/unit/translator-resp-openai-responses.test.ts index dc7132ef85..3659a82fb5 100644 --- a/tests/unit/translator-resp-openai-responses.test.ts +++ b/tests/unit/translator-resp-openai-responses.test.ts @@ -611,6 +611,9 @@ test("OpenAI -> Responses: parallel tool calls with mixed content survive transl }, ], }, + // #6906: real providers may send no separate usage chunk at all — the stream-end + // flush is what finalizes response.completed in that case. + null, ]); const doneEvents = events.filter(