diff --git a/CHANGELOG.md b/CHANGELOG.md index 6aeae5c..1b36457 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,48 @@ All notable changes to this project are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/). +## [0.2.45] - 2026-08-21 + +### Fixed +- **Open WebUI (and every strict OpenAI client) can finally chat with the gateway.** + `POST /v1/chat/completions` rejected perfectly ordinary payloads with a bare HTTP 422 — + a plain-text body no OpenAI client can parse, which is why the chat window just spun with + no answer and no error. The request parser now accepts every shape real clients send: + `content` as a string, `null`, or an array of content parts (text parts concatenated, + images skipped), `tools: null`, `stream: null`, a string `reasoning_effort` + (`low`/`medium`/`high`) as well as a numeric budget, `max_completion_tokens` as an alias + for `max_tokens`, `messages[].name` and `messages[].tool_call_id` proxied through, and a + missing `Content-Type`. A body that genuinely cannot be read now returns an OpenAI error + envelope with HTTP 400. +- **Strict OpenAI response JSON, no optional `null`s.** Absent tool calls no longer emit + `"tool_calls": null` (the key is omitted entirely); `finish_reason` is mapped to the + spec's closed set, so `cortiq serve`'s `max_tokens` reaches clients as `length`; the + response always carries a non-empty `id`. +- **`model` echoes the id the client requested** (`cortiq-auto` stays `cortiq-auto`, not the + upstream's `cortiq`), matching OpenAI's own behaviour. The model that actually answered + stays in `cortiq.selected_model` and `X-Cortiq-Selected-Model`. Applies to + `/v1/chat/completions`, `/v1/completions` and `/v1/messages`. +- **Streaming is re-normalised instead of forwarded verbatim.** Every chunk is reissued as a + strict `chat.completion.chunk`: the stream opens with a `role` delta, `message`-shaped + chunks are converted to `delta`, empty deltas are dropped, and it always closes with one + terminal `finish_reason` and `data: [DONE]` — even when the upstream stops early or dies + mid-flight, in which case the failure arrives as an SSE `error` event rather than a client + waiting forever. SSE responses also carry `X-Accel-Buffering: no` so nginx/Traefik cannot + buffer the stream into a hang. +- **Client bookkeeping keys are no longer proxied upstream.** `chat_id`, `session_id`, + `metadata`, `features`, `background_tasks` and friends (plus a non-string `user`) are + stripped before the provider call, so strict backends stop rejecting the request. + +### Added +- **`GET /v1/models` lists managed local CMF models.** Every `[[cmf.servers]]` entry now + appears alongside `cortiq-auto` and the `[[models]]` pool, with live `running`/`healthy` + state in a non-standard `cortiq` field, so Open WebUI and other clients can pin a local + `.cmf` directly. Entries also carry `created`. + +### Changed +- Dropped leftover `[to_wire]` debug prints from the OpenAI provider — they wrote to stderr + on every single request. + ## [0.2.44] - 2026-08-08 ### Added diff --git a/Cargo.lock b/Cargo.lock index 95e3bb1..86aff42 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -285,7 +285,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "cortiq-gateway" -version = "0.2.44" +version = "0.2.45" dependencies = [ "anyhow", "arc-swap", diff --git a/Cargo.toml b/Cargo.toml index 8f1c1bb..2780a19 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cortiq-gateway" -version = "0.2.44" +version = "0.2.45" edition = "2021" rust-version = "1.75" license = "Apache-2.0" diff --git a/docs/PROTOCOLS.md b/docs/PROTOCOLS.md index aecd61b..79a88b7 100644 --- a/docs/PROTOCOLS.md +++ b/docs/PROTOCOLS.md @@ -53,7 +53,17 @@ Routing rule common to all protocols: ``` Standard OpenAI fields are supported; unknown fields are proxied to the provider unchanged -(forward-compatible). +(forward-compatible), minus the bookkeeping keys chat UIs attach to the payload +(`chat_id`, `session_id`, `metadata`, `features`, …), which are dropped so strict upstreams +do not reject the call. + +The parser is deliberately lenient about the shapes real clients send: +`content` may be a string, `null`, or an array of content parts (text parts are +concatenated, images and other non-text parts are skipped); `tools` may be `null`; +`reasoning_effort` may be a string (`low` | `medium` | `high`) or a token budget; +`max_completion_tokens` is accepted as an alias for `max_tokens`. A body that still +cannot be read comes back as an OpenAI error envelope with HTTP 400 — never a bare 422, +which OpenAI clients cannot parse (they show no answer at all). ### Response (non-streaming) @@ -62,12 +72,12 @@ Standard OpenAI fields are supported; unknown fields are proxied to the provider "id": "chatcmpl-9f...", "object": "chat.completion", "created": 1718800000, - "model": "local-qwen", // the ACTUALLY selected model + "model": "cortiq-auto", // the model the CLIENT asked for, echoed back "choices": [ { "index": 0, "message": { "role": "assistant", "content": "x = 2 or x = 3" }, - "finish_reason": "stop" + "finish_reason": "stop" // always one of stop|length|tool_calls|content_filter } ], "usage": { @@ -86,10 +96,24 @@ Standard OpenAI fields are supported; unknown fields are proxied to the provider } ``` +> `model` echoes the id the client requested, exactly as OpenAI does — clients match the +> reply against the id they sent. The model that actually answered is in +> `cortiq.selected_model` and in the `X-Cortiq-Selected-Model` header. + +> Optional fields are omitted rather than sent as `null`: absent tool calls mean no +> `tool_calls` key at all. `"tool_calls": null` is not in the OpenAI schema and trips +> strict clients. + > The `cortiq` field is an optional extension. OpenAI SDKs ignore it. If the client does > not want extra fields, disable `cortiq.echo` — the metadata will remain in the response > headers. +Streaming (`stream: true`) emits strict `chat.completion.chunk` events: the stream always +opens with a `role` delta, carries exactly one terminal `finish_reason`, and ends with +`data: [DONE]` — even when the upstream stops early or dies mid-flight, in which case the +failure arrives as an SSE `error` event. A client can therefore never be left waiting on a +half-formed stream. + ### Metadata in response headers (always present) ``` @@ -164,20 +188,27 @@ single user message and follows the same pipeline. Response `object` is `"text_c ## 4. OpenAI Models `GET /v1/models` — auto-discovery for clients. Returns the gateway's virtual models -(`cortiq-auto`, `cortiq-embed`) and all real models in the pool. +(`cortiq-auto`, `cortiq-embed`), every model in the `[[models]]` pool, and every managed +local `[[cmf.servers]]` model, so a client can pin a local `.cmf` directly. ```jsonc { "object": "list", "data": [ - { "id": "cortiq-auto", "object": "model", "owned_by": "cortiq-gateway" }, - { "id": "local-qwen", "object": "model", "owned_by": "cortiq-gateway" }, - { "id": "gpt-4o-mini", "object": "model", "owned_by": "cortiq-gateway" }, - { "id": "claude-opus", "object": "model", "owned_by": "cortiq-gateway" } + { "id": "cortiq-auto", "object": "model", "created": 1718800000, "owned_by": "cortiq" }, + { "id": "gpt-4o-mini", "object": "model", "created": 1718800000, "owned_by": "openai" }, + { "id": "claude-opus", "object": "model", "created": 1718800000, "owned_by": "anthropic" }, + { "id": "cmf-qwen3-4b", "object": "model", "created": 1718800000, "owned_by": "cortiq-cmf", + "cortiq": { "kind": "cmf", "local": true, "running": true, "healthy": true } } ] } ``` +Managed CMF models are listed as soon as they are configured, with their live state in the +non-standard `cortiq` field (clients ignore it). They are not filtered by health: a large +`.cmf` takes minutes to load, and a model that disappears from the list while it warms up +is worse than one marked "not ready yet". + --- ## 5. Anthropic Messages @@ -198,7 +229,7 @@ single user message and follows the same pipeline. Response `object` is `"text_c "id": "msg_...", "type": "message", "role": "assistant", - "model": "local-qwen", + "model": "cortiq-auto", "content": [ { "type": "text", "text": "x = 2 or x = 3" } ], "stop_reason": "end_turn", "usage": { "input_tokens": 24, "output_tokens": 11 } diff --git a/docs/PROTOCOLS.ru.md b/docs/PROTOCOLS.ru.md index 3f356b7..ff0e4c8 100644 --- a/docs/PROTOCOLS.ru.md +++ b/docs/PROTOCOLS.ru.md @@ -53,7 +53,17 @@ ``` Поддерживаются стандартные поля OpenAI; неизвестные поля проксируются к провайдеру -без изменений (forward-compatible). +без изменений (forward-compatible) — кроме служебных ключей, которые чат-интерфейсы +добавляют в payload для собственных нужд (`chat_id`, `session_id`, `metadata`, +`features`, …): они отбрасываются, чтобы строгий upstream не отклонил вызов. + +Парсер намеренно терпим к тем формам, которые реально шлют клиенты: `content` может быть +строкой, `null` или массивом content-частей (текстовые части склеиваются, изображения и +прочие нетекстовые пропускаются); `tools` может быть `null`; `reasoning_effort` — строкой +(`low` | `medium` | `high`) или бюджетом токенов; `max_completion_tokens` принимается как +синоним `max_tokens`. Если тело всё-таки не читается, возвращается OpenAI-конверт ошибки с +HTTP 400 — но не «голый» 422, который OpenAI-клиенты не умеют разбирать (для пользователя +это выглядит как отсутствие ответа). ### Ответ (без стриминга) @@ -62,12 +72,12 @@ "id": "chatcmpl-9f...", "object": "chat.completion", "created": 1718800000, - "model": "local-qwen", // РЕАЛЬНО выбранная модель + "model": "cortiq-auto", // модель, которую ЗАПРОСИЛ клиент — возвращается как есть "choices": [ { "index": 0, "message": { "role": "assistant", "content": "x = 2 или x = 3" }, - "finish_reason": "stop" + "finish_reason": "stop" // всегда одно из stop|length|tool_calls|content_filter } ], "usage": { @@ -86,9 +96,22 @@ } ``` +> `model` возвращает тот id, который запросил клиент, — ровно как это делает OpenAI: +> клиенты сопоставляют ответ с отправленным id. Модель, которая реально ответила, — в +> `cortiq.selected_model` и в заголовке `X-Cortiq-Selected-Model`. + +> Необязательные поля не сериализуются вместо того, чтобы приходить как `null`: если +> tool-вызовов нет, ключа `tool_calls` в ответе нет вообще. `"tool_calls": null` не входит +> в схему OpenAI и ломает строгие клиенты. + > Поле `cortiq` — необязательное расширение. OpenAI-SDK его игнорируют. Если клиент > не хочет «лишних» полей — выключите `cortiq.echo`, метаданные останутся в заголовках. +Стриминг (`stream: true`) отдаёт строгие события `chat.completion.chunk`: поток всегда +открывается дельтой с `role`, содержит ровно один финальный `finish_reason` и завершается +`data: [DONE]` — даже если upstream оборвался или упал на середине (тогда ошибка приходит +SSE-событием `error`). Клиент никогда не остаётся ждать недосказанный поток. + ### Метаданные в заголовках (всегда) ``` @@ -163,20 +186,27 @@ data: [DONE] ## 4. OpenAI Models `GET /v1/models` — автодискавери для клиентов. Возвращает виртуальные модели шлюза -(`cortiq-auto`, `cortiq-embed`) и все реальные модели пула. +(`cortiq-auto`, `cortiq-embed`), все модели пула `[[models]]` и все управляемые локальные +модели `[[cmf.servers]]` — чтобы клиент мог напрямую закрепить локальный `.cmf`. ```jsonc { "object": "list", "data": [ - { "id": "cortiq-auto", "object": "model", "owned_by": "cortiq-gateway" }, - { "id": "local-qwen", "object": "model", "owned_by": "cortiq-gateway" }, - { "id": "gpt-4o-mini", "object": "model", "owned_by": "cortiq-gateway" }, - { "id": "claude-opus", "object": "model", "owned_by": "cortiq-gateway" } + { "id": "cortiq-auto", "object": "model", "created": 1718800000, "owned_by": "cortiq" }, + { "id": "gpt-4o-mini", "object": "model", "created": 1718800000, "owned_by": "openai" }, + { "id": "claude-opus", "object": "model", "created": 1718800000, "owned_by": "anthropic" }, + { "id": "cmf-qwen3-4b", "object": "model", "created": 1718800000, "owned_by": "cortiq-cmf", + "cortiq": { "kind": "cmf", "local": true, "running": true, "healthy": true } } ] } ``` +Управляемые CMF-модели попадают в список сразу, как только настроены; их живое состояние — +в нестандартном поле `cortiq` (клиенты его игнорируют). По здоровью они не фильтруются: +крупный `.cmf` грузится минутами, и модель, исчезающая из списка на время прогрева, хуже +модели, помеченной «ещё не готова». + --- ## 5. Anthropic Messages diff --git a/src/admin/mod.rs b/src/admin/mod.rs index caf901d..5582cf9 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -870,6 +870,7 @@ async fn probe_model(State(state): State, Path(id): Path) - role: "user".into(), content: "ping".into(), tool_calls: vec![], + ..Default::default() }], tools: vec![], params: GenParams { diff --git a/src/model.rs b/src/model.rs index aa840a2..27e0b1e 100644 --- a/src/model.rs +++ b/src/model.rs @@ -15,13 +15,53 @@ pub enum RoutingDirective { Pinned { model_id: String }, } -#[derive(Clone, Debug, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct Message { pub role: String, // system | user | assistant | tool - #[serde(default)] + /// Always a plain string internally. On the wire OpenAI clients legitimately + /// send `null` (assistant turns that only carry tool calls) or an array of + /// content parts (Open WebUI and every other multimodal UI), so + /// [`de_content`] flattens both shapes instead of rejecting the request. + #[serde(default, deserialize_with = "de_content")] pub content: String, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub tool_calls: Vec, + /// Speaker name (`messages[].name`) — proxied through when the client sends it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub name: Option, + /// Set on `role = "tool"` replies; dropping it breaks multi-turn tool calling. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_call_id: Option, +} + +/// Deserialize `messages[].content` from any shape an OpenAI-compatible client +/// may send: a string, `null`, or an array of content parts. Text parts are +/// concatenated; non-text parts (images, audio, files) are skipped — this +/// gateway routes text, and a picture must not turn into a 422. +fn de_content<'de, D>(d: D) -> std::result::Result +where + D: serde::Deserializer<'de>, +{ + use serde::Deserialize as _; + Ok(match serde_json::Value::deserialize(d)? { + serde_json::Value::Null => String::new(), + serde_json::Value::String(s) => s, + serde_json::Value::Array(parts) => { + let mut out = String::new(); + for part in parts { + // {"type":"text","text":"..."} / {"type":"input_text","text":"..."} + let text = part + .get("text") + .and_then(|t| t.as_str()) + .or_else(|| part.as_str()); + if let Some(t) = text { + out.push_str(t); + } + } + out + } + other => other.to_string(), + }) } #[derive(Clone, Debug, Default)] diff --git a/src/pipeline.rs b/src/pipeline.rs index a32d1ec..0721156 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -430,6 +430,7 @@ impl Pipeline { role: "system".into(), content: sys.clone(), tool_calls: Vec::new(), + ..Default::default() }, ); } @@ -457,6 +458,7 @@ impl Pipeline { role: "system".into(), content: sys.clone(), tool_calls: Vec::new(), + ..Default::default() }, ); } @@ -564,6 +566,7 @@ impl Pipeline { role: "system".into(), content: sys.clone(), tool_calls: Vec::new(), + ..Default::default() }, ); } @@ -591,6 +594,7 @@ impl Pipeline { role: "system".into(), content: sys.clone(), tool_calls: Vec::new(), + ..Default::default() }, ); } @@ -734,6 +738,7 @@ fn cached_response(hit: &crate::cache::CacheHit) -> ChatResponse { role: "assistant".into(), content: hit.content.clone(), tool_calls: vec![], + ..Default::default() }, finish_reason: "stop".into(), }], @@ -956,6 +961,7 @@ fn shadow_req(model_id: &str, content: String, max_tokens: u32) -> ChatRequest { role: "user".to_string(), content, tool_calls: Vec::new(), + ..Default::default() }], tools: Vec::new(), params: GenParams { diff --git a/src/protocols/anthropic_messages.rs b/src/protocols/anthropic_messages.rs index 332cf60..342aa4d 100644 --- a/src/protocols/anthropic_messages.rs +++ b/src/protocols/anthropic_messages.rs @@ -5,7 +5,7 @@ use crate::error::{GatewayError, Result}; use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective}; -use crate::protocols::openai_chat::cortiq_headers; +use crate::protocols::openai_chat::{cortiq_headers, sse_headers}; use crate::state::SharedState; use axum::response::{IntoResponse, Response}; use axum::{extract::State, routing::post, Json, Router}; @@ -84,6 +84,7 @@ async fn handler( role: "system".into(), content: sys, tool_calls: vec![], + ..Default::default() }); } for m in &req.messages { @@ -91,9 +92,12 @@ async fn handler( role: m["role"].as_str().unwrap_or("user").to_string(), content: content_to_string(&m["content"]), tool_calls: vec![], + ..Default::default() }); } + // Echoed back as-is: clients match the reply against the id they asked for. + let requested_model = req.model.clone(); let canonical = ChatRequest { routing: parse_routing(&req.model), messages, @@ -116,13 +120,12 @@ async fn handler( if req.stream { let (info, stream) = state.pipeline.run_stream(canonical, &state).await?; - let mut headers = cortiq_headers(&info); - headers.insert( - axum::http::header::CONTENT_TYPE, - axum::http::HeaderValue::from_static("text/event-stream"), - ); - let translated = openai_to_anthropic_sse(stream); - return Ok((headers, axum::body::Body::from_stream(translated)).into_response()); + let translated = openai_to_anthropic_sse(stream, requested_model); + return Ok(( + sse_headers(&info), + axum::body::Body::from_stream(translated), + ) + .into_response()); } let resp = state.pipeline.run(canonical, &state).await?; @@ -136,7 +139,7 @@ async fn handler( "id": resp.id, "type": "message", "role": "assistant", - "model": resp.model_used, + "model": requested_model, "content": [{ "type": "text", "text": answer }], "stop_reason": "end_turn", "stop_sequence": serde_json::Value::Null, @@ -152,6 +155,7 @@ async fn handler( /// events, so Anthropic-native clients can stream regardless of the upstream provider. fn openai_to_anthropic_sse( stream: crate::providers::ChatStream, + model: String, ) -> impl futures::Stream> + Send + 'static { use futures::StreamExt; async_stream::stream! { @@ -162,7 +166,7 @@ fn openai_to_anthropic_sse( yield Ok(evt("message_start", serde_json::json!({ "type": "message_start", "message": {"id": "msg_gw", "type": "message", "role": "assistant", - "content": [], "model": "cortiq-auto", "stop_reason": null, + "content": [], "model": model, "stop_reason": null, "usage": {"input_tokens": 0, "output_tokens": 0}} }))); yield Ok(evt("content_block_start", serde_json::json!({ diff --git a/src/protocols/mcp.rs b/src/protocols/mcp.rs index e2e1284..3bf3966 100644 --- a/src/protocols/mcp.rs +++ b/src/protocols/mcp.rs @@ -94,6 +94,7 @@ async fn handle_tool_call( role: "user".into(), content: prompt.to_string(), tool_calls: vec![], + ..Default::default() }], tools: vec![], params: GenParams::default(), diff --git a/src/protocols/openai_chat.rs b/src/protocols/openai_chat.rs index 7f1c1a8..6b29f70 100644 --- a/src/protocols/openai_chat.rs +++ b/src/protocols/openai_chat.rs @@ -1,6 +1,17 @@ //! Incoming adapter for OpenAI Chat Completions: `POST /v1/chat/completions`. //! Translates the OpenAI request body into a canonical [`ChatRequest`], passes it //! to the pipeline, then converts the canonical [`ChatResponse`] back to OpenAI format. +//! +//! Wire-compatibility rules this adapter enforces (see docs/PROTOCOLS.md §2): +//! * the request body is parsed leniently — every shape a real OpenAI client +//! sends (`content: null`, content-part arrays, `tools: null`, a string +//! `reasoning_effort`, UI-private extras) is accepted, and a body we truly +//! cannot read comes back as an OpenAI error envelope, never a bare 422; +//! * the response is strict OpenAI JSON — no optional `null` fields, a +//! spec-legal `finish_reason`, and the *requested* model id echoed back; +//! * the SSE stream is re-normalised chunk by chunk and always terminates +//! with `data: [DONE]`, so a client can never sit waiting on a half-formed +//! stream from an upstream that improvises. use crate::error::{GatewayError, Result}; use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RouteInfo, RoutingDirective}; @@ -34,32 +45,113 @@ pub(crate) fn cortiq_headers(c: &RouteInfo) -> axum::http::HeaderMap { h } +/// Headers every SSE response carries. `X-Accel-Buffering: no` matters when the +/// gateway sits behind nginx/Traefik — without it the proxy buffers the whole +/// stream and the client's chat window just spins. +pub(crate) fn sse_headers(info: &RouteInfo) -> axum::http::HeaderMap { + use axum::http::HeaderValue; + let mut headers = cortiq_headers(info); + headers.insert( + axum::http::header::CONTENT_TYPE, + HeaderValue::from_static("text/event-stream; charset=utf-8"), + ); + headers.insert( + axum::http::header::CACHE_CONTROL, + HeaderValue::from_static("no-cache"), + ); + headers.insert("X-Accel-Buffering", HeaderValue::from_static("no")); + headers +} + pub fn routes() -> Router { Router::new().route("/v1/chat/completions", post(handler)) } +/// `reasoning_effort` is a string in the OpenAI spec (`low` | `medium` | `high`) +/// but a token budget in several local runtimes. Accept both. +#[derive(Deserialize)] +#[serde(untagged)] +enum Effort { + Budget(u32), + Level(String), +} + +impl Effort { + fn to_budget(&self) -> Option { + match self { + Self::Budget(n) => Some(*n), + Self::Level(s) => match s.trim().to_ascii_lowercase().as_str() { + "none" | "off" | "minimal" => Some(0), + "low" => Some(512), + "medium" | "default" | "auto" => Some(2048), + "high" | "max" => Some(8192), + _ => None, + }, + } + } +} + #[derive(Deserialize)] struct OpenAiChatRequest { - model: String, + model: Option, messages: Vec, #[serde(default)] temperature: Option, #[serde(default)] max_tokens: Option, + /// The modern spelling of `max_tokens`; clients send one or the other. + #[serde(default)] + max_completion_tokens: Option, #[serde(default)] top_p: Option, #[serde(default)] think_budget: Option, #[serde(default)] - reasoning_effort: Option, + reasoning_effort: Option, + /// `null` is a legal value here for several clients, hence `Option`. #[serde(default)] - stream: bool, + stream: Option, #[serde(default)] - tools: Vec, + tools: Option>, #[serde(flatten)] rest: serde_json::Map, } +/// Keys chat UIs attach to the payload for their own bookkeeping. They are not +/// part of the OpenAI request schema, and forwarding them to a strict upstream +/// (llama.cpp, vLLM, `cortiq serve`) is a needless way to earn a 400. +const CLIENT_PRIVATE_KEYS: &[&str] = &[ + "chat_id", + "session_id", + "id", + "metadata", + "background_tasks", + "features", + "variables", + "model_item", + "tool_ids", + "filter_ids", + "files", + "citations", + "params", + "direct", +]; + +/// Strip client-private bookkeeping from the passthrough map, and drop a +/// non-string `user` (Open WebUI sends an object there for pipelines, which the +/// OpenAI schema does not allow). +fn sanitize_passthrough( + mut rest: serde_json::Map, +) -> serde_json::Map { + for k in CLIENT_PRIVATE_KEYS { + rest.remove(*k); + } + if rest.get("user").map(|u| !u.is_string()).unwrap_or(false) { + rest.remove("user"); + } + rest +} + /// Parse the `model` field value into a routing directive. /// `cortiq-auto[:profile]` → Auto; otherwise → Pinned(real id). fn parse_routing(model: &str) -> RoutingDirective { @@ -73,10 +165,43 @@ fn parse_routing(model: &str) -> RoutingDirective { } } +fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +/// Map whatever an upstream calls its stop reason onto the closed set the +/// OpenAI schema defines. Clients that validate the field (LiteLLM, the +/// official SDKs, several agent frameworks) reject anything else. +pub(crate) fn normalize_finish_reason(raw: &str) -> &'static str { + match raw.trim().to_ascii_lowercase().as_str() { + "length" | "max_tokens" | "max_new_tokens" | "token_limit" => "length", + "tool_calls" | "tool_use" | "function_call" => "tool_calls", + "content_filter" | "safety" => "content_filter", + _ => "stop", + } +} + +/// A response id every client can key on. Upstreams that omit `id` (or send an +/// empty one) would otherwise leave the field blank, which some clients treat +/// as a malformed completion. +fn response_id(upstream: &str) -> String { + let trimmed = upstream.trim(); + if !trimmed.is_empty() { + return trimmed.to_string(); + } + format!("chatcmpl-{}", crate::admin::random_token(12)) +} + async fn handler( State(state): State, acct: Option>, - Json(req): Json, + // Taken as raw bytes rather than `Json`: axum's extractor rejection is a + // plain-text 422 that OpenAI clients cannot parse, and a client that cannot + // parse the failure has no answer to show — it just hangs. + body: axum::body::Bytes, ) -> Result { // hot protocol toggle: if the adapter is disabled in config — return 404 if !state.live().cfg.protocols.openai_chat { @@ -84,25 +209,43 @@ async fn handler( "openai_chat protocol is disabled".into(), )); } + let req: OpenAiChatRequest = serde_json::from_slice(&body) + .map_err(|e| GatewayError::InvalidRequest(format!("invalid request body: {e}")))?; + if req.messages.is_empty() { return Err(GatewayError::InvalidRequest( "messages must not be empty".into(), )); } + // Echoed back verbatim: OpenAI returns the model the caller asked for, and + // clients (Open WebUI among them) match the reply against their own id. The + // model that actually answered travels in `cortiq.selected_model` and the + // `X-Cortiq-Selected-Model` header. + let requested_model = req + .model + .as_deref() + .map(str::trim) + .filter(|m| !m.is_empty()) + .unwrap_or("cortiq-auto") + .to_string(); + let stream = req.stream.unwrap_or(false); + let canonical = ChatRequest { - routing: parse_routing(&req.model), + routing: parse_routing(&requested_model), messages: req.messages, - tools: req.tools, + tools: req.tools.unwrap_or_default(), params: GenParams { temperature: req.temperature, - max_tokens: req.max_tokens, + max_tokens: req.max_tokens.or(req.max_completion_tokens), top_p: req.top_p, - think_budget: req.think_budget.or(req.reasoning_effort), + think_budget: req + .think_budget + .or_else(|| req.reasoning_effort.as_ref().and_then(Effort::to_budget)), stop: Vec::new(), - passthrough: req.rest, + passthrough: sanitize_passthrough(req.rest), }, - stream: req.stream, + stream, meta: RequestMeta { protocol: "openai_chat".into(), account: acct.map(|e| e.0 .0.clone()).unwrap_or_default(), @@ -110,43 +253,51 @@ async fn handler( }, }; - // streaming: forward provider SSE chunks verbatim with X-Cortiq-* headers - if req.stream { - let (info, stream) = state.pipeline.run_stream(canonical, &state).await?; - let mut headers = cortiq_headers(&info); - headers.insert( - axum::http::header::CONTENT_TYPE, - axum::http::HeaderValue::from_static("text/event-stream"), - ); - headers.insert( - axum::http::header::CACHE_CONTROL, - axum::http::HeaderValue::from_static("no-cache"), - ); - return Ok((headers, axum::body::Body::from_stream(stream)).into_response()); + // streaming: provider SSE is re-normalised into strict OpenAI chunks + if stream { + let (info, raw) = state.pipeline.run_stream(canonical, &state).await?; + let headers = sse_headers(&info); + let normalized = normalize_sse(raw, requested_model); + return Ok((headers, axum::body::Body::from_stream(normalized)).into_response()); } let resp = state.pipeline.run(canonical, &state).await?; let headers = cortiq_headers(&resp.cortiq); - let mut body = serde_json::json!({ - "id": resp.id, - "object": "chat.completion", - "created": std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(), - "model": resp.model_used, - "choices": resp.choices.iter().map(|c| { + let choices: Vec = resp + .choices + .iter() + .map(|c| { + let mut message = serde_json::Map::new(); + message.insert("role".into(), serde_json::json!(c.message.role)); + message.insert("content".into(), serde_json::json!(c.message.content)); + // Absent tool calls are omitted entirely — `"tool_calls": null` is + // not in the schema and trips strict clients. + if !c.message.tool_calls.is_empty() { + message.insert( + "tool_calls".into(), + serde_json::Value::Array(c.message.tool_calls.clone()), + ); + } + let finish = if c.message.tool_calls.is_empty() { + normalize_finish_reason(&c.finish_reason) + } else { + "tool_calls" + }; serde_json::json!({ "index": c.index, - "message": { - "role": c.message.role, - "content": c.message.content, - "tool_calls": if c.message.tool_calls.is_empty() { serde_json::Value::Null } else { serde_json::Value::Array(c.message.tool_calls.clone()) }, - }, - "finish_reason": c.finish_reason, + "message": serde_json::Value::Object(message), + "finish_reason": finish, }) - }).collect::>(), + }) + .collect(); + + let mut body = serde_json::json!({ + "id": response_id(&resp.id), + "object": "chat.completion", + "created": now_secs(), + "model": requested_model, + "choices": choices, "usage": { "prompt_tokens": resp.usage.prompt_tokens, "completion_tokens": resp.usage.completion_tokens, @@ -155,7 +306,7 @@ async fn handler( }); if state.live().cfg.cortiq.echo { - body["cortiq"] = serde_json::json!({ + let mut echo = serde_json::json!({ "task_label": resp.cortiq.task_label, "complexity": { "score": resp.cortiq.complexity_score, @@ -163,10 +314,230 @@ async fn handler( }, "selected_model": resp.cortiq.selected_model, "route_source": resp.cortiq.route_source, - "router_request_id": resp.cortiq.router_request_id, "cost_usd": resp.cortiq.cost_usd, }); + if let Some(rid) = &resp.cortiq.router_request_id { + echo["router_request_id"] = serde_json::json!(rid); + } + body["cortiq"] = echo; } Ok((headers, Json(body)).into_response()) } + +/// Rewrite an upstream SSE stream into strict OpenAI `chat.completion.chunk` +/// events. +/// +/// Every chunk is reissued with an id, `created`, and the *requested* model id; +/// `message`-shaped chunks are converted to `delta`; stop reasons are mapped to +/// the spec's closed set; empty deltas are dropped. The stream is guaranteed to +/// open with a `role` delta, to carry exactly one terminal `finish_reason`, and +/// to end with `data: [DONE]` — even when the upstream dies mid-flight, in +/// which case the failure is delivered as an SSE `error` event so the client +/// shows a message instead of waiting forever. +fn normalize_sse( + stream: crate::providers::ChatStream, + model: String, +) -> impl futures::Stream> + Send + 'static { + use futures::StreamExt; + async_stream::stream! { + let fallback_id = response_id(""); + let created = now_secs(); + let mut buf = String::new(); + let mut id: Option = None; + // Which choice indices already carried a `role` delta (n > 1 gets one each). + let mut roled: std::collections::HashSet = std::collections::HashSet::new(); + let mut sent_finish = false; + let mut failure: Option = None; + + // Emit one normalised event; `None` means "nothing worth sending". + macro_rules! chunk { + ($choices:expr, $extra:expr) => {{ + let mut evt = serde_json::json!({ + "id": id.clone().unwrap_or_else(|| fallback_id.clone()), + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": $choices, + }); + for (k, v) in $extra { + evt[k] = v; + } + bytes::Bytes::from(format!("data: {evt}\n\n")) + }}; + } + + futures::pin_mut!(stream); + 'outer: while let Some(item) = stream.next().await { + let bytes = match item { + Ok(b) => b, + Err(e) => { + failure = Some(e.to_string()); + break 'outer; + } + }; + buf.push_str(&String::from_utf8_lossy(&bytes)); + + while let Some(idx) = buf.find("\n\n") { + let event: String = buf.drain(..idx + 2).collect(); + for line in event.lines() { + let Some(data) = line.trim_start().strip_prefix("data:") else { continue }; + let data = data.trim(); + if data.is_empty() || data == "[DONE]" { + continue; + } + let Ok(v) = serde_json::from_str::(data) else { continue }; + + // An upstream error delivered inside the stream: pass it on + // and stop — the client needs to see it, not time out. + if let Some(err) = v.get("error").filter(|e| !e.is_null()) { + failure = Some(err.to_string()); + break 'outer; + } + + if id.is_none() { + id = v["id"].as_str().filter(|s| !s.is_empty()).map(str::to_string); + } + + let mut choices = Vec::new(); + if let Some(arr) = v["choices"].as_array() { + for (i, c) in arr.iter().enumerate() { + // Some upstreams send a full `message` per chunk + // instead of a `delta`; both mean the same thing here. + let src = if c["delta"].is_object() { &c["delta"] } else { &c["message"] }; + let index = c["index"].as_u64().unwrap_or(i as u64); + let mut delta = serde_json::Map::new(); + if roled.insert(index) { + delta.insert("role".into(), serde_json::json!("assistant")); + } + if let Some(text) = src["content"].as_str() { + if !text.is_empty() { + delta.insert("content".into(), serde_json::json!(text)); + } + } + // Reasoning traces and citations are not in the base + // schema but every client that understands them + // ignores them safely, so they are worth keeping. + for key in ["reasoning_content", "reasoning", "thinking", "refusal", "annotations"] { + if let Some(val) = src.get(key).filter(|v| !v.is_null()) { + delta.insert(key.into(), val.clone()); + } + } + if let Some(tc) = src["tool_calls"].as_array().filter(|a| !a.is_empty()) { + delta.insert("tool_calls".into(), serde_json::Value::Array(tc.clone())); + } + + let finish = c["finish_reason"].as_str().map(normalize_finish_reason); + if finish.is_none() && delta.is_empty() { + continue; + } + let mut choice = serde_json::json!({ + "index": index, + "delta": serde_json::Value::Object(delta), + }); + if let Some(f) = finish { + choice["finish_reason"] = serde_json::json!(f); + sent_finish = true; + } + choices.push(choice); + } + } + + // 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"] { + if let Some(val) = v.get(key).filter(|v| !v.is_null()) { + extra.push((key, val.clone())); + } + } + + if !choices.is_empty() || !extra.is_empty() { + yield Ok(chunk!(choices, extra)); + } + } + } + } + + if let Some(message) = failure { + let evt = serde_json::json!({ + "error": { "message": message, "type": "upstream_unavailable", "code": "stream_failed" } + }); + yield Ok(bytes::Bytes::from(format!("data: {evt}\n\n"))); + } + + // Close the stream properly whatever the upstream did: a client that + // never sees a stop reason and a [DONE] keeps its spinner turning. + if !sent_finish { + let mut delta = serde_json::Map::new(); + if roled.insert(0) { + delta.insert("role".into(), serde_json::json!("assistant")); + } + let choices = vec![serde_json::json!({ + "index": 0, + "delta": serde_json::Value::Object(delta), + "finish_reason": "stop", + })]; + let extra: Vec<(&str, serde_json::Value)> = Vec::new(); + yield Ok(chunk!(choices, extra)); + } + yield Ok(bytes::Bytes::from("data: [DONE]\n\n")); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn content_parts_and_null_are_accepted() { + let body = serde_json::json!({ + "model": "cortiq-auto", + "messages": [ + {"role": "user", "content": [{"type": "text", "text": "hello "}, {"type": "image_url", "image_url": {"url": "x"}}, {"type": "text", "text": "world"}]}, + {"role": "assistant", "content": null}, + {"role": "tool", "content": "42", "tool_call_id": "call_1"} + ], + "tools": null, + "reasoning_effort": "high", + "chat_id": "owui-123" + }); + let req: OpenAiChatRequest = serde_json::from_value(body).expect("must parse"); + assert_eq!(req.messages[0].content, "hello world"); + assert_eq!(req.messages[1].content, ""); + assert_eq!(req.messages[2].tool_call_id.as_deref(), Some("call_1")); + assert!(req.tools.unwrap_or_default().is_empty()); + assert_eq!( + req.reasoning_effort.as_ref().and_then(Effort::to_budget), + Some(8192) + ); + assert!(!sanitize_passthrough(req.rest).contains_key("chat_id")); + } + + #[test] + fn finish_reasons_map_to_the_openai_set() { + assert_eq!(normalize_finish_reason("max_tokens"), "length"); + assert_eq!(normalize_finish_reason("eos"), "stop"); + assert_eq!(normalize_finish_reason("tool_use"), "tool_calls"); + } + + #[tokio::test] + async fn stream_is_closed_even_when_the_upstream_stops_early() { + use futures::StreamExt; + 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()) + .map(|b| String::from_utf8_lossy(&b.unwrap()).to_string()) + .collect() + .await; + let joined = out.join(""); + assert!(joined.contains("\"role\":\"assistant\"")); + assert!(joined.contains("\"model\":\"cortiq-auto\"")); + assert!(joined.contains("\"finish_reason\":\"stop\"")); + assert!(joined.trim_end().ends_with("data: [DONE]")); + } +} diff --git a/src/protocols/openai_completions.rs b/src/protocols/openai_completions.rs index a180609..96d97ad 100644 --- a/src/protocols/openai_completions.rs +++ b/src/protocols/openai_completions.rs @@ -4,7 +4,7 @@ use crate::error::{GatewayError, Result}; use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective}; -use crate::protocols::openai_chat::cortiq_headers; +use crate::protocols::openai_chat::{cortiq_headers, normalize_finish_reason, sse_headers}; use crate::state::SharedState; use axum::response::{IntoResponse, Response}; use axum::{extract::State, routing::post, Json, Router}; @@ -71,12 +71,15 @@ async fn handler( )); } + // Echoed back as-is: clients match the reply against the id they asked for. + let requested_model = req.model.clone(); let canonical = ChatRequest { routing: parse_routing(&req.model), messages: vec![Message { role: "user".into(), content: prompt, tool_calls: vec![], + ..Default::default() }], tools: vec![], params: GenParams { @@ -97,13 +100,8 @@ async fn handler( if req.stream { let (info, stream) = state.pipeline.run_stream(canonical, &state).await?; - let mut headers = cortiq_headers(&info); - headers.insert( - axum::http::header::CONTENT_TYPE, - axum::http::HeaderValue::from_static("text/event-stream"), - ); return Ok(( - headers, + sse_headers(&info), axum::body::Body::from_stream(chat_to_completions_sse(stream)), ) .into_response()); @@ -115,11 +113,12 @@ async fn handler( .first() .map(|c| c.message.content.clone()) .unwrap_or_default(); - let finish = resp - .choices - .first() - .map(|c| c.finish_reason.clone()) - .unwrap_or_else(|| "stop".into()); + let finish = normalize_finish_reason( + resp.choices + .first() + .map(|c| c.finish_reason.as_str()) + .unwrap_or("stop"), + ); let headers = cortiq_headers(&resp.cortiq); let body = serde_json::json!({ "id": resp.id, @@ -128,7 +127,7 @@ async fn handler( .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs(), - "model": resp.model_used, + "model": requested_model, "choices": [{ "text": answer, "index": 0, "logprobs": serde_json::Value::Null, "finish_reason": finish }], "usage": { "prompt_tokens": resp.usage.prompt_tokens, diff --git a/src/protocols/openai_models.rs b/src/protocols/openai_models.rs index 43326f1..1c7e06c 100644 --- a/src/protocols/openai_models.rs +++ b/src/protocols/openai_models.rs @@ -1,5 +1,10 @@ //! Inbound OpenAI Models adapter: `GET /v1/models`. -//! Lists the configured pool (plus the virtual `cortiq-auto`) for client discovery. +//! Lists everything a client can address: the virtual `cortiq-auto`, the +//! configured `[[models]]` pool, and the managed local `[[cmf.servers]]`. +//! +//! The managed CMF servers belong here because they are routable model ids like +//! any other — omitting them left OpenAI clients (Open WebUI among them) unable +//! to pin a local model even though the gateway would happily serve it. use crate::error::{GatewayError, Result}; use crate::state::SharedState; @@ -9,6 +14,13 @@ pub fn routes() -> Router { Router::new().route("/v1/models", get(handler)) } +fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + async fn handler(State(state): State) -> Result { let live = state.live(); if !live.cfg.protocols.openai_models { @@ -16,17 +28,51 @@ async fn handler(State(state): State) -> Result "openai_models protocol is disabled".into(), )); } + let created = now_secs(); let mut data = vec![serde_json::json!({ "id": "cortiq-auto", "object": "model", + "created": created, "owned_by": "cortiq", })]; + let mut seen: std::collections::HashSet = + std::collections::HashSet::from(["cortiq-auto".to_string()]); + for m in &live.cfg.models { + if !seen.insert(m.id.clone()) { + continue; + } data.push(serde_json::json!({ "id": m.id, "object": "model", + "created": created, "owned_by": m.provider, })); } + + // Managed local CMF servers. Health is reported alongside (a non-standard + // field clients ignore) rather than used as a filter: a large `.cmf` takes + // minutes to load, and a model that vanishes from the list while it warms up + // is worse than one marked "not ready yet". + let cmf = state.cmf.status(); + for s in live.cfg.cmf.effective_servers() { + if !seen.insert(s.id.clone()) { + continue; // a static [[models]] entry with the same id wins + } + let status = cmf.servers.iter().find(|st| st.id == s.id); + data.push(serde_json::json!({ + "id": s.id, + "object": "model", + "created": created, + "owned_by": "cortiq-cmf", + "cortiq": { + "kind": "cmf", + "local": true, + "running": status.map(|st| st.running).unwrap_or(false), + "healthy": status.map(|st| st.healthy).unwrap_or(false), + }, + })); + } + Ok(Json(serde_json::json!({ "object": "list", "data": data }))) } diff --git a/src/providers/anthropic.rs b/src/providers/anthropic.rs index fcab748..bad2299 100644 --- a/src/providers/anthropic.rs +++ b/src/providers/anthropic.rs @@ -146,6 +146,7 @@ impl Provider for AnthropicProvider { role: "assistant".into(), content, tool_calls: vec![], + ..Default::default() }, finish_reason: v["stop_reason"].as_str().unwrap_or("stop").to_string(), }], diff --git a/src/providers/openai.rs b/src/providers/openai.rs index 2c3ade6..c0d1123 100644 --- a/src/providers/openai.rs +++ b/src/providers/openai.rs @@ -54,13 +54,11 @@ impl OpenAiProvider { if let Some(m) = req.params.max_tokens { body["max_tokens"] = m.into(); } - eprintln!("[to_wire] think_budget={:?}", req.params.think_budget); if let Some(tb) = req.params.think_budget { body["think_budget"] = tb.into(); body["reasoning_effort"] = tb.into(); // cortiq serve uses enable_thinking (bool): 0 = off, >0 = on body["enable_thinking"] = (tb > 0).into(); - eprintln!("[to_wire] SET enable_thinking={}", tb > 0); } if !req.tools.is_empty() { body["tools"] = serde_json::Value::Array(req.tools.clone()); @@ -69,10 +67,6 @@ impl OpenAiProvider { for (k, v) in &req.params.passthrough { body[k] = v.clone(); } - eprintln!( - "[to_wire] enable_thinking={:?}", - body.get("enable_thinking") - ); body } } @@ -141,6 +135,7 @@ impl Provider for OpenAiProvider { role, content, tool_calls, + ..Default::default() }, finish_reason: c["finish_reason"].as_str().unwrap_or("stop").to_string(), });