From 426d8636bc152a58c5c288758a4466c9399b4ea1 Mon Sep 17 00:00:00 2001 From: diegosouzapw Date: Thu, 19 Mar 2026 11:02:13 -0300 Subject: [PATCH] fix(stream): extract usage from remaining buffer in flush handler (#480) --- open-sse/utils/stream.ts | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 72e1758c3c..e58adc0bf7 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -518,6 +518,33 @@ export function createSSEStream(options: StreamOptions = {}) { if (buffer.trim()) { const parsed = parseSSELine(buffer.trim()); if (parsed && !parsed.done) { + // Extract usage from remaining buffer — if the usage-bearing event + // (e.g. response.completed) is the last SSE line, it ends up here + // in the flush handler where extractUsage was not called. + // Non-destructive merge: some providers send usage across multiple + // events (e.g. prompt_tokens in message_start, completion_tokens + // in message_delta). Direct assignment would lose earlier data. + const extracted = extractUsage(parsed); + if (extracted) { + if (!state.usage) { + state.usage = extracted; + } else { + if (extracted.prompt_tokens > 0) + state.usage.prompt_tokens = extracted.prompt_tokens; + if (extracted.completion_tokens > 0) + state.usage.completion_tokens = extracted.completion_tokens; + if (extracted.total_tokens > 0) state.usage.total_tokens = extracted.total_tokens; + if (extracted.cache_read_input_tokens > 0) + state.usage.cache_read_input_tokens = extracted.cache_read_input_tokens; + if (extracted.cache_creation_input_tokens > 0) + state.usage.cache_creation_input_tokens = extracted.cache_creation_input_tokens; + if (extracted.cached_tokens > 0) + state.usage.cached_tokens = extracted.cached_tokens; + if (extracted.reasoning_tokens > 0) + state.usage.reasoning_tokens = extracted.reasoning_tokens; + } + } + const translated = translateResponse(targetFormat, sourceFormat, parsed, state); // Log OpenAI intermediate chunks