Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 53 additions & 6 deletions open-sse/utils/stream.ts
Original file line number Diff line number Diff line change
Expand Up @@ -799,6 +799,12 @@ export function createSSEStream(options: StreamOptions = {}) {
const clientPayloadCollector = createStructuredSSECollector({
stage: "client_response",
});
const requestRecord = asRecord(body);
const requestStreamOptions = asRecord(
requestRecord.stream_options ?? requestRecord.streamOptions
);
const expectsOpenAIUsageOnlyChunk =
requestStreamOptions.include_usage === true || requestStreamOptions.includeUsage === true;

// Per-stream instances to avoid shared state with concurrent streams
const decoder = new TextDecoder();
Expand Down Expand Up @@ -1589,12 +1595,49 @@ export function createSSEStream(options: StreamOptions = {}) {
// GOOD LUCK
// ----------------------------------------------------
// Chat Completions: full sanitization pipeline
// Hardening: detect upstream returning empty choices array
// which breaks OpenAI-compatible clients (e.g. Copilot Chat).
// We drop these chunks entirely rather than injecting an error,
// as injecting a chunk with finish_reason: "stop" will prematurely
// terminate the stream for the client.

// OpenAI-compatible streaming with `stream_options.include_usage=true`
// ends with a usage-only chunk where `choices` is deliberately `[]`.
// Forward that standards-compliant chunk instead of turning it into an
// empty-response error. Keep the existing hardening for malformed empty
// choices chunks without valid usage.
if (Array.isArray(parsed.choices) && parsed.choices.length === 0) {
const emptyChoicesUsage = extractUsage(parsed) ?? parsed.usage;
if (hasValidUsage(emptyChoicesUsage)) {
usage = emptyChoicesUsage;
output = `data: ${JSON.stringify(parsed)}\n`;
injectedUsage = true;
clientPayload = parsed;
clientPayloadCollector.push(clientPayload);
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
continue;
}

console.warn(
`[STREAM] Upstream returned empty choices array (${provider || "provider"}:${model || "unknown"}) — emitting error chunk`
);
const errorChunk = {
id: parsed.id || `omniroute-empty-choices-${Date.now()}`,
object: "chat.completion.chunk",
created: parsed.created || Math.floor(Date.now() / 1000),
model: parsed.model || model || "unknown",
choices: [
{
index: 0,
delta: {
role: "assistant",
content: "[OmniRoute] Upstream returned an empty response. Please retry.",
},
finish_reason: "stop",
},
],
};
output = `data: ${JSON.stringify(errorChunk)}\n`;
injectedUsage = true;
clientPayload = errorChunk;
reqLogger?.appendConvertedChunk?.(output);
controller.enqueue(encoder.encode(output));
continue;
}

Expand Down Expand Up @@ -1774,7 +1817,11 @@ export function createSSEStream(options: StreamOptions = {}) {
injectedUsage = true;
}
}
if (isFinishChunk && !hasValidUsage(parsed.usage)) {
if (
isFinishChunk &&
!hasValidUsage(parsed.usage) &&
!expectsOpenAIUsageOnlyChunk
) {
const estimated = estimateUsage(body, totalContentLength, FORMATS.OPENAI);
parsed.usage = filterUsageForFormat(estimated, FORMATS.OPENAI);
output = `data: ${JSON.stringify(parsed)}\n`;
Expand Down
63 changes: 59 additions & 4 deletions tests/unit/stream-utils.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1656,16 +1656,71 @@ test("createSSEStream passthrough drops empty choices array chunks", async () =>
}
);

// The empty choices chunk should have been dropped entirely
assert.doesNotMatch(text, /\[OmniRoute\] Upstream returned an empty response/);

// Subsequent valid chunks should still be present and correctly processed
// Empty choices without usage still get replaced with a synthetic error chunk.
assert.match(text, /\[OmniRoute\] Upstream returned an empty response/);
assert.match(text, /"finish_reason":"stop"/);
// Subsequent valid chunks should still be present
assert.match(text, /"content":"Hello"/);
assert.match(text, /"finish_reason":"stop"/);
assert.equal(onCompletePayload.status, 200);
assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Hello");
});

test("createSSEStream passthrough forwards OpenAI usage-only empty choices chunks", async () => {
let onCompletePayload = null;
const usage = { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 };
const text = await readTransformed(
[
`data: ${JSON.stringify({
id: "chatcmpl_usage_only",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4.1-mini",
choices: [{ index: 0, delta: { role: "assistant", content: "Hello" } }],
usage: null,
})}\n\n`,
`data: ${JSON.stringify({
id: "chatcmpl_usage_only",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4.1-mini",
choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
usage: null,
})}\n\n`,
`data: ${JSON.stringify({
id: "chatcmpl_usage_only",
object: "chat.completion.chunk",
created: 1,
model: "gpt-4.1-mini",
choices: [],
usage,
})}\n\n`,
],
{
mode: "passthrough",
sourceFormat: FORMATS.OPENAI,
provider: "openai-compatible",
model: "gpt-4.1-mini",
body: {
messages: [{ role: "user", content: "hello" }],
stream_options: { include_usage: true },
},
onComplete(payload) {
onCompletePayload = payload;
},
}
);

assert.doesNotMatch(text, /\[OmniRoute\] Upstream returned an empty response/);
assert.match(text, /"choices":\[\]/);
assert.match(text, /"usage":\{"prompt_tokens":7,"completion_tokens":3,"total_tokens":10\}/);
assert.equal(onCompletePayload.status, 200);
assert.equal(onCompletePayload.usage.prompt_tokens, 7);
assert.equal(onCompletePayload.usage.completion_tokens, 3);
assert.equal(onCompletePayload.responseBody.choices[0].message.content, "Hello");
assert.deepEqual(onCompletePayload.responseBody.usage, usage);
});

test("createSSEStream passthrough logs empty response after tool_calls completion", async () => {
let onCompletePayload = null;
const text = await readTransformed(
Expand Down