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.

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
-                    // as OpenAI ships `usage` when `stream_options` asks for it.
-                    let mut extra: Vec<(&str, serde_json::Value)> = Vec::new();
-                    if let Some(u) = v.get("usage").filter(|u| !u.is_null()) {
-                        extra.push(("usage", u.clone()));
-                    }
-                    for key in ["cortiq_cost_usd", "cortiq_estimated"] {
+                    // Terminal metadata is held back and shipped once, after the
+                    // stop reason — the order OpenAI uses. Emitting it mid-stream
+                    // (as several runtimes do) puts choice-less chunks between the
+                    // deltas and the finish, which is legal but unlike every
+                    // reference implementation.
+                    for key in ["usage", "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() {
-                        yield Ok(chunk!(choices, extra));
+                    if !choices.is_empty() {
+                        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;