From 7b2e4b4837bf3e8dcc486ab2b24933c0b803d700 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza Date: Sun, 2 Aug 2026 02:29:21 -0300 Subject: [PATCH] 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 --- open-sse/utils/passthroughTailProcessor.ts | 10 +- open-sse/utils/responsesStreamHelpers.ts | 43 +++++++ open-sse/utils/stream.ts | 7 +- ...ponses-commentary-passthrough-6199.test.ts | 111 +++++++++++++++++- 4 files changed, 163 insertions(+), 8 deletions(-) diff --git a/open-sse/utils/passthroughTailProcessor.ts b/open-sse/utils/passthroughTailProcessor.ts index 845bc6e352..732e09cb4f 100644 --- a/open-sse/utils/passthroughTailProcessor.ts +++ b/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`; } diff --git a/open-sse/utils/responsesStreamHelpers.ts b/open-sse/utils/responsesStreamHelpers.ts index bbc2b14911..85eff8ccd4 100644 --- a/open-sse/utils/responsesStreamHelpers.ts +++ b/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", diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 884c510695..4b4243b8f2 100644 --- a/open-sse/utils/stream.ts +++ b/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) { diff --git a/tests/unit/responses-commentary-passthrough-6199.test.ts b/tests/unit/responses-commentary-passthrough-6199.test.ts index 5346f46615..cb7f34ec7f 100644 --- a/tests/unit/responses-commentary-passthrough-6199.test.ts +++ b/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); +});