diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index aea5d7d41a..16572b18b3 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -771,6 +771,7 @@ export function createSSEStream(options: StreamOptions = {}) { const passthroughResponsesOutputItems: unknown[] = []; const passthroughResponsesPendingFunctionCalls = new Map(); let passthroughResponsesId: string | null = null; + let passthroughLastChatId: string | null = null; let passthroughResponsesCurrentFunctionCallKey: string | null = null; const passthroughResponsesReasoningSummarySeen = new Set(); // #6199 — commentary-phase items announced via `response.output_item.added` are @@ -1955,6 +1956,16 @@ export function createSSEStream(options: StreamOptions = {}) { const isFinishChunk = parsed.choices?.[0]?.finish_reason; + // Remember the upstream's chat-completion id so synthetic chunks + // emitted at flush (e.g. the estimated usage-only chunk) carry the + // stream's real string id instead of null on the chat path + // (passthroughResponsesId is only ever set on the Responses path). + if (typeof parsed.id === "string" && parsed.id) { + passthroughLastChatId = parsed.id; + } else if (typeof parsed.id === "number") { + passthroughLastChatId = String(parsed.id); + } + if (isFinishChunk) { passthroughSawFinishReason = true; } @@ -1973,28 +1984,21 @@ export function createSSEStream(options: StreamOptions = {}) { parsed.choices[0].finish_reason !== "tool_calls" ) { parsed.choices[0].finish_reason = "tool_calls"; - // If we modify it, we must output the modified object - if (!injectedUsage && hasValidUsage(parsed.usage)) { - output = `data: ${JSON.stringify(parsed)}\n\n`; - injectedUsage = true; - } + // If we modify it, we must output the modified object. This used to + // piggyback on the estimated-usage rewrite below; with the estimate + // moved to flush() (#12151 follow-up) the rewrite must happen here. + // injectedUsage doubles as the "output already rewritten" latch — + // without it the raw line overwrites this rewrite further down. + output = `data: ${JSON.stringify(parsed)}\n\n`; + injectedUsage = true; } - if ( - isFinishChunk && - !passthroughForwardedUsage && - !hasValidUsage(parsed.usage) && - !hasValidUsage(usage) && - totalContentLength > 0 - ) { - const estimated = estimateUsage(body, totalContentLength, sourceFormat || FORMATS.OPENAI); - if (hasValidUsage(estimated)) { - parsed.usage = filterUsageForFormat(estimated, sourceFormat || FORMATS.OPENAI); - output = `data: ${JSON.stringify(parsed)}\n\n`; - usage = estimated; - passthroughForwardedUsage = true; - injectedUsage = true; - } - } else if (isFinishChunk && hasValidUsage(usage) && !passthroughForwardedUsage) { + // #12151 follow-up: do NOT inject estimated usage into the finish chunk. + // A genuine OpenAI upstream sends its usage in a trailing empty-choices + // chunk AFTER the finish; estimating here marked passthroughForwardedUsage + // and made the real trailing block get dropped in favor of the estimate + // (billing regression pinned by tests/unit/stream-utils.test.ts). The + // estimate is now emitted in flush(), only when the upstream stayed silent. + if (isFinishChunk && hasValidUsage(usage) && !passthroughForwardedUsage) { const buffered = addBufferToUsage(usage); parsed.usage = filterUsageForFormat(buffered, sourceFormat || FORMATS.OPENAI); output = `data: ${JSON.stringify(parsed)}\n\n`; @@ -2510,6 +2514,30 @@ export function createSSEStream(options: StreamOptions = {}) { forward(controller, encoder.encode(finishOutput)); clientPayloadCollector.push(syntheticFinishChunk); } + // #12151: upstream never reported usage — emit the estimate as a + // canonical OpenAI trailing usage-only chunk (empty choices) before + // [DONE], so metered clients still see token counts. When the + // 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), + model, + choices: [], + usage: filterUsageForFormat(usage, sourceFormat || FORMATS.OPENAI), + }; + const usageOutput = `data: ${JSON.stringify(usageOnlyChunk)}\n\n`; + reqLogger?.appendConvertedChunk?.(usageOutput); + forward(controller, encoder.encode(usageOutput)); + clientPayloadCollector.push(usageOnlyChunk); + passthroughForwardedUsage = true; + } await emitFinalSseMetadata(controller, usage); doneSent = true; if (shouldEmitDoneTerminator) { diff --git a/src/i18n/messages/en.json b/src/i18n/messages/en.json index aa7f379693..8f5791d9c9 100644 --- a/src/i18n/messages/en.json +++ b/src/i18n/messages/en.json @@ -13381,6 +13381,9 @@ "colProvider": "Provider", "colModel": "Model", "colQuota": "Quota", + "colLimits": "Limits", + "trainsOnPrompts": "Trains on prompts", + "trainsOnPromptsHelp": "This provider discloses that it may use your prompts to train models", "colContext": "Context", "colCapabilities": "Capabilities", "colTos": "ToS Risk", diff --git a/src/i18n/messages/pt-BR.json b/src/i18n/messages/pt-BR.json index df8baf6d4c..b167757ca9 100644 --- a/src/i18n/messages/pt-BR.json +++ b/src/i18n/messages/pt-BR.json @@ -13382,12 +13382,12 @@ "colProvider": "Provedor", "colModel": "Modelo", "colQuota": "Cota", + "colLimits": "Limites", + "trainsOnPrompts": "Treina com prompts", + "trainsOnPromptsHelp": "Este provedor declara que pode usar seus prompts para treinar modelos", "colContext": "Contexto", "colCapabilities": "Capacidades", "colTos": "Risco ToS", - "colLimits": "__MISSING__:Rate limits", - "trainsOnPrompts": "__MISSING__:Trains on prompts", - "trainsOnPromptsHelp": "__MISSING__:This provider's terms state it may train on the prompts you send. Models without this badge either state they do not, or do not document it — an absent statement is not a guarantee.", "newBadge": "novo", "setupGuide": "Guia de configuração", "disabledByFeed": "Desativado pelo feed Radar", diff --git a/src/i18n/messages/vi.json b/src/i18n/messages/vi.json index 1c016d98d7..d353b50d2b 100644 --- a/src/i18n/messages/vi.json +++ b/src/i18n/messages/vi.json @@ -13382,12 +13382,12 @@ "colProvider": "Nhà cung cấp", "colModel": "Mô hình", "colQuota": "Hạn ngạch", + "colLimits": "Giới hạn", + "trainsOnPrompts": "Huấn luyện bằng prompt", + "trainsOnPromptsHelp": "Nhà cung cấp này công bố có thể dùng prompt của bạn để huấn luyện mô hình", "colContext": "Ngữ cảnh", "colCapabilities": "Khả năng", "colTos": "Rủi ro ToS", - "colLimits": "__MISSING__:Rate limits", - "trainsOnPrompts": "__MISSING__:Trains on prompts", - "trainsOnPromptsHelp": "__MISSING__:This provider's terms state it may train on the prompts you send. Models without this badge either state they do not, or do not document it — an absent statement is not a guarantee.", "newBadge": "mới", "setupGuide": "Hướng dẫn thiết lập", "disabledByFeed": "Bị vô hiệu hóa bởi nguồn cấp dữ liệu Radar", diff --git a/tests/unit/stream-passthrough-usage-estimation.test.ts b/tests/unit/stream-passthrough-usage-estimation.test.ts index 8d6f1d80de..5a62f234c4 100644 --- a/tests/unit/stream-passthrough-usage-estimation.test.ts +++ b/tests/unit/stream-passthrough-usage-estimation.test.ts @@ -34,23 +34,6 @@ test("passthrough no fake: tool_only contentLength==0 -> no estimate (tool_calls import { createSSEStream } from "../../open-sse/utils/stream.ts"; -function collectSSE(stream: TransformStream) { - return async (writable: WritableStream, readable: ReadableStream) => { - const chunks: string[] = []; - const decoder = new TextDecoder(); - const reader = readable.getReader(); - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - chunks.push(decoder.decode(value, { stream: true })); - } - } finally { - reader.releaseLock(); - } - return chunks.join(""); - }; -} function parseSSEUsage(sseText: string): unknown[] { return sseText @@ -107,7 +90,7 @@ test("passthrough SSE: finish stop without usage + include_usage:true -> emits u assert.ok(typeof usage.completion_tokens === "number" && usage.completion_tokens > 0); }); -test("passthrough SSE: trailing choices:[] valid after estimated finish -> trailing is dropped (estimated wins)", async () => { +test("passthrough SSE: real trailing choices:[] usage is forwarded; no estimate is emitted (real wins)", async () => { const body = { model: "m", messages: [{ role: "user", content: "hi" }], stream: true, stream_options: { include_usage: true } }; const stream = createSSEStream({ mode: "passthrough" as const, @@ -129,17 +112,23 @@ test("passthrough SSE: trailing choices:[] valid after estimated finish -> trail })(); const enc = new TextEncoder(); 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`)); - // finish without usage -> should estimate (injectedUsage=false at that point) + // finish without usage -> passes through untouched (estimate only happens at flush, and only if no usage ever arrives) await writer.write(enc.encode(`data: ${JSON.stringify({ id: "chatcmpl-1", object: "chat.completion.chunk", choices: [{ index: 0, delta: {}, finish_reason: "stop" }] })}\n\n`)); - // trailing choices:[] with valid usage 50ms after -> inside empty-choices block hasValid(emptyChoicesUsage)&&!injectedUsage is now false, so chunk is dropped (warn path) + // trailing choices:[] with valid usage -> forwarded verbatim (marks passthroughForwardedUsage, so flush skips the estimate) await writer.write(enc.encode(`data: ${JSON.stringify({ id: "chatcmpl-1", object: "chat.completion.chunk", choices: [], usage: { prompt_tokens: 8, completion_tokens: 6, total_tokens: 14 } })}\n\n`)); await writer.write(enc.encode("data: [DONE]\n\n")); await writer.close(); const text = await readAll; const parsed = parseSSEUsage(text); const withUsage = parsed.filter((p: unknown) => (p as Record).usage); - // With the guard, the trailing valid is dropped (estimated was already sent on finish). Without guard we would see 2 (double). We assert drop. - // If upstream ever sends real include_usage trailing, this documents the v1 tradeoff: estimated wins, valid is dropped. - assert.equal(withUsage.length, 1, `expected 1 usage (estimated, trailing dropped), got ${withUsage.length} — usages: ${JSON.stringify(withUsage.map((p) => (p as Record).usage))}`); - assert.equal((withUsage[0] as Record & { usage: Record }).usage.estimated, true); + // v2 contract (#12151 follow-up): the upstream's REAL trailing usage block is forwarded + // and wins; the estimate exists only for upstreams that never report usage (emitted at + // flush). Exactly one usage block ever reaches the client — never two, never estimated + // when a real one arrived (the v1 "estimated wins" tradeoff was a billing regression). + assert.equal(withUsage.length, 1, `expected 1 usage (the real trailing block), got ${withUsage.length} — usages: ${JSON.stringify(withUsage.map((p) => (p as Record).usage))}`); + const forwarded = (withUsage[0] as Record & { usage: Record }).usage; + assert.equal(forwarded.estimated, undefined); + assert.equal(forwarded.prompt_tokens, 8); + assert.equal(forwarded.completion_tokens, 6); + assert.equal(forwarded.total_tokens, 14); });