From 43ca41d088b2e3a8b6022149c461cbba535ab499 Mon Sep 17 00:00:00 2001 From: "ScrewTSW (public-projects)" Date: Wed, 19 Aug 2026 00:07:28 +0200 Subject: [PATCH 1/3] fix(openai-adapters): don't swallow stream chunks that carry usage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chatCompletionStream deferred any chunk with a `usage` field so that usage could be re-emitted after all content. That assumes usage appears only on a terminal chunk, which holds for the OpenAI API but not in general. llama.cpp-based servers can attach a running `usage` counter to every chunk. In that case the deferral branch matched on all of them, each overwriting lastChunkWithUsage, and only the final chunk was ever yielded — so the entire response was discarded and the assistant message rendered empty. Observed against a local orchestrator: 51 chunks in, 1 out, all reasoning_content and content deltas lost. Only defer chunks that are genuinely usage-only: usage present, no finish_reason, and an empty delta. Chunks carrying a payload are yielded immediately, and a usage-bearing content chunk clears the deferred chunk so it is not re-emitted as a duplicate. Tests cover both regimes (per-chunk running usage and OpenAI's terminal usage-only chunk) and are mutation-verified: reverting the predicate to `!!result.usage` fails them with empty content. Co-Authored-By: Claude Opus 5 --- packages/openai-adapters/src/apis/OpenAI.ts | 15 +- .../src/apis/OpenAIStreamUsage.test.ts | 140 ++++++++++++++++++ 2 files changed, 153 insertions(+), 2 deletions(-) create mode 100644 packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts diff --git a/packages/openai-adapters/src/apis/OpenAI.ts b/packages/openai-adapters/src/apis/OpenAI.ts index 3f750110c3d..b12186053f0 100644 --- a/packages/openai-adapters/src/apis/OpenAI.ts +++ b/packages/openai-adapters/src/apis/OpenAI.ts @@ -168,11 +168,22 @@ export class OpenAIApi implements BaseLlmApi { ); let lastChunkWithUsage: ChatCompletionChunk | undefined; for await (const result of response) { - // Check if this chunk contains usage information - if (result.usage) { + // Defer usage-only chunks so usage is reported after all content. + // Some servers (e.g. llama.cpp with incremental usage) attach a running + // `usage` object to EVERY chunk. Withholding on `usage` alone would then + // swallow the entire stream, so only defer chunks that carry no payload. + const choice = result.choices?.[0]; + const isUsageOnly = + !!result.usage && + !choice?.finish_reason && + Object.keys(choice?.delta ?? {}).length === 0; + if (isUsageOnly) { // Store it to emit after all content chunks lastChunkWithUsage = result; } else { + if (result.usage) { + lastChunkWithUsage = undefined; + } yield result; } } diff --git a/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts b/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts new file mode 100644 index 00000000000..88b47548337 --- /dev/null +++ b/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts @@ -0,0 +1,140 @@ +import { ChatCompletionChunk } from "openai/resources/index"; +import { describe, expect, it, vi } from "vitest"; + +import { OpenAIApi } from "./OpenAI.js"; + +/** + * Regression tests for streaming chunks that carry `usage`. + * + * The adapter defers usage chunks so that usage is reported after content. + * OpenAI sends usage exactly once, in a terminal chunk with empty `choices`. + * llama.cpp-based servers can instead attach a *running* usage object to every + * chunk; deferring on the presence of `usage` alone swallowed those streams + * entirely, producing empty assistant messages. + */ + +function chunk( + delta: Record, + usage?: { completion_tokens: number }, + finish_reason: string | null = null, +): ChatCompletionChunk { + return { + id: "chatcmpl-test", + object: "chat.completion.chunk", + created: 0, + model: "test-model", + choices: [{ index: 0, delta, finish_reason }], + ...(usage + ? { + usage: { + prompt_tokens: 10, + total_tokens: 10 + usage.completion_tokens, + ...usage, + }, + } + : {}), + } as unknown as ChatCompletionChunk; +} + +function apiYielding(chunks: ChatCompletionChunk[]) { + const api = new OpenAIApi({ + provider: "openai", + apiKey: "test-key", + apiBase: "http://192.168.1.2:58108/v1/", + }); + vi.spyOn(api["openai"].chat.completions, "create").mockResolvedValue({ + async *[Symbol.asyncIterator]() { + for (const c of chunks) { + yield c; + } + }, + } as any); + return api; +} + +async function collect(api: OpenAIApi, body: any) { + const out: ChatCompletionChunk[] = []; + for await (const c of api.chatCompletionStream( + body, + new AbortController().signal, + )) { + out.push(c); + } + return out; +} + +const body = { + model: "test-model", + messages: [{ role: "user" as const, content: "hi" }], + stream: true as const, +}; + +describe("chatCompletionStream usage handling", () => { + it("preserves content when every chunk carries a running usage object", async () => { + // Shape observed from the local orchestrator: usage increments per token. + const api = apiYielding([ + chunk({ role: "assistant", content: null }, { completion_tokens: 1 }), + chunk({ reasoning_content: "think" }, { completion_tokens: 2 }), + chunk({ content: "Hello" }, { completion_tokens: 3 }), + chunk({ content: " world" }, { completion_tokens: 4 }), + chunk({}, { completion_tokens: 5 }, "stop"), + ]); + + const out = await collect(api, body); + + const text = out + .map((c) => (c.choices?.[0]?.delta as any)?.content ?? "") + .join(""); + expect(text).toBe("Hello world"); + + const reasoning = out + .map((c) => (c.choices?.[0]?.delta as any)?.reasoning_content ?? "") + .join(""); + expect(reasoning).toBe("think"); + + // Nothing is dropped, and the finish_reason chunk still arrives. + expect(out).toHaveLength(5); + expect(out.at(-1)?.choices?.[0]?.finish_reason).toBe("stop"); + }); + + it("still defers a terminal usage-only chunk to the end", async () => { + // OpenAI shape with stream_options.include_usage: usage arrives alone. + const api = apiYielding([ + chunk({ content: "Hello" }), + chunk({ content: " world" }), + { + id: "chatcmpl-test", + object: "chat.completion.chunk", + created: 0, + model: "test-model", + choices: [], + usage: { prompt_tokens: 10, completion_tokens: 2, total_tokens: 12 }, + } as unknown as ChatCompletionChunk, + ]); + + const out = await collect(api, body); + + expect(out).toHaveLength(3); + expect(out.at(-1)?.usage?.completion_tokens).toBe(2); + expect( + out + .slice(0, 2) + .map((c) => (c.choices?.[0]?.delta as any)?.content) + .join(""), + ).toBe("Hello world"); + }); + + it("does not emit a duplicate trailing chunk when usage rides on content", async () => { + const api = apiYielding([ + chunk({ content: "A" }, { completion_tokens: 1 }), + chunk({ content: "B" }, { completion_tokens: 2 }, "stop"), + ]); + + const out = await collect(api, body); + + expect(out).toHaveLength(2); + expect( + out.map((c) => (c.choices?.[0]?.delta as any)?.content).join(""), + ).toBe("AB"); + }); +}); From ee935ac12f02c2f87c61dcf7a81afe7197d1d2dd Mon Sep 17 00:00:00 2001 From: ScrewTSW Date: Wed, 19 Aug 2026 22:48:23 +0200 Subject: [PATCH 2/3] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts b/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts index 88b47548337..7f7078fe2dd 100644 --- a/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts +++ b/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts @@ -40,7 +40,7 @@ function apiYielding(chunks: ChatCompletionChunk[]) { const api = new OpenAIApi({ provider: "openai", apiKey: "test-key", - apiBase: "http://192.168.1.2:58108/v1/", + apiBase: "http://custom:8080/v1/", }); vi.spyOn(api["openai"].chat.completions, "create").mockResolvedValue({ async *[Symbol.asyncIterator]() { From 201cc79ee00b1631c80bf78d6c9cbd28482d0c48 Mon Sep 17 00:00:00 2001 From: "ScrewTSW (public-projects)" Date: Wed, 19 Aug 2026 23:29:53 +0200 Subject: [PATCH 3/3] fix(openai-adapters): classify usage-only chunks across all choices `isUsageOnly` inspected `choices[0]` only. With `n > 1`, a chunk whose first choice is empty but whose second carries content was classified as usage-only and deferred, dropping the other choice's payload. Use `every` across all choices instead. Also strengthens the duplicate-emission test: it began with a content-bearing chunk, so no deferred chunk ever existed and an implementation that failed to clear one would still have passed. It now opens with a usage-only chunk. Mutation-verified: the `n > 1` test fails against the old `choices[0]` logic, and the duplicate test fails when the clearing branch is removed. Co-Authored-By: Claude Opus 5 --- packages/openai-adapters/src/apis/OpenAI.ts | 11 +++- .../src/apis/OpenAIStreamUsage.test.ts | 58 +++++++++++++++++++ 2 files changed, 66 insertions(+), 3 deletions(-) diff --git a/packages/openai-adapters/src/apis/OpenAI.ts b/packages/openai-adapters/src/apis/OpenAI.ts index b12186053f0..b19efd05707 100644 --- a/packages/openai-adapters/src/apis/OpenAI.ts +++ b/packages/openai-adapters/src/apis/OpenAI.ts @@ -172,11 +172,16 @@ export class OpenAIApi implements BaseLlmApi { // Some servers (e.g. llama.cpp with incremental usage) attach a running // `usage` object to EVERY chunk. Withholding on `usage` alone would then // swallow the entire stream, so only defer chunks that carry no payload. - const choice = result.choices?.[0]; + // Check every choice, not just the first: with `n > 1` an empty + // choices[0] alongside a content-bearing choices[1] would otherwise + // classify the chunk as usage-only and drop the other choice's payload. const isUsageOnly = !!result.usage && - !choice?.finish_reason && - Object.keys(choice?.delta ?? {}).length === 0; + (result.choices ?? []).every( + (choice) => + !choice?.finish_reason && + Object.keys(choice?.delta ?? {}).length === 0, + ); if (isUsageOnly) { // Store it to emit after all content chunks lastChunkWithUsage = result; diff --git a/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts b/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts index 7f7078fe2dd..6d09c0fb58a 100644 --- a/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts +++ b/packages/openai-adapters/src/apis/OpenAIStreamUsage.test.ts @@ -36,6 +36,33 @@ function chunk( } as unknown as ChatCompletionChunk; } +/** Multi-choice chunk, for `n > 1` requests. */ +function multiChoiceChunk( + deltas: Record[], + usage?: { completion_tokens: number }, +): ChatCompletionChunk { + return { + id: "chatcmpl-test", + object: "chat.completion.chunk", + created: 0, + model: "test-model", + choices: deltas.map((delta, index) => ({ + index, + delta, + finish_reason: null, + })), + ...(usage + ? { + usage: { + prompt_tokens: 10, + total_tokens: 10 + usage.completion_tokens, + ...usage, + }, + } + : {}), + } as unknown as ChatCompletionChunk; +} + function apiYielding(chunks: ChatCompletionChunk[]) { const api = new OpenAIApi({ provider: "openai", @@ -126,6 +153,9 @@ describe("chatCompletionStream usage handling", () => { it("does not emit a duplicate trailing chunk when usage rides on content", async () => { const api = apiYielding([ + // Leading usage-only chunk, so a deferred chunk actually exists to be + // cleared. Without it this test passes even if clearing is broken. + chunk({}, { completion_tokens: 0 }), chunk({ content: "A" }, { completion_tokens: 1 }), chunk({ content: "B" }, { completion_tokens: 2 }, "stop"), ]); @@ -137,4 +167,32 @@ describe("chatCompletionStream usage handling", () => { out.map((c) => (c.choices?.[0]?.delta as any)?.content).join(""), ).toBe("AB"); }); + + it("does not defer a chunk when a later choice carries content (n > 1)", async () => { + const api = apiYielding([ + // choices[0] is empty but choices[1] has content: inspecting only the + // first choice would classify this as usage-only and drop "B". + multiChoiceChunk([{}, { content: "B" }], { completion_tokens: 1 }), + chunk({ content: "A" }, { completion_tokens: 2 }, "stop"), + ]); + + const out = await collect(api, body); + + expect(out).toHaveLength(2); + expect((out[0].choices?.[1]?.delta as any)?.content).toBe("B"); + }); + + it("still defers a chunk when every choice is empty (n > 1)", async () => { + const api = apiYielding([ + chunk({ content: "A" }, undefined, "stop"), + multiChoiceChunk([{}, {}], { completion_tokens: 3 }), + ]); + + const out = await collect(api, body); + + // The usage-only chunk is deferred, then emitted last. + expect(out).toHaveLength(2); + expect((out[0].choices?.[0]?.delta as any)?.content).toBe("A"); + expect(out[1].usage?.completion_tokens).toBe(3); + }); });