fix(sse): emit trailing usage estimate in translate streams when upstream stays silent (#12828)

Validado numa worktree combinada com a onda de streaming desta leva sobre `release/v3.8.51`: typecheck:core limpo, check-api-typecheck OK (289), check-file-size OK após rebaseline, 88/88 nos testes focados.

O #12151 cobriu só metade: passthrough emitia o chunk final de usage, translate calculava a estimativa **depois** de fechar o stream, então o número só chegava ao log do servidor e nunca ao cliente. Fechar essa metade é o que faz a feature existir de fato.

Não emitir segundo chunk quando o upstream já mandou usage real é o detalhe que impede a correção de virar contagem dobrada.

changelog.d/fixes/12828-translate-usage-trailing-chunk.md

@@ -0,0 +1 @@
+- **fix(sse):** translate-mode streams now emit the estimated token counts as a trailing usage-only chunk before `[DONE]` when the upstream stays silent, so metered chat clients see totals instead of nothing ([#12828](https://github.com/diegosouzapw/OmniRoute/pull/12828)) — thanks @maxmad64bis

config/quality/file-size-baseline.json

@@ -1,4 +1,5 @@
 {
+  "_rebaseline_2026_09_10_12828_translate_usage_chunk": "PR #12828 own growth: open-sse/utils/stream.ts 3072->3080 (+8). Translate-mode streams now send the estimated usage as the canonical trailing usage-only chunk before [DONE] when the upstream stays silent (parity with the #12151 passthrough flush), with a latch so a finish chunk that already carried the estimate is not doubled. The chunk builder is shared with the passthrough flush in open-sse/utils/usageOnlyChunk.ts (under cap); what remains is the flush-site wiring. Covered by tests/unit/stream-translate-usage-trailing.test.ts.",
   "_rebaseline_2026_09_06_runtime_quotagroup_nodemap": "Own growth: src/app/(dashboard)/dashboard/runtime/RuntimePageClient.tsx 1201->1222 (+21, check-file-size split-newline). QuotaGroup is a module-level sibling and was reading nodeMap from RuntimePageClient's closure; that identifier is not in scope, so a quota monitor with status error/exhausted/alerting throws ReferenceError. Fix threads nodeMap as a prop (3 call sites + parameter + ProviderNodeEntry import). Prettier wraps the long import and the three QuotaGroup JSX tags. Covered by tests/unit/ui/runtime-page-client.test.tsx (empty monitors stay green; error+exhausted fixtures mount QuotaGroup).",
   "_rebaseline_2026_09_05_claude_extra_usage_preflight": "Own growth: open-sse/services/combo.ts 4080->4084 (+4). buildAutoCandidates now forwards connection.providerSpecificData into evaluateQuotaCutoff so a Claude account with blockExtraUsage=false is not dropped at the 5h bar. Irreducible at the existing cutoff call site; the helper lives in claudeExtraUsage.ts (under cap). Covered by tests/unit/quota-preflight.test.ts.",
   "_rebaseline_2026_09_04_12697_combo_pin_allowlist": "PR #12697 own growth: src/sse/handlers/chat.ts 2454->2458 (+4). checkModelAvailable preflight and handleSingleModelChat now call comboPinAllowlist so a pin-only combo step cannot scan the provider pool after 502/429. Helper lives in src/lib/combos/steps.ts under cap. Covered by tests/unit/combo-pin-implicit-allowlist.test.ts (11/11).",
@@ -436,7 +437,7 @@
     "open-sse/translator/response/openai-responses.ts": 1466,
     "open-sse/utils/cursorAgentProtobuf.ts": 1547,
     "open-sse/utils/proxyFetch.ts": 1271,
-    "open-sse/utils/stream.ts": 3072,
+    "open-sse/utils/stream.ts": 3080,
     "open-sse/vendor/codex-chatgpt-web/adapters/chatgpt-web/browser-worker.ts": 4398,
     "open-sse/vendor/codex-chatgpt-web/bridge.ts": 1335,
     "src/app/(dashboard)/dashboard/HomePageClient.tsx": 1344,

open-sse/utils/stream.ts

@@ -88,6 +88,7 @@ import { restoreClaudeToolName } from "../services/claudeCodeToolRemapper.ts";
 import { normalizeFinalOpenAIStreamChunk } from "./openAIStreamChunk.ts";
 import { collectClaudeDelta } from "./streamClaudeDelta.ts";
 import { createStreamTiming, type StreamTiming } from "./streamTiming.ts";
+import { buildUsageOnlyChunk } from "./usageOnlyChunk.ts";
 
 /**
  * Race a response body read against a timeout.
@@ -768,6 +769,8 @@ export function createSSEStream(options: StreamOptions = {}) {
   let passthroughBufferedTextualToolCallContent = "";
   /** Passthrough: whether a usage block was already forwarded to the client (prevents double). */
   let passthroughForwardedUsage = false;
+  /** Translate: usage already reached the client, or no trailing usage chunk applies. */
+  let translateForwardedUsage = sourceFormat !== FORMATS.OPENAI || !shouldEmitDoneTerminator;
   // Passthrough Responses SSE: snapshots of items seen via `response.output_item.done`,
   // used to backfill `response.completed.response.output` when upstream returns it
   // empty (which happens when `store: false` — see backfillResponsesCompletedOutput).
@@ -1038,9 +1041,11 @@ export function createSSEStream(options: StreamOptions = {}) {
       const estimated = estimateUsage(body, totalContentLength, sourceFormat);
       itemSanitized.usage = timing.withTps(filterUsageForFormat(estimated, sourceFormat));
       state.usage = estimated;
+      if (hasValidUsage(estimated)) translateForwardedUsage = true; // finish chunk carries it
     } else if (state?.finishReason && isFinishChunk && state.usage) {
       const buffered = addBufferToUsage(state.usage);
       itemSanitized.usage = timing.withTps(filterUsageForFormat(buffered, sourceFormat));
+      translateForwardedUsage = true;
     }
 
     if (
@@ -2565,14 +2570,11 @@ export function createSSEStream(options: StreamOptions = {}) {
               // upstream DID send usage (trailing or in-band), it was forwarded
               // already and passthroughForwardedUsage guards this off.
               if (shouldEmitDoneTerminator && !passthroughForwardedUsage && hasValidUsage(usage)) {
-                const usageOnlyChunk = {
-                  id: passthroughLastChatId ?? passthroughResponsesId ?? `chatcmpl-${Date.now()}`,
-                  object: "chat.completion.chunk",
-                  created: Math.floor(Date.now() / 1000),
+                const usageOnlyChunk = buildUsageOnlyChunk(
+                  passthroughLastChatId ?? passthroughResponsesId,
                   model,
-                  choices: [],
-                  usage: timing.withTps(filterUsageForFormat(usage, sourceFormat || FORMATS.OPENAI)),
-                };
+                  timing.withTps(filterUsageForFormat(usage, sourceFormat || FORMATS.OPENAI))
+                );
                 const usageOutput = `data: ${JSON.stringify(usageOnlyChunk)}\n\n`;
                 reqLogger?.appendConvertedChunk?.(usageOutput);
                 forward(controller, encoder.encode(usageOutput));
@@ -2842,8 +2844,26 @@ export function createSSEStream(options: StreamOptions = {}) {
            * emitted once at stream end when merged into the final translated chunk.
            */
 
+          // Estimate usage if provider didn't return valid usage (for translate mode)
+          if (!hasValidUsage(state?.usage) && totalContentLength > 0) {
+            state.usage = estimateUsage(body, totalContentLength, sourceFormat);
+          }
+
           // Send [DONE] (only if not already sent during transform)
           if (!doneSent) {
+            // Upstream stayed silent on usage: send the estimate as the canonical
+            // trailing usage-only chunk before [DONE], like the passthrough flush.
+            if (!translateForwardedUsage && hasValidUsage(state?.usage)) {
+              const usageOnlyChunk = buildUsageOnlyChunk(
+                (state as unknown as Record<string, unknown>)?.chatId,
+                model,
+                timing.withTps(filterUsageForFormat(state.usage, sourceFormat))
+              );
+              const usageOutput = `data: ${JSON.stringify(usageOnlyChunk)}\n\n`;
+              reqLogger?.appendConvertedChunk?.(usageOutput);
+              forward(controller, encoder.encode(usageOutput));
+              clientPayloadCollector.push(usageOnlyChunk);
+            }
             await emitFinalSseMetadata(controller, state?.usage as Record<string, unknown> | null);
             doneSent = true;
             if (shouldEmitDoneTerminator) {
@@ -2854,11 +2874,6 @@ export function createSSEStream(options: StreamOptions = {}) {
             }
           }
 
-          // Estimate usage if provider didn't return valid usage (for translate mode)
-          if (!hasValidUsage(state?.usage) && totalContentLength > 0) {
-            state.usage = estimateUsage(body, totalContentLength, sourceFormat);
-          }
-
           if (hasValidUsage(state?.usage)) {
             logUsage(state.provider || targetFormat, state.usage, model, connectionId, apiKeyInfo);
           } else {

open-sse/utils/usageOnlyChunk.ts

@@ -0,0 +1,15 @@
+/**
+ * Canonical OpenAI trailing usage-only chunk (`choices: []`) sent before `[DONE]`
+ * when the upstream never reported usage, so metered chat clients still see
+ * token counts (#12151). Shared by the passthrough and translate flushes.
+ */
+export function buildUsageOnlyChunk(id: unknown, model: unknown, usage: unknown) {
+  return {
+    id: id ?? `chatcmpl-${Date.now()}`,
+    object: "chat.completion.chunk",
+    created: Math.floor(Date.now() / 1000),
+    model,
+    choices: [],
+    usage,
+  };
+}

tests/unit/stream-translate-usage-trailing.test.ts

@@ -0,0 +1,151 @@
+// Translate-mode trailing usage: when a chat client talks to a Responses
+// upstream that stays silent on usage, the flush must emit the estimate as a
+// canonical trailing usage-only chunk (empty choices) before [DONE] — the only
+// token counts an OpenAI-compatible client ever reads.
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { createSSEStream } from "../../open-sse/utils/stream.ts";
+import { FORMATS } from "../../open-sse/translator/formats.ts";
+
+const enc = new TextEncoder();
+
+async function readStream(stream: TransformStream<Uint8Array, Uint8Array>): Promise<string> {
+  const reader = stream.readable.getReader();
+  const chunks: Uint8Array[] = [];
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    chunks.push(value);
+  }
+  return Buffer.concat(chunks).toString("utf8");
+}
+
+async function runTranslate(upstreamLines: string[]): Promise<string> {
+  const stream = createSSEStream({
+    mode: "translate" as const,
+    targetFormat: FORMATS.OPENAI_RESPONSES,
+    sourceFormat: FORMATS.OPENAI,
+    provider: "testprov",
+    model: "m",
+    body: {
+      model: "m",
+      messages: [{ role: "user", content: "hello world, please answer at length" }],
+    },
+  });
+  const writer = stream.writable.getWriter();
+  const reading = readStream(stream);
+  for (const line of upstreamLines) {
+    await writer.write(enc.encode(line));
+  }
+  await writer.close();
+  return reading;
+}
+
+function parseDataPayloads(text: string): unknown[] {
+  return text
+    .split("\n")
+    .filter((line) => line.startsWith("data: "))
+    .map((line) => line.slice("data: ".length))
+    .filter((data) => data && data !== "[DONE]")
+    .map((data) => JSON.parse(data));
+}
+
+test("translate silent upstream emits a canonical usage-only chunk before [DONE]", async () => {
+  // Comments stay disabled (default): the assertion reads the real client
+  // data flux, never the opt-in SSE metadata comments.
+  const text = await runTranslate([
+    `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "Hello there, this is a streamed answer." })}\n\n`,
+    "data: [DONE]\n\n",
+  ]);
+
+  assert.ok(text.includes("data: [DONE]"), `stream must still terminate, got: ${text.slice(-300)}`);
+  const payloads = parseDataPayloads(text);
+  const usageChunks = payloads.filter(
+    (p) =>
+      p &&
+      typeof p === "object" &&
+      Array.isArray((p as Record<string, unknown>).choices) &&
+      (p as Record<string, unknown>).choices.length === 0 &&
+      (p as Record<string, unknown>).usage != null
+  );
+  assert.equal(
+    usageChunks.length,
+    1,
+    `expected exactly one usage-only chunk, got ${usageChunks.length} — flux: ${text.slice(-800)}`
+  );
+  const usage = (usageChunks[0] as Record<string, unknown>).usage as Record<string, unknown>;
+  assert.ok(
+    typeof usage.prompt_tokens === "number" && usage.prompt_tokens > 0,
+    `estimated usage must carry input tokens: ${JSON.stringify(usage)}`
+  );
+  assert.ok(
+    typeof usage.completion_tokens === "number" && usage.completion_tokens > 0,
+    `estimated usage must carry output tokens: ${JSON.stringify(usage)}`
+  );
+  const doneIdx = text.indexOf("data: [DONE]");
+  const usageIdx = text.lastIndexOf('"usage"');
+  assert.ok(usageIdx > 0 && usageIdx < doneIdx, "usage-only chunk must precede [DONE]");
+});
+
+test("translate real upstream usage is forwarded without a duplicate estimate", async () => {
+  const text = await runTranslate([
+    `data: ${JSON.stringify({ type: "response.output_text.delta", delta: "Hello there" })}\n\n`,
+    `data: ${JSON.stringify({ type: "response.completed", response: { usage: { input_tokens: 2249, output_tokens: 123, total_tokens: 2372 } } })}\n\n`,
+    "data: [DONE]\n\n",
+  ]);
+
+  const payloads = parseDataPayloads(text);
+  const withUsage = payloads.filter(
+    (p) => p && typeof p === "object" && (p as Record<string, unknown>).usage != null
+  );
+  assert.equal(
+    withUsage.length,
+    1,
+    `real upstream usage must reach the client exactly once, got ${withUsage.length} — usages: ${JSON.stringify(withUsage.map((p) => (p as Record<string, unknown>).usage))}`
+  );
+  const forwarded = (withUsage[0] as Record<string, unknown>).usage as Record<string, unknown>;
+  const prompt = forwarded.prompt_tokens ?? forwarded.input_tokens;
+  const completion = forwarded.completion_tokens ?? forwarded.output_tokens;
+  assert.equal(prompt, 2249);
+  assert.equal(completion, 123);
+  assert.equal(forwarded.estimated, undefined);
+});
+
+test("passthrough silent upstream still emits its trailing usage-only chunk", async () => {
+  const body = {
+    model: "m",
+    messages: [{ role: "user", content: "hi" }],
+    stream: true,
+    stream_options: { include_usage: true },
+  };
+  const stream = createSSEStream({
+    mode: "passthrough" as const,
+    body,
+    sourceFormat: FORMATS.OPENAI,
+    clientResponseFormat: FORMATS.OPENAI,
+    provider: "test",
+  });
+  const writer = stream.writable.getWriter();
+  const reading = readStream(stream);
+  await writer.write(
+    enc.encode(
+      `data: ${JSON.stringify({ id: "chatcmpl-1", object: "chat.completion.chunk", choices: [{ index: 0, delta: { content: "hello world" }, finish_reason: null }] })}\n\n`
+    )
+  );
+  await writer.write(
+    enc.encode(
+      `data: ${JSON.stringify({ id: "chatcmpl-1", object: "chat.completion.chunk", choices: [{ index: 0, delta: {}, finish_reason: "stop" }] })}\n\n`
+    )
+  );
+  await writer.write(enc.encode("data: [DONE]\n\n"));
+  await writer.close();
+  const text = await reading;
+
+  const payloads = parseDataPayloads(text);
+  const withUsage = payloads.filter(
+    (p) => p && typeof p === "object" && (p as Record<string, unknown>).usage != null
+  );
+  assert.ok(withUsage.length >= 1, `passthrough must still emit usage, got: ${text.slice(0, 600)}`);
+  const last = withUsage[withUsage.length - 1] as Record<string, unknown>;
+  assert.equal((last.usage as Record<string, unknown>).estimated, true);
+});