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.

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)

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;
+            }
           }
         }
       },

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);

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<string, unknown> }> = [];
+  const emit = (eventType: string, data: Record<string, unknown>) => {
+    data.sequence_number = ++state.seq;
+    events.push({ event: eventType, data });
+  };
+  return { events, emit };
+}

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");

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: <type>\ndata: <json>\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"
+  );
+});

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");

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(