fix(stream): no choice-less chunks unless the client asks for usage

Every stream used to carry two chunks with an empty choices array: usage,
which the upstream emits BEFORE the stop reason, and a cortiq_cost_usd chunk
the gateway added itself. OpenAI ships the usage chunk only when the request
sets stream_options: {"include_usage": true}, and chat UIs do not set it -
which is why the same tool, in the same Open WebUI, sees neither chunk from
Groq and two from us.

Honour the flag. With no stream_options a stream is now exactly
role -> content... -> tool_calls -> finish_reason -> [DONE], with no
choice-less chunks at all. When usage is requested it ships once, after the
stop reason, with cortiq_cost_usd folded into the same chunk rather than
trailing behind it.

Token accounting, cost and the X-Cortiq-* headers are untouched: those numbers
were never read back off the wire.

Verified: the default stream matches the reference shape byte for byte, the
opt-in stream matches OpenAI's, the official OpenAI SDK reassembles the
streamed tool call, and the full Open WebUI 0.11.0 tool loop still runs the
tool and answers from its result.

CHANGELOG.md

@@ -3,6 +3,22 @@
 All notable changes to this project are documented here.
 The format is based on [Keep a Changelog](https://keepachangelog.com/).
 
+## [0.2.49] - 2026-08-21
+
+### Fixed
+- **The SSE stream now has the same shape every hosted provider sends.** Two choice-less
+  chunks used to ride along on every stream: `usage` (emitted by the upstream *before* the
+  stop reason) and a `cortiq_cost_usd` chunk of the gateway's own. OpenAI ships the usage
+  chunk only when the client sets `stream_options: {"include_usage": true}` — and chat UIs
+  do not set it, which is why a Groq stream carries neither. The gateway now honours that
+  flag: with no `stream_options`, a stream is exactly
+  `role → content… → tool_calls → finish_reason → [DONE]` with no choice-less chunks at
+  all. When usage *is* requested it ships once, **after** the stop reason, with
+  `cortiq_cost_usd` folded into the same chunk instead of trailing behind it.
+
+  Token accounting, cost and the `X-Cortiq-*` headers are unchanged — the numbers were
+  never taken from those chunks.
+
 ## [0.2.48] - 2026-08-21
 
 Verified end to end: Open WebUI 0.11.0 → gateway → managed CMF/Qwen, native

Cargo.lock

@@ -285,7 +285,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
 
 [[package]]
 name = "cortiq-gateway"
-version = "0.2.48"
+version = "0.2.49"
 dependencies = [
  "anyhow",
  "arc-swap",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "cortiq-gateway"
-version = "0.2.48"
+version = "0.2.49"
 edition = "2021"
 rust-version = "1.75"
 license = "Apache-2.0"

docs/PROTOCOLS.md

@@ -121,6 +121,12 @@ opens with a `role` delta, carries exactly one terminal `finish_reason`, and end
 failure arrives as an SSE `error` event. A client can therefore never be left waiting on a
 half-formed stream.
 
+Streaming carries no choice-less chunks unless asked: `usage` is emitted only when the
+request sets `stream_options: {"include_usage": true}`, and then exactly once, after the
+`finish_reason` chunk, with `cortiq_cost_usd` folded in. Without it a stream is
+`role → content… → tool_calls → finish_reason → [DONE]` and nothing else — the shape the
+hosted providers send.
+
 ### Tool calls
 
 `tools` and `tool_calls` are proxied through, with the wire kept clean:

docs/PROTOCOLS.ru.md

@@ -119,6 +119,12 @@ HTTP 400 — но не «голый» 422, который OpenAI-клиенты
 `data: [DONE]` — даже если upstream оборвался или упал на середине (тогда ошибка приходит
 SSE-событием `error`). Клиент никогда не остаётся ждать недосказанный поток.
 
+В стриме нет чанков без `choices`, если их не просили: `usage` отдаётся, только когда
+запрос содержит `stream_options: {"include_usage": true}`, и тогда ровно один раз — после
+чанка с `finish_reason`, с вложенным `cortiq_cost_usd`. Без него поток выглядит как
+`role → content… → tool_calls → finish_reason → [DONE]` и ничего больше — ровно так же,
+как у хостовых провайдеров.
+
 ### Tool calls
 
 `tools` и `tool_calls` проксируются, но провод держится чистым:

src/protocols/openai_chat.rs

@@ -233,6 +233,16 @@ async fn handler(
         .unwrap_or("cortiq-auto")
         .to_string();
     let stream = req.stream.unwrap_or(false);
+    // OpenAI only emits the trailing usage chunk when the client asks for it,
+    // and chat UIs do not ask. Mirroring that keeps our stream byte-shaped like
+    // every other provider's instead of carrying two extra choice-less chunks
+    // that each client has to know to ignore.
+    let include_usage = req
+        .rest
+        .get("stream_options")
+        .and_then(|o| o.get("include_usage"))
+        .and_then(|v| v.as_bool())
+        .unwrap_or(false);
 
     let canonical = ChatRequest {
         routing: parse_routing(&requested_model),
@@ -260,7 +270,7 @@ async fn handler(
     if stream {
         let (info, raw) = state.pipeline.run_stream(canonical, &state).await?;
         let headers = sse_headers(&info);
-        let normalized = normalize_sse(raw, requested_model);
+        let normalized = normalize_sse(raw, requested_model, include_usage);
         return Ok((headers, axum::body::Body::from_stream(normalized)).into_response());
     }
 
@@ -350,6 +360,7 @@ async fn handler(
 fn normalize_sse(
     stream: crate::providers::ChatStream,
     model: String,
+    include_usage: bool,
 ) -> impl futures::Stream<Item = Result<bytes::Bytes>> + Send + 'static {
     use futures::StreamExt;
     async_stream::stream! {
@@ -368,6 +379,9 @@ fn normalize_sse(
             std::collections::HashMap::new();
         let mut tool_indices: std::collections::HashSet<u64> = std::collections::HashSet::new();
         let mut emitted_tool_calls = false;
+        // `usage` / `cortiq_*` collected along the way, shipped after the finish.
+        let mut tail_extras: std::collections::BTreeMap<&str, serde_json::Value> =
+            std::collections::BTreeMap::new();
 
         // Emit one normalised event; `None` means "nothing worth sending".
         macro_rules! chunk {
@@ -508,20 +522,20 @@ fn normalize_sse(
                         }
                     }
 
-                    // Terminal metadata rides on a choice-less chunk, exactly
+                    // Terminal metadata is held back and shipped once, after the
-                    // as OpenAI ships `usage` when `stream_options` asks for it.
+                    // stop reason — the order OpenAI uses. Emitting it mid-stream
-                    let mut extra: Vec<(&str, serde_json::Value)> = Vec::new();
+                    // (as several runtimes do) puts choice-less chunks between the
-                    if let Some(u) = v.get("usage").filter(|u| !u.is_null()) {
+                    // deltas and the finish, which is legal but unlike every
-                        extra.push(("usage", u.clone()));
+                    // reference implementation.
-                    }
+                    for key in ["usage", "cortiq_cost_usd", "cortiq_estimated"] {
-                    for key in ["cortiq_cost_usd", "cortiq_estimated"] {
                         if let Some(val) = v.get(key).filter(|v| !v.is_null()) {
-                            extra.push((key, val.clone()));
+                            tail_extras.insert(key, val.clone());
                         }
                     }
 
-                    if !choices.is_empty() || !extra.is_empty() {
+                    if !choices.is_empty() {
-                        yield Ok(chunk!(choices, extra));
+                        let none: Vec<(&str, serde_json::Value)> = Vec::new();
+                        yield Ok(chunk!(choices, none));
                     }
                 }
             }
@@ -563,6 +577,15 @@ fn normalize_sse(
             let extra: Vec<(&str, serde_json::Value)> = Vec::new();
             yield Ok(chunk!(choices, extra));
         }
+        // One trailing chunk, and only when the client asked for usage.
+        if include_usage {
+            if let Some(usage) = tail_extras.remove("usage") {
+                let mut extra: Vec<(&str, serde_json::Value)> = vec![("usage", usage)];
+                extra.extend(tail_extras.into_iter());
+                let no_choices: Vec<serde_json::Value> = Vec::new();
+                yield Ok(chunk!(no_choices, extra));
+            }
+        }
         yield Ok(bytes::Bytes::from("data: [DONE]\n\n"));
     }
 }
@@ -612,7 +635,7 @@ mod tests {
                 .map(|c| Ok(bytes::Bytes::from(c)))
                 .collect::<Vec<_>>(),
         );
-        normalize_sse(Box::pin(upstream), "cortiq-auto".into())
+        normalize_sse(Box::pin(upstream), "cortiq-auto".into(), false)
             .map(|b| String::from_utf8_lossy(&b.unwrap()).to_string())
             .collect::<Vec<_>>()
             .await
@@ -667,6 +690,63 @@ mod tests {
         );
     }
 
+    async fn drain_with_usage(chunks: Vec<&'static str>, include_usage: bool) -> String {
+        use futures::StreamExt;
+        let upstream = futures::stream::iter(
+            chunks
+                .into_iter()
+                .map(|c| Ok(bytes::Bytes::from(c)))
+                .collect::<Vec<_>>(),
+        );
+        normalize_sse(Box::pin(upstream), "cortiq-auto".into(), include_usage)
+            .map(|b| String::from_utf8_lossy(&b.unwrap()).to_string())
+            .collect::<Vec<_>>()
+            .await
+            .join("")
+    }
+
+    /// The upstream ships usage BEFORE the stop reason and the gateway adds a
+    /// cost chunk of its own. A client that did not ask for usage must see
+    /// neither — the same stream shape every hosted provider sends.
+    const USAGE_STREAM: &[&str] = &[
+        "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
+        "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":1,\"total_tokens\":4}}\n\n",
+        "data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
+        "data: {\"choices\":[],\"cortiq_cost_usd\":0.0}\n\n",
+        "data: [DONE]\n\n",
+    ];
+
+    #[tokio::test]
+    async fn without_stream_options_no_choice_less_chunks_are_sent() {
+        let out = drain_with_usage(USAGE_STREAM.to_vec(), false).await;
+        assert!(!out.contains("\"usage\""), "unrequested usage chunk: {out}");
+        assert!(!out.contains("cortiq_cost_usd"), "stray cost chunk: {out}");
+        assert!(
+            !out.contains("\"choices\":[]"),
+            "choice-less chunk leaked: {out}"
+        );
+        assert!(out.contains("\"finish_reason\":\"stop\""));
+        assert!(out.trim_end().ends_with("data: [DONE]"));
+    }
+
+    #[tokio::test]
+    async fn requested_usage_is_one_chunk_after_the_finish_reason() {
+        let out = drain_with_usage(USAGE_STREAM.to_vec(), true).await;
+        assert_eq!(
+            out.matches("\"usage\"").count(),
+            1,
+            "usage must ship once: {out}"
+        );
+        let finish = out.find("\"finish_reason\"").expect("finish chunk");
+        let usage = out.find("\"usage\"").expect("usage chunk");
+        assert!(usage > finish, "usage must follow the stop reason: {out}");
+        assert!(
+            out.contains("cortiq_cost_usd"),
+            "cost folded into usage chunk: {out}"
+        );
+        assert!(out.trim_end().ends_with("data: [DONE]"));
+    }
+
     #[tokio::test]
     async fn ordinary_text_is_untouched_by_the_tool_filter() {
         let out = drain(vec![
@@ -685,7 +765,7 @@ mod tests {
         let upstream = futures::stream::iter(vec![Ok(bytes::Bytes::from(
             "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
         ))]);
-        let out: Vec<String> = normalize_sse(Box::pin(upstream), "cortiq-auto".into())
+        let out: Vec<String> = normalize_sse(Box::pin(upstream), "cortiq-auto".into(), false)
             .map(|b| String::from_utf8_lossy(&b.unwrap()).to_string())
             .collect()
             .await;