diff --git a/crates/engines/src/base.rs b/crates/engines/src/base.rs index df68750b..6385ffa6 100644 --- a/crates/engines/src/base.rs +++ b/crates/engines/src/base.rs @@ -332,7 +332,7 @@ pub(crate) fn parse_json_reply(reply: UpstreamResponse) -> GResult<(u16, Value)> )); }; let v: Value = serde_json::from_slice(bytes) - .map_err(|e| GatewayError::internal("parse upstream response").with_source(e))?; + .map_err(|e| crate::engine::unparsed_reply(reply.status, "parse upstream response", e))?; if let Some(err) = crate::engine::vendor_error(reply.status, &v) { return Err(err); } diff --git a/crates/engines/src/claude_engine.rs b/crates/engines/src/claude_engine.rs index 2ab0abeb..33622e4f 100644 --- a/crates/engines/src/claude_engine.rs +++ b/crates/engines/src/claude_engine.rs @@ -2,13 +2,13 @@ //! blocks, and streaming (the standard anthropic SSE event sequence). Marks //! `is_messages_protocol` so the usage extractor applies the Anthropic map. -use gw_models::{GResult, GatewayError, GatewayResponse}; +use gw_models::{GResult, GatewayResponse}; use gw_protocol::object; use gw_protocol::reasoning::{ThinkingDialect, is_thinking_block}; use serde_json::{Map, Value, json}; use crate::base::base_engine; -use crate::engine::{EngineOutcome, ModelEngine, StreamChunk}; +use crate::engine::{EngineOutcome, ModelEngine, StreamChunk, unparsed_reply}; use crate::transport::{UpstreamBody, UpstreamRequest}; const DEFAULT_MAX_TOKENS: i64 = 1024; @@ -238,7 +238,7 @@ impl ClaudeEngine { fn parse_json(&self, status: u16, bytes: &[u8]) -> GResult { let v: Value = serde_json::from_slice(bytes) - .map_err(|e| GatewayError::internal("parse anthropic response").with_source(e))?; + .map_err(|e| unparsed_reply(status, "parse anthropic response", e))?; self.parse_value(status, v) } @@ -367,7 +367,7 @@ impl ModelEngine for ClaudeEngine { match reply.body { UpstreamBody::Json(b) => { let v: Value = serde_json::from_slice(&b).map_err(|e| { - GatewayError::internal("parse converse response").with_source(e) + unparsed_reply(reply.status, "parse converse response", e) })?; if let Some(err) = crate::engine::vendor_error(reply.status, &v) { return Err(err); @@ -490,8 +490,7 @@ impl<'a> SseState<'a> { } } "content_block_delta" => { - // the native event is forwarded whole, so its text is read; the chat - // surface renders from the chunk instead, so its text is moved + // the native event keeps its delta; the chat surface moves it out if self.preserve_native { if let Some(t) = v["delta"]["text"].as_str() { self.full.push_str(t); @@ -510,7 +509,6 @@ impl<'a> SseState<'a> { } native_chunk.delta = t; } - // the native event keeps its delta; the chat surface moves it out if self.preserve_native { if let Some(t) = v["delta"]["thinking"].as_str() { self.append_thinking(t); diff --git a/crates/engines/src/engine.rs b/crates/engines/src/engine.rs index 1296db66..2a489e04 100644 --- a/crates/engines/src/engine.rs +++ b/crates/engines/src/engine.rs @@ -131,6 +131,15 @@ pub fn vendor_error(http_status: u16, v: &Value) -> Option { }) } +pub(crate) fn unparsed_reply( + status: u16, + what: impl Into, + e: serde_json::Error, +) -> GatewayError { + vendor_error(status, &Value::Null) + .unwrap_or_else(|| GatewayError::internal(what).with_source(e)) +} + /// MiniMax reports business errors as `base_resp.status_code != 0` on an HTTP 200. pub(crate) fn reject_minimax_error(v: &Value) -> GResult<()> { let code = v["base_resp"]["status_code"].as_i64().unwrap_or(0); diff --git a/crates/engines/src/factory.rs b/crates/engines/src/factory.rs index aca70e9c..8a18572c 100644 --- a/crates/engines/src/factory.rs +++ b/crates/engines/src/factory.rs @@ -53,7 +53,7 @@ pub fn get_engine( 400, format!( "realtime model `{}` is served on the /v1/realtime websocket surface, not the chat surface", - p.as_str() + request.model_param_v2.unwrap_or_default().model_name ), )); } @@ -92,6 +92,17 @@ mod tests { assert_eq!(dispatched, Protocol::ALL.len() - 1); } + #[test] + fn a_realtime_refusal_names_the_model() { + let t: SharedTransport = Arc::new(MockTransport); + let request = GatewayRequest { + model_param_v2: Some(ModelParamV2::with_name(Protocol::Realtime, "gpt-rt")), + ..Default::default() + }; + let err = get_engine(request, t).err().unwrap(); + assert!(err.message.contains("`gpt-rt`"), "{}", err.message); + } + #[test] fn rejects_missing_param() { let t: SharedTransport = Arc::new(MockTransport); diff --git a/crates/engines/src/families.rs b/crates/engines/src/families.rs index 6953e153..6f3cf73c 100644 --- a/crates/engines/src/families.rs +++ b/crates/engines/src/families.rs @@ -1397,7 +1397,7 @@ impl ResponsesEngine { /// Non-streaming Responses reply: full `output` array + `usage`. fn parse_json(&self, status: u16, bytes: &[u8]) -> GResult { let mut v: Value = serde_json::from_slice(bytes) - .map_err(|e| GatewayError::internal("parse responses reply").with_source(e))?; + .map_err(|e| crate::engine::unparsed_reply(status, "parse responses reply", e))?; if let Some(err) = crate::engine::vendor_error(status, &v) { return Err(err); } diff --git a/crates/engines/src/mock_transport.rs b/crates/engines/src/mock_transport.rs index e0e43c4f..4186be4c 100644 --- a/crates/engines/src/mock_transport.rs +++ b/crates/engines/src/mock_transport.rs @@ -908,8 +908,6 @@ impl Transport for MockTransport { self.ernie_reply(&req) } else if u.contains("minimax") { self.minimax_reply(&req) - } else if u.contains("meta.llama") { - self.llama_reply(&req) } else if u.contains("/messages") { self.anthropic_reply(&req) } else if u.contains(":generateContent") || u.contains(":streamGenerateContent") { diff --git a/crates/engines/src/openai_engine.rs b/crates/engines/src/openai_engine.rs index 0f99bfff..71031b52 100644 --- a/crates/engines/src/openai_engine.rs +++ b/crates/engines/src/openai_engine.rs @@ -3,7 +3,7 @@ use std::borrow::Cow; -use gw_models::{GResult, GatewayError, GatewayResponse}; +use gw_models::{GResult, GatewayResponse}; use gw_protocol::object; use gw_protocol::reasoning::is_thinking_block; use serde_json::{Map, Value, json}; @@ -143,7 +143,7 @@ impl OpenAiEngine { fn parse_json(&self, status: u16, body: &[u8]) -> GResult { let mut v: Value = serde_json::from_slice(body) - .map_err(|e| GatewayError::internal("parse openai response").with_source(e))?; + .map_err(|e| crate::engine::unparsed_reply(status, "parse openai response", e))?; // surface vendor error envelopes instead of silently returning empty if let Some(err) = crate::engine::vendor_error(status, &v) { return Err(err); diff --git a/crates/engines/src/pump.rs b/crates/engines/src/pump.rs index 3066d390..84581909 100644 --- a/crates/engines/src/pump.rs +++ b/crates/engines/src/pump.rs @@ -30,7 +30,7 @@ pub struct PumpResult { pub(crate) fn reject_json_error(what: &str, status: u16, body: &UpstreamBody) -> GResult<()> { if let UpstreamBody::Json(b) = body { let v: Value = serde_json::from_slice(b) - .map_err(|e| GatewayError::internal(format!("parse {what} reply")).with_source(e))?; + .map_err(|e| crate::engine::unparsed_reply(status, format!("parse {what} reply"), e))?; if let Some(err) = crate::engine::vendor_error(status, &v) { return Err(err); } diff --git a/crates/engines/src/transport.rs b/crates/engines/src/transport.rs index c63f7472..a2fded5e 100644 --- a/crates/engines/src/transport.rs +++ b/crates/engines/src/transport.rs @@ -14,8 +14,8 @@ pub use crate::mock_transport::MockTransport; /// Fixed "created" timestamp for deterministic mock payloads. pub const MOCK_CREATED: i64 = 1_720_000_000; -/// 1x1 PNG-ish placeholder bytes, base64. Deterministic image/audio payload. -pub const MOCK_B64: &str = "TU9DS0JZVEVT"; // "MOCKBYTES" +/// Deterministic image/audio payload: base64 of "MOCKBYTES". +pub const MOCK_B64: &str = "TU9DS0JZVEVT"; pub(crate) const DEFAULT_CONNECT_RETRIES: u32 = 1; /// Wire headers an engine attaches; names are always literals. diff --git a/crates/handler/tests/fallback.rs b/crates/handler/tests/fallback.rs index b070db14..17e4dc40 100644 --- a/crates/handler/tests/fallback.rs +++ b/crates/handler/tests/fallback.rs @@ -28,6 +28,8 @@ impl Transport for Vendor { let (status, body): (u16, &'static [u8]) = match body["model"].as_str() { Some("broken") => (503, br#"{"error":{"message":"vendor down"}}"#), Some("throttled") => (429, br#"{"error":{"message":"rate limited"}}"#), + Some("proxied") => (502, b"502 Bad Gateway"), + Some("oversized") => (413, b"Request Entity Too Large"), Some("healthy") => ( 200, br#"{"model":"healthy","choices":[{"message":{"content":"ok"}}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}"#, @@ -55,6 +57,8 @@ models: - {{name: unavailable, protocol: openai-chat, provider: absent, fallback_models: [broken, throttled, healthy]}} - {{name: broken, protocol: openai-chat, fallback_models: [throttled, healthy]}} - {{name: throttled, protocol: openai-chat}} + - {{name: proxied, protocol: openai-chat, fallback_models: [healthy]}} + - {{name: oversized, protocol: openai-chat, fallback_models: [healthy]}} - {{name: healthy, protocol: openai-chat, qpm: {model_qpm}}} accounts: [{{name: a, provider: p, protocols: [openai-chat]}}]" ); @@ -179,6 +183,29 @@ async fn an_exhausted_chain_reports_the_last_upstream_error() { assert!(err.message.contains("rate limited"), "{}", err.message); } +#[tokio::test] +async fn a_non_json_error_body_keeps_the_vendor_status() { + let (h, vendor) = handler(100.0, "tenants: [{name: t}]", 100).expect("fallback config"); + let ak = h.state().auth.authenticate("k").await.expect("access key"); + let ctx = h + .run(request("proxied", true), ak.clone()) + .await + .expect("a proxy's 502 page falls back"); + let trail = ctx.decisions_line(); + assert!(trail.contains("fallback: proxied -> healthy"), "{trail}"); + let err = h + .run(request("oversized", true), ak) + .await + .err() + .expect("a 413 page is the vendor's refusal"); + assert_eq!((err.http_status, err.original_status()), (413, Some(413))); + assert_eq!( + vendor.calls(), + 3, + "a 4xx page neither fails over nor falls back" + ); +} + #[tokio::test] async fn fallback_consumes_each_request_limit_once() { for (qps, limits, denial) in [ diff --git a/crates/models/src/request.rs b/crates/models/src/request.rs index e0c6ba3c..efe50388 100644 --- a/crates/models/src/request.rs +++ b/crates/models/src/request.rs @@ -6,8 +6,7 @@ pub use domain::*; /// Everything an engine needs to serve one request. #[derive(Debug, Default, Clone)] pub struct GatewayRequest { - /// Arc'd: selected from the shared pool per request and cloned along the - /// engine seam — sharing beats copying six strings per hop. + /// Arc'd: selected from the shared pool per request and cloned along the engine seam. pub account: Option>, pub message: Vec, pub stream: bool, diff --git a/crates/state/src/store.rs b/crates/state/src/store.rs index 3973bd7f..7ef1ebe4 100644 --- a/crates/state/src/store.rs +++ b/crates/state/src/store.rs @@ -67,13 +67,10 @@ pub struct BillingRecord { pub tenant: String, /// Effective end user: the key's `owner` if set, else request metadata; empty /// when neither is present. The precise per-user billing dimension. - #[serde(default)] pub user_id: String, /// Ingress correlation id, joins this row to the access log and audit events. - #[serde(default)] pub request_id: String, /// Unix seconds the call was billed — the billing-period axis. - #[serde(default)] pub created_at_epoch_secs: i64, /// Public model the caller requested. pub model: String, @@ -88,18 +85,14 @@ pub struct BillingRecord { pub total_tokens: i64, pub cost_micros: i64, /// What the serving account's vendor charged us (zero = untracked). - #[serde(default)] pub vendor_cost_micros: i64, /// Non-token units billed (TTS characters, transcription seconds, rerank /// search units); their price is folded into `cost_micros`. - #[serde(default)] pub billed_units: i64, /// PTU spilled over to a paygo account (a failover occurred). - #[serde(default)] pub ptu_spillover: bool, /// Token counts were estimated (an aborted stream billed from delivered /// text), not read from a vendor usage payload. - #[serde(default)] pub estimated: bool, }