Compare commits

...
1 Commits
Author SHA1 Message Date
infosave2007 b339c2b622 fix(openai): strict OpenAI wire compatibility — Open WebUI can chat again
Docker / publish (push) Canceled after 0s
Release / crates-io (push) Canceled after 0s
Release / release (macos-14, aarch64-apple-darwin) (push) Canceled after 0s
Release / release (ubuntu-latest, x86_64-unknown-linux-gnu) (push) Canceled after 0s
Release / release (windows-latest, x86_64-pc-windows-msvc) (push) Canceled after 0s
/v1/chat/completions answered ordinary client payloads with a bare HTTP 422
(a plain-text body no OpenAI client can parse), so Open WebUI showed neither
an answer nor an error — it just hung. Parse the body leniently instead:
content as string/null/content-part array, tools: null, stream: null, a
string reasoning_effort, max_completion_tokens, messages[].name and
tool_call_id, and a missing Content-Type. Unreadable bodies now come back as
an OpenAI error envelope with 400.

Responses are strict OpenAI JSON: no "tool_calls": null, finish_reason mapped
to the spec's closed set (cortiq serve's max_tokens -> length), never an empty
id, and `model` echoes the id the client requested — the model that actually
answered stays in cortiq.selected_model and X-Cortiq-Selected-Model.

Streaming is re-normalised rather than forwarded verbatim: a leading role
delta, message-shaped chunks converted to delta, one terminal finish_reason,
always a trailing data: [DONE], and an SSE error event when the upstream dies
mid-flight. Adds X-Accel-Buffering: no so a fronting proxy cannot buffer the
stream into a hang.

GET /v1/models now lists managed [[cmf.servers]] models with their live
running/healthy state, so clients can pin a local .cmf directly.

Also strips client bookkeeping keys (chat_id, session_id, metadata, features,
...) from the upstream call and drops the leftover [to_wire] debug prints.
2026-08-21 09:10:31 +03:00
15 changed files with 660 additions and 93 deletions
+42
View File
@@ -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
Generated
+1 -1
View File
@@ -285,7 +285,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
[[package]]
name = "cortiq-gateway"
version = "0.2.44"
version = "0.2.45"
dependencies = [
"anyhow",
"arc-swap",
+1 -1
View File
@@ -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"
+40 -9
View File
@@ -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 }
+38 -8
View File
@@ -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
+1
View File
@@ -870,6 +870,7 @@ async fn probe_model(State(state): State<SharedState>, Path(id): Path<String>) -
role: "user".into(),
content: "ping".into(),
tool_calls: vec![],
..Default::default()
}],
tools: vec![],
params: GenParams {
+42 -2
View File
@@ -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<serde_json::Value>,
/// Speaker name (`messages[].name`) — proxied through when the client sends it.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub name: Option<String>,
/// 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<String>,
}
/// 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<String, D::Error>
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)]
+6
View File
@@ -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 {
+14 -10
View File
@@ -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<Item = Result<bytes::Bytes>> + 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!({
+1
View File
@@ -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(),
+413 -42
View File
@@ -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<SharedState> {
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<u32> {
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<String>,
messages: Vec<Message>,
#[serde(default)]
temperature: Option<f32>,
#[serde(default)]
max_tokens: Option<u32>,
/// The modern spelling of `max_tokens`; clients send one or the other.
#[serde(default)]
max_completion_tokens: Option<u32>,
#[serde(default)]
top_p: Option<f32>,
#[serde(default)]
think_budget: Option<u32>,
#[serde(default)]
reasoning_effort: Option<u32>,
reasoning_effort: Option<Effort>,
/// `null` is a legal value here for several clients, hence `Option`.
#[serde(default)]
stream: bool,
stream: Option<bool>,
#[serde(default)]
tools: Vec<serde_json::Value>,
tools: Option<Vec<serde_json::Value>>,
#[serde(flatten)]
rest: serde_json::Map<String, serde_json::Value>,
}
/// 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<String, serde_json::Value>,
) -> serde_json::Map<String, serde_json::Value> {
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<SharedState>,
acct: Option<axum::Extension<super::AccountTag>>,
Json(req): Json<OpenAiChatRequest>,
// Taken as raw bytes rather than `Json<T>`: 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<Response> {
// 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<serde_json::Value> = 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::<Vec<_>>(),
})
.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<Item = Result<bytes::Bytes>> + 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<String> = None;
// Which choice indices already carried a `role` delta (n > 1 gets one each).
let mut roled: std::collections::HashSet<u64> = std::collections::HashSet::new();
let mut sent_finish = false;
let mut failure: Option<String> = 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::<serde_json::Value>(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<String> = 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]"));
}
}
+12 -13
View File
@@ -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,
+47 -1
View File
@@ -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<SharedState> {
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<SharedState>) -> Result<impl IntoResponse> {
let live = state.live();
if !live.cfg.protocols.openai_models {
@@ -16,17 +28,51 @@ async fn handler(State(state): State<SharedState>) -> Result<impl IntoResponse>
"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<String> =
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 })))
}
+1
View File
@@ -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(),
}],
+1 -6
View File
@@ -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(),
});