Compare commits

..
4 Commits
Author SHA1 Message Date
infosave2007 0b071279ac fix(stream): no choice-less chunks unless the client asks for usage
CI / lint (push) Canceled after 0s
CI / test (macos-14) (push) Canceled after 0s
CI / test (ubuntu-latest) (push) Canceled after 0s
CI / test (windows-latest) (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
Docker / publish (push) Canceled after 0s
Every stream used to carry two chunks with an empty choices array: usage,
which the upstream emits BEFORE the stop reason, and a cortiq_cost_usd chunk
the gateway added itself. OpenAI ships the usage chunk only when the request
sets stream_options: {"include_usage": true}, and chat UIs do not set it -
which is why the same tool, in the same Open WebUI, sees neither chunk from
Groq and two from us.

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

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

Verified: the default stream matches the reference shape byte for byte, the
opt-in stream matches OpenAI's, the official OpenAI SDK reassembles the
streamed tool call, and the full Open WebUI 0.11.0 tool loop still runs the
tool and answers from its result.
2026-08-21 21:24:11 +03:00
infosave2007 9fd4bdaa85 fix(tools): never deliver a tool call twice, and make an unparsed one executable
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
Qwen-family templates write a function call into the generated text as
<tool_call>{...}</tool_call>. Runtimes disagree on what to do with it: some
parse it into structured tool_calls AND still stream the raw markup as
content - the client then gets the same call twice, once as prose it must not
show and once as the real thing - while others never parse it at all, so a
perfectly good call arrives as text no tool loop will run.

Settle it in the gateway. A streaming-aware filter strips the markup from
content (handling a marker split across SSE chunks, leaving ordinary text
containing "<" alone, and handing an unterminated block back as text), and
when nothing structured came with it the markup is promoted into a real
tool_calls entry with a generated call_... id. Markup that is not a JSON
object with a name is ignored so a stray marker cannot become a bogus call.

Streamed tool_calls are also brought to the OpenAI shape: index on every
entry, id and type exactly once per call so continuation deltas carrying only
an argument fragment stay legal, and function.arguments always a string. A
reply that carries tool calls now always ends with finish_reason "tool_calls",
whatever the upstream called it. The same cleanup runs on the non-streaming
path, so every protocol adapter benefits.

Verified end to end: Open WebUI 0.11.0 -> gateway -> managed CMF/Qwen with
native function calling. The model returns get_weather({"city":"Paris"}), Open
WebUI executes the tool, feeds the result back, and the assistant answers from
it - no markup in content, finish_reason tool_calls, stream closed with [DONE].
2026-08-21 18:49:19 +03:00
infosave2007 9c15109f26 chore(router): address the hosted router by hostname, TLS verification on
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
router.allaigate.com now resolves and serves a valid certificate, so the
hosted routing API no longer has to be reached by bare IP with certificate
verification switched off. Make https://router.allaigate.com the default
[router].url with verify_tls = true.

Existing configs still pointing at 138.226.222.209 (with or without a scheme
or a trailing slash) are moved to the hostname on load and have verification
restored, then saved: it is the same machine, and this is the difference
between a verified connection and a blind one. A self-hosted router URL is
left untouched.

README, docs/ROUTER.md (both languages), config/gateway.example.toml and
bench/accuracy.py follow.

Verified live: the migration rewrites a legacy config on startup, admin
health reports the hostname reachable over verified TLS (389 ms), and real
requests route through it - "math"/medium 0.416 and "translation"/low 0.268
with route_source=router.
2026-08-21 13:21:24 +03:00
infosave2007 f65f131c41 fix(routing): offered tools must not make a local model unroutable
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
Verified against a real Open WebUI container: it attaches its ~26 built-in
tools (knowledge bases, notes, calendar, ...) to every single message, and
resolve() treated that offer as a hard capability requirement. On a pool of
tools-less local models a plain "hello" therefore failed with "No healthy
model available that satisfies required capabilities", and a pinned local
model was reported as "not available - not in the pool (deleted?)".

In the OpenAI contract `tools` is an offer; only tool_choice: "required" (or
a named function) makes the call impossible without them. Keep preferring a
tool-capable model, but fall back to serving the request with the tools
stripped - together with tool_choice/parallel_tool_calls, which a provider
rejects on their own. A pin whose only shortcoming is tool support is now
honoured; an unknown pin and tool_choice: "required" still fail loudly.

The downgrade is never silent: X-Cortiq-Tools-Dropped: true, plus
cortiq.tools_dropped in the echo block.
2026-08-21 09:59:28 +03:00
19 changed files with 939 additions and 50 deletions
+80
View File
@@ -3,6 +3,86 @@
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
function calling, tool executed and its result fed back into the reply.
### Fixed
- **A tool call is never delivered twice.** Qwen-family templates write the call into the
generated text as `<tool_call>{…}</tool_call>`; some runtimes parse it into structured
`tool_calls` *and* still stream the raw markup as `content`, so the client sees the call
once as prose it must not show and once as the real thing. The gateway now strips the
markup from `content` — including when the marker is split across SSE chunks — while
passing the structured call through untouched. Ordinary text containing `<` is not
affected, and an unterminated block is handed back as text rather than swallowed.
- **`finish_reason` agrees with the reply.** A response carrying tool calls now always ends
with `finish_reason: "tool_calls"`, whatever the upstream called it — clients branch on
that field to decide whether to run the tool.
- **Streamed `tool_calls` are brought to the OpenAI shape**: `index` on every entry, `id`
and `type` exactly once per call (continuation deltas that carry only an argument
fragment are left alone, as the spec requires), and `function.arguments` always a string
even when the runtime sends an object.
### Added
- **Unparsed `<tool_call>` markup is promoted into a real call.** When the upstream never
parsed it, the gateway turns it into a proper `tool_calls` entry with a generated
`call_…` id, so the call is executable instead of decorative. Markup that is not a JSON
object with a `name` is ignored — a stray marker must not become a bogus call.
## [0.2.47] - 2026-08-21
### Changed
- **The hosted allaigate router is addressed by hostname:
`https://router.allaigate.com`** is now the default `[router].url`, and
`verify_tls` defaults to `true`. The endpoint used to be reachable only by bare IP,
whose certificate is issued for the hostname — so every hosted setup had to run with
certificate verification switched off. It no longer does.
- **Existing configs are migrated on load.** A `[router].url` still pointing at the old
`138.226.222.209` (with or without a scheme or trailing slash) is rewritten to the
hostname, `verify_tls` is turned back on, and the file is saved — the endpoint is the
same machine, and this is the difference between a verified connection and a blind one.
A self-hosted router URL is never touched.
- README, `docs/ROUTER.md` (both languages), `config/gateway.example.toml` and
`bench/accuracy.py` updated to the hostname.
## [0.2.46] - 2026-08-21
Verified end to end against a real Open WebUI container talking to the gateway.
### Fixed
- **Offered `tools` no longer make a local model unroutable.** Open WebUI attaches its ~26
built-in tools (knowledge bases, notes, calendar, …) to *every* message, and the gateway
treated that offer as a hard capability requirement — so a plain "hello" against a pool of
tools-less local models failed with «No healthy model available that satisfies required
capabilities». In the OpenAI contract `tools` is an offer: only `tool_choice: "required"`
(or a named function) makes the call impossible without them. The gateway now still
prefers a tool-capable model, but falls back to serving the request with the tools
stripped (along with `tool_choice`/`parallel_tool_calls`, which a provider rejects on
their own). `tool_choice: "required"` still fails loudly, as it must.
- **A pinned model that merely lacks tool support is honoured** instead of being reported as
«not available — it is not in the pool (deleted?)». An unknown pin still fails loudly.
### Added
- `X-Cortiq-Tools-Dropped: true` (and `cortiq.tools_dropped` in the echo block) whenever the
gateway answers without the tools the client offered, so the downgrade is never silent.
## [0.2.45] - 2026-08-21
### Fixed
Generated
+1 -1
View File
@@ -285,7 +285,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
[[package]]
name = "cortiq-gateway"
version = "0.2.45"
version = "0.2.49"
dependencies = [
"anyhow",
"arc-swap",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "cortiq-gateway"
version = "0.2.45"
version = "0.2.49"
edition = "2021"
rust-version = "1.75"
license = "Apache-2.0"
+3 -3
View File
@@ -175,9 +175,9 @@ print(resp.choices[0].message.content)
`model: "cortiq-auto"` enables routing. Any **real** model name from the config
(`"gpt-4o-mini"`, `"local-qwen"`) is a direct passthrough, without routing.
> **Using the hosted allaigate router?** Set `url = "https://138.226.222.209"`,
> `verify_tls = false`, `taxonomy_id = "data-assistant"`, and a `cortiq_…` key in
> `CORTIQ_ROUTER_KEY`. On hard prompts the router escalates to an oracle (~10 s), so
> **Using the hosted allaigate router?** It is the default: `url =
> "https://router.allaigate.com"` with `verify_tls = true`. Set `taxonomy_id =
> "data-assistant"` and a `cortiq_…` key in `CORTIQ_ROUTER_KEY`. On hard prompts the router escalates to an oracle (~10 s), so
> set `timeout_ms = 12000+` — otherwise the gateway gracefully degrades to the default model.
---
+3 -3
View File
@@ -167,9 +167,9 @@ print(resp.choices[0].message.content)
`model: "cortiq-auto"` включает роутинг. Любое **реальное** имя модели из конфига
(`"gpt-4o-mini"`, `"local-qwen"`) — это passthrough напрямую, без роутинга.
> **Используете хостовый allaigate-роутер?** Укажите `url = "https://138.226.222.209"`,
> `verify_tls = false`, `taxonomy_id = "data-assistant"` и ключ `cortiq_…` в
> `CORTIQ_ROUTER_KEY`. На сложных запросах роутер эскалирует к oracle (~10 с), поэтому
> **Используете хостовый allaigate-роутер?** Он стоит по умолчанию: `url =
> "https://router.allaigate.com"`, `verify_tls = true`. Задайте `taxonomy_id =
> "data-assistant"` и ключ `cortiq_…` в `CORTIQ_ROUTER_KEY`. На сложных запросах роутер эскалирует к oracle (~10 с), поэтому
> ставьте `timeout_ms = 12000+` — иначе шлюз мягко деградирует на default-модель.
---
+2 -2
View File
@@ -4,12 +4,12 @@ heuristic (what you get without a semantic classifier, e.g. with LiteLLM/Portkey
CORTIQ_ROUTER_KEY=cortiq_... python3 bench/accuracy.py
Env: CORTIQ_ROUTER_KEY (required), ROUTER_URL (default https://138.226.222.209),
Env: CORTIQ_ROUTER_KEY (required), ROUTER_URL (default https://router.allaigate.com),
TAXONOMY (default data-assistant).
"""
import json, os, ssl, sys, urllib.request
ROUTER_URL = os.environ.get("ROUTER_URL", "https://138.226.222.209").rstrip("/")
ROUTER_URL = os.environ.get("ROUTER_URL", "https://router.allaigate.com").rstrip("/")
TAXONOMY = os.environ.get("TAXONOMY", "data-assistant")
KEY = os.environ.get("CORTIQ_ROUTER_KEY", "")
CTX = ssl._create_unverified_context()
+2 -2
View File
@@ -8,13 +8,13 @@
listen = "0.0.0.0:9000"
# --- connection to cortiq-router (decides task type / complexity) -----------
# Hosted allaigate router: url = "https://138.226.222.209", verify_tls = false
# Hosted allaigate router: url = "https://router.allaigate.com" (the default)
# (cert is bound to an IP), taxonomy_id = "data-assistant", a cortiq_… key in
# CORTIQ_ROUTER_KEY.
# NOTE: on hard prompts the router escalates to an oracle LLM (~10 s) — for the
# hosted router set timeout_ms = 12000+, otherwise the gateway degrades to default.
[router]
url = "https://138.226.222.209" # hosted allaigate router (or your own cortiq-router URL)
url = "https://router.allaigate.com" # hosted allaigate router (or your own cortiq-router URL)
api_key_env = "CORTIQ_ROUTER_KEY" # name of the env var holding the router key
timeout_ms = 12000 # hard prompts escalate to an oracle (~10 s)
taxonomy_id = "data-assistant" # the hosted router's task taxonomy
+27
View File
@@ -65,6 +65,13 @@ concatenated, images and other non-text parts are skipped); `tools` may be `null
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).
`tools` is treated as an **offer**, not a requirement. The gateway prefers a tool-capable
model, but when none is available it answers without the tools rather than failing, and says
so via `X-Cortiq-Tools-Dropped: true` (chat UIs attach their built-in tools to every message,
so the strict reading made ordinary chat impossible on a tools-less local pool).
`tool_choice: "required"` — or a named function — is a real requirement and still fails with
`502 no_healthy_model` when no model can honour it.
### Response (non-streaming)
```jsonc
@@ -114,6 +121,26 @@ 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:
* raw `<tool_call>…</tool_call>` markup (how Qwen-family templates express a call in the
generated text) is stripped from `content`, so a client never receives the same call
twice — once as prose and once structured. Markers split across SSE chunks are handled;
ordinary text containing `<` is not touched;
* when the upstream never parsed that markup, it is promoted into a real `tool_calls`
entry with a generated `call_…` id, so the call is executable rather than decorative;
* streamed entries carry `index` always, `id`/`type` once per call, and
`function.arguments` as a string even if the runtime sent an object;
* a reply carrying tool calls always ends with `finish_reason: "tool_calls"`.
### Metadata in response headers (always present)
```
+28
View File
@@ -65,6 +65,13 @@
HTTP 400 — но не «голый» 422, который OpenAI-клиенты не умеют разбирать (для пользователя
это выглядит как отсутствие ответа).
`tools` трактуется как **предложение**, а не требование. Шлюз предпочитает модель с
поддержкой tools, но если такой нет — отвечает без них, а не падает, и сообщает об этом
заголовком `X-Cortiq-Tools-Dropped: true` (чат-интерфейсы добавляют свои встроенные tools в
каждое сообщение, поэтому строгое прочтение делало обычный чат на локальном пуле без tools
невозможным). `tool_choice: "required"` — или именованная функция — это уже настоящее
требование, и оно по-прежнему завершается `502 no_healthy_model`, если его некому выполнить.
### Ответ (без стриминга)
```jsonc
@@ -112,6 +119,27 @@ 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` проксируются, но провод держится чистым:
* сырая разметка `<tool_call>…</tool_call>` (так Qwen-подобные шаблоны выражают вызов
прямо в тексте) вырезается из `content` — клиент никогда не получает один и тот же
вызов дважды, прозой и структурой. Маркер, разорванный между SSE-чанками, обрабатывается
корректно; обычный текст с символом `<` не трогается;
* если апстрим эту разметку не распарсил, она повышается до настоящего элемента
`tool_calls` со сгенерированным `call_…` id — вызов становится исполнимым, а не
декоративным;
* в стриме у каждого элемента всегда есть `index`, `id`/`type` — ровно один раз на вызов,
а `function.arguments` — всегда строка, даже если рантайм прислал объект;
* ответ с tool-вызовами всегда завершается `finish_reason: "tool_calls"`.
### Метаданные в заголовках (всегда)
```
+4 -4
View File
@@ -41,11 +41,11 @@ minted and shown to you.
```toml
[router]
url = "https://138.226.222.209" # allaigate routing API
verify_tls = false # cert is bound to an IP
url = "https://router.allaigate.com" # allaigate routing API (the default)
verify_tls = true # valid certificate — keep verification on
taxonomy_id = "data-assistant"
api_key_env = "CORTIQ_ROUTER_KEY" # your cortiq_… key
timeout_ms = 15000 # allow for oracle escalation on hard prompts
api_key_env = "CORTIQ_ROUTER_KEY" # your cortiq_… key
timeout_ms = 15000 # allow for oracle escalation on hard prompts
```
```bash
+4 -4
View File
@@ -40,11 +40,11 @@ Cortiq Gateway — тонкий и быстрый прокси; **интелле
```toml
[router]
url = "https://138.226.222.209" # routing API allaigate
verify_tls = false # серт привязан к IP
url = "https://router.allaigate.com" # routing API allaigate (значение по умолчанию)
verify_tls = true # валидный сертификат — проверку не выключаем
taxonomy_id = "data-assistant"
api_key_env = "CORTIQ_ROUTER_KEY" # ваш ключ cortiq_…
timeout_ms = 15000 # запас под oracle-эскалацию на сложных запросах
api_key_env = "CORTIQ_ROUTER_KEY" # ваш ключ cortiq_…
timeout_ms = 15000 # запас под oracle-эскалацию на сложных запросах
```
```bash
+71 -3
View File
@@ -318,6 +318,34 @@ fn default_verify_tls() -> bool {
true
}
/// The hosted allaigate routing API.
pub const DEFAULT_ROUTER_URL: &str = "https://router.allaigate.com";
/// The bare IP the hosted router used to be addressed by. It served a
/// certificate for `router.allaigate.com`, so every config pointing at it had to
/// turn TLS verification off. The hostname now resolves, so configs written
/// before it existed are moved over on load — the endpoint is the same machine,
/// and this is the difference between a verified connection and a blind one.
const LEGACY_ROUTER_HOSTS: &[&str] = &[
"https://138.226.222.209",
"http://138.226.222.209",
"138.226.222.209",
];
impl RouterCfg {
/// Point a legacy IP-addressed router config at the hostname and restore TLS
/// verification. Returns true when something changed.
fn migrate_hosted_endpoint(&mut self) -> bool {
let url = self.url.trim().trim_end_matches('/');
if !LEGACY_ROUTER_HOSTS.contains(&url) {
return false;
}
self.url = DEFAULT_ROUTER_URL.to_string();
self.verify_tls = true;
true
}
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct RouteCfg {
#[serde(default = "default_text_strategy")]
@@ -882,9 +910,9 @@ impl Default for Config {
// models and the console asks for one; nothing breaks.
router: RouterCfg {
enabled: true,
url: "https://138.226.222.209".to_string(),
url: DEFAULT_ROUTER_URL.to_string(),
timeout_ms: 15_000, // hard prompts escalate to an oracle (~10 s)
verify_tls: false, // the hosted endpoint is IP-bound
verify_tls: true,
api_key_env: Some("CORTIQ_ROUTER_KEY".to_string()),
taxonomy_id: Some("data-assistant".to_string()),
},
@@ -954,7 +982,14 @@ impl Config {
pub fn load(path: &str) -> anyhow::Result<Self> {
let raw = std::fs::read_to_string(path)
.map_err(|e| anyhow::anyhow!("cannot read config {path}: {e}"))?;
let cfg: Config = toml::from_str(&raw)?;
let mut cfg: Config = toml::from_str(&raw)?;
if cfg.router.migrate_hosted_endpoint() {
tracing::info!(
url = DEFAULT_ROUTER_URL,
"router endpoint moved from the bare IP to the hostname; TLS verification is on"
);
let _ = cfg.save(path);
}
cfg.validate()?;
Ok(cfg)
}
@@ -1091,6 +1126,39 @@ impl Config {
mod tests {
use super::*;
#[test]
fn the_hosted_router_defaults_to_the_hostname_with_tls_on() {
let cfg = Config::default();
assert_eq!(cfg.router.url, "https://router.allaigate.com");
assert!(cfg.router.verify_tls);
}
#[test]
fn a_legacy_ip_router_config_moves_to_the_hostname() {
for legacy in [
"https://138.226.222.209",
"https://138.226.222.209/",
"138.226.222.209",
] {
let mut r = Config::default().router;
r.url = legacy.to_string();
r.verify_tls = false;
assert!(r.migrate_hosted_endpoint(), "{legacy} must migrate");
assert_eq!(r.url, DEFAULT_ROUTER_URL);
assert!(r.verify_tls, "TLS verification must be restored");
}
}
#[test]
fn a_self_hosted_router_url_is_left_alone() {
let mut r = Config::default().router;
r.url = "http://127.0.0.1:8081".to_string();
r.verify_tls = false;
assert!(!r.migrate_hosted_endpoint());
assert_eq!(r.url, "http://127.0.0.1:8081");
assert!(!r.verify_tls);
}
const SAMPLE: &str = r#"
listen = "0.0.0.0:9000"
+1
View File
@@ -29,6 +29,7 @@ mod secrets;
mod state;
mod stats;
mod sysmon;
mod toolcalls;
use axum::{routing::get, Router};
use clap::Parser;
+5
View File
@@ -113,6 +113,11 @@ pub struct RouteInfo {
pub cost_usd: f64,
#[serde(default)]
pub failover: bool,
/// The client offered `tools` but the model that answered cannot use them,
/// so they were dropped from the upstream call (see `Pipeline::resolve`).
/// Surfaced as `X-Cortiq-Tools-Dropped` so a silent downgrade is debuggable.
#[serde(default)]
pub tools_dropped: bool,
}
#[derive(Clone, Debug, Serialize)]
+137 -8
View File
@@ -186,7 +186,7 @@ impl Pipeline {
// Tool support detected via the local servers' healthz handshake —
// treated as equivalent to declared caps when filtering candidates.
detected_tools: &std::collections::HashSet<String>,
) -> Result<(RouteDecision, String, Vec<String>)> {
) -> Result<(RouteDecision, String, Vec<String>, bool)> {
// 1. Extract text for classification
let text = self.extract_text(
&req.messages,
@@ -277,6 +277,7 @@ impl Pipeline {
// Filter candidates by capabilities (e.g. tools support)
let needs_tools = !req.tools.is_empty();
let tools_mandatory = needs_tools && tool_choice_is_mandatory(&req.params.passthrough);
let has_tools = |id: &str, prov: &std::sync::Arc<dyn crate::providers::Provider>| {
prov.caps().tools || detected_tools.contains(id)
};
@@ -294,6 +295,31 @@ impl Pipeline {
}
}
// `tools` is an OFFER in the OpenAI contract, not a demand: only
// `tool_choice: "required"` (or a named function) makes the call
// impossible without them. Chat UIs lean on that — Open WebUI attaches
// its ~26 built-in tools to every single message — so treating the offer
// as a hard requirement made an ordinary "hello" unanswerable on a pool
// of tools-less local models. Serve the request without the tools and
// flag the downgrade (`X-Cortiq-Tools-Dropped`) instead of failing.
let mut tools_dropped = false;
// A pin stays a pin: when the pinned model exists and merely lacks tool
// support, honour the pin and drop the offer rather than reporting the
// model as missing.
if final_candidates.is_empty() && needs_tools && !tools_mandatory {
if let RoutingDirective::Pinned { model_id } = &req.routing {
if live.registry.get(model_id).is_some() {
final_candidates.push(model_id.clone());
tools_dropped = true;
tracing::info!(
model = %model_id,
"pinned model does not support the offered tools — serving without them"
);
}
}
}
// An explicit pin must never silently answer from a different model: if
// the pinned id is unknown or was filtered out, fail loudly so the caller
// knows their choice was not honoured (instead of getting a stray reply
@@ -326,7 +352,7 @@ impl Pipeline {
// (an agentic request is worth a capable model), then the rest of the
// pool. Coding agents (OpenCode, Cline, …) always send tools; without
// this every low-tier verdict was a hard failure.
if final_candidates.is_empty() {
let widened_order = || -> Vec<String> {
let mut widened: Vec<String> = Vec::new();
for tier in ["medium", "high"] {
for id in live.routing.candidates(tier) {
@@ -340,7 +366,10 @@ impl Pipeline {
widened.push(id.clone());
}
}
for id in widened {
widened
};
if final_candidates.is_empty() {
for id in widened_order() {
if let Some(prov) = live.registry.get(&id) {
if needs_tools && !has_tools(&id, &prov) {
continue;
@@ -355,20 +384,40 @@ impl Pipeline {
);
}
}
// Same relaxation for an AUTO route once every narrower pass came up empty.
if final_candidates.is_empty() && needs_tools && !tools_mandatory {
for id in widened_order() {
if live.registry.get(&id).is_some() {
final_candidates.push(id);
}
}
if !final_candidates.is_empty() {
tools_dropped = true;
tracing::info!(
chosen = %final_candidates[0],
"no model in the pool supports the offered tools — serving without them"
);
}
}
if final_candidates.is_empty() {
return Err(GatewayError::UpstreamUnavailable(
"No healthy model available that satisfies required capabilities (e.g. tools)"
.to_string(),
));
}
Ok((decision, route_source, final_candidates))
Ok((decision, route_source, final_candidates, tools_dropped))
}
async fn run_inner(&self, req: ChatRequest, state: &SharedState) -> Result<ChatResponse> {
async fn run_inner(&self, mut req: ChatRequest, state: &SharedState) -> Result<ChatResponse> {
let live = state.live();
let detected_tools = state.cmf.detected_tools.lock().unwrap().clone();
let (decision, route_source, mut final_candidates) =
let (decision, route_source, mut final_candidates, tools_dropped) =
self.resolve(&req, &live, &detected_tools).await?;
if tools_dropped {
strip_tools(&mut req);
}
// Serving-when-promoted (gated): a promoted task-label is served by the
// local model FIRST (failover to cloud on error). Per-request VETO uses
@@ -478,6 +527,7 @@ impl Pipeline {
router_request_id: decision.router_request_id.clone(),
cost_usd: cost,
failover: attempt > 0,
tools_dropped,
};
return Ok(resp);
}
@@ -504,7 +554,7 @@ impl Pipeline {
/// recorded when the stream completes (tokens taken from the final usage chunk).
pub async fn run_stream(
&self,
req: ChatRequest,
mut req: ChatRequest,
state: &SharedState,
) -> Result<(RouteInfo, crate::providers::ChatStream)> {
let started = Instant::now();
@@ -518,7 +568,7 @@ impl Pipeline {
let live = state.live();
let detected_tools = state.cmf.detected_tools.lock().unwrap().clone();
let (decision, route_source, final_candidates) =
let (decision, route_source, final_candidates, tools_dropped) =
match self.resolve(&req, &live, &detected_tools).await {
Ok(t) => t,
Err(e) => {
@@ -526,6 +576,9 @@ impl Pipeline {
return Err(e);
}
};
if tools_dropped {
strip_tools(&mut req);
}
let mut last_err = None;
let mut attempt = 0usize;
@@ -612,6 +665,7 @@ impl Pipeline {
router_request_id: decision.router_request_id.clone(),
cost_usd: 0.0,
failover: attempt > 0,
tools_dropped,
};
// for the client-facing usage estimate when the provider
// sends none: the prompt size is known here, the reply is
@@ -689,6 +743,26 @@ impl Pipeline {
}
}
/// Remove the tool offer from a request the chosen model cannot honour. The
/// companion fields must go too: an upstream given `tool_choice` with no
/// `tools` answers 400.
fn strip_tools(req: &mut ChatRequest) {
req.tools.clear();
req.params.passthrough.remove("tool_choice");
req.params.passthrough.remove("parallel_tool_calls");
}
/// Is the client's `tool_choice` a demand rather than a preference?
/// `"required"` and `{"type":"function", …}` mean the caller cannot be served
/// without tools; `"auto"`, `"none"` and an absent field are just an offer.
fn tool_choice_is_mandatory(passthrough: &serde_json::Map<String, serde_json::Value>) -> bool {
match passthrough.get("tool_choice") {
Some(serde_json::Value::String(s)) => s.eq_ignore_ascii_case("required"),
Some(serde_json::Value::Object(_)) => true,
_ => false,
}
}
fn now_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
@@ -756,6 +830,7 @@ fn cached_response(hit: &crate::cache::CacheHit) -> ChatResponse {
router_request_id: None,
cost_usd: 0.0,
failover: false,
tools_dropped: false,
},
}
}
@@ -1067,3 +1142,57 @@ async fn shadow_capture(
judge_tokens,
});
}
#[cfg(test)]
mod tests {
use super::*;
fn map(json: serde_json::Value) -> serde_json::Map<String, serde_json::Value> {
json.as_object().cloned().unwrap_or_default()
}
#[test]
fn an_offered_tool_list_is_not_a_demand() {
// Open WebUI attaches its built-in tools with no tool_choice at all.
assert!(!tool_choice_is_mandatory(&map(serde_json::json!({}))));
assert!(!tool_choice_is_mandatory(&map(
serde_json::json!({"tool_choice": "auto"})
)));
assert!(!tool_choice_is_mandatory(&map(
serde_json::json!({"tool_choice": "none"})
)));
}
#[test]
fn required_and_named_tool_choice_are_demands() {
assert!(tool_choice_is_mandatory(&map(
serde_json::json!({"tool_choice": "required"})
)));
assert!(tool_choice_is_mandatory(&map(
serde_json::json!({"tool_choice": {"type": "function", "function": {"name": "f"}}})
)));
}
#[test]
fn stripping_tools_also_drops_their_companion_fields() {
let mut req = ChatRequest {
routing: RoutingDirective::Auto { profile: None },
messages: vec![],
tools: vec![serde_json::json!({"type": "function"})],
params: GenParams {
passthrough: map(
serde_json::json!({"tool_choice": "auto", "parallel_tool_calls": true, "top_k": 40}),
),
..Default::default()
},
stream: false,
meta: RequestMeta::default(),
};
strip_tools(&mut req);
assert!(req.tools.is_empty());
assert!(!req.params.passthrough.contains_key("tool_choice"));
assert!(!req.params.passthrough.contains_key("parallel_tool_calls"));
// unrelated passthrough must survive
assert!(req.params.passthrough.contains_key("top_k"));
}
}
+251 -16
View File
@@ -42,6 +42,9 @@ pub(crate) fn cortiq_headers(c: &RouteInfo) -> axum::http::HeaderMap {
if let Some(id) = &c.router_request_id {
put(&mut h, "X-Cortiq-Request-Id", id);
}
if c.tools_dropped {
put(&mut h, "X-Cortiq-Tools-Dropped", "true");
}
h
}
@@ -230,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),
@@ -257,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());
}
@@ -319,6 +332,9 @@ async fn handler(
if let Some(rid) = &resp.cortiq.router_request_id {
echo["router_request_id"] = serde_json::json!(rid);
}
if resp.cortiq.tools_dropped {
echo["tools_dropped"] = serde_json::json!(true);
}
body["cortiq"] = echo;
}
@@ -335,9 +351,16 @@ async fn handler(
/// 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.
///
/// Tool calls are settled here too (see [`crate::toolcalls`]): raw
/// `<tool_call>` markup never reaches `content`, structured deltas are brought
/// to the OpenAI shape, markup is promoted to a real call when the upstream
/// produced none, and a stream that emitted tool calls always finishes with
/// `finish_reason: "tool_calls"`.
fn normalize_sse(
stream: crate::providers::ChatStream,
model: String,
include_usage: bool,
) -> impl futures::Stream<Item = Result<bytes::Bytes>> + Send + 'static {
use futures::StreamExt;
async_stream::stream! {
@@ -349,6 +372,16 @@ fn normalize_sse(
let mut roled: std::collections::HashSet<u64> = std::collections::HashSet::new();
let mut sent_finish = false;
let mut failure: Option<String> = None;
// Tool-call hygiene: one markup filter per choice index, the set of
// tool-call indices already opened, and whether anything structured
// was emitted at all.
let mut markup: std::collections::HashMap<u64, crate::toolcalls::ToolMarkupFilter> =
std::collections::HashMap::new();
let mut tool_indices: std::collections::HashSet<u64> = std::collections::HashSet::new();
let mut emitted_tool_calls = false;
// `usage` / `cortiq_*` collected along the way, shipped after the finish.
let mut tail_extras: std::collections::BTreeMap<&str, serde_json::Value> =
std::collections::BTreeMap::new();
// Emit one normalised event; `None` means "nothing worth sending".
macro_rules! chunk {
@@ -411,8 +444,15 @@ fn normalize_sse(
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));
// Strip any `<tool_call>` markup before it can
// reach the client as prose — the structured
// call is the only form a tool loop may see.
let visible = markup
.entry(index)
.or_default()
.push(text);
if !visible.is_empty() {
delta.insert("content".into(), serde_json::json!(visible));
}
}
// Reasoning traces and citations are not in the base
@@ -424,10 +464,45 @@ fn normalize_sse(
}
}
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 normalized =
crate::toolcalls::normalize_delta_tool_calls(tc, &mut tool_indices);
if !normalized.is_empty() {
emitted_tool_calls = true;
delta.insert(
"tool_calls".into(),
serde_json::Value::Array(normalized),
);
}
}
let finish = c["finish_reason"].as_str().map(normalize_finish_reason);
// Last chance to promote `<tool_call>` markup into a
// real call: an upstream that never parsed it would
// otherwise hand the client prose no tool loop runs.
if finish.is_some() && !emitted_tool_calls {
if let Some(filter) = markup.get_mut(&index) {
let leftover = filter.finish();
if !leftover.is_empty() {
let merged = match delta.get("content").and_then(|v| v.as_str()) {
Some(prev) => format!("{prev}{leftover}"),
None => leftover,
};
delta.insert("content".into(), serde_json::json!(merged));
}
let calls = crate::toolcalls::markup_to_tool_calls(filter.captured(), 0);
if !calls.is_empty() {
emitted_tool_calls = true;
let tc = serde_json::json!({
"index": index,
"delta": { "tool_calls": calls },
});
let extra: Vec<(&str, serde_json::Value)> = Vec::new();
yield Ok(chunk!(vec![tc], extra));
}
}
}
if finish.is_none() && delta.is_empty() {
continue;
}
@@ -436,6 +511,10 @@ fn normalize_sse(
"delta": serde_json::Value::Object(delta),
});
if let Some(f) = finish {
// A reply that carries tool calls stops for that
// reason, whatever the upstream called it —
// clients branch on this field to run the tool.
let f = if emitted_tool_calls { "tool_calls" } else { f };
choice["finish_reason"] = serde_json::json!(f);
sent_finish = true;
}
@@ -443,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));
}
}
}
@@ -476,14 +555,37 @@ fn normalize_sse(
if roled.insert(0) {
delta.insert("role".into(), serde_json::json!("assistant"));
}
// Held-back text and unparsed markup must not be lost with the stream.
for (index, filter) in markup.iter_mut() {
let leftover = filter.finish();
if !leftover.is_empty() && *index == 0 {
delta.insert("content".into(), serde_json::json!(leftover));
}
if !emitted_tool_calls {
let calls = crate::toolcalls::markup_to_tool_calls(filter.captured(), 0);
if !calls.is_empty() {
emitted_tool_calls = true;
delta.insert("tool_calls".into(), serde_json::Value::Array(calls));
}
}
}
let choices = vec![serde_json::json!({
"index": 0,
"delta": serde_json::Value::Object(delta),
"finish_reason": "stop",
"finish_reason": if emitted_tool_calls { "tool_calls" } else { "stop" },
})];
let extra: Vec<(&str, serde_json::Value)> = Vec::new();
yield Ok(chunk!(choices, extra));
}
// One trailing chunk, and only when the client asked for usage.
if include_usage {
if let Some(usage) = tail_extras.remove("usage") {
let mut extra: Vec<(&str, serde_json::Value)> = vec![("usage", usage)];
extra.extend(tail_extras.into_iter());
let no_choices: Vec<serde_json::Value> = Vec::new();
yield Ok(chunk!(no_choices, extra));
}
}
yield Ok(bytes::Bytes::from("data: [DONE]\n\n"));
}
}
@@ -524,13 +626,146 @@ mod tests {
assert_eq!(normalize_finish_reason("tool_use"), "tool_calls");
}
/// Collect a normalised stream into one string for assertions.
async fn drain(chunks: Vec<&'static str>) -> String {
use futures::StreamExt;
let upstream = futures::stream::iter(
chunks
.into_iter()
.map(|c| Ok(bytes::Bytes::from(c)))
.collect::<Vec<_>>(),
);
normalize_sse(Box::pin(upstream), "cortiq-auto".into(), false)
.map(|b| String::from_utf8_lossy(&b.unwrap()).to_string())
.collect::<Vec<_>>()
.await
.join("")
}
#[tokio::test]
async fn tool_markup_is_not_duplicated_into_content() {
// The reported shape: the upstream streams the call as prose AND then
// sends it properly structured.
let out = drain(vec![
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"Sure. <tool_call>\"}}]}\n\n",
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"{\\\"name\\\":\\\"get_weather\\\",\\\"arguments\\\":{\\\"city\\\":\\\"Paris\\\"}}\"}}]}\n\n",
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"</tool_call>\"}}]}\n\n",
"data: {\"choices\":[{\"index\":0,\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"get_weather\",\"arguments\":\"{\\\"city\\\":\\\"Paris\\\"}\"}}]}}]}\n\n",
"data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n",
])
.await;
assert!(
!out.contains("tool_call>"),
"markup leaked into content: {out}"
);
assert!(
out.contains("\"id\":\"call_1\""),
"structured call missing: {out}"
);
// exactly one structured call, and the stop reason agrees with it
assert_eq!(out.matches("\"tool_calls\":[").count(), 1);
assert!(out.contains("\"finish_reason\":\"tool_calls\""));
assert!(out.trim_end().ends_with("data: [DONE]"));
}
#[tokio::test]
async fn unparsed_markup_is_promoted_to_a_real_tool_call() {
// An upstream that never parses the markup: the call must still be
// executable by the client rather than arriving as prose.
let out = drain(vec![
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"<tool_call>{\\\"name\\\":\\\"get_weather\\\",\\\"arguments\\\":{\\\"city\\\":\\\"Paris\\\"}}</tool_call>\"}}]}\n\n",
"data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n",
])
.await;
assert!(!out.contains("tool_call>"), "markup leaked: {out}");
assert!(
out.contains("\"name\":\"get_weather\""),
"call not promoted: {out}"
);
assert!(
out.contains("\"finish_reason\":\"tool_calls\""),
"wrong stop reason: {out}"
);
}
async fn drain_with_usage(chunks: Vec<&'static str>, include_usage: bool) -> String {
use futures::StreamExt;
let upstream = futures::stream::iter(
chunks
.into_iter()
.map(|c| Ok(bytes::Bytes::from(c)))
.collect::<Vec<_>>(),
);
normalize_sse(Box::pin(upstream), "cortiq-auto".into(), include_usage)
.map(|b| String::from_utf8_lossy(&b.unwrap()).to_string())
.collect::<Vec<_>>()
.await
.join("")
}
/// The upstream ships usage BEFORE the stop reason and the gateway adds a
/// cost chunk of its own. A client that did not ask for usage must see
/// neither — the same stream shape every hosted provider sends.
const USAGE_STREAM: &[&str] = &[
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n",
"data: {\"choices\":[],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":1,\"total_tokens\":4}}\n\n",
"data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: {\"choices\":[],\"cortiq_cost_usd\":0.0}\n\n",
"data: [DONE]\n\n",
];
#[tokio::test]
async fn without_stream_options_no_choice_less_chunks_are_sent() {
let out = drain_with_usage(USAGE_STREAM.to_vec(), false).await;
assert!(!out.contains("\"usage\""), "unrequested usage chunk: {out}");
assert!(!out.contains("cortiq_cost_usd"), "stray cost chunk: {out}");
assert!(
!out.contains("\"choices\":[]"),
"choice-less chunk leaked: {out}"
);
assert!(out.contains("\"finish_reason\":\"stop\""));
assert!(out.trim_end().ends_with("data: [DONE]"));
}
#[tokio::test]
async fn requested_usage_is_one_chunk_after_the_finish_reason() {
let out = drain_with_usage(USAGE_STREAM.to_vec(), true).await;
assert_eq!(
out.matches("\"usage\"").count(),
1,
"usage must ship once: {out}"
);
let finish = out.find("\"finish_reason\"").expect("finish chunk");
let usage = out.find("\"usage\"").expect("usage chunk");
assert!(usage > finish, "usage must follow the stop reason: {out}");
assert!(
out.contains("cortiq_cost_usd"),
"cost folded into usage chunk: {out}"
);
assert!(out.trim_end().ends_with("data: [DONE]"));
}
#[tokio::test]
async fn ordinary_text_is_untouched_by_the_tool_filter() {
let out = drain(vec![
"data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"a < b and 2<3\"}}]}\n\n",
"data: {\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n",
"data: [DONE]\n\n",
])
.await;
assert!(out.contains("a < b and 2<3"), "text mangled: {out}");
assert!(out.contains("\"finish_reason\":\"stop\""));
}
#[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())
let out: Vec<String> = normalize_sse(Box::pin(upstream), "cortiq-auto".into(), false)
.map(|b| String::from_utf8_lossy(&b.unwrap()).to_string())
.collect()
.await;
+1
View File
@@ -164,6 +164,7 @@ impl Provider for AnthropicProvider {
router_request_id: None,
cost_usd: 0.0,
failover: false,
tools_dropped: false,
},
})
}
+18 -3
View File
@@ -122,13 +122,23 @@ impl Provider for OpenAiProvider {
for (i, c) in choices_arr.iter().enumerate() {
let msg_val = &c["message"];
let role = msg_val["role"].as_str().unwrap_or("assistant").to_string();
let content = msg_val["content"].as_str().unwrap_or("").to_string();
let raw_content = msg_val["content"].as_str().unwrap_or("");
// Qwen-style templates put the call in the text as `<tool_call>…`.
// Strip it: the structured array is the only form a client may act
// on, and prose that merely looks like a call is worse than none.
let (content, markup) = crate::toolcalls::split_tool_markup(raw_content);
let tool_calls = msg_val["tool_calls"]
let mut tool_calls = msg_val["tool_calls"]
.as_array()
.cloned()
.unwrap_or_default();
if tool_calls.is_empty() {
// Upstream never parsed the markup — promote it, so the call is
// executable instead of decorative.
tool_calls = crate::toolcalls::markup_to_tool_calls(&markup, 0);
}
let tool_calls_present = !tool_calls.is_empty();
choices.push(crate::model::Choice {
index: c["index"].as_u64().unwrap_or(i as u64) as u32,
message: crate::model::Message {
@@ -137,7 +147,11 @@ impl Provider for OpenAiProvider {
tool_calls,
..Default::default()
},
finish_reason: c["finish_reason"].as_str().unwrap_or("stop").to_string(),
finish_reason: if tool_calls_present {
"tool_calls".to_string()
} else {
c["finish_reason"].as_str().unwrap_or("stop").to_string()
},
});
}
@@ -167,6 +181,7 @@ impl Provider for OpenAiProvider {
router_request_id: None,
cost_usd: 0.0,
failover: false,
tools_dropped: false,
},
})
}
+300
View File
@@ -0,0 +1,300 @@
//! Tool-call hygiene for OpenAI-compatible responses.
//!
//! Qwen-family chat templates express a function call as literal markup in the
//! generated text:
//!
//! ```text
//! <tool_call>
//! {"name": "get_weather", "arguments": {"city": "Paris"}}
//! </tool_call>
//! ```
//!
//! Runtimes handle that inconsistently. Some parse it into structured
//! `tool_calls` **and** still stream the raw markup as `content` — the client
//! then sees the call twice, once as prose it must not show and once as the real
//! thing, and tool-loop implementations (Open WebUI's among them) get confused.
//! Others never parse it at all, so a perfectly good call reaches the client as
//! text no tool loop will ever execute.
//!
//! This module makes the gateway the place where that is settled: markup is
//! stripped from the text, and when nothing structured came with it the markup
//! is promoted into a real `tool_calls` array.
const OPEN: &str = "<tool_call>";
const CLOSE: &str = "</tool_call>";
/// Longest suffix of `s` that is a proper prefix of `pat`. Used to hold back the
/// tail of a streamed chunk that might be the beginning of a marker split across
/// chunk boundaries (`"<tool"` + `"_call>"`).
fn partial_marker_len(s: &str, pat: &str) -> usize {
let max = pat.len().saturating_sub(1).min(s.len());
(1..=max)
.rev()
.find(|&n| {
s.is_char_boundary(s.len() - n)
&& pat.as_bytes().starts_with(&s.as_bytes()[s.len() - n..])
})
.unwrap_or(0)
}
/// Incremental `<tool_call>` filter for SSE deltas.
///
/// Feed it content fragments in order; it returns the text that is safe to
/// forward and keeps the markup payloads for [`Self::captured`].
#[derive(Default)]
pub struct ToolMarkupFilter {
/// Text held back: either an unterminated marker prefix or the inside of an
/// open `<tool_call>` block.
pending: String,
inside: bool,
captured: Vec<String>,
}
impl ToolMarkupFilter {
pub fn new() -> Self {
Self::default()
}
/// Consume one content fragment; returns the part that may be forwarded now.
pub fn push(&mut self, chunk: &str) -> String {
self.pending.push_str(chunk);
let mut out = String::new();
loop {
if self.inside {
let Some(end) = self.pending.find(CLOSE) else {
// still inside the block — nothing to forward
return out;
};
let payload: String = self.pending[..end].to_string();
self.captured.push(payload);
self.pending = self.pending[end + CLOSE.len()..].to_string();
self.inside = false;
continue;
}
match self.pending.find(OPEN) {
Some(start) => {
out.push_str(&self.pending[..start]);
self.pending = self.pending[start + OPEN.len()..].to_string();
self.inside = true;
}
None => {
// Hold back a trailing partial marker so a split like
// "…<tool" / "_call>…" is not forwarded as visible text.
let hold = partial_marker_len(&self.pending, OPEN);
let cut = self.pending.len() - hold;
out.push_str(&self.pending[..cut]);
self.pending = self.pending[cut..].to_string();
return out;
}
}
}
}
/// End of stream: an unterminated block was never a tool call after all, so
/// give the text back rather than swallowing output.
pub fn finish(&mut self) -> String {
let mut out = String::new();
if self.inside {
out.push_str(OPEN);
self.inside = false;
}
out.push_str(&std::mem::take(&mut self.pending));
out
}
/// Payloads found between `<tool_call>` and `</tool_call>`.
pub fn captured(&self) -> &[String] {
&self.captured
}
}
/// Strip every complete `<tool_call>` block from a finished string.
/// Returns the cleaned text and the payloads that were removed.
pub fn split_tool_markup(text: &str) -> (String, Vec<String>) {
let mut f = ToolMarkupFilter::new();
let mut out = f.push(text);
out.push_str(&f.finish());
(out, f.captured)
}
/// Turn `<tool_call>` payloads into OpenAI `tool_calls` entries, numbered from
/// `start_index`. Payloads that are not a JSON object with a `name` are skipped —
/// a stray marker must not become a bogus call.
pub fn markup_to_tool_calls(payloads: &[String], start_index: usize) -> Vec<serde_json::Value> {
let mut out = Vec::new();
for payload in payloads {
let Ok(v) = serde_json::from_str::<serde_json::Value>(payload.trim()) else {
continue;
};
let Some(name) = v.get("name").and_then(|n| n.as_str()) else {
continue;
};
// `arguments` is a JSON string on the wire; templates emit an object.
let arguments = match v.get("arguments").or_else(|| v.get("parameters")) {
Some(serde_json::Value::String(s)) => s.clone(),
Some(other) => other.to_string(),
None => "{}".to_string(),
};
out.push(serde_json::json!({
"index": start_index + out.len(),
"id": new_call_id(),
"type": "function",
"function": { "name": name, "arguments": arguments },
}));
}
out
}
pub fn new_call_id() -> String {
format!("call_{}", crate::admin::random_token(12))
}
/// Bring one streamed `tool_calls` array up to the OpenAI shape.
///
/// `seen` tracks which choice-local indices already had their opening delta, so
/// the identity fields (`id`, `type`) are added once and continuation deltas —
/// which legitimately carry nothing but an argument fragment — are left alone.
pub fn normalize_delta_tool_calls(
raw: &[serde_json::Value],
seen: &mut std::collections::HashSet<u64>,
) -> Vec<serde_json::Value> {
let mut out = Vec::new();
for (i, tc) in raw.iter().enumerate() {
let index = tc.get("index").and_then(|v| v.as_u64()).unwrap_or(i as u64);
let mut entry = serde_json::Map::new();
entry.insert("index".into(), serde_json::json!(index));
let first = seen.insert(index);
if let Some(id) = tc
.get("id")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
{
entry.insert("id".into(), serde_json::json!(id));
} else if first {
entry.insert("id".into(), serde_json::json!(new_call_id()));
}
if first {
let kind = tc
.get("type")
.and_then(|v| v.as_str())
.unwrap_or("function");
entry.insert("type".into(), serde_json::json!(kind));
}
if let Some(f) = tc.get("function") {
let mut func = serde_json::Map::new();
if let Some(name) = f.get("name").and_then(|v| v.as_str()) {
func.insert("name".into(), serde_json::json!(name));
} else if first {
func.insert("name".into(), serde_json::json!(""));
}
match f.get("arguments") {
// Arguments are a string on the wire; some runtimes send the object.
Some(serde_json::Value::String(s)) => {
func.insert("arguments".into(), serde_json::json!(s));
}
Some(serde_json::Value::Null) | None => {
if first {
func.insert("arguments".into(), serde_json::json!(""));
}
}
Some(other) => {
func.insert("arguments".into(), serde_json::json!(other.to_string()));
}
}
entry.insert("function".into(), serde_json::Value::Object(func));
}
out.push(serde_json::Value::Object(entry));
}
out
}
#[cfg(test)]
mod tests {
use super::*;
const CALL: &str = r#"<tool_call>
{"name": "get_weather", "arguments": {"city": "Paris"}}
</tool_call>"#;
#[test]
fn markup_is_stripped_from_finished_text() {
let (clean, payloads) = split_tool_markup(&format!("Sure!{CALL}"));
assert_eq!(clean, "Sure!");
assert_eq!(payloads.len(), 1);
}
#[test]
fn markup_split_across_stream_chunks_never_leaks() {
let mut f = ToolMarkupFilter::new();
let mut seen = String::new();
// the marker itself is broken across three fragments
for chunk in [
"Let me check. <tool",
"_call>",
"{\"name\":\"get_weather\",",
"\"arguments\":{\"city\":\"Paris\"}}",
"</tool",
"_call>",
" done",
] {
seen.push_str(&f.push(chunk));
}
seen.push_str(&f.finish());
assert_eq!(seen, "Let me check. done");
assert!(!seen.contains("tool_call"));
assert_eq!(f.captured().len(), 1);
}
#[test]
fn captured_markup_becomes_a_real_tool_call() {
let (_, payloads) = split_tool_markup(CALL);
let calls = markup_to_tool_calls(&payloads, 0);
assert_eq!(calls.len(), 1);
assert_eq!(calls[0]["type"], "function");
assert_eq!(calls[0]["index"], 0);
assert_eq!(calls[0]["function"]["name"], "get_weather");
// arguments must be a STRING, not an object
assert_eq!(
calls[0]["function"]["arguments"].as_str().unwrap(),
r#"{"city":"Paris"}"#
);
assert!(calls[0]["id"].as_str().unwrap().starts_with("call_"));
}
#[test]
fn a_stray_marker_does_not_become_a_call() {
let calls = markup_to_tool_calls(&["not json at all".into(), "{\"no_name\":1}".into()], 0);
assert!(calls.is_empty());
}
#[test]
fn an_unterminated_block_is_given_back_as_text() {
let mut f = ToolMarkupFilter::new();
let mut out = f.push("thinking <tool_call>{\"name\":\"x\"");
out.push_str(&f.finish());
assert_eq!(out, "thinking <tool_call>{\"name\":\"x\"");
assert!(f.captured().is_empty());
}
#[test]
fn delta_tool_calls_get_identity_once_and_object_arguments_become_strings() {
let mut seen = std::collections::HashSet::new();
let opening = normalize_delta_tool_calls(
&[serde_json::json!({"index": 0, "function": {"name": "f", "arguments": {"a": 1}}})],
&mut seen,
);
assert_eq!(opening[0]["type"], "function");
assert!(opening[0]["id"].as_str().unwrap().starts_with("call_"));
assert_eq!(opening[0]["function"]["arguments"], r#"{"a":1}"#);
// a continuation delta must not be handed a second id
let cont = normalize_delta_tool_calls(
&[serde_json::json!({"index": 0, "function": {"arguments": "\"}"}})],
&mut seen,
);
assert!(cont[0].get("id").is_none());
assert!(cont[0].get("type").is_none());
}
}