diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b03b2a..d0994e1 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/Cargo.lock b/Cargo.lock index d1ff057..e0943a3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -285,7 +285,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "cortiq-gateway" -version = "0.2.48" +version = "0.2.49" dependencies = [ "anyhow", "arc-swap", diff --git a/Cargo.toml b/Cargo.toml index 3f9ec28..466927f 100644 --- a/Cargo.toml +++ b/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" diff --git a/docs/PROTOCOLS.md b/docs/PROTOCOLS.md index 76925aa..05d6bc1 100644 --- a/docs/PROTOCOLS.md +++ b/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: diff --git a/docs/PROTOCOLS.ru.md b/docs/PROTOCOLS.ru.md index 6569512..fcfdab7 100644 --- a/docs/PROTOCOLS.ru.md +++ b/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` проксируются, но провод держится чистым: diff --git a/src/protocols/openai_chat.rs b/src/protocols/openai_chat.rs index 784a6be..b19918b 100644 --- a/src/protocols/openai_chat.rs +++ b/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> + 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 = 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 = 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::>(), ); - 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::>() .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::>(), + ); + normalize_sse(Box::pin(upstream), "cortiq-auto".into(), include_usage) + .map(|b| String::from_utf8_lossy(&b.unwrap()).to_string()) + .collect::>() + .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 = normalize_sse(Box::pin(upstream), "cortiq-auto".into()) + let out: Vec = normalize_sse(Box::pin(upstream), "cortiq-auto".into(), false) .map(|b| String::from_utf8_lossy(&b.unwrap()).to_string()) .collect() .await;