diff --git a/Cargo.lock b/Cargo.lock index cad1926f..f54820bd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2944,7 +2944,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2 0.6.5", + "socket2 0.5.10", "thiserror 2.0.21", "tokio", "tracing", @@ -2983,7 +2983,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.5", + "socket2 0.5.10", "tracing", "windows-sys 0.61.2", ] @@ -4837,8 +4837,8 @@ dependencies = [ [[package]] name = "tw-bedrock" -version = "0.81.0" -source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.81.0#aae0e486da837c94189046a7bcf5095d4eb1948f" +version = "0.82.0" +source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?rev=14f52f25a434faf1e7336746b57c9ad8d8aebb96#14f52f25a434faf1e7336746b57c9ad8d8aebb96" dependencies = [ "aws-credential-types", "aws-sigv4", @@ -4852,16 +4852,16 @@ dependencies = [ [[package]] name = "tw-breaker" -version = "0.81.0" -source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.81.0#aae0e486da837c94189046a7bcf5095d4eb1948f" +version = "0.82.0" +source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?rev=14f52f25a434faf1e7336746b57c9ad8d8aebb96#14f52f25a434faf1e7336746b57c9ad8d8aebb96" dependencies = [ "serde", ] [[package]] name = "tw-dialect" -version = "0.81.0" -source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.81.0#aae0e486da837c94189046a7bcf5095d4eb1948f" +version = "0.82.0" +source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?rev=14f52f25a434faf1e7336746b57c9ad8d8aebb96#14f52f25a434faf1e7336746b57c9ad8d8aebb96" dependencies = [ "memchr", "serde", @@ -4870,8 +4870,8 @@ dependencies = [ [[package]] name = "tw-guard" -version = "0.81.0" -source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?tag=v0.81.0#aae0e486da837c94189046a7bcf5095d4eb1948f" +version = "0.82.0" +source = "git+https://github.com/ThinkWatchProject/ThinkWatch-Core.git?rev=14f52f25a434faf1e7336746b57c9ad8d8aebb96#14f52f25a434faf1e7336746b57c9ad8d8aebb96" dependencies = [ "aho-corasick", "base64 0.22.1", diff --git a/Cargo.toml b/Cargo.toml index 58592396..2c5de095 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -70,10 +70,10 @@ opt-level = 3 # never re-exported through a local shim. And the reverse: something only # this side uses (the at-rest crypto, IMDSv2 credentials, the gateway error) # lives here, not in core. -tw-bedrock = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.81.0" } -tw-breaker = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.81.0" } -tw-dialect = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.81.0" } -tw-guard = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", tag = "v0.81.0" } +tw-bedrock = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", rev = "14f52f25a434faf1e7336746b57c9ad8d8aebb96" } +tw-breaker = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", rev = "14f52f25a434faf1e7336746b57c9ad8d8aebb96" } +tw-dialect = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", rev = "14f52f25a434faf1e7336746b57c9ad8d8aebb96" } +tw-guard = { git = "https://github.com/ThinkWatchProject/ThinkWatch-Core.git", rev = "14f52f25a434faf1e7336746b57c9ad8d8aebb96" } # Web framework axum = { version = "0.8", features = ["macros", "ws"] } diff --git a/crates/common/src/models/provider.rs b/crates/common/src/models/provider.rs index 8c8a481f..cc02a0e7 100644 --- a/crates/common/src/models/provider.rs +++ b/crates/common/src/models/provider.rs @@ -65,6 +65,10 @@ pub struct Model { /// limit. #[serde(default)] pub max_output_tokens: Option, + /// The reasoning setting: `off`, or a level (`low` … `max`) for + /// requests whose reasoning is on. `None` ⇒ requests go out as sent. + #[serde(default)] + pub reasoning_effort: Option, } fn default_enabled() -> bool { diff --git a/crates/gateway/src/lifecycle/mod.rs b/crates/gateway/src/lifecycle/mod.rs index 40773b2e..f57cd9b3 100644 --- a/crates/gateway/src/lifecycle/mod.rs +++ b/crates/gateway/src/lifecycle/mod.rs @@ -185,6 +185,9 @@ pub(crate) struct Sent { /// Waiting to go out again: the attempt that failed was counted /// against the route already, and nothing is under way. pub waiting: bool, + /// The attempt under way sent a different reasoning effort than the + /// caller asked (see `proxy::effort`). + pub effort: Option, } /// Build the streaming pump: forward the upstream's bytes to the caller — @@ -284,6 +287,7 @@ pub(crate) fn build_chat_pump( retries: 0, last_sent: std::time::Instant::now(), waiting: false, + effort: None, })); // Gemini clients get no keep-alive comments: the google-genai SDK reads // every SSE line as JSON. @@ -314,7 +318,7 @@ pub(crate) fn build_chat_pump( } let sent_at = std::time::Instant::now(); if let Ok(mut s) = sent.lock() { - *s = Sent { retries: resend.budget.retries(), last_sent: sent_at, waiting: false }; + *s = Sent { retries: resend.budget.retries(), last_sent: sent_at, waiting: false, effort: None }; } // What an attempt that failed had read is not this one's: its // usage is not billed, its error not the request's. @@ -365,6 +369,9 @@ pub(crate) fn build_chat_pump( return; } }; + if let Ok(mut s) = sent.lock() { + s.effort = wire.effort; + } if let Ok(mut r) = readers.lock() { *r = Readers { sniffer: Some(tw_dialect::usage::Sniffer::new()), @@ -683,6 +690,7 @@ pub(crate) fn build_chat_pump( retries: 0, last_sent: started_at, waiting: false, + effort: None, }); let invoked = Invoked { client_ip: identity.ip_address.clone(), @@ -1099,6 +1107,7 @@ fn resending( retries: retry, last_sent: std::time::Instant::now(), waiting: true, + effort: None, }; } tracing::info!( @@ -1524,9 +1533,12 @@ impl Surface for ChatCompletionSurface { deps.request.request_started_at.elapsed().as_millis() as i64, logged_status, with_usage_detail( - retried_detail( - pinned_detail(error_detail, deps.route.pinned), - deps.route.sel_record.retries, + effort_detail( + retried_detail( + pinned_detail(error_detail, deps.route.pinned), + deps.route.sel_record.retries, + ), + deps.route.sel_record.effort, ), &usage, usage_estimated(&invoked.view), @@ -1594,6 +1606,23 @@ fn retried_detail(detail: Option, retries: u32) -> Option, + effort: Option, +) -> Option { + let Some(effort) = effort else { + return detail; + }; + let mut d = match detail { + Some(serde_json::Value::Object(d)) => d, + _ => serde_json::Map::new(), + }; + d.insert("effort".into(), effort.detail()); + Some(serde_json::Value::Object(d)) +} + /// The usage a captured view bills — shared by `record_usage` and /// `emit_audit` so the budget and the audit row can never disagree. fn extract_usage(view: &CapturedView) -> tw_dialect::usage::Usage { @@ -1822,6 +1851,7 @@ mod tests { retries: 0, last_sent: std::time::Instant::now(), waiting: false, + effort: None, }); resending( "up", @@ -1987,6 +2017,24 @@ mod tests { assert!(r.sealed_refusal().is_none()); } + #[test] + fn a_changed_reasoning_effort_is_said_on_the_log_row() { + use crate::proxy::effort::{EffortChange, Reason}; + assert_eq!(effort_detail(None, None), None); + let change = EffortChange { + asked: "xhigh", + sent: Some("high"), + reason: Reason::Rejected, + }; + assert_eq!( + effort_detail(Some(serde_json::json!({"retries": 1})), Some(change)), + Some(serde_json::json!({ + "retries": 1, + "effort": {"asked": "xhigh", "sent": "high", "reason": "rejected"}, + })) + ); + } + #[test] fn a_pinned_route_is_said_on_the_log_row() { assert_eq!(pinned_detail(None, false), None); diff --git a/crates/gateway/src/proxy/effort.rs b/crates/gateway/src/proxy/effort.rs new file mode 100644 index 00000000..945c5026 --- /dev/null +++ b/crates/gateway/src/proxy/effort.rs @@ -0,0 +1,700 @@ +//! Reasoning effort: the model's setting, and what an upstream refuses. +//! +//! # The setting (`models.reasoning_effort`) +//! +//! `off` turns reasoning off in a request that has it on (OpenAI's formats +//! write `none`; the others take the switch out, with the effort). A level +//! (`low` … `max`) changes the effort of a request **whose reasoning is +//! on**; a request without it — a title, a summary, a client that never +//! asked — goes out untouched: reasoning is never turned on for a caller. +//! Unset: requests go out as sent. It applies to each hop as it is +//! addressed, to the caller's request in its own format and for the model +//! named upstream (Gemini's budgets are capped per model); a request +//! converted for another format carries the changed effort the way that +//! format writes it. What reasoning is on, and where each format writes the +//! effort, is `tw_dialect::params`, shared with the desktop gateway. +//! +//! # What an upstream refuses +//! +//! Not every model takes every level, and an upstream that does not says +//! so with a 400 (`tw_dialect::think::effort_rejection`; for a `none` it +//! does not take, `none_rejection`). Both are read against the effort the +//! hop **wrote** (`tw_dialect::params::written_effort`), whether or not +//! reasoning is on — Claude with thinking off still refuses `xhigh` — and a +//! request that only carries a budget has none to refuse. The request then +//! goes to the **same upstream** once more, with the level it named or, +//! when it named none, without the effort (the model's default). That is +//! not a retry of a failure of the moment: it does not wait and does not +//! count against `gateway.retry_wait_secs` (`super::retry`). A second +//! refusal is the answer. +//! +//! A resend that is answered is remembered in Redis, per provider, format +//! sent and model named upstream, for [`MEMORY_TTL_SECS`]: later requests +//! that would write the refused level write the one that worked straight +//! away. A remembered fix that is refused itself is forgotten (and replaced +//! if the resend that follows is answered); so is one whose resend was +//! answered at the level first refused. After the week the refused level +//! is tried again — a model can gain one. A provider's entries go when its +//! URL or credential changes, a route of it relearns its protocol, or it +//! is deleted ([`forget_provider`]). +//! +//! What a hop sent differently from what it would have is said on the +//! request's `gateway_logs` row, as `"effort"` ([`EffortChange::detail`]). + +use std::collections::BTreeMap; + +use fred::interfaces::{HashesInterface, KeysInterface}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use tw_dialect::ir::{Dialect, Effort}; +use tw_dialect::params::WrittenEffort; +use tw_dialect::think::EffortFix; +use uuid::Uuid; + +use crate::error::GatewayError; + +/// A model's reasoning setting (`models.reasoning_effort`). Unset is no +/// setting at all: `Option`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReasoningSetting { + /// Turn reasoning off in requests that have it on. + Off, + /// Change the effort of requests whose reasoning is on. + Level(Effort), +} + +/// The values `models.reasoning_effort` may hold, in display order. +pub const SETTINGS: [&str; 6] = ["off", "low", "medium", "high", "xhigh", "max"]; + +impl ReasoningSetting { + /// One of [`SETTINGS`]. `minimal` is not one: few models take it, and + /// the ones that do not refuse it outright. + pub fn parse(s: &str) -> Option { + Some(match s { + "off" => Self::Off, + "low" => Self::Level(Effort::Low), + "medium" => Self::Level(Effort::Medium), + "high" => Self::Level(Effort::High), + "xhigh" => Self::Level(Effort::XHigh), + "max" => Self::Level(Effort::Max), + _ => return None, + }) + } + + /// As stored: one of [`SETTINGS`]. + pub fn slug(self) -> &'static str { + match self { + Self::Off => "off", + Self::Level(e) => tw_dialect::think::openai(e), + } + } + + /// Apply to `body`, a request in `dialect` for `model` (the name sent + /// upstream). Whether anything changed. + pub(crate) fn apply(self, dialect: Dialect, body: &mut Value, model: &str) -> bool { + match self { + Self::Off => { + if !tw_dialect::params::reasoning_on(dialect, body) { + return false; + } + let before = body.clone(); + tw_dialect::params::disable_thinking(dialect, body); + *body != before + } + Self::Level(e) => tw_dialect::params::set_reasoning_effort(dialect, body, e, model), + } + } +} + +/// What the caller's request asks for reasoning, read once. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct Asked { + /// The model's setting, applied to each hop. + pub setting: Option, + /// The effort the caller asked for, with reasoning on: what a hop the + /// setting changed is said against. + pub effort: Option, + /// A hop can carry an effort an upstream could refuse: the caller has + /// reasoning on or writes one. Otherwise no hop asks the memory. + pub any: bool, +} + +impl Asked { + pub(crate) fn of(setting: Option, dialect: Dialect, body: &Value) -> Self { + Self { + setting, + effort: tw_dialect::params::reasoning_effort(dialect, body), + any: tw_dialect::params::reasoning_on(dialect, body) + || tw_dialect::params::written_effort(dialect, body).is_some(), + } + } + + /// What applying the setting to a hop changed, said against what the + /// caller asked: `after` is the hop's body in `dialect` once applied. + /// Nothing to say when the caller named no effort or the hop sends the + /// same. + pub(crate) fn change(&self, dialect: Dialect, after: &Value) -> Option { + let asked = slug(self.effort?); + let sent = shown(dialect, after); + (sent != Some(asked)).then_some(EffortChange { + asked, + sent, + reason: Reason::Setting, + }) + } +} + +/// Why a hop sent another effort. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Reason { + /// The model's setting. + Setting, + /// The upstream refused what went out first, and the request went + /// out again. + Rejected, + /// The upstream refused that level before ([`Memory`]). + Remembered, +} + +/// A hop that sent another reasoning effort. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct EffortChange { + /// What the hop would have sent before this change: for the setting, + /// what the caller asked; for a refusal or the memory, the level it + /// replaced. `none` is OpenAI's reasoning turned off. + pub asked: &'static str, + /// What it sent; `None`: the effort left out. + pub sent: Option<&'static str>, + pub reason: Reason, +} + +impl EffortChange { + /// As said on the log row: `{"asked": "xhigh", "sent": "high", + /// "reason": "rejected"}`, `sent` `null` when the effort was left out. + pub(crate) fn detail(&self) -> Value { + serde_json::json!({ + "asked": self.asked, + "sent": self.sent, + "reason": match self.reason { + Reason::Setting => "setting", + Reason::Rejected => "rejected", + Reason::Remembered => "remembered", + }, + }) + } +} + +/// `minimal` … `max`, as OpenAI writes them. +fn slug(e: Effort) -> &'static str { + tw_dialect::think::openai(e) +} + +/// The effort `v`, in `dialect`, sends: the field written, else the one +/// its reasoning implies (a budget read as a level). +pub(crate) fn shown(dialect: Dialect, v: &Value) -> Option<&'static str> { + tw_dialect::params::written_effort(dialect, v) + .map(WrittenEffort::slug) + .or_else(|| tw_dialect::params::reasoning_effort(dialect, v).map(slug)) +} + +/// The effort written in `body`, a hop in `dialect`. +pub(crate) fn written(dialect: Dialect, body: &[u8]) -> Option { + let v: Value = serde_json::from_slice(body).ok()?; + tw_dialect::params::written_effort(dialect, &v) +} + +/// Rewrite the effort written in `body` as `fix` says. What the hop sends +/// then ([`shown`]) when anything changed; `None` leaves `body` as it was. +pub(crate) fn apply_fix( + dialect: Dialect, + body: &mut Vec, + fix: EffortFix, +) -> Option> { + let mut v: Value = serde_json::from_slice(body).ok()?; + let changed = match fix { + EffortFix::Use(e) => tw_dialect::params::replace_reasoning_effort(dialect, &mut v, e), + EffortFix::Drop => tw_dialect::params::remove_reasoning_effort(dialect, &mut v), + }; + if !changed { + return None; + } + let bytes = serde_json::to_vec(&v).ok()?; + *body = bytes; + Some(shown(dialect, &v)) +} + +/// Did the upstream refuse the effort `sent`? How to send it again. +/// `label` is the upstream's name, which the error's message starts with +/// and which is not what it said. +pub(crate) fn rejection(err: &GatewayError, label: &str, sent: WrittenEffort) -> Option { + let GatewayError::ProviderHttpError { + status, message, .. + } = err + else { + return None; + }; + let said = message + .strip_prefix(label) + .and_then(|m| m.strip_prefix(": ")) + .unwrap_or(message); + match sent { + WrittenEffort::Level(e) => tw_dialect::think::effort_rejection(*status, said, e), + WrittenEffort::Off => { + tw_dialect::think::none_rejection(*status, said).then_some(EffortFix::Drop) + } + } +} + +// ───────────────────────────────────────────── memory + +/// How long a refused level is remembered: 7 days. +pub const MEMORY_TTL_SECS: i64 = 7 * 24 * 60 * 60; + +/// One provider's memory: a hash, a field per format sent and model named +/// upstream ([`field`]). Hash-tagged on the provider, as the route keys +/// are on the route. +fn key(provider_id: Uuid) -> String { + format!("effort_memory:{{provider:{provider_id}}}") +} + +/// `openai-chat:gpt-5`: the format first, which has no `:` in it. +fn field(dialect: Dialect, model: &str) -> String { + format!("{}:{model}", dialect.slug()) +} + +/// What one model, asked in one format, refused: the written effort refused +/// (`none` too) → what worked instead. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(transparent)] +pub(crate) struct Memory(BTreeMap); + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +struct Entry { + /// The level that was answered; `None`: the effort left out. + #[serde(rename = "use")] + instead: Option, + /// Unix seconds. + at: i64, +} + +impl Memory { + fn parse(value: &str) -> Self { + serde_json::from_str(value).unwrap_or_default() + } + + /// How to rewrite a hop that writes `written`, as of `now`. Entries + /// older than [`MEMORY_TTL_SECS`] are no longer taken. + pub(crate) fn instead_of(&self, written: WrittenEffort, now: i64) -> Option { + let entry = self.0.get(written.slug())?; + if now - entry.at >= MEMORY_TTL_SECS { + return None; + } + Some(match &entry.instead { + Some(s) => EffortFix::Use(tw_dialect::think::parse_openai(s).flatten()?), + None => EffortFix::Drop, + }) + } + + /// `rejected` was refused and the resend `fix` answered, at `now`. A + /// fix that is `rejected` itself (the upstream took it after all) is + /// forgotten instead. Expired entries go. + pub(crate) fn note(&mut self, rejected: WrittenEffort, fix: EffortFix, now: i64) { + self.0.retain(|_, e| now - e.at < MEMORY_TTL_SECS); + let instead = match fix { + EffortFix::Use(e) if WrittenEffort::Level(e) == rejected => { + self.forget(rejected); + return; + } + EffortFix::Use(e) => Some(slug(e).to_string()), + EffortFix::Drop => None, + }; + self.0 + .insert(rejected.slug().to_string(), Entry { instead, at: now }); + } + + /// The fix remembered for `rejected` was refused itself. + pub(crate) fn forget(&mut self, rejected: WrittenEffort) { + self.0.remove(rejected.slug()); + } + + pub(crate) fn is_empty(&self) -> bool { + self.0.is_empty() + } +} + +pub(crate) fn now() -> i64 { + chrono::Utc::now().timestamp() +} + +/// What `provider_id` refused for `model` asked in `dialect`. A Redis that +/// cannot be read remembers nothing: the request goes out as it would have. +pub(crate) async fn recall( + redis: &fred::clients::Client, + provider_id: Uuid, + dialect: Dialect, + model: &str, +) -> Memory { + match redis + .hget::, _, _>(key(provider_id), field(dialect, model)) + .await + { + Ok(Some(v)) => Memory::parse(&v), + Ok(None) => Memory::default(), + Err(e) => { + tracing::debug!(error = %e, "could not read the reasoning effort memory"); + Memory::default() + } + } +} + +/// Change what `provider_id` remembers for `model` in `dialect`, and write +/// it back: emptied, the field goes. Best-effort. +async fn update( + redis: &fred::clients::Client, + provider_id: Uuid, + dialect: Dialect, + model: &str, + change: impl FnOnce(&mut Memory), +) { + let mut memory = recall(redis, provider_id, dialect, model).await; + let before = memory.clone(); + change(&mut memory); + if memory == before { + return; + } + let (key, field) = (key(provider_id), field(dialect, model)); + let written = async { + if memory.is_empty() { + return redis.hdel::<(), _, _>(&key, &field).await; + } + let value = serde_json::to_string(&memory).unwrap_or_default(); + redis + .hset::<(), _, _>(&key, (field.as_str(), value)) + .await?; + redis.expire::<(), _>(&key, MEMORY_TTL_SECS, None).await + } + .await; + if let Err(e) = written { + tracing::warn!(error = %e, "could not write the reasoning effort memory"); + } +} + +/// `provider_id` refused `rejected` for `model` in `dialect`, and the +/// resend `fix` was answered. +pub(crate) async fn remember( + redis: &fred::clients::Client, + provider_id: Uuid, + dialect: Dialect, + model: &str, + rejected: WrittenEffort, + fix: EffortFix, +) { + update(redis, provider_id, dialect, model, |m| { + m.note(rejected, fix, now()) + }) + .await; +} + +/// The fix remembered for `rejected` was refused itself. +pub(crate) async fn forget( + redis: &fred::clients::Client, + provider_id: Uuid, + dialect: Dialect, + model: &str, + rejected: WrittenEffort, +) { + update(redis, provider_id, dialect, model, |m| m.forget(rejected)).await; +} + +/// Forget what `provider_id` refused: it may be another upstream now. +/// Best-effort. +pub async fn forget_provider(redis: &fred::clients::Client, provider_id: Uuid) { + if let Err(e) = redis.del::<(), _>(key(provider_id)).await { + tracing::warn!(error = %e, %provider_id, "could not clear the reasoning effort memory"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + const L: fn(Effort) -> WrittenEffort = WrittenEffort::Level; + + #[test] + fn the_settings_are_off_and_five_levels() { + for s in SETTINGS { + let setting = ReasoningSetting::parse(s).unwrap(); + assert_eq!(setting.slug(), s); + } + assert_eq!(ReasoningSetting::parse("off"), Some(ReasoningSetting::Off)); + assert_eq!( + ReasoningSetting::parse("xhigh"), + Some(ReasoningSetting::Level(Effort::XHigh)) + ); + for bad in ["minimal", "none", "on", "", "MAX"] { + assert_eq!(ReasoningSetting::parse(bad), None, "{bad}"); + } + } + + /// The setting applied to `body` (`dialect`), as one hop sends it, and + /// what is said about it. + fn applied(setting: &str, dialect: Dialect, mut body: Value) -> (Value, Option) { + let setting = ReasoningSetting::parse(setting); + let asked = Asked::of(setting, dialect, &body); + if let Some(s) = setting { + s.apply(dialect, &mut body, "gpt-5"); + } + let change = asked.change(dialect, &body); + (body, change) + } + + #[test] + fn a_level_changes_only_a_request_with_reasoning_on() { + let (body, change) = applied( + "max", + Dialect::Chat, + json!({"model": "m", "reasoning_effort": "xhigh", "messages": []}), + ); + assert_eq!(body["reasoning_effort"], "max"); + assert_eq!( + change.unwrap().detail(), + json!({"asked": "xhigh", "sent": "max", "reason": "setting"}) + ); + + // No reasoning: a title request goes out as sent. + let title = json!({"model": "m", "messages": []}); + assert_eq!(applied("max", Dialect::Chat, title.clone()), (title, None)); + + // Reasoning turned off by the caller stays off. + let none = json!({"model": "m", "reasoning_effort": "none"}); + assert_eq!(applied("max", Dialect::Chat, none.clone()), (none, None)); + + // Already at that level: nothing to say. + let same = json!({"model": "m", "reasoning": {"effort": "max"}, "input": "hi"}); + assert_eq!( + applied("max", Dialect::Responses, same.clone()), + (same, None) + ); + } + + #[test] + fn a_level_is_written_where_the_callers_format_puts_it() { + // Anthropic adaptive thinking, no effort named: one is added, and + // there was nothing asked to say it against. + let (body, change) = applied( + "high", + Dialect::Anthropic, + json!({"model": "m", "max_tokens": 64000, + "thinking": {"type": "adaptive"}, "messages": []}), + ); + assert_eq!(body["output_config"]["effort"], "high"); + assert_eq!(change, None); + + let (body, _) = applied( + "high", + Dialect::Responses, + json!({"model": "m", "reasoning": {"effort": "low"}, "input": "hi"}), + ); + assert_eq!(body["reasoning"]["effort"], "high"); + } + + #[test] + fn off_turns_reasoning_off_as_each_format_says_it() { + let (claude, change) = applied( + "off", + Dialect::Anthropic, + json!({"model": "m", "max_tokens": 64000, + "thinking": {"type": "adaptive"}, + "output_config": {"effort": "max"}, "messages": []}), + ); + assert!(claude.get("thinking").is_none(), "{claude}"); + assert!(claude.get("output_config").is_none(), "{claude}"); + assert_eq!( + change.unwrap().detail(), + json!({"asked": "max", "sent": null, "reason": "setting"}) + ); + + // OpenAI's formats say `none`: left out, the model's default applies. + let (chat, change) = applied( + "off", + Dialect::Chat, + json!({"model": "m", "reasoning_effort": "high"}), + ); + assert_eq!(chat["reasoning_effort"], "none"); + assert_eq!( + change.unwrap().detail(), + json!({"asked": "high", "sent": "none", "reason": "setting"}) + ); + + // Nothing to turn off. + let plain = json!({"model": "m", "messages": []}); + assert_eq!(applied("off", Dialect::Chat, plain.clone()), (plain, None)); + } + + #[test] + fn a_fix_rewrites_the_effort_as_written() { + let mut body = + serde_json::to_vec(&json!({"model": "m", "reasoning_effort": "xhigh"})).unwrap(); + assert_eq!(written(Dialect::Chat, &body), Some(L(Effort::XHigh))); + assert_eq!( + apply_fix(Dialect::Chat, &mut body, EffortFix::Use(Effort::High)), + Some(Some("high")) + ); + assert_eq!(written(Dialect::Chat, &body), Some(L(Effort::High))); + assert_eq!( + apply_fix(Dialect::Chat, &mut body, EffortFix::Drop), + Some(None) + ); + assert_eq!(written(Dialect::Chat, &body), None); + let before = body.clone(); + assert_eq!(apply_fix(Dialect::Chat, &mut body, EffortFix::Drop), None); + assert_eq!(body, before); + + // Written while thinking is off: still the field the upstream reads. + let mut off = serde_json::to_vec(&json!({"model": "m", "max_tokens": 9, + "thinking": {"type": "disabled"}, "output_config": {"effort": "xhigh"}})) + .unwrap(); + assert_eq!(written(Dialect::Anthropic, &off), Some(L(Effort::XHigh))); + assert_eq!( + apply_fix(Dialect::Anthropic, &mut off, EffortFix::Use(Effort::High)), + Some(Some("high")) + ); + + // A budget only: no written effort. + let budget = serde_json::to_vec(&json!({"model": "m", "max_tokens": 64000, + "thinking": {"type": "enabled", "budget_tokens": 32000}})) + .unwrap(); + assert_eq!(written(Dialect::Anthropic, &budget), None); + } + + fn http(status: u16, said: &str) -> GatewayError { + GatewayError::ProviderHttpError { + status, + message: format!("relay: {said}"), + retry_after: None, + } + } + + #[test] + fn a_refusal_of_the_effort_is_read_off_the_upstreams_words() { + let said = r#"{"error":{"message":"Unsupported value: 'xhigh' is not supported with this model. Supported values are: 'low', 'medium', and 'high'.","param":"reasoning_effort"}}"#; + assert_eq!( + rejection(&http(400, said), "relay", L(Effort::XHigh)), + Some(EffortFix::Use(Effort::High)) + ); + // Claude with thinking off: a ceiling. + let claude = r#"{"type":"error","error":{"type":"invalid_request_error","message":"effort 'xhigh' is not supported when thinking is disabled. Use effort 'high' or below."}}"#; + assert_eq!( + rejection(&http(400, claude), "relay", L(Effort::XHigh)), + Some(EffortFix::Use(Effort::High)) + ); + // `none` not taken: left out. + let none = r#"{"error":{"message":"Invalid value for reasoning_effort: 'none'"}}"#; + assert_eq!( + rejection(&http(400, none), "relay", WrittenEffort::Off), + Some(EffortFix::Drop) + ); + // Not a 400: an upstream failing is not a refusal. + assert_eq!(rejection(&http(500, said), "relay", L(Effort::XHigh)), None); + // A refusal of something else. + assert_eq!( + rejection( + &http( + 400, + r#"{"message":"text.verbosity must be one of low, medium, high"}"# + ), + "relay", + L(Effort::XHigh) + ), + None + ); + // The upstream's own name is not what it said. + assert_eq!( + rejection( + &GatewayError::ProviderHttpError { + status: 400, + message: "high effort reasoning relay: messages: Field required".into(), + retry_after: None, + }, + "high effort reasoning relay", + L(Effort::High) + ), + None + ); + assert_eq!( + rejection( + &GatewayError::UpstreamAuthError { + status: 403, + message: "reasoning effort".into() + }, + "relay", + L(Effort::High) + ), + None + ); + } + + #[test] + fn the_memory_keeps_each_refused_level_for_a_week() { + let t0 = 1_800_000_000; + let mut m = Memory::default(); + m.note(L(Effort::XHigh), EffortFix::Use(Effort::High), t0); + m.note(WrittenEffort::Off, EffortFix::Drop, t0 + 10); + assert_eq!( + m.instead_of(L(Effort::XHigh), t0 + 60), + Some(EffortFix::Use(Effort::High)) + ); + assert_eq!( + m.instead_of(WrittenEffort::Off, t0 + 60), + Some(EffortFix::Drop) + ); + assert_eq!(m.instead_of(L(Effort::High), t0 + 60), None); + // A week on, the refused level is tried again. + assert_eq!(m.instead_of(L(Effort::XHigh), t0 + MEMORY_TTL_SECS), None); + + // As stored, and read back. + let stored = serde_json::to_string(&m).unwrap(); + assert_eq!( + serde_json::from_str::(&stored).unwrap(), + json!({"xhigh": {"use": "high", "at": t0}, "none": {"use": null, "at": t0 + 10}}) + ); + assert_eq!(Memory::parse(&stored), m); + // Anything else reads as nothing remembered. + assert_eq!(Memory::parse("not json"), Memory::default()); + + // A fix refused itself is forgotten; replaced when another works. + m.forget(L(Effort::XHigh)); + assert_eq!(m.instead_of(L(Effort::XHigh), t0 + 60), None); + m.note(L(Effort::XHigh), EffortFix::Use(Effort::Medium), t0 + 20); + assert_eq!( + m.instead_of(L(Effort::XHigh), t0 + 60), + Some(EffortFix::Use(Effort::Medium)) + ); + // Answered at the level first refused: nothing to remember. + m.note(L(Effort::XHigh), EffortFix::Use(Effort::XHigh), t0 + 30); + assert_eq!(m.instead_of(L(Effort::XHigh), t0 + 60), None); + + // Noting drops what expired. + m.note( + L(Effort::Low), + EffortFix::Use(Effort::Medium), + t0 + MEMORY_TTL_SECS + 20, + ); + assert_eq!(m.0.len(), 1, "{m:?}"); + } + + #[test] + fn the_memory_is_keyed_on_the_provider_the_format_and_the_model() { + let id = Uuid::nil(); + assert_eq!( + key(id), + "effort_memory:{provider:00000000-0000-0000-0000-000000000000}" + ); + assert_eq!(field(Dialect::Chat, "gpt-5"), "openai-chat:gpt-5"); + assert_ne!( + field(Dialect::Anthropic, "m"), + field(Dialect::Responses, "m") + ); + } +} diff --git a/crates/gateway/src/proxy/generate.rs b/crates/gateway/src/proxy/generate.rs index 6f356e80..cd9b7d73 100644 --- a/crates/gateway/src/proxy/generate.rs +++ b/crates/gateway/src/proxy/generate.rs @@ -46,6 +46,7 @@ use tw_dialect::ir::{Dialect, Target}; use super::body_capture::prepare_body_capture; use super::early_cancel::EarlyCancelSlot; +use super::effort::{self, EffortChange, Reason}; use super::headers::{ request_id_header, resolve_conversation_id, resolve_session_id, resolve_trace_id, }; @@ -59,6 +60,8 @@ use super::routing::{ use super::sealed::Conversation; use super::shaper::{StreamShaper, rewrite_model}; use super::{GatewayErrorResponse, GatewayRequestIdentity, GatewayState}; +use tw_dialect::params::WrittenEffort; +use tw_dialect::think::EffortFix; use crate::cache::ResponseCache; use crate::guards::Caller; @@ -249,6 +252,9 @@ pub(crate) struct Outbound { /// The model's output cap (`models.max_output_tokens`), applied to each /// hop as it is addressed (see [`Outbound::cap_output`]). pub max_output_tokens: Option, + /// The request's reasoning: the model's setting, applied to each hop + /// as it is addressed, and what the caller asked (see `super::effort`). + pub reasoning: effort::Asked, } /// The request as it goes out to one upstream, and what it takes to read @@ -275,6 +281,14 @@ pub(crate) struct Wire { /// the caller marked none. An upstream that refuses them gets the /// request again without them (see `proxy::cache_marks`). pub auto_cache: bool, + /// This hop sends another reasoning effort than it would have, and why + /// (see `super::effort`). + pub effort: Option, + /// The effort written before the memory rewrote it: what a refusal of + /// the rewritten one is remembered under. `None`: not rewritten. + pub pre: Option, + /// The memory rewrote the effort. + pub remembered: bool, } impl Wire { @@ -293,6 +307,41 @@ impl Wire { } } + /// The reasoning effort this hop writes, as its own format writes it. + fn written_effort(&self) -> Option { + effort::written(self.dialect, &self.body) + } + + /// Rewrite the written effort as `fix` says, in place of `replaced`. + /// False when nothing changed. + fn fix_effort(&mut self, fix: EffortFix, replaced: WrittenEffort, reason: Reason) -> bool { + let Some(sent) = effort::apply_fix(self.dialect, &mut self.body, fix) else { + return false; + }; + self.effort = Some(EffortChange { + asked: replaced.slug(), + sent, + reason, + }); + true + } + + /// This hop with the effort `memory` says the upstream takes instead + /// of the one it writes. + fn remembered(mut self, memory: &effort::Memory) -> Wire { + if memory.is_empty() { + return self; + } + if let Some(written) = self.written_effort() + && let Some(fix) = memory.instead_of(written, effort::now()) + && self.fix_effort(fix, written, Reason::Remembered) + { + self.pre = Some(written); + self.remembered = true; + } + self + } + async fn send_to( &self, upstream: &super::transport::Upstream, @@ -414,10 +463,22 @@ impl Outbound { .map_err(|r| GatewayError::TransformError(r.0)) }; + // The model's reasoning setting, for the model named upstream, on + // the caller's request in its own format. + let mut source = std::borrow::Cow::Borrowed(&self.body); + let mut setting_change = None; + if let Some(setting) = self.reasoning.setting { + let mut body = self.body.clone(); + if setting.apply(client, &mut body, model) { + setting_change = self.reasoning.change(client, &body); + source = std::borrow::Cow::Owned(body); + } + } + if protocol.dialect() == client { // Forwarded as sent. Only the model changes, and a Chat // stream always asks for its usage (see `hides_usage`). - let mut body = self.body.clone(); + let mut body = source.into_owned(); let (path, query) = if client == Dialect::Gemini { // Gemini names the model in the path, and is always asked // for SSE (see `GEMINI_SSE`). @@ -473,10 +534,13 @@ impl Outbound { collect, ledger, auto_cache: false, + effort: setting_change, + pre: None, + remembered: false, }); } - let mut decoded = decode(&self.body)?; + let mut decoded = decode(&source)?; decoded.request.model = model.to_string(); let prepared = decoded.encode(&target(protocol.dialect())); if !prepared.dropped.is_empty() { @@ -514,6 +578,9 @@ impl Outbound { collect: prepared.session, ledger, auto_cache, + effort: setting_change, + pre: None, + remembered: false, }) } } @@ -525,15 +592,22 @@ impl Outbound { /// added gets the request once more without them, and is remembered as /// refusing them for this model (see `proxy::cache_marks`). /// -/// A rejected dialect or breakpoint is known before a single byte of body -/// arrives — the status check happens inside `send` — so a stream needs -/// no special handling: nothing has reached the client yet when the retry -/// happens. +/// An upstream that refuses the reasoning effort gets the request once +/// more with the one it named, or without one, and is remembered as +/// refusing it for this model when that is answered (see `proxy::effort`). +/// A level it refused before goes out as what worked then. +/// +/// A rejected dialect, breakpoint or effort is known before a single byte +/// of body arrives — the status check happens inside `send` — so a stream +/// needs no special handling: nothing has reached the client yet when the +/// retry happens. None of these count as a retry (`proxy::retry`): the +/// upstream did not fail, it refused something that can be left out. pub(crate) async fn send( entry: &RouteEntry, outbound: &Outbound, call_ctx: &CallCtx, db: &sqlx::PgPool, + redis: &fred::clients::Client, model: &str, ) -> Result<(reqwest::Response, Wire), GatewayError> { let upstream = &entry.upstream; @@ -541,25 +615,86 @@ pub(crate) async fn send( let first = outbound .address(entry.protocol, model, official)? .for_upstream(upstream, model); + // Only a request that can carry an effort asks the memory. + let memory = if outbound.reasoning.any { + effort::recall(redis, entry.provider_id, first.dialect, model).await + } else { + effort::Memory::default() + }; + let first = first.remembered(&memory); + if first.remembered { + tracing::info!( + provider = %entry.provider_name, + model, + effort = %first.effort.map(|c| c.detail()).unwrap_or_default(), + "Sending the reasoning effort the upstream took before instead of one it refused" + ); + } // The format of the last request sent, to read its refusal in. let mut wire_dialect = first.dialect; let mut last = match first.send_to(upstream, call_ctx).await { Ok(resp) => return Ok((resp, first)), Err(e) => e, }; + let mut wire = first; - if first.auto_cache && super::cache_marks::is_refusal(&last, &upstream.label) { - let mut wire = first; - if wire.drop_added_marks() { - upstream.cache_marks.note(model); + if wire.auto_cache + && super::cache_marks::is_refusal(&last, &upstream.label) + && wire.drop_added_marks() + { + upstream.cache_marks.note(model); + tracing::info!( + provider = %entry.provider_name, + model, + "Upstream refused the cache breakpoints added on conversion — sending again without them" + ); + match wire.send_to(upstream, call_ctx).await { + Ok(resp) => return Ok((resp, wire)), + Err(e) => last = e, + } + } + + // Read only off a 400: the body is parsed for it. A budget is no + // written effort, and never refused as one. + if matches!(last, GatewayError::ProviderHttpError { status: 400, .. }) + && let Some(sent) = wire.written_effort() + && let Some(fix) = effort::rejection(&last, &upstream.label, sent) + { + // What is remembered is keyed on what was written before the + // memory rewrote it. + let pre = wire.pre.unwrap_or(sent); + let remembered = wire.remembered; + if wire.fix_effort(fix, pre, Reason::Rejected) { tracing::info!( provider = %entry.provider_name, model, - "Upstream refused the cache breakpoints added on conversion — sending again without them" + effort = %wire.effort.map(|c| c.detail()).unwrap_or_default(), + "Upstream refused the reasoning effort — sending again with one it takes" ); - match wire.send_to(upstream, call_ctx).await { - Ok(resp) => return Ok((resp, wire)), - Err(e) => last = e, + let resent = wire.send_to(upstream, call_ctx).await; + // Answered: remembered, in place of a remembered fix that was + // refused itself. Refused again: that fix is forgotten. + let (redis, provider_id, dialect, model) = ( + redis.clone(), + entry.provider_id, + wire.dialect, + model.to_string(), + ); + match resent { + Ok(resp) => { + tokio::spawn(async move { + effort::remember(&redis, provider_id, dialect, &model, pre, fix).await; + }); + return Ok((resp, wire)); + } + Err(e) => { + if remembered { + tokio::spawn(async move { + effort::forget(&redis, provider_id, dialect, &model, pre).await; + }); + } + last = e; + } } } } @@ -576,10 +711,18 @@ pub(crate) async fn send( let wire = outbound .address(*protocol, model, official)? .for_upstream(upstream, model); + let wire = if outbound.reasoning.any { + let memory = effort::recall(redis, entry.provider_id, wire.dialect, model).await; + wire.remembered(&memory) + } else { + wire + }; wire_dialect = wire.dialect; match wire.send_to(upstream, call_ctx).await { Ok(resp) => { super::protocol_relearn::persist(db, entry.route_id, *protocol).await; + // What it refused was said in another format. + effort::forget_provider(redis, entry.provider_id).await; return Ok((resp, wire)); } Err(e) => { @@ -823,15 +966,17 @@ async fn run( })? }; - // 6. The model's output cap. Applied to each hop as it is addressed - // (`Outbound::cap_output`): whether to fill one in depends on the - // upstream model, and the field on the upstream's format. The - // upstream stops there by itself; the answer is not measured. - let max_output_tokens = state - .router - .load() - .config_for(&mapped_model) - .max_output_tokens; + // 6. The model's settings. The output cap is applied to each hop as it + // is addressed (`Outbound::cap_output`): whether to fill one in + // depends on the upstream model, and the field on the upstream's + // format. The upstream stops there by itself; the answer is not + // measured. The reasoning setting is applied to each hop too, for + // the model named upstream, on the caller's request in its own format + // (see `super::effort`): a hop converted for another format carries + // the changed effort the way that one writes it. + let model_config = state.router.load().config_for(&mapped_model); + let max_output_tokens = model_config.max_output_tokens; + let reasoning = effort::Asked::of(model_config.reasoning, surface.dialect, &outbound_body); let call_ctx = CallCtx::new( Some(trace_id.clone()), @@ -845,11 +990,15 @@ async fn run( // limit. let cache_fingerprint = if surface.caches { ResponseCache::fingerprint(&outbound_body).map(|mut fp| { - // The cap is applied after this, per hop: an answer made under - // one cap must not be served under another. + // The cap and the reasoning setting are applied after this, per + // hop: an answer made under one must not be served under + // another. if let Some(cap) = max_output_tokens { fp.extend_from_slice(format!("\nmax_output_tokens={cap}").as_bytes()); } + if let Some(setting) = reasoning.setting { + fp.extend_from_slice(format!("\nreasoning_effort={}", setting.slug()).as_bytes()); + } fp }) } else { @@ -1041,6 +1190,7 @@ async fn run( redaction: guards.redaction.clone(), ledger: ledger.clone(), max_output_tokens, + reasoning, }; let snapshot = |route: &RouteEntry, sel_record| crate::lifecycle::ChatPostInvokeDeps { state: state.clone(), @@ -1104,19 +1254,23 @@ async fn run( let outbound = std::sync::Arc::new(outbound); let call_ctx = call_ctx.clone(); let db = state.db.clone(); + let redis = state.redis.clone(); let model = entry .upstream_model .clone() .unwrap_or_else(|| mapped_model.clone()); Box::new(move || { - let (entry, outbound, call_ctx, db, model) = ( + let (entry, outbound, call_ctx, db, redis, model) = ( entry.clone(), outbound.clone(), call_ctx.clone(), db.clone(), + redis.clone(), model.clone(), ); - Box::pin(async move { send(&entry, &outbound, &call_ctx, &db, &model).await }) + Box::pin( + async move { send(&entry, &outbound, &call_ctx, &db, &redis, &model).await }, + ) }) }; @@ -1317,6 +1471,7 @@ fn json_response(body: Vec) -> axum::response::Response { #[cfg(test)] mod tests { use super::*; + use tw_dialect::ir::Effort; #[test] fn a_gemini_path_names_the_model_and_whether_to_stream() { @@ -1349,9 +1504,136 @@ mod tests { redaction, ledger, max_output_tokens: cap, + reasoning: effort::Asked::default(), } } + /// `body` from a `surface` caller, under the reasoning `setting`. + fn with_setting(surface: ClientSurface, path: &str, body: Value, setting: &str) -> Outbound { + let setting = effort::ReasoningSetting::parse(setting); + let reasoning = effort::Asked::of(setting, surface.dialect, &body); + Outbound { + reasoning, + ..outbound(surface, path, body, None) + } + } + + fn sent(wire: &Wire) -> Value { + serde_json::from_slice(&wire.body).unwrap() + } + + #[test] + fn the_reasoning_setting_goes_out_on_every_hop_in_its_format() { + let ask = serde_json::json!({ + "model": "gpt-5", "reasoning_effort": "xhigh", + "messages": [{"role": "user", "content": "ping"}] + }); + let outbound = with_setting(CHAT, "/v1/chat/completions", ask.clone(), "max"); + // The caller's request stays as sent: the setting is per hop. + assert_eq!(outbound.body, ask); + // As sent, to a Chat upstream. + let wire = outbound + .address(UpstreamProtocol::OpenAiChat, "gpt-5", false) + .unwrap(); + assert_eq!(sent(&wire)["reasoning_effort"], "max"); + assert_eq!( + wire.effort.map(|c| c.detail()), + Some(serde_json::json!({"asked": "xhigh", "sent": "max", "reason": "setting"})) + ); + // Converted, to a Responses upstream. + let wire = outbound + .address(UpstreamProtocol::OpenAiResponses, "gpt-5", false) + .unwrap(); + assert_eq!(sent(&wire)["reasoning"]["effort"], "max"); + + // Off, for OpenAI's formats: `none`. + let ask = serde_json::json!({ + "model": "gpt-5", "reasoning": {"effort": "high"}, "input": "ping" + }); + let outbound = with_setting(RESPONSES, "/v1/responses", ask, "off"); + let wire = outbound + .address(UpstreamProtocol::OpenAiResponses, "gpt-5", false) + .unwrap(); + assert_eq!(sent(&wire)["reasoning"]["effort"], "none"); + + // A request without reasoning goes out as without the setting. + let ask = serde_json::json!({"model": "gpt-5", "messages": []}); + let outbound = with_setting(CHAT, "/v1/chat/completions", ask, "max"); + assert!(!outbound.reasoning.any); + let wire = outbound + .address(UpstreamProtocol::OpenAiChat, "gpt-5", false) + .unwrap(); + assert_eq!(wire.effort, None); + assert!(sent(&wire).get("reasoning_effort").is_none()); + } + + #[test] + fn a_refused_effort_is_sent_again_as_the_upstream_says() { + let ask = serde_json::json!({ + "model": "gpt-5", "reasoning_effort": "xhigh", + "messages": [{"role": "user", "content": "ping"}] + }); + let outbound = with_setting(CHAT, "/v1/chat/completions", ask, ""); + let mut wire = outbound + .address(UpstreamProtocol::OpenAiChat, "gpt-5", false) + .unwrap(); + assert_eq!(wire.effort, None); + let xhigh = WrittenEffort::Level(Effort::XHigh); + assert_eq!(wire.written_effort(), Some(xhigh)); + + assert!(wire.fix_effort(EffortFix::Use(Effort::High), xhigh, Reason::Rejected)); + assert_eq!(sent(&wire)["reasoning_effort"], "high"); + assert_eq!( + wire.effort.map(|c| c.detail()), + Some(serde_json::json!({"asked": "xhigh", "sent": "high", "reason": "rejected"})) + ); + // Left out: the model's default. + assert!(wire.fix_effort(EffortFix::Drop, xhigh, Reason::Rejected)); + assert_eq!( + wire.effort.map(|c| c.detail()), + Some(serde_json::json!({"asked": "xhigh", "sent": null, "reason": "rejected"})) + ); + // Nothing left to change. + assert!(!wire.fix_effort(EffortFix::Drop, xhigh, Reason::Rejected)); + } + + #[test] + fn a_level_refused_before_goes_out_as_what_worked() { + let ask = serde_json::json!({ + "model": "gpt-5", "reasoning_effort": "xhigh", + "messages": [{"role": "user", "content": "ping"}] + }); + let outbound = with_setting(CHAT, "/v1/chat/completions", ask, ""); + let xhigh = WrittenEffort::Level(Effort::XHigh); + let mut memory = effort::Memory::default(); + memory.note(xhigh, EffortFix::Use(Effort::High), effort::now()); + let wire = outbound + .address(UpstreamProtocol::OpenAiChat, "gpt-5", false) + .unwrap() + .remembered(&memory); + assert_eq!(sent(&wire)["reasoning_effort"], "high"); + assert!(wire.remembered); + assert_eq!(wire.pre, Some(xhigh)); + assert_eq!( + wire.effort.map(|c| c.detail()), + Some(serde_json::json!({"asked": "xhigh", "sent": "high", "reason": "remembered"})) + ); + // Another level: as sent. + let mut other = effort::Memory::default(); + other.note( + WrittenEffort::Level(Effort::Max), + EffortFix::Drop, + effort::now(), + ); + let wire = outbound + .address(UpstreamProtocol::OpenAiChat, "gpt-5", false) + .unwrap() + .remembered(&other); + assert_eq!(wire.written_effort(), Some(xhigh)); + assert!(!wire.remembered); + assert_eq!(wire.effort, None); + } + /// The body a Chat caller's request goes out with to a Chat upstream, /// under a model cap of `cap`. fn sent_to_chat(ask: Value, cap: u32, official: bool) -> Value { diff --git a/crates/gateway/src/proxy/mod.rs b/crates/gateway/src/proxy/mod.rs index d0fa85a9..a72563e1 100644 --- a/crates/gateway/src/proxy/mod.rs +++ b/crates/gateway/src/proxy/mod.rs @@ -23,6 +23,7 @@ mod accounting; mod body_capture; pub(crate) mod cache_marks; mod early_cancel; +pub mod effort; pub(crate) mod generate; mod headers; mod identity; diff --git a/crates/gateway/src/proxy/pipeline.rs b/crates/gateway/src/proxy/pipeline.rs index 5d7c3d45..77f9731d 100644 --- a/crates/gateway/src/proxy/pipeline.rs +++ b/crates/gateway/src/proxy/pipeline.rs @@ -139,6 +139,7 @@ pub(super) fn launch_stream_pump( deps.route.sel_record.started_at = sent.last_sent; deps.route.sel_record.retries = sent.retries; } + deps.route.sel_record.effort = sent.effort; // The client left while the stream waited to go out again: the // attempt that failed was counted against the route when the retry // was decided, and nothing went out since. diff --git a/crates/gateway/src/proxy/routing.rs b/crates/gateway/src/proxy/routing.rs index e130e2fa..1cb9639a 100644 --- a/crates/gateway/src/proxy/routing.rs +++ b/crates/gateway/src/proxy/routing.rs @@ -298,6 +298,10 @@ pub(crate) struct SelectionRecord { /// failure before this route answered (see `super::retry`). Said on /// the log row as `"retries"`. pub retries: u32, + /// The hop that answered sent a different reasoning effort than the + /// caller asked (see `super::effort`). Said on the log row as + /// `"effort"`. + pub effort: Option, /// The request's last attempt was already counted against the route: /// a stream whose client left while it waited to go out again. pub counted: bool, @@ -475,6 +479,7 @@ pub(super) async fn select_route_with_failover<'a>( outbound, call_ctx, &ctx.state.db, + &ctx.state.redis, upstream_model, ) .await @@ -486,7 +491,7 @@ pub(super) async fn select_route_with_failover<'a>( outbound.input_estimate, ) .await - .map(|answer| (answer, wire.ledger)), + .map(|answer| (answer, wire.ledger, wire.effort)), Err(e) => Err(e), } }; @@ -524,7 +529,7 @@ pub(super) async fn select_route_with_failover<'a>( }; match result { - Ok((response, ledger)) => { + Ok((response, ledger, effort)) => { set_affinity( &ctx.state.redis, ctx.user_id, @@ -543,6 +548,7 @@ pub(super) async fn select_route_with_failover<'a>( started_at, picked_latency_ms: Some(attempt_latency_ms), retries: budget.retries(), + effort, counted: false, }, )); @@ -654,6 +660,7 @@ pub(super) async fn select_route_for_stream<'a>( // picked route's wall time for the streaming case. picked_latency_ms: None, retries: 0, + effort: None, counted: false, }, )); diff --git a/crates/gateway/src/router.rs b/crates/gateway/src/router.rs index 56c12b3b..b733bdd9 100644 --- a/crates/gateway/src/router.rs +++ b/crates/gateway/src/router.rs @@ -78,6 +78,9 @@ pub struct ModelRoutingConfig { /// limit. Carried alongside the routing overrides so the proxy reads /// everything model-scoped in one HashMap lookup. pub max_output_tokens: Option, + /// The model's reasoning setting (`models.reasoning_effort`): see + /// `proxy::effort`. `None` ⇒ requests go out as sent. + pub reasoning: Option, } /// Affinity scope — see `proxy.rs` for the runtime semantics. @@ -334,6 +337,7 @@ mod tests { affinity_mode: Some(AffinityMode::None), affinity_ttl_secs: Some(60), max_output_tokens: None, + reasoning: None, }, ); let cfg = router.config_for("gpt-4o-mini"); diff --git a/crates/server/src/app.rs b/crates/server/src/app.rs index ee513950..6c449749 100644 --- a/crates/server/src/app.rs +++ b/crates/server/src/app.rs @@ -1252,10 +1252,11 @@ pub(crate) async fn load_providers_into_router( affinity_mode: Option, affinity_ttl_secs: Option, max_output_tokens: Option, + reasoning_effort: Option, } let model_rows = sqlx::query_as::<_, ModelRow>( r#"SELECT model_id, routing_strategy, affinity_mode, affinity_ttl_secs, - max_output_tokens + max_output_tokens, reasoning_effort FROM models"#, ) .fetch_all(&state.db) @@ -1281,6 +1282,10 @@ pub(crate) async fn load_providers_into_router( .max_output_tokens .and_then(|v| u32::try_from(v).ok()) .filter(|v| *v > 0), + reasoning: m + .reasoning_effort + .as_deref() + .and_then(think_watch_gateway::proxy::effort::ReasoningSetting::parse), }; // Skip storing the all-default config (saves a HashMap entry // per model that's just inheriting global defaults). @@ -1288,6 +1293,7 @@ pub(crate) async fn load_providers_into_router( || cfg.affinity_mode.is_some() || cfg.affinity_ttl_secs.is_some() || cfg.max_output_tokens.is_some() + || cfg.reasoning.is_some() { router.set_model_config(&m.model_id, cfg); } diff --git a/crates/server/src/handlers/models.rs b/crates/server/src/handlers/models.rs index 5bad2cfe..0efe9083 100644 --- a/crates/server/src/handlers/models.rs +++ b/crates/server/src/handlers/models.rs @@ -124,6 +124,12 @@ pub struct CreateModelRequest { /// for nothing gets it. NULL/missing = no limit. #[serde(default)] pub max_output_tokens: Option, + /// The reasoning setting: `off` takes reasoning out of requests that + /// have it on; `low` / `medium` / `high` / `xhigh` / `max` change the + /// effort of requests whose reasoning is on. NULL/missing = requests + /// go out as sent. + #[serde(default)] + pub reasoning_effort: Option, /// Removed: see [`refuse_output_guardrails`]. #[serde(default)] #[schema(value_type = Option>, deprecated)] @@ -175,6 +181,7 @@ pub async fn create_model( )?; refuse_output_guardrails(req.output_guardrails.as_ref())?; let max_output_tokens = max_output_tokens(req.max_output_tokens)?; + let reasoning_effort = reasoning_effort(req.reasoning_effort)?; let model = repo::insert( &state.db, @@ -188,6 +195,7 @@ pub async fn create_model( affinity_ttl_secs: req.affinity_ttl_secs, tags: req.tags.as_deref(), max_output_tokens, + reasoning_effort: reasoning_effort.as_deref(), cache_weights: cache, }, ) @@ -207,9 +215,9 @@ pub async fn create_model( // the 5-min TTL elapses. state.weight_cache.invalidate_all().await; - // The output cap is read from the router's per-model config; a route - // naming this model id may already exist. - if max_output_tokens.is_some() { + // The output cap and the reasoning setting are read from the router's + // per-model config; a route naming this model id may already exist. + if max_output_tokens.is_some() || reasoning_effort.is_some() { crate::app::rebuild_gateway_router(&state).await; } @@ -254,6 +262,11 @@ pub struct UpdateModelRequest { #[serde(default, deserialize_with = "deserialize_some")] #[schema(value_type = Option)] pub max_output_tokens: Option>, + /// PATCH-clearable reasoning setting. Absent = unchanged, JSON `null` + /// = clear (requests go out as sent), string = one of off / low / + /// medium / high / xhigh / max. + #[serde(default, deserialize_with = "deserialize_some")] + pub reasoning_effort: Option>, /// Removed: see [`refuse_output_guardrails`]. #[serde(default)] #[schema(value_type = Option>, deprecated)] @@ -292,6 +305,20 @@ pub(crate) fn max_output_tokens(n: Option) -> Result, AppError> .transpose() } +/// A model's reasoning setting as stored: one of +/// `think_watch_gateway::proxy::effort::SETTINGS`, or none — the column's +/// CHECK, as a useful 400. +pub(crate) fn reasoning_effort(value: Option) -> Result, AppError> { + use think_watch_gateway::proxy::effort::SETTINGS; + match value { + Some(v) if !SETTINGS.contains(&v.as_str()) => Err(AppError::BadRequest(format!( + "reasoning_effort must be one of: {}", + SETTINGS.join(", ") + ))), + v => Ok(v), + } +} + /// Cache weights may be zero (an upstream that does not bill cache /// reads) but not negative — the column's CHECK, as a useful 400. fn validate_cache_weights(weights: &[Option]) -> Result<(), AppError> { @@ -402,6 +429,10 @@ pub async fn update_model( None => existing.max_output_tokens, Some(inner) => max_output_tokens(inner)?, }; + let new_reasoning_effort: Option = match req.reasoning_effort { + None => existing.reasoning_effort.clone(), + Some(inner) => reasoning_effort(inner)?, + }; validate_routing_overrides( new_strategy.as_deref(), new_affinity_mode.as_deref(), @@ -423,6 +454,7 @@ pub async fn update_model( affinity_ttl_secs: new_affinity_ttl, tags: new_tags.as_deref(), max_output_tokens: new_max_output_tokens, + reasoning_effort: new_reasoning_effort.as_deref(), cache_weights: cache, }, req.enabled.unwrap_or(existing.enabled), @@ -1648,5 +1680,28 @@ mod tests { let set: UpdateModelRequest = serde_json::from_value(serde_json::json!({"max_output_tokens": 4096})).unwrap(); assert_eq!(set.max_output_tokens, Some(Some(4096))); + + assert_eq!(absent.reasoning_effort, None); + let cleared: UpdateModelRequest = + serde_json::from_value(serde_json::json!({"reasoning_effort": null})).unwrap(); + assert_eq!(cleared.reasoning_effort, Some(None)); + let set: UpdateModelRequest = + serde_json::from_value(serde_json::json!({"reasoning_effort": "max"})).unwrap(); + assert_eq!(set.reasoning_effort, Some(Some("max".into()))); + } + + #[test] + fn the_reasoning_setting_is_off_or_a_level() { + assert_eq!(reasoning_effort(None).unwrap(), None); + for ok in ["off", "low", "medium", "high", "xhigh", "max"] { + assert_eq!( + reasoning_effort(Some(ok.into())).unwrap().as_deref(), + Some(ok) + ); + } + for bad in ["minimal", "none", "on", "", "High"] { + let e = reasoning_effort(Some(bad.into())).unwrap_err(); + assert!(e.to_string().contains("reasoning_effort"), "{bad}: {e}"); + } } } diff --git a/crates/server/src/handlers/providers.rs b/crates/server/src/handlers/providers.rs index 2211274b..f54a59be 100644 --- a/crates/server/src/handlers/providers.rs +++ b/crates/server/src/handlers/providers.rs @@ -386,6 +386,9 @@ pub async fn update_provider( // path back for an operator who fixed access upstream and // re-saved their credentials. let forgotten = crate::protocol_probe::clear_for_provider(&state.db, id).await; + // And for the reasoning efforts it refused: another upstream may + // take them. + think_watch_gateway::proxy::effort::forget_provider(&state.redis, id).await; if cleared > 0 || forgotten > 0 { tracing::info!( provider = %existing.name, @@ -464,6 +467,7 @@ pub async fn delete_provider( .await?; let name = repo::name_of(&state.db, id).await?; let routes_deleted = repo::soft_delete(&state.db, id).await?; + think_watch_gateway::proxy::effort::forget_provider(&state.redis, id).await; state.audit.log( auth_user diff --git a/crates/server/src/services/model_repository.rs b/crates/server/src/services/model_repository.rs index 79ef440e..15147ef6 100644 --- a/crates/server/src/services/model_repository.rs +++ b/crates/server/src/services/model_repository.rs @@ -52,6 +52,8 @@ pub struct ModelRow { /// The most output tokens a request to this model may ask for. /// `None` ⇒ no limit. pub max_output_tokens: Option, + /// The reasoning setting (`off`, `low` … `max`). `None` ⇒ unchanged. + pub reasoning_effort: Option, } #[derive(Debug, Serialize, sqlx::FromRow, utoipa::ToSchema)] @@ -92,7 +94,7 @@ pub struct ModelRouteRow { const MODEL_COLUMNS: &str = "id, model_id, display_name, input_weight, output_weight, \ cache_read_weight, cache_write_weight, cache_write_1h_weight, \ routing_strategy, affinity_mode, affinity_ttl_secs, tags, enabled, \ - max_output_tokens"; + max_output_tokens, reasoning_effort"; /// One page of the catalog, and the total matching `search` / `status`. /// @@ -144,7 +146,7 @@ pub async fn list( m.enabled, COALESCE(rc.providers, '{{}}'::text[]) AS providers, m.routing_strategy, m.affinity_mode, m.affinity_ttl_secs, - m.max_output_tokens + m.max_output_tokens, m.reasoning_effort FROM models m LEFT JOIN LATERAL ( SELECT COUNT(*) AS route_count, @@ -186,6 +188,7 @@ pub struct ModelFields<'a> { pub affinity_ttl_secs: Option, pub tags: Option<&'a [String]>, pub max_output_tokens: Option, + pub reasoning_effort: Option<&'a str>, /// Read, 5-minute write, 1-hour write. pub cache_weights: [Option; 3], } @@ -196,8 +199,9 @@ pub async fn insert(pool: &PgPool, model_id: &str, f: &ModelFields<'_>) -> Resul (model_id, display_name, input_weight, output_weight, routing_strategy, affinity_mode, affinity_ttl_secs, tags, max_output_tokens, - cache_read_weight, cache_write_weight, cache_write_1h_weight) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) + cache_read_weight, cache_write_weight, cache_write_1h_weight, + reasoning_effort) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) RETURNING {MODEL_COLUMNS}"# ); Ok(sqlx::query_as::<_, Model>(AssertSqlSafe(sql)) @@ -213,6 +217,7 @@ pub async fn insert(pool: &PgPool, model_id: &str, f: &ModelFields<'_>) -> Resul .bind(f.cache_weights[0]) .bind(f.cache_weights[1]) .bind(f.cache_weights[2]) + .bind(f.reasoning_effort) .fetch_one(pool) .await?) } @@ -245,7 +250,8 @@ pub async fn update( max_output_tokens = $10, cache_read_weight = $11, cache_write_weight = $12, - cache_write_1h_weight = $13 + cache_write_1h_weight = $13, + reasoning_effort = $14 WHERE id = $1 RETURNING {MODEL_COLUMNS}"# ); @@ -263,6 +269,7 @@ pub async fn update( .bind(f.cache_weights[0]) .bind(f.cache_weights[1]) .bind(f.cache_weights[2]) + .bind(f.reasoning_effort) .fetch_one(pool) .await?) } diff --git a/crates/test-support/tests/gateway_proxy.rs b/crates/test-support/tests/gateway_proxy.rs index f8207697..1ab61029 100644 --- a/crates/test-support/tests/gateway_proxy.rs +++ b/crates/test-support/tests/gateway_proxy.rs @@ -665,3 +665,302 @@ async fn an_upstream_that_refuses_added_cache_breakpoints_gets_the_request_witho assert_eq!(sent.len(), 3); assert!(!String::from_utf8_lossy(&sent[2].body).contains("cache_control")); } + +/// A Chat upstream that takes reasoning efforts up to `high`: a request +/// asking for more is refused with a 400 naming the ones it takes, the +/// way OpenAI words it. It does not take `none` either. +async fn chat_upstream_up_to_high() -> MockProvider { + use wiremock::matchers::{method, path}; + use wiremock::{Mock, Request, ResponseTemplate}; + let upstream = MockProvider { + server: wiremock::MockServer::start().await, + }; + upstream + .mount( + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(|req: &Request| { + let body: Value = serde_json::from_slice(&req.body).unwrap_or_default(); + let effort = body["reasoning_effort"].as_str().unwrap_or(""); + if effort == "none" { + return ResponseTemplate::new(400).set_body_json(json!({"error": { + "message": "Invalid value for 'reasoning_effort': 'none'.", + "type": "invalid_request_error", + }})); + } + if matches!(effort, "xhigh" | "max") { + return ResponseTemplate::new(400).set_body_json(json!({"error": { + "message": format!("Unsupported value: '{effort}' is not supported with this model. Supported values are: 'low', 'medium', and 'high'."), + "type": "invalid_request_error", + "param": "reasoning_effort", + "code": "unsupported_value", + }})); + } + let usage = json!({"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}); + if body["stream"] == true { + let chunk = |choices: Value, usage: Value| { + format!( + "data: {}\n\n", + json!({"id": "c", "object": "chat.completion.chunk", "created": 0, + "model": "m", "choices": choices, "usage": usage}) + ) + }; + let sse = [ + chunk( + json!([{"index": 0, "delta": {"role": "assistant", "content": "hi"}, "finish_reason": null}]), + Value::Null, + ), + chunk(json!([{"index": 0, "delta": {}, "finish_reason": "stop"}]), Value::Null), + chunk(json!([]), usage), + "data: [DONE]\n\n".to_string(), + ] + .concat(); + ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream") + } else { + ResponseTemplate::new(200).set_body_json(json!({ + "id": "c", "object": "chat.completion", "created": 0, "model": "m", + "choices": [{"index": 0, "finish_reason": "stop", + "message": {"role": "assistant", "content": "hi"}}], + "usage": usage, + })) + } + }), + ) + .await; + upstream +} + +/// The reasoning efforts `upstream` received, in order. +async fn efforts_sent(upstream: &MockProvider) -> Vec> { + upstream + .received_requests() + .await + .iter() + .map(|r| { + let body: Value = r.body_json().unwrap(); + body["reasoning_effort"].as_str().map(str::to_string) + }) + .collect() +} + +/// `model` on `upstream` (OpenAI Chat), with the reasoning `setting`, and a +/// key: the provider's id and the key. +async fn seed_reasoning( + app: &TestApp, + upstream: &str, + model: &str, + setting: Option<&str>, +) -> (Uuid, String) { + let user = fixtures::create_random_user(&app.db).await.unwrap(); + let provider = + fixtures::create_provider(&app.db, &unique_name("effort"), "openai", upstream, None) + .await + .unwrap(); + fixtures::create_model_and_route(&app.db, provider.id, model) + .await + .unwrap(); + sqlx::query("UPDATE models SET reasoning_effort = $1 WHERE model_id = $2") + .bind(setting) + .bind(model) + .execute(&app.db) + .await + .unwrap(); + app.rebuild_gateway_router().await; + let key = + fixtures::create_api_key(&app.db, user.user.id, "effort", &["ai_gateway"], None, None) + .await + .unwrap() + .plaintext; + (provider.id, key) +} + +/// A refused reasoning effort goes to the same upstream once more with the +/// level it named, the client gets the answer, and later requests send +/// that level straight away — streamed or not. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn a_refused_reasoning_effort_is_resent_once_and_remembered() { + use fred::interfaces::HashesInterface; + // The provider's new base URL is the loopback mock again. + let app = TestApp::spawn_reaching_loopback().await; + let upstream = chat_upstream_up_to_high().await; + let model = "effort-refused"; + let (provider_id, key) = seed_reasoning(&app, &upstream.uri(), model, None).await; + let gw = app.gateway_client(); + gw.set_bearer(&key); + // Sampled at a temperature: a repeated request is not a cache hit. + let ask = |stream: bool| { + json!({"model": model, "stream": stream, "temperature": 0.5, + "reasoning_effort": "xhigh", + "messages": [{"role": "user", "content": "ping"}]}) + }; + + let resp = gw.post("/v1/chat/completions", ask(false)).await.unwrap(); + resp.assert_ok(); + assert_eq!( + efforts_sent(&upstream).await, + [Some("xhigh".into()), Some("high".into())] + ); + + // Remembered once the resend was answered (written in the background). + let memory_key = format!("effort_memory:{{provider:{provider_id}}}"); + let field = format!("openai-chat:{model}"); + let mut remembered: Option = None; + for _ in 0..50 { + remembered = app.state.redis.hget(&memory_key, &field).await.unwrap(); + if remembered.is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + let remembered: Value = serde_json::from_str(&remembered.expect("remembered")).unwrap(); + assert_eq!(remembered["xhigh"]["use"], "high", "{remembered}"); + + // Now straight to `high`, for a whole answer and a stream alike. + for stream in [false, true] { + let before = upstream.received_requests().await.len(); + let resp = gw.post("/v1/chat/completions", ask(stream)).await.unwrap(); + resp.assert_ok(); + assert_eq!( + efforts_sent(&upstream).await[before..], + [Some("high".to_string())], + "stream={stream}" + ); + } + + // A changed endpoint forgets: another upstream may take `xhigh`. + let (console, _) = admin_session_with_user(&app).await; + console + .patch( + &format!("/api/admin/providers/{provider_id}"), + json!({"base_url": upstream.uri()}), + ) + .await + .unwrap() + .assert_ok(); + let gone: Option = app.state.redis.hget(&memory_key, &field).await.unwrap(); + assert_eq!(gone, None); + + // A remembered fix the upstream refuses itself is replaced by the one + // that is answered next. + let now = chrono::Utc::now().timestamp(); + let _: () = app + .state + .redis + .hset( + &memory_key, + ( + field.as_str(), + json!({"xhigh": {"use": "max", "at": now}}).to_string(), + ), + ) + .await + .unwrap(); + let before = upstream.received_requests().await.len(); + gw.post("/v1/chat/completions", ask(false)) + .await + .unwrap() + .assert_ok(); + assert_eq!( + efforts_sent(&upstream).await[before..], + [Some("max".to_string()), Some("high".to_string())] + ); + let mut replaced = Value::Null; + for _ in 0..50 { + let v: Option = app.state.redis.hget(&memory_key, &field).await.unwrap(); + replaced = v + .map(|v| serde_json::from_str(&v).unwrap()) + .unwrap_or_default(); + if replaced["xhigh"]["use"] == "high" { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + assert_eq!(replaced["xhigh"]["use"], "high", "{replaced}"); +} + +/// A refused effort refused again is the answer: one resend, no more. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn an_effort_refused_twice_is_the_answer() { + use wiremock::matchers::{method, path}; + use wiremock::{Mock, ResponseTemplate}; + let app = TestApp::spawn().await; + let upstream = MockProvider { + server: wiremock::MockServer::start().await, + }; + upstream + .mount( + Mock::given(method("POST")) + .and(path("/v1/chat/completions")) + .respond_with(ResponseTemplate::new(400).set_body_json(json!({"error": { + "message": "reasoning_effort: only 'low' or 'medium' are allowed", + "type": "invalid_request_error"}}))), + ) + .await; + let (_, key) = seed_reasoning(&app, &upstream.uri(), "effort-twice", None).await; + let gw = app.gateway_client(); + gw.set_bearer(&key); + let resp = gw + .post( + "/v1/chat/completions", + json!({"model": "effort-twice", "reasoning_effort": "high", + "messages": [{"role": "user", "content": "ping"}]}), + ) + .await + .unwrap(); + resp.assert_status(400); + assert_eq!( + efforts_sent(&upstream).await, + [Some("high".into()), Some("medium".into())] + ); +} + +/// The model's setting: a level changes requests that have reasoning on +/// and leaves the rest alone; `off` writes OpenAI's `none`, and an upstream +/// that does not take it gets the request without the field — remembered. +#[ignore = "integration test — run via `make test-it`"] +#[tokio::test] +async fn the_models_reasoning_setting_rewrites_requests_with_reasoning_on() { + let app = TestApp::spawn().await; + let upstream = chat_upstream_up_to_high().await; + let (_, key) = seed_reasoning(&app, &upstream.uri(), "effort-medium", Some("medium")).await; + let (_, off_key) = seed_reasoning(&app, &upstream.uri(), "effort-off", Some("off")).await; + let gw = app.gateway_client(); + + gw.set_bearer(&key); + for (n, effort) in [(1, Some("low")), (2, None)] { + let mut body = json!({"model": "effort-medium", "temperature": 0.1 * f64::from(n), + "messages": [{"role": "user", "content": "ping"}]}); + if let Some(e) = effort { + body["reasoning_effort"] = json!(e); + } + gw.post("/v1/chat/completions", body) + .await + .unwrap() + .assert_ok(); + } + // `low` became `medium`; the request without reasoning stayed without. + assert_eq!(efforts_sent(&upstream).await, [Some("medium".into()), None]); + + gw.set_bearer(&off_key); + for n in [1, 2] { + gw.post( + "/v1/chat/completions", + json!({"model": "effort-off", "reasoning_effort": "high", + "temperature": 0.1 * f64::from(n), + "messages": [{"role": "user", "content": "ping"}]}), + ) + .await + .unwrap() + .assert_ok(); + if n == 1 { + // Remembered in the background. + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + } + assert_eq!( + efforts_sent(&upstream).await[2..], + [Some("none".into()), None, None] + ); +} diff --git a/db/schema.sql b/db/schema.sql index 877519f4..00eb4a63 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -385,6 +385,13 @@ CREATE TABLE IF NOT EXISTS models ( -- is lowered to it, and a request without one gets it. NULL ⇒ no -- limit. See crates/gateway/src/proxy/generate.rs. max_output_tokens INTEGER CHECK (max_output_tokens IS NULL OR max_output_tokens > 0), + -- The reasoning setting: 'off' takes reasoning out of requests that + -- have it on; a level changes the effort of requests whose reasoning + -- is on. NULL ⇒ requests go out as sent. See + -- crates/gateway/src/proxy/effort.rs. + reasoning_effort TEXT + CHECK (reasoning_effort IS NULL + OR reasoning_effort IN ('off', 'low', 'medium', 'high', 'xhigh', 'max')), created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); @@ -401,6 +408,9 @@ ALTER TABLE models ADD COLUMN IF NOT EXISTS cache_write_1h_weight -- after this file (crates/common/src/guard_policy/legacy.rs). ALTER TABLE models ADD COLUMN IF NOT EXISTS max_output_tokens INTEGER CHECK (max_output_tokens IS NULL OR max_output_tokens > 0); +ALTER TABLE models ADD COLUMN IF NOT EXISTS reasoning_effort + TEXT CHECK (reasoning_effort IS NULL + OR reasoning_effort IN ('off', 'low', 'medium', 'high', 'xhigh', 'max')); -- Platform-wide per-token pricing baseline. Single-row singleton -- (PK pinned to 1 via CHECK). `cost($) = tokens × weight × baseline`. diff --git a/web/src/components/segmented.tsx b/web/src/components/segmented.tsx index 4136ee05..3697176a 100644 --- a/web/src/components/segmented.tsx +++ b/web/src/components/segmented.tsx @@ -22,6 +22,7 @@ export function Segmented({ label, disabled = false, size = 'default', + fill = false, className, }: { value: T; @@ -31,6 +32,8 @@ export function Segmented({ label: string; disabled?: boolean; size?: 'sm' | 'default'; + /** Take the whole row, the options sharing it equally. */ + fill?: boolean; className?: string; }) { const refs = useRef<(HTMLButtonElement | null)[]>([]); @@ -57,7 +60,8 @@ export function Segmented({ aria-disabled={disabled || undefined} onKeyDown={onKeyDown} className={cn( - 'inline-flex items-center gap-0.5 rounded-md border bg-muted/30 p-0.5', + fill ? 'flex w-full' : 'inline-flex', + 'items-center gap-0.5 rounded-md border bg-muted/30 p-0.5', size === 'sm' ? 'text-xs' : 'text-sm', className, )} @@ -77,7 +81,8 @@ export function Segmented({ disabled={disabled || o.disabled} onClick={() => !checked && onChange(o.value)} className={cn( - 'rounded px-2.5 font-medium whitespace-nowrap transition-colors outline-none focus-visible:ring-2 focus-visible:ring-ring/50 disabled:cursor-not-allowed disabled:opacity-50', + fill ? 'flex-auto px-1.5' : 'px-2.5', + 'rounded font-medium whitespace-nowrap transition-colors outline-none focus-visible:ring-2 focus-visible:ring-ring/50 disabled:cursor-not-allowed disabled:opacity-50', size === 'sm' ? 'py-0.5' : 'py-1', checked ? 'bg-background text-foreground shadow-sm' diff --git a/web/src/i18n/en.json b/web/src/i18n/en.json index 7006d45a..d290e064 100644 --- a/web/src/i18n/en.json +++ b/web/src/i18n/en.json @@ -864,13 +864,19 @@ "cacheReadWeight": "Cache read weight", "cacheWriteWeight": "Cache write weight", "cacheWrite1hWeight": "1-hour cache write weight", - "maxOutputTokens": "Max output tokens" + "maxOutputTokens": "Max output tokens", + "reasoningEffort": "Reasoning" }, "useGlobalDefault": "Use global default", "unlimited": "Unlimited", "routingOverrideTitle": "Routing overrides", "routingOverrideHint": "Override the gateway-wide defaults for this model. Leave blank to inherit from Settings → Gateway.", "maxOutputTokensHint": "The most output tokens a request to this model can ask for. A request that sets a limit gets the smaller of the two; one that sets none gets this value if it is within the model family's default (32,000 for Claude, 8,192 for others), and the model's own default otherwise. Leave blank for no limit.", + "reasoningEffortHint": "A level applies only to requests that already have reasoning on; requests without reasoning (such as title generation) are left unchanged.", + "reasoning": { + "keep": "Unchanged", + "off": "Off" + }, "strategy": { "weighted": "Weighted random", "latency": "Latency-first", diff --git a/web/src/i18n/zh.json b/web/src/i18n/zh.json index 6f9af408..2ace1a87 100644 --- a/web/src/i18n/zh.json +++ b/web/src/i18n/zh.json @@ -864,13 +864,19 @@ "cacheReadWeight": "缓存读取权重", "cacheWriteWeight": "缓存写入权重", "cacheWrite1hWeight": "1 小时缓存写入权重", - "maxOutputTokens": "最大输出 token" + "maxOutputTokens": "最大输出 token", + "reasoningEffort": "推理" }, "useGlobalDefault": "使用全局默认", "unlimited": "不限制", "routingOverrideTitle": "路由覆盖", "routingOverrideHint": "针对此模型覆盖网关全局默认。留空则继承「设置 → 网关」中的配置。", "maxOutputTokensHint": "发往此模型的请求最多输出这么多 token。请求写了上限时取两者中较小的；未写时，此值不超过该模型系列的默认上限（Claude 32,000，其他 8,192）则按此值发送，否则按模型默认。留空则不限制。", + "reasoningEffortHint": "指定档位时，仅改写已开启推理的请求；未开启推理的请求（如生成标题）保持不变。", + "reasoning": { + "keep": "不改变", + "off": "关闭" + }, "strategy": { "weighted": "加权随机", "latency": "延迟优先", diff --git a/web/src/routes/gateway/models/ModelDetailSheet.tsx b/web/src/routes/gateway/models/ModelDetailSheet.tsx index 19b816e0..53fe83f5 100644 --- a/web/src/routes/gateway/models/ModelDetailSheet.tsx +++ b/web/src/routes/gateway/models/ModelDetailSheet.tsx @@ -164,6 +164,16 @@ export function ModelDetailSheet({ +
+
{t('models.field.reasoningEffort')}
+
+ {model.reasoning_effort == null + ? t('models.reasoning.keep') + : model.reasoning_effort === 'off' + ? t('models.reasoning.off') + : model.reasoning_effort} +
+
{CACHE_WEIGHTS.map((k) => (
diff --git a/web/src/routes/gateway/models/ModelEditorDialog.test.tsx b/web/src/routes/gateway/models/ModelEditorDialog.test.tsx index 0513142c..9da31cb8 100644 --- a/web/src/routes/gateway/models/ModelEditorDialog.test.tsx +++ b/web/src/routes/gateway/models/ModelEditorDialog.test.tsx @@ -86,3 +86,39 @@ describe('ModelEditorDialog max output tokens', () => { expect(apiPatch).not.toHaveBeenCalled() }) }) + +describe('ModelEditorDialog reasoning', () => { + it('leaves requests unchanged by default and sends a chosen level', async () => { + const user = userEvent.setup() + renderEditor(model) + + const group = screen.getByRole('radiogroup', { name: 'Reasoning' }) + expect(screen.getByRole('radio', { name: 'Unchanged' })).toHaveAttribute('aria-checked', 'true') + expect(group).toBeInTheDocument() + await user.click(screen.getByRole('radio', { name: 'max' })) + await user.click(screen.getByRole('button', { name: 'Save' })) + + await waitFor(() => + expect(apiPatch).toHaveBeenCalledWith( + '/api/admin/models/m-1', + expect.objectContaining({ reasoning_effort: 'max' }), + ), + ) + }) + + it('shows the stored setting and clears it back to unchanged', async () => { + const user = userEvent.setup() + renderEditor({ ...model, reasoning_effort: 'off' }) + + expect(screen.getByRole('radio', { name: 'Off' })).toHaveAttribute('aria-checked', 'true') + await user.click(screen.getByRole('radio', { name: 'Unchanged' })) + await user.click(screen.getByRole('button', { name: 'Save' })) + + await waitFor(() => + expect(apiPatch).toHaveBeenCalledWith( + '/api/admin/models/m-1', + expect.objectContaining({ reasoning_effort: null }), + ), + ) + }) +}) diff --git a/web/src/routes/gateway/models/ModelEditorDialog.tsx b/web/src/routes/gateway/models/ModelEditorDialog.tsx index 2d8d0f6f..debc86f4 100644 --- a/web/src/routes/gateway/models/ModelEditorDialog.tsx +++ b/web/src/routes/gateway/models/ModelEditorDialog.tsx @@ -13,6 +13,7 @@ import { } from '@/components/ui/dialog'; import { Input } from '@/components/ui/input'; import { Label } from '@/components/ui/label'; +import { Segmented } from '@/components/segmented'; import { Select, SelectContent, @@ -28,6 +29,7 @@ import { AFFINITY_MODES, CACHE_WEIGHTS, MAX_OUTPUT_TOKENS_CEILING, + REASONING_LEVELS, ROUTING_STRATEGIES, derivedCacheWeight, emptyModelForm, @@ -37,6 +39,7 @@ import { type ModelFormState, type ModelRow, type PlatformPricing, + type ReasoningEffort, type RoutingStrategy, } from './types'; @@ -89,6 +92,7 @@ export function ModelEditorDialog({ affinity_mode: (model.affinity_mode ?? '') as ModelFormState['affinity_mode'], affinity_ttl_secs: model.affinity_ttl_secs == null ? '' : String(model.affinity_ttl_secs), max_output_tokens: model.max_output_tokens == null ? '' : String(model.max_output_tokens), + reasoning_effort: model.reasoning_effort ?? '', }); } else { setForm(emptyModelForm); @@ -143,6 +147,7 @@ export function ModelEditorDialog({ affinity_mode: form.affinity_mode === '' ? null : form.affinity_mode, affinity_ttl_secs: ttlNum, max_output_tokens: maxOutputTokens, + reasoning_effort: form.reasoning_effort === '' ? null : form.reasoning_effort, }; setSaving(true); try { @@ -336,6 +341,24 @@ export function ModelEditorDialog({ />

{t('models.maxOutputTokensHint')}

+
+ +
+ + size="sm" + fill + label={t('models.field.reasoningEffort')} + value={form.reasoning_effort === '' ? 'keep' : form.reasoning_effort} + options={[ + { value: 'keep', label: t('models.reasoning.keep') }, + { value: 'off', label: t('models.reasoning.off') }, + ...REASONING_LEVELS.map((l) => ({ value: l, label: l })), + ]} + onChange={(v) => setForm({ ...form, reasoning_effort: v === 'keep' ? '' : v })} + /> +
+

{t('models.reasoningEffortHint')}

+
{error && ( diff --git a/web/src/routes/gateway/models/types.ts b/web/src/routes/gateway/models/types.ts index af32b694..e354107c 100644 --- a/web/src/routes/gateway/models/types.ts +++ b/web/src/routes/gateway/models/types.ts @@ -37,8 +37,21 @@ export interface ModelRow { /// gets it when it is within the model family's default (32,000 for /// Claude, 8,192 for others). null ⇒ no limit. max_output_tokens?: number | null; + /// Reasoning setting; null ⇒ requests go out as sent. + reasoning_effort?: ReasoningEffort | null; } +/// `off` takes reasoning out of requests that have it on; a level +/// changes the effort of requests whose reasoning is on. Mirrors +/// `SETTINGS` in `crates/gateway/src/proxy/effort.rs`. +export type ReasoningEffort = 'off' | 'low' | 'medium' | 'high' | 'xhigh' | 'max'; + +export const REASONING_EFFORTS: ReasoningEffort[] = ['off', 'low', 'medium', 'high', 'xhigh', 'max']; + +/// The five levels are shown as the APIs write them; only "off" is +/// translated. +export const REASONING_LEVELS: ReasoningEffort[] = ['low', 'medium', 'high', 'xhigh', 'max']; + /// Upper bound the form accepts for `max_output_tokens` — the largest /// value a Postgres `INTEGER` column holds. export const MAX_OUTPUT_TOKENS_CEILING = 2_147_483_647; @@ -156,6 +169,8 @@ export interface ModelFormState { affinity_ttl_secs: string; /// Empty string ⇒ no limit (stored as NULL). max_output_tokens: string; + /// Empty string ⇒ unchanged (stored as NULL). + reasoning_effort: '' | ReasoningEffort; } export interface RouteFormState { @@ -183,6 +198,7 @@ export const emptyModelForm: ModelFormState = { affinity_mode: '', affinity_ttl_secs: '', max_output_tokens: '', + reasoning_effort: '', }; export const emptyRouteForm: RouteFormState = {