fix(responses): normalize terminal usage for Codex (#9192)

* fix(responses): normalize terminal usage for Codex

* refactor(responses): reduce stream gate growth

* refactor(responses): keep stream within size ratchet

---------

Co-authored-by: diegosouzapw <diegosouzapw@users.noreply.github.com>

open-sse/utils/passthroughTailProcessor.ts

@@ -3,6 +3,7 @@ import { parseSSEDataPayload } from "./streamHelpers.ts";
 import {
   backfillResponsesCompletedOutput,
   normalizeResponsesSseIds,
+  normalizeResponsesCompletedUsage,
   pushUniqueResponsesOutputItems,
   stringifyIdValue,
   stripResponsesLifecycleEcho,
@@ -174,13 +175,20 @@ function handleResponsesTailPayload(
   const outputPayload = textualToolCallBackfilled
     ? context.toResponsesCompletedWithToolCalls(parsed)
     : parsed;
+  const usageNormalized = normalizeResponsesCompletedUsage(outputPayload);
   const stripped = stripResponsesLifecycleEcho(outputPayload);
   const backfilled = backfillResponsesCompletedOutput(
     outputPayload,
     context.passthroughResponsesOutputItems
   );
 
-  if (stripped || backfilled || textualToolCallBackfilled || responsesIdsNormalized) {
+  if (
+    stripped ||
+    backfilled ||
+    textualToolCallBackfilled ||
+    responsesIdsNormalized ||
+    usageNormalized
+  ) {
     output = `data: ${JSON.stringify(outputPayload)}\n\n`;
   }
 

open-sse/utils/responsesStreamHelpers.ts

@@ -138,6 +138,49 @@ export function backfillResponsesCompletedOutput(
   return true;
 }
 
+/**
+ * Keep the terminal Responses payload compatible with strict clients such as Codex.
+ * Upstreams may expose only input/output counts (or omit usage entirely), while the
+ * client deserializer requires all three canonical token fields.
+ */
+export function normalizeResponsesCompletedUsage(parsed: unknown): boolean {
+  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return false;
+  const obj = parsed as JsonRecord;
+  if (obj.type !== "response.completed") return false;
+  if (!obj.response || typeof obj.response !== "object" || Array.isArray(obj.response)) {
+    return false;
+  }
+
+  const response = obj.response as JsonRecord;
+  const current =
+    response.usage && typeof response.usage === "object" && !Array.isArray(response.usage)
+      ? (response.usage as JsonRecord)
+      : {};
+  const finiteNumber = (value: unknown): number | null => {
+    const numeric = Number(value);
+    return Number.isFinite(numeric) ? numeric : null;
+  };
+  const inputTokens =
+    finiteNumber(current.input_tokens) ?? finiteNumber(current.prompt_tokens) ?? 0;
+  const outputTokens =
+    finiteNumber(current.output_tokens) ?? finiteNumber(current.completion_tokens) ?? 0;
+  const totalTokens = finiteNumber(current.total_tokens) ?? inputTokens + outputTokens;
+
+  const normalized: JsonRecord = {
+    ...current,
+    input_tokens: inputTokens,
+    output_tokens: outputTokens,
+  };
+  normalized.total_tokens = totalTokens;
+  const changed =
+    !response.usage ||
+    current.input_tokens !== inputTokens ||
+    current.output_tokens !== outputTokens ||
+    current.total_tokens !== totalTokens;
+  response.usage = normalized;
+  return changed;
+}
+
 const RESPONSES_LIFECYCLE_EVENT_TYPES = new Set([
   "response.created",
   "response.in_progress",

open-sse/utils/stream.ts

@@ -49,6 +49,7 @@ import {
 } from "../services/sessionManager.ts";
 import {
   backfillResponsesCompletedOutput,
+  normalizeResponsesCompletedUsage as normalizeUsage,
   normalizeResponsesSseIds,
   pushUniqueResponsesOutputItems,
   stringifyIdValue,
@@ -1585,7 +1586,8 @@ export function createSSEStream(options: StreamOptions = {}) {
                     stripped ||
                     backfilled ||
                     textualToolCallBackfilled ||
-                    responsesIdsNormalized
+                    responsesIdsNormalized ||
+                    normalizeUsage(parsed)
                   ) {
                     output = `data: ${JSON.stringify(parsed)}\n\n`;
                     injectedUsage = true;
@@ -2273,7 +2275,6 @@ export function createSSEStream(options: StreamOptions = {}) {
                 }
                 clientPayloadCollector.push(bufferedPayload);
 
-                // Normalize numeric IDs for final buffered data: chunk (same as transform path)
                 if (typeof bufferedPayload === "object" && !Array.isArray(bufferedPayload)) {
                   const flushedParsed = bufferedPayload as JsonRecord;
                   const flushedType =
@@ -2281,7 +2282,7 @@ export function createSSEStream(options: StreamOptions = {}) {
                   const isResponses = flushedType.startsWith("response.");
                   const isClaude = isClaudeEventPayload(flushedParsed);
                   if (isResponses) {
-                    if (normalizeResponsesSseIds(flushedParsed)) {
+                    if (normalizeResponsesSseIds(flushedParsed) || normalizeUsage(flushedParsed)) {
                       output = `data: ${JSON.stringify(flushedParsed)}\n\n`;
                     }
                   } else if (!isClaude) {

tests/unit/responses-commentary-passthrough-6199.test.ts

@@ -173,10 +173,7 @@ test("commentary-phase output text is NOT forwarded when dropping is enabled (#6
     "commentary-phase text must be dropped from the passthrough stream"
   );
   // The commentary item announcement / completion must not leak either.
-  assert.ok(
-    !output.includes("msg_commentary"),
-    "commentary item events must be dropped entirely"
-  );
+  assert.ok(!output.includes("msg_commentary"), "commentary item events must be dropped entirely");
   // The real answer must always be forwarded.
   assert.ok(output.includes(FINAL_TEXT), "the final answer text must be forwarded");
   assert.ok(output.includes("msg_final"), "the final answer item must be forwarded");
@@ -194,3 +191,109 @@ test("commentary passes through when dropping is disabled (gate/regression) (#61
   );
   assert.ok(output.includes(FINAL_TEXT), "the final answer text must still be forwarded");
 });
+
+test("response.completed always includes total_tokens for strict Codex clients", async () => {
+  const output = await readTransformed(
+    [
+      sse({
+        type: "response.completed",
+        response: {
+          id: "resp_codex_usage",
+          status: "completed",
+          output: [],
+          usage: {
+            input_tokens: 88,
+            output_tokens: 6,
+            input_tokens_details: { cached_tokens: 0 },
+            output_tokens_details: { reasoning_tokens: 0 },
+          },
+        },
+      }),
+    ],
+    PASSTHROUGH_RESPONSES_OPTIONS
+  );
+
+  const completedLine = output
+    .split(/\r?\n/)
+    .find((line) => line.startsWith("data:") && line.includes('"response.completed"'));
+  assert.ok(completedLine, "the terminal Responses event must be forwarded");
+
+  const completed = JSON.parse(completedLine.slice(5).trim());
+  assert.deepEqual(completed.response.usage, {
+    input_tokens: 88,
+    output_tokens: 6,
+    total_tokens: 94,
+    input_tokens_details: { cached_tokens: 0 },
+    output_tokens_details: { reasoning_tokens: 0 },
+  });
+});
+
+test("response.completed synthesizes zero usage when upstream omits usage", async () => {
+  const completed = {
+    type: "response.completed",
+    response: { id: "resp_codex_no_usage", status: "completed", output: [] },
+  };
+  const output = await readTransformed(
+    [`data: ${JSON.stringify(completed)}`],
+    PASSTHROUGH_RESPONSES_OPTIONS
+  );
+  const completedLine = output
+    .split(/\r?\n/)
+    .find((line) => line.startsWith("data:") && line.includes('"response.completed"'));
+  assert.ok(completedLine, "the terminal Responses event must be forwarded");
+  const forwarded = JSON.parse(completedLine.slice(5).trim());
+  assert.deepEqual(forwarded.response.usage, {
+    input_tokens: 0,
+    output_tokens: 0,
+    total_tokens: 0,
+  });
+});
+
+test("Claude to Responses translation includes canonical Codex usage", async () => {
+  const output = await readTransformed(
+    [
+      sse({
+        type: "message_start",
+        message: {
+          id: "msg_agentrouter_claude",
+          type: "message",
+          role: "assistant",
+          model: "gpt-5.6-sol",
+          usage: { input_tokens: 88, output_tokens: 0 },
+        },
+      }),
+      sse({
+        type: "content_block_start",
+        index: 0,
+        content_block: { type: "text", text: "" },
+      }),
+      sse({
+        type: "content_block_delta",
+        index: 0,
+        delta: { type: "text_delta", text: "probe-ok" },
+      }),
+      sse({ type: "content_block_stop", index: 0 }),
+      sse({
+        type: "message_delta",
+        delta: { stop_reason: "end_turn", stop_sequence: null },
+        usage: { output_tokens: 6 },
+      }),
+      sse({ type: "message_stop" }),
+    ],
+    {
+      mode: "translate",
+      targetFormat: "claude",
+      sourceFormat: "openai-responses",
+      provider: "agentrouter",
+      body: { model: "agentrouter/gpt-5.6-sol" },
+    }
+  );
+  const completedLine = output
+    .split(/\r?\n/)
+    .find((line) => line.startsWith("data:") && line.includes('"response.completed"'));
+  assert.ok(completedLine, "the translated terminal event must be emitted");
+  const completed = JSON.parse(completedLine.slice(5).trim());
+  assert.equal(completed.response.usage.input_tokens, 88);
+  assert.equal(completed.response.usage.output_tokens, 6);
+  assert.equal(completed.response.usage.total_tokens, 94);
+});