diff --git a/crates/consts/src/error_class.rs b/crates/consts/src/error_class.rs new file mode 100644 index 0000000..5f0f68f --- /dev/null +++ b/crates/consts/src/error_class.rs @@ -0,0 +1,258 @@ +//! Client-facing error classification: the Bedrock-derived closed set from +//! the gateway error contract. Every rendered error carries exactly one +//! class; the numeric `ErrCode` stays internal (logs/metrics only). + +use crate::ErrCode; + +/// One classification from the contract's closed set. `ModelStreamError` is +/// in-stream only and never renders at the HTTP phase. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErrClass { + Validation, + /// Hard quota or balance exhaustion; retry only after quota becomes available. + ServiceQuotaExceeded, + UnrecognizedClient, + AccessDenied, + ResourceNotFound, + ModelTimeout, + Conflict, + RequestEntityTooLarge, + ModelError, + Throttling, + InternalServer, + ServiceUnavailable, + ModelStreamError, +} + +impl ErrClass { + /// PascalCase exception name, the `x-amzn-errortype` header value. + pub const fn name(self) -> &'static str { + match self { + ErrClass::Validation => "ValidationException", + ErrClass::ServiceQuotaExceeded => "ServiceQuotaExceededException", + ErrClass::UnrecognizedClient => "UnrecognizedClientException", + ErrClass::AccessDenied => "AccessDeniedException", + ErrClass::ResourceNotFound => "ResourceNotFoundException", + ErrClass::ModelTimeout => "ModelTimeoutException", + ErrClass::Conflict => "ConflictException", + ErrClass::RequestEntityTooLarge => "RequestEntityTooLargeException", + ErrClass::ModelError => "ModelErrorException", + ErrClass::Throttling => "ThrottlingException", + ErrClass::InternalServer => "InternalServerException", + ErrClass::ServiceUnavailable => "ServiceUnavailableException", + ErrClass::ModelStreamError => "ModelStreamErrorException", + } + } + + /// snake_case of [`name`](Self::name), the body `code` field. + pub const fn code(self) -> &'static str { + match self { + ErrClass::Validation => "validation_exception", + ErrClass::ServiceQuotaExceeded => "service_quota_exceeded_exception", + ErrClass::UnrecognizedClient => "unrecognized_client_exception", + ErrClass::AccessDenied => "access_denied_exception", + ErrClass::ResourceNotFound => "resource_not_found_exception", + ErrClass::ModelTimeout => "model_timeout_exception", + ErrClass::Conflict => "conflict_exception", + ErrClass::RequestEntityTooLarge => "request_entity_too_large_exception", + ErrClass::ModelError => "model_error_exception", + ErrClass::Throttling => "throttling_exception", + ErrClass::InternalServer => "internal_server_exception", + ErrClass::ServiceUnavailable => "service_unavailable_exception", + ErrClass::ModelStreamError => "model_stream_error_exception", + } + } + + /// External HTTP status rendered to clients. + pub const fn status(self) -> u16 { + match self { + ErrClass::Validation | ErrClass::ServiceQuotaExceeded => 400, + ErrClass::UnrecognizedClient => 401, + ErrClass::AccessDenied => 403, + ErrClass::ResourceNotFound => 404, + ErrClass::ModelTimeout => 408, + ErrClass::Conflict => 409, + ErrClass::RequestEntityTooLarge => 413, + // ModelStreamError never renders at the HTTP phase; 424 nominal + ErrClass::ModelError | ErrClass::ModelStreamError => 424, + ErrClass::Throttling => 429, + ErrClass::InternalServer => 500, + ErrClass::ServiceUnavailable => 503, + } + } + + /// Coarse `type` for the OpenAI envelope, for SDKs that key on it. + pub const fn openai_type(self) -> &'static str { + match self { + ErrClass::Validation + | ErrClass::ServiceQuotaExceeded + | ErrClass::Conflict + | ErrClass::RequestEntityTooLarge => "invalid_request_error", + ErrClass::UnrecognizedClient => "authentication_error", + ErrClass::AccessDenied => "permission_denied_error", + ErrClass::ResourceNotFound => "not_found_error", + ErrClass::ModelTimeout => "timeout_error", + ErrClass::ModelError | ErrClass::ModelStreamError => "model_error", + ErrClass::Throttling => "rate_limit_error", + ErrClass::InternalServer | ErrClass::ServiceUnavailable => "server_error", + } + } + + /// `type` for the Anthropic envelope — the discriminator its SDKs + /// dispatch exceptions on. + pub const fn anthropic_type(self) -> &'static str { + match self { + ErrClass::Validation | ErrClass::ServiceQuotaExceeded | ErrClass::Conflict => { + "invalid_request_error" + } + ErrClass::UnrecognizedClient => "authentication_error", + ErrClass::AccessDenied => "permission_error", + ErrClass::ResourceNotFound => "not_found_error", + ErrClass::RequestEntityTooLarge => "request_too_large", + ErrClass::Throttling => "rate_limit_error", + ErrClass::ServiceUnavailable => "overloaded_error", + ErrClass::ModelTimeout + | ErrClass::ModelError + | ErrClass::ModelStreamError + | ErrClass::InternalServer => "api_error", + } + } + + /// Classify an internal error for rendering. `None` for 499 + /// (client-closed): the peer is gone, nothing is rendered. + /// + /// Upstream-family codes classify by code first: their `status` is the + /// real vendor status (or a pinned 502), which must not be mistaken for + /// the client's own auth/validation failure. + pub fn classify(code: ErrCode, status: u16) -> Option { + if status == 499 { + return None; + } + let class = match code { + ErrCode::FED_RESP_TIMEOUT => ErrClass::ModelTimeout, + ErrCode::FED_RESP_UNKNOWN + | ErrCode::FED_RESP_RPC_FAILED + | ErrCode::FED_RESP_NIL + | ErrCode::FED_RESP_STATUS_NOT_ZERO + | ErrCode::PARSE_FED_RESP + | ErrCode::GEN_RES_NOT_NULL + // a terminal upstream 429/503 keeps its transient retry + // semantics instead of collapsing into the 424 "don't retry" + | ErrCode::EMPTY_RESP => match status { + 408 => ErrClass::ModelTimeout, + 429 => ErrClass::Throttling, + 503 => ErrClass::ServiceUnavailable, + _ => ErrClass::ModelError, + }, + ErrCode::STOP_LIMIT_MSG => ErrClass::Throttling, + ErrCode::QUOTA_EXHAUSTED => ErrClass::ServiceQuotaExceeded, + ErrCode::PERMISSION_CHECK => ErrClass::AccessDenied, + ErrCode::REQ_JSON | ErrCode::REQ_NON_CHAT => ErrClass::Validation, + _ => return Self::from_status(status), + }; + Some(class) + } + + /// Classify by external-facing status alone (ad-hoc error sites and the + /// open `ErrCode` tail). `None` for 499. + pub fn from_status(status: u16) -> Option { + let class = match status { + 401 => ErrClass::UnrecognizedClient, + 403 => ErrClass::AccessDenied, + 404 => ErrClass::ResourceNotFound, + 408 => ErrClass::ModelTimeout, + 409 => ErrClass::Conflict, + 413 => ErrClass::RequestEntityTooLarge, + 424 => ErrClass::ModelError, + 429 => ErrClass::Throttling, + 499 => return None, + 503 => ErrClass::ServiceUnavailable, + s if s >= 500 => ErrClass::InternalServer, + // 400, 405, 415, 422, 501 and the remaining 4xx tail: the request + // as sent cannot be served + _ => ErrClass::Validation, + }; + Some(class) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn code_is_snake_of_name() { + for c in [ + ErrClass::Validation, + ErrClass::ServiceQuotaExceeded, + ErrClass::UnrecognizedClient, + ErrClass::AccessDenied, + ErrClass::ResourceNotFound, + ErrClass::ModelTimeout, + ErrClass::Conflict, + ErrClass::RequestEntityTooLarge, + ErrClass::ModelError, + ErrClass::Throttling, + ErrClass::InternalServer, + ErrClass::ServiceUnavailable, + ErrClass::ModelStreamError, + ] { + let mut snake = String::new(); + for (i, ch) in c.name().chars().enumerate() { + if ch.is_uppercase() { + if i > 0 { + snake.push('_'); + } + snake.extend(ch.to_lowercase()); + } else { + snake.push(ch); + } + } + assert_eq!(c.code(), snake, "{}", c.name()); + } + } + + #[test] + fn upstream_family_classifies_by_code_not_status() { + // a vendor 401 is the upstream's failure, not the client's + let c = ErrClass::classify(ErrCode::FED_RESP_STATUS_NOT_ZERO, 401).unwrap(); + assert_eq!(c, ErrClass::ModelError); + assert_eq!(c.status(), 424); + // terminal vendor throttle/capacity keep their transient semantics + let c = ErrClass::classify(ErrCode::FED_RESP_STATUS_NOT_ZERO, 429).unwrap(); + assert_eq!(c, ErrClass::Throttling); + let c = ErrClass::classify(ErrCode::FED_RESP_STATUS_NOT_ZERO, 503).unwrap(); + assert_eq!(c, ErrClass::ServiceUnavailable); + // deadline family + let c = ErrClass::classify(ErrCode::FED_RESP_TIMEOUT, 502).unwrap(); + assert_eq!((c, c.status()), (ErrClass::ModelTimeout, 408)); + let c = ErrClass::classify(ErrCode::QUOTA_EXHAUSTED, 400).unwrap(); + assert_eq!(c, ErrClass::ServiceQuotaExceeded); + } + + #[test] + fn client_closed_is_not_rendered() { + assert!(ErrClass::classify(ErrCode::SYSTEM_ERROR, 499).is_none()); + assert!(ErrClass::from_status(499).is_none()); + } + + #[test] + fn own_faults_classify_by_status() { + assert_eq!( + ErrClass::classify(ErrCode::SYSTEM_ERROR, 503), + Some(ErrClass::ServiceUnavailable) + ); + assert_eq!( + ErrClass::classify(ErrCode::SYSTEM_ERROR, 500), + Some(ErrClass::InternalServer) + ); + assert_eq!( + ErrClass::classify(ErrCode::REQ_PARAM, 404), + Some(ErrClass::ResourceNotFound) + ); + assert_eq!( + ErrClass::classify(ErrCode::REQ_PARAM, 400), + Some(ErrClass::Validation) + ); + } +} diff --git a/crates/consts/src/error_code.rs b/crates/consts/src/error_code.rs index 289182a..67b12b5 100644 --- a/crates/consts/src/error_code.rs +++ b/crates/consts/src/error_code.rs @@ -32,6 +32,9 @@ impl ErrCode { pub const FED_RESP_STATUS_NOT_ZERO: ErrCode = ErrCode(2008); pub const PARSE_FED_RESP: ErrCode = ErrCode(2009); pub const GEN_RES_NOT_NULL: ErrCode = ErrCode(2010); + /// Upstream deadline exceeded (header, body, or stream-idle); kept at an + /// internal 5xx status so failover still runs, rendered externally as 408. + pub const FED_RESP_TIMEOUT: ErrCode = ErrCode(2012); pub const REQ_JSON: ErrCode = ErrCode(3001); pub const REQ_PARAM: ErrCode = ErrCode(3002); @@ -40,6 +43,7 @@ impl ErrCode { pub const EMPTY_RESP: ErrCode = ErrCode(4003); pub const STOP_LIMIT_MSG: ErrCode = ErrCode(4004); + pub const QUOTA_EXHAUSTED: ErrCode = ErrCode(4005); #[inline] pub const fn value(self) -> i64 { diff --git a/crates/consts/src/lib.rs b/crates/consts/src/lib.rs index afee66e..11b783a 100644 --- a/crates/consts/src/lib.rs +++ b/crates/consts/src/lib.rs @@ -3,9 +3,11 @@ //! Layer L0: depends on nothing internal. Holds the error-code model and the //! `Protocol` enum the engine factory dispatches on. +pub mod error_class; pub mod error_code; pub mod protocol; +pub use error_class::ErrClass; pub use error_code::ErrCode; pub use protocol::Protocol; diff --git a/crates/dag/src/nodes.rs b/crates/dag/src/nodes.rs index 7c9d500..b99262f 100644 --- a/crates/dag/src/nodes.rs +++ b/crates/dag/src/nodes.rs @@ -15,11 +15,16 @@ const DEFAULT_COMPLETION_RESERVE: i64 = 256; /// `max_tokens: i64::MAX` can't overflow the estimate or corrupt the counter. const MAX_RESERVE: i64 = 1_000_000; -/// A shared admission denial as the wire error every limit answers with. +/// The shared 429 for throttling-style denials (rate/QPM/TPM); hard quota +/// exhaustion answers with [`quota_denied`] instead. fn limit_denied(msg: String) -> GatewayError { GatewayError::new(ErrCode::STOP_LIMIT_MSG, 429, msg) } +fn quota_denied(msg: String) -> GatewayError { + GatewayError::new(ErrCode::QUOTA_EXHAUSTED, 400, msg) +} + /// preprocess/model_quota: per-(AK, model) daily token cap — AK override, else /// tenant default, else unmetered (the per-AK daily cap backstops). Over-quota /// degrades to the tenant's fallback model when one is configured. Runs before @@ -298,7 +303,7 @@ impl DagNode for QuotaCheck { let at = gw_state::epoch_secs(); admission::reserve_daily(ctx.state.governance.as_ref(), &ctx.ak, est, at) .await - .map_err(limit_denied)?; + .map_err(quota_denied)?; ctx.quota_reserved = Some(est); ctx.quota_at = at; ctx.decide("quota_check", format!("reserved {est}")); @@ -461,7 +466,7 @@ impl DagNode for UserBudgetGate { ctx.effective_user_id(), ) .await - .map_err(limit_denied) + .map_err(quota_denied) } } @@ -534,7 +539,7 @@ impl DagNode for CallEngine { ctx.state .avail .record(requested_model(ctx.request.model_param_v2.as_ref()), false); - return Err(first_err); + return Err(named(first_err, ctx)); }; let spillover = failed.is_ptu() && !next.is_ptu(); ctx.decide( @@ -564,15 +569,24 @@ impl DagNode for CallEngine { .avail .record(requested_model(ctx.request.model_param_v2.as_ref()), false); note_failure(ctx, &next.name, threshold, cooldown).await; - Err(e) + Err(named(e, ctx)) } } } - Err(e) => Err(e), + Err(e) => Err(named(e, ctx)), } } } +/// Attach the requested model to a terminal engine-call error, for the +/// contract's 424 `resource_name` extra. Error path only. +fn named(mut e: GatewayError, ctx: &DagContext) -> GatewayError { + if e.resource.is_none() { + e.resource = Some(requested_model(ctx.request.model_param_v2.as_ref()).to_owned()); + } + e +} + /// Record an account failure; on the cooldown transition, alert and note the /// decision — one implementation for both engine attempts, so the fleet-backed /// `record_failure` call and its alerting can't drift apart. diff --git a/crates/engines/src/engine.rs b/crates/engines/src/engine.rs index 179f2ef..302730b 100644 --- a/crates/engines/src/engine.rs +++ b/crates/engines/src/engine.rs @@ -91,15 +91,18 @@ pub fn vendor_error(http_status: u16, v: &Value) -> Option { let err = v.get("error").filter(|e| e.is_object()); let message = match (err, http_status >= 400) { (Some(e), _) => e["message"].as_str().unwrap_or("upstream error"), + // `Message`: the AWS auth layer capitalizes where modeled Bedrock + // errors use lowercase `message`. (None, true) => v["message"] .as_str() .or_else(|| v["msg"].as_str()) + .or_else(|| v["Message"].as_str()) .unwrap_or("upstream error"), (None, false) => return None, } .to_owned(); - let status = if http_status >= 400 { - http_status + let original_status = if http_status >= 400 { + Some(http_status) } else { err.and_then(|e| { e["http_code"] @@ -109,13 +112,16 @@ pub fn vendor_error(http_status: u16, v: &Value) -> Option { .or_else(|| e["code"].as_str().and_then(|s| s.parse::().ok())) }) .filter(|c| *c >= 400) - .unwrap_or(502) }; - Some(GatewayError::new( + let error = GatewayError::new( ErrCode::FED_RESP_STATUS_NOT_ZERO, - status, + original_status.unwrap_or(502), message, - )) + ); + Some(match original_status { + Some(status) => error.with_original_status(status), + None => error, + }) } /// Fill `total_tokens` from prompt + completion when the vendor omitted it. @@ -136,8 +142,13 @@ mod tests { let v = json!({"error": {"message": "overloaded", "code": "529"}}); let e = vendor_error(200, &v).unwrap(); assert_eq!((e.http_status, e.message.as_str()), (529, "overloaded")); + assert_eq!(e.original_status(), Some(529)); let e = vendor_error(503, &v).unwrap(); assert_eq!(e.http_status, 503); + assert_eq!(e.original_status(), Some(503)); + let e = vendor_error(200, &json!({"error": {"message": "overloaded"}})).unwrap(); + assert_eq!(e.http_status, 502); + assert_eq!(e.original_status(), None); } #[test] @@ -146,6 +157,9 @@ mod tests { let e = vendor_error(403, &bedrock).unwrap(); assert_eq!(e.http_status, 403); assert_eq!(e.message, "The security token is invalid."); + let aws_auth = json!({"Message": "Invalid API Key format"}); + let e = vendor_error(403, &aws_auth).unwrap(); + assert_eq!(e.message, "Invalid API Key format"); let dashscope = json!({"code": "Throttling", "message": "rate exceeded", "request_id": "r"}); let e = vendor_error(429, &dashscope).unwrap(); diff --git a/crates/engines/src/factory.rs b/crates/engines/src/factory.rs index cbd9399..5088c26 100644 --- a/crates/engines/src/factory.rs +++ b/crates/engines/src/factory.rs @@ -1,6 +1,6 @@ //! Engine factory: one engine per [`Protocol`], matched exhaustively at //! compile time. Realtime is not a chat-pipeline engine — it is served on the -//! /v1/realtime WebSocket surface, so the factory answers 501-with-pointer for +//! /v1/realtime WebSocket surface, so the factory answers 400-with-pointer for //! it on the chat path. use gw_consts::{ErrCode, Protocol}; @@ -48,8 +48,8 @@ pub fn get_engine( Protocol::Dashscope => Box::new(DashScopeEngine::new(request, transport)), Protocol::Realtime => { return Err(GatewayError::new( - ErrCode::INTERNAL_UNKNOWN, - 501, + ErrCode::REQ_NON_CHAT, + 400, format!( "realtime model `{}` is served on the /v1/realtime websocket surface, not the chat surface", p.as_str() @@ -82,7 +82,7 @@ mod tests { for &p in Protocol::ALL { let got = get_engine(req(p), t.clone()); if p == Protocol::Realtime { - assert_eq!(got.err().map(|e| e.http_status), Some(501), "{p}"); + assert_eq!(got.err().map(|e| e.http_status), Some(400), "{p}"); } else { assert!(got.is_ok(), "no engine for {p}"); dispatched += 1; diff --git a/crates/engines/src/http_transport.rs b/crates/engines/src/http_transport.rs index 4dd21b3..a16b7e8 100644 --- a/crates/engines/src/http_transport.rs +++ b/crates/engines/src/http_transport.rs @@ -11,7 +11,10 @@ use std::time::Duration; use gw_models::{GResult, GatewayError}; -use crate::transport::{MockTransport, Transport, UpstreamBody, UpstreamRequest, UpstreamResponse}; +use crate::transport::{ + MockTransport, StreamFault, Transport, UpstreamBody, UpstreamRequest, UpstreamResponse, + upstream_fault_code, +}; const RETRY_BACKOFF: Duration = Duration::from_millis(100); // A hung connect (black-holed SYN) must surface as a connect error — which the @@ -121,7 +124,7 @@ impl Transport for HttpTransport { Ok(r) => r, Err(_) => { return Err(GatewayError::new( - gw_consts::ErrCode::FED_RESP_RPC_FAILED, + gw_consts::ErrCode::FED_RESP_TIMEOUT, 502, format!( "upstream request failed: no response headers within {:?}", @@ -146,7 +149,7 @@ impl Transport for HttpTransport { } Err(e) => { return Err(GatewayError::new( - gw_consts::ErrCode::FED_RESP_RPC_FAILED, + upstream_fault_code(e.is_timeout()), 502, format!("upstream request failed: {e}"), )); @@ -162,7 +165,10 @@ impl Transport for HttpTransport { .unwrap_or(false); if is_sse { use futures::TryStreamExt; - let stream = Box::pin(resp.bytes_stream().map_err(|e| e.to_string())); + let stream = Box::pin(resp.bytes_stream().map_err(|e| StreamFault { + timeout: e.is_timeout(), + message: e.to_string(), + })); return Ok(UpstreamResponse { status, body: idle_capped(stream, policy.timeout), @@ -174,7 +180,7 @@ impl Transport for HttpTransport { .await .map_err(|_| { GatewayError::new( - gw_consts::ErrCode::FED_RESP_RPC_FAILED, + gw_consts::ErrCode::FED_RESP_TIMEOUT, 502, format!("upstream body not read within {:?}", policy.timeout), ) @@ -182,8 +188,16 @@ impl Transport for HttpTransport { } else { read.await }; - let bytes = - bytes.map_err(|e| GatewayError::internal("read upstream body").with_source(e))?; + let bytes = bytes.map_err(|e| { + // both stay 502 (failover-eligible); the code split preserves the + // external 408-vs-424 classification + let what = if e.is_timeout() { + "read upstream body timed out" + } else { + "read upstream body failed" + }; + GatewayError::new(upstream_fault_code(e.is_timeout()), 502, what).with_source(e) + })?; Ok(UpstreamResponse { status, body: UpstreamBody::Json(bytes), @@ -195,7 +209,7 @@ impl Transport for HttpTransport { /// one terminal error item — no gap between chunks may exceed the policy /// timeout, but an actively flowing stream lives as long as the generation. fn idle_capped( - stream: futures::stream::BoxStream<'static, Result>, + stream: futures::stream::BoxStream<'static, Result>, gap: Duration, ) -> UpstreamBody { use futures::StreamExt; @@ -206,7 +220,13 @@ fn idle_capped( match tokio::time::timeout(gap, s.next()).await { Ok(Some(item)) => Some((item, Some(s))), Ok(None) => None, - Err(_) => Some((Err(format!("stream idle for {gap:?}")), None)), + Err(_) => Some(( + Err(StreamFault { + timeout: true, + message: format!("stream idle for {gap:?}"), + }), + None, + )), } }, ))) diff --git a/crates/engines/src/pump.rs b/crates/engines/src/pump.rs index 3d58c5a..8975189 100644 --- a/crates/engines/src/pump.rs +++ b/crates/engines/src/pump.rs @@ -1,11 +1,11 @@ //! Shared SSE pump: the one buffered/live decode loop every streaming engine //! drives its vendor frames through. -use gw_models::{GResult, GatewayError, StreamChunk}; +use gw_models::{GResult, GatewayError, StreamChunk, StreamError}; use serde_json::Value; use crate::sse::SseDecoder; -use crate::transport::UpstreamBody; +use crate::transport::{StreamFault, UpstreamBody}; /// What a pump run produced. #[derive(Debug, Default)] @@ -41,20 +41,24 @@ pub(crate) fn reject_json_error(what: &str, status: u16, body: &UpstreamBody) -> Ok(()) } -/// A mid-stream transport/decode fault: after bytes reached the client it is a -/// committed abort — keep what was delivered, no failover (`None`); before -/// that it is a plain upstream failure (`Some(err)`). -fn stream_fault(vendor: &'static str, e: &str, sent_any: bool) -> Option { - if sent_any { - tracing::warn!(vendor, error = %e, "upstream stream failed mid-response"); - None - } else { - Some(GatewayError::new( - gw_consts::ErrCode::FED_RESP_RPC_FAILED, - 502, - format!("upstream stream failed: {e}"), - )) +/// Deliver the terminal error frame for a committed abort (best effort — the +/// client may already be gone) and mark the pump result aborted. Committed +/// implies a live channel: `sent_any` only ever turns true with `tx` attached. +async fn abort_frame( + tx: &Option>, + out: &mut PumpResult, + error: StreamError, +) { + debug_assert!(tx.is_some(), "committed abort implies a live channel"); + if let Some(sender) = tx { + let _ = sender + .send(StreamChunk { + error: Some(Box::new(error)), + ..Default::default() + }) + .await; } + out.aborted = true; } pub async fn pump_sse( @@ -88,23 +92,32 @@ where let mut dec = SseDecoder::default(); let mut sent_any = false; while let Some(item) = s.next().await { - let bytes = match item.map_err(|e| stream_fault(vendor, &e, sent_any)) { + // A fault after bytes reached the client is a committed abort, + // not a failover signal: emit the terminal error frame and keep + // what was delivered. Before commit it is a plain upstream + // failure, eligible for failover. + let bytes = match item { Ok(b) => b, - Err(Some(err)) => return Err(err), - Err(None) => { - out.aborted = true; + Err(fault) if sent_any => { + tracing::warn!(vendor, error = %fault.message, "upstream stream failed mid-response"); + abort_frame(&tx, &mut out, fault.stream_error()).await; break; } + Err(fault) => return Err(fault.into_error()), }; - let events = match dec - .feed(&bytes) - .map_err(|e| stream_fault(vendor, &e, sent_any)) - { + let events = match dec.feed(&bytes) { Ok(events) => events, - Err(Some(err)) => return Err(err), - Err(None) => { - out.aborted = true; - break; + Err(e) => { + let fault = StreamFault { + timeout: false, + message: e, + }; + if sent_any { + tracing::warn!(vendor, error = %fault.message, "upstream stream failed mid-response"); + abort_frame(&tx, &mut out, fault.stream_error()).await; + break; + } + return Err(fault.into_error()); } }; for data in events { @@ -119,7 +132,11 @@ where Ok(c) => c, Err(e) if sent_any => { tracing::warn!(vendor, error = %e, "vendor error frame after commit"); - out.aborted = true; + if let Some(error) = StreamError::from_committed_error(e) { + abort_frame(&tx, &mut out, error).await; + } else { + out.aborted = true; + } out.streamed_live = tx.is_some(); return Ok(out); } diff --git a/crates/engines/src/transport.rs b/crates/engines/src/transport.rs index 4b65658..4b9adb9 100644 --- a/crates/engines/src/transport.rs +++ b/crates/engines/src/transport.rs @@ -7,7 +7,7 @@ use std::sync::Arc; use gw_consts::Protocol; -use gw_models::{GResult, GatewayError}; +use gw_models::{GResult, GatewayError, StreamError}; use serde_json::{Value, json}; /// Fixed "created" timestamp for deterministic mock payloads. @@ -28,12 +28,55 @@ pub struct UpstreamRequest { pub account: String, } +/// A live-stream item failure: a transport fault or one of our deadlines. +/// `timeout` selects the ModelTimeout classification downstream. +#[derive(Debug)] +pub struct StreamFault { + pub timeout: bool, + pub message: String, +} + +impl StreamFault { + /// The pre-commit form: an ordinary upstream failure eligible for + /// failover (internal 502 keeps it above the 5xx threshold). + pub fn into_error(self) -> GatewayError { + GatewayError::new( + upstream_fault_code(self.timeout), + 502, + format!("upstream stream failed: {}", self.message), + ) + } + + /// The post-commit form: the terminal in-stream error frame. + pub fn stream_error(&self) -> StreamError { + StreamError { + class: if self.timeout { + gw_consts::ErrClass::ModelTimeout + } else { + gw_consts::ErrClass::ModelStreamError + }, + message: format!("upstream stream failed: {}", self.message), + original_status: None, + } + } +} + +/// The upstream-fault code: deadlines classify as ModelTimeout downstream, +/// everything else as the generic RPC failure. +pub(crate) fn upstream_fault_code(timeout: bool) -> gw_consts::ErrCode { + if timeout { + gw_consts::ErrCode::FED_RESP_TIMEOUT + } else { + gw_consts::ErrCode::FED_RESP_RPC_FAILED + } +} + /// Body of an upstream response: buffered JSON, buffered SSE bytes, or live /// SSE bytes yielded as the vendor sends them. pub enum UpstreamBody { Json(bytes::Bytes), Sse(Vec), - SseStream(futures::stream::BoxStream<'static, Result>), + SseStream(futures::stream::BoxStream<'static, Result>), } impl std::fmt::Debug for UpstreamBody { @@ -62,13 +105,7 @@ impl UpstreamResponse { use futures::StreamExt; let mut buf = Vec::new(); while let Some(item) = s.next().await { - let bytes = item.map_err(|e| { - GatewayError::new( - gw_consts::ErrCode::FED_RESP_RPC_FAILED, - 502, - format!("upstream stream failed: {e}"), - ) - })?; + let bytes = item.map_err(StreamFault::into_error)?; buf.extend_from_slice(&bytes); if buf.len() > MAX_BUFFERED_SSE { return Err(GatewayError::new( diff --git a/crates/engines/tests/http_transport_wire.rs b/crates/engines/tests/http_transport_wire.rs index 20e76bd..3fa859b 100644 --- a/crates/engines/tests/http_transport_wire.rs +++ b/crates/engines/tests/http_transport_wire.rs @@ -12,11 +12,20 @@ use axum::routing::post; use axum::{Json, Router}; use gw_consts::Protocol; use gw_engines::http_transport::{DispatchTransport, HttpTransport}; -use gw_engines::transport::{Transport, UpstreamBody, UpstreamRequest}; +use gw_engines::transport::{StreamFault, Transport, UpstreamBody, UpstreamRequest}; use gw_engines::{ModelEngine, OpenAiEngine}; use gw_models::{ChatMsg, GatewayRequest, ModelParamV2}; use serde_json::{Value, json}; +async fn serve_router(app: Router) -> std::net::SocketAddr { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + addr +} + async fn spawn_vendor() -> String { let app = Router::new().route( "/v1/chat/completions", @@ -52,11 +61,7 @@ async fn spawn_vendor() -> String { } }), ); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - tokio::spawn(async move { - axum::serve(listener, app).await.unwrap(); - }); + let addr = serve_router(app).await; format!("http://{addr}") } @@ -84,6 +89,98 @@ async fn http_transport_json_over_real_socket() { assert_eq!(v["choices"][0]["message"]["content"], "srv:wire"); } +async fn spawn_stalled_json_vendor() -> String { + let app = Router::new().route( + "/slow-json", + post(|| async { + let body = futures::stream::unfold(0, |step| async move { + match step { + 0 => Some(( + Ok::<_, std::convert::Infallible>(bytes::Bytes::from_static(b"{")), + 1, + )), + 1 => { + tokio::time::sleep(Duration::from_secs(2)).await; + Some((Ok(bytes::Bytes::from_static(b"}")), 2)) + } + _ => None, + } + }); + axum::response::Response::builder() + .header("content-type", "application/json") + .body(axum::body::Body::from_stream(body)) + .unwrap() + }), + ); + let addr = serve_router(app).await; + format!("http://{addr}/slow-json") +} + +#[tokio::test] +async fn non_stream_body_timeout_classifies_as_model_timeout() { + let transport = HttpTransport::new(Duration::from_millis(300)).unwrap(); + let err = transport + .send(UpstreamRequest { + protocol: Protocol::OpenaiChat, + method: "POST".into(), + url: spawn_stalled_json_vendor().await, + headers: vec![("content-type".into(), "application/json".into())], + body: b"{}".to_vec(), + stream: false, + account: "slow-body".into(), + }) + .await + .expect_err("the non-stream body must share the request deadline"); + assert_eq!(err.code, gw_consts::ErrCode::FED_RESP_TIMEOUT); + assert_eq!(err.http_status, 502); + assert_eq!( + gw_consts::ErrClass::classify(err.code, err.http_status), + Some(gw_consts::ErrClass::ModelTimeout) + ); + assert!(err.message.contains("read upstream body")); +} + +async fn spawn_breaking_json_vendor() -> String { + let app = Router::new().route( + "/broken-json", + post(|| async { + let body = futures::stream::iter(vec![ + Ok(bytes::Bytes::from_static(b"{\"partial\":")), + Err(std::io::Error::other("mid-body reset")), + ]); + axum::response::Response::builder() + .header("content-type", "application/json") + .body(axum::body::Body::from_stream(body)) + .unwrap() + }), + ); + let addr = serve_router(app).await; + format!("http://{addr}/broken-json") +} + +#[tokio::test] +async fn non_stream_body_break_classifies_as_model_error() { + let transport = HttpTransport::new(Duration::from_secs(5)).unwrap(); + let err = transport + .send(UpstreamRequest { + protocol: Protocol::OpenaiChat, + method: "POST".into(), + url: spawn_breaking_json_vendor().await, + headers: vec![("content-type".into(), "application/json".into())], + body: b"{}".to_vec(), + stream: false, + account: "broken-body".into(), + }) + .await + .expect_err("a mid-body break must surface as an error"); + assert_eq!(err.code, gw_consts::ErrCode::FED_RESP_RPC_FAILED); + assert_eq!(err.http_status, 502, "failover-eligible upstream fault"); + assert_eq!( + gw_consts::ErrClass::classify(err.code, err.http_status), + Some(gw_consts::ErrClass::ModelError) + ); +} + #[tokio::test] async fn http_transport_sse_over_real_socket() { let base = spawn_vendor().await; @@ -266,11 +363,7 @@ async fn spawn_paced_vendor(frames: usize, gap: Duration) -> String { .unwrap() }), ); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - tokio::spawn(async move { - axum::serve(listener, app).await.unwrap(); - }); + let addr = serve_router(app).await; format!("http://{addr}/paced") } @@ -318,6 +411,11 @@ async fn stalled_stream_errors_at_the_idle_gap_instead_of_hanging() { Err(e) => e, }; assert_eq!(err.http_status, 502); + assert_eq!( + err.code, + gw_consts::ErrCode::FED_RESP_TIMEOUT, + "idle gap must carry the timeout code so it classifies as ModelTimeout" + ); assert!( started.elapsed() < Duration::from_secs(5), "must fail at the idle gap, not wait out the vendor: {:?}", @@ -377,11 +475,14 @@ async fn midstream_upstream_error_after_send_aborts_without_failover() { #[async_trait::async_trait] impl Transport for FlakyStream { async fn send(&self, _req: UpstreamRequest) -> gw_models::GResult { - let frames: Vec> = vec![ + let frames: Vec> = vec![ Ok(bytes::Bytes::from( "data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n", )), - Err("connection reset".to_string()), + Err(StreamFault { + timeout: false, + message: "connection reset".to_string(), + }), ]; Ok(UpstreamResponse { status: 200, @@ -391,7 +492,13 @@ async fn midstream_upstream_error_after_send_aborts_without_failover() { } let (tx, mut rx) = tokio::sync::mpsc::channel(8); - tokio::spawn(async move { while rx.recv().await.is_some() {} }); + let collect = tokio::spawn(async move { + let mut got = Vec::new(); + while let Some(c) = rx.recv().await { + got.push(c); + } + got + }); let mut request = GatewayRequest { message: vec![ChatMsg::text("user", "hi")], model_param_v2: Some(ModelParamV2::with_name(Protocol::OpenaiChat, "gpt")), @@ -407,6 +514,14 @@ async fn midstream_upstream_error_after_send_aborts_without_failover() { assert!(out.response.aborted, "post-send break must mark aborted"); assert!(out.streamed_live); assert_eq!(out.response.message, "partial"); + let got = collect.await.unwrap(); + let err = got + .iter() + .find_map(|c| c.error.as_ref()) + .expect("committed abort must deliver a terminal error frame"); + assert_eq!(err.class, gw_consts::ErrClass::ModelStreamError); + assert!(err.message.contains("connection reset")); + assert_eq!(err.original_status, None); } #[tokio::test] @@ -419,7 +534,7 @@ async fn vendor_error_frame_after_send_aborts_without_failover() { #[async_trait::async_trait] impl Transport for ErrorFrameStream { async fn send(&self, _req: UpstreamRequest) -> gw_models::GResult { - let frames: Vec> = vec![ + let frames: Vec> = vec![ Ok(bytes::Bytes::from( "data: {\"choices\":[{\"delta\":{\"content\":\"partial\"}}]}\n\n", )), @@ -435,7 +550,13 @@ async fn vendor_error_frame_after_send_aborts_without_failover() { } let (tx, mut rx) = tokio::sync::mpsc::channel(8); - tokio::spawn(async move { while rx.recv().await.is_some() {} }); + let collect = tokio::spawn(async move { + let mut got = Vec::new(); + while let Some(c) = rx.recv().await { + got.push(c); + } + got + }); let mut request = GatewayRequest { message: vec![ChatMsg::text("user", "hi")], model_param_v2: Some(ModelParamV2::with_name(Protocol::OpenaiChat, "gpt")), @@ -451,4 +572,12 @@ async fn vendor_error_frame_after_send_aborts_without_failover() { assert!(out.response.aborted, "vendor error after send must abort"); assert!(out.streamed_live); assert_eq!(out.response.message, "partial"); + let got = collect.await.unwrap(); + let err = got + .iter() + .find_map(|c| c.error.as_ref()) + .expect("committed vendor error must deliver a terminal error frame"); + assert_eq!(err.class, gw_consts::ErrClass::ModelStreamError); + assert_eq!(err.message, "overloaded"); + assert_eq!(err.original_status, None); } diff --git a/crates/handler/src/lib.rs b/crates/handler/src/lib.rs index d10cf88..11984cb 100644 --- a/crates/handler/src/lib.rs +++ b/crates/handler/src/lib.rs @@ -889,8 +889,8 @@ mod tests { #[tokio::test] async fn abuse_tiers_suspend_after_repeated_rejections() { - // daily_token_quota 1: every request rejects at reserve time (429) - let yaml = "listen: {host: h, port: 1}\nabuse: {tiers: [{rejects: 2, suspend_hours: 2}]}\nmodels: [{name: gpt-4o, protocol: openai-chat}]\naccounts: [{name: a1, provider: openai, protocols: ['openai-chat']}]\naccess_keys: [{ak: k1, product: p, qps: 100, daily_token_quota: 1}]"; + // qps 0: every request is a true throttling rejection. + let yaml = "listen: {host: h, port: 1}\nabuse: {tiers: [{rejects: 2, suspend_hours: 2}]}\nmodels: [{name: gpt-4o, protocol: openai-chat}]\naccounts: [{name: a1, provider: openai, protocols: ['openai-chat']}]\naccess_keys: [{ak: k1, product: p, qps: 0, daily_token_quota: 100000}]"; let cfg = Arc::new(GatewayConfig::from_yaml(yaml).unwrap()); let state = Arc::new(GatewayState::from_config(&cfg)); let h = OnlineHandler::new( @@ -900,10 +900,6 @@ mod tests { let mut alerts = h.state().alerts.take_receiver().expect("receiver"); let key = h.state().auth.authenticate("k1").await.unwrap(); let reject = |k: AkInfo| h.run(chat_req("gpt-4o", "hi"), k); - assert!( - reject(key.clone()).await.is_ok(), - "first request admits and burns the 1-token quota" - ); assert_eq!( reject(key.clone()).await.err().map(|e| e.http_status), Some(429) @@ -938,6 +934,30 @@ mod tests { assert_eq!((ev.kind, ev.subject.as_str()), ("abuse_suspend", "k1")); } + #[tokio::test] + async fn hard_quota_rejection_does_not_trip_abuse_tier() { + let yaml = "listen: {host: h, port: 1}\nabuse: {tiers: [{rejects: 1, suspend_hours: 2}]}\nmodels: [{name: gpt-4o, protocol: openai-chat}]\naccounts: [{name: a1, provider: openai, protocols: ['openai-chat']}]\naccess_keys: [{ak: k1, product: p, qps: 100, daily_token_quota: 1}]"; + let cfg = Arc::new(GatewayConfig::from_yaml(yaml).unwrap()); + let state = Arc::new(GatewayState::from_config(&cfg)); + let h = OnlineHandler::new( + gw_state::SharedConfig::new(cfg, state), + Arc::new(gw_engines::MockTransport), + ); + let key = h.state().auth.authenticate("k1").await.unwrap(); + assert!(h.run(chat_req("gpt-4o", "hi"), key.clone()).await.is_ok()); + let err = h + .run(chat_req("gpt-4o", "hi"), key) + .await + .err() + .expect("hard quota must reject"); + assert_eq!(err.code, gw_consts::ErrCode::QUOTA_EXHAUSTED); + let fresh = h.state().auth.authenticate("k1").await.unwrap(); + assert_eq!( + fresh.status_at(gw_state::epoch_secs()), + gw_state::KeyStatus::Active + ); + } + #[tokio::test] async fn quota_fallback_skips_variant_select() { let yaml = "listen: {host: h, port: 1}\nmodels: [{name: pub-m, protocol: openai-chat, variants: [{model: canary-m, weight: 1}]}, {name: canary-m, protocol: openai-chat}, {name: fb-m, protocol: openai-chat}]\naccounts: [{name: a1, provider: openai, protocols: ['openai-chat']}]\ntenants: [{name: t1, models: [pub-m, canary-m, fb-m], fallback_model: fb-m, model_quotas: {pub-m: 1}}]\naccess_keys: [{ak: k1, tenant: t1, product: p, qps: 100, daily_token_quota: 100000}]"; @@ -997,7 +1017,8 @@ mod tests { .await .err() .expect("second denied by the per-user budget"); - assert_eq!(err.http_status, 429); + assert_eq!(err.http_status, 400); + assert_eq!(err.code, gw_consts::ErrCode::QUOTA_EXHAUSTED); } #[tokio::test] @@ -1092,11 +1113,14 @@ mod tests { _req: gw_engines::transport::UpstreamRequest, ) -> GResult { use futures::StreamExt; - let frames: Vec> = vec![ + let frames: Vec> = vec![ Ok(bytes::Bytes::from( "data: {\"choices\":[{\"delta\":{\"content\":\"partial answer\"}}]}\n\n", )), - Err("connection reset".to_owned()), + Err(gw_engines::transport::StreamFault { + timeout: false, + message: "connection reset".to_owned(), + }), ]; Ok(gw_engines::transport::UpstreamResponse { status: 200, @@ -1153,14 +1177,17 @@ mod tests { _req: gw_engines::transport::UpstreamRequest, ) -> GResult { use futures::StreamExt; - let frames: Vec> = vec![ + let frames: Vec> = vec![ Ok(bytes::Bytes::from( "data: {\"type\":\"message_start\",\"message\":{\"model\":\"claude-sonnet\",\"usage\":{\"input_tokens\":100}}}\n\n", )), Ok(bytes::Bytes::from( "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"delivered words here\"}}\n\n", )), - Err("connection reset".to_owned()), + Err(gw_engines::transport::StreamFault { + timeout: false, + message: "connection reset".to_owned(), + }), ]; Ok(gw_engines::transport::UpstreamResponse { status: 200, @@ -1210,7 +1237,7 @@ mod tests { _req: gw_engines::transport::UpstreamRequest, ) -> GResult { use futures::StreamExt; - let frames: Vec> = vec![ + let frames: Vec> = vec![ Ok(bytes::Bytes::from( "data: {\"choices\":[{\"delta\":{\"content\":\"reach me at jane@corp.com now\"}}]}\n\n", )), @@ -1268,7 +1295,7 @@ mod tests { _req: gw_engines::transport::UpstreamRequest, ) -> GResult { use futures::StreamExt; - let frames: Vec> = vec![ + let frames: Vec> = vec![ Ok(bytes::Bytes::from( "data: {\"choices\":[{\"delta\":{\"content\":\"key sk-abcdefghijklmnopqrstuvwxyz012345 ok\"}}]}\n\n", )), diff --git a/crates/models/src/error.rs b/crates/models/src/error.rs index 2dfd567..ec864de 100644 --- a/crates/models/src/error.rs +++ b/crates/models/src/error.rs @@ -17,6 +17,10 @@ pub struct GatewayError { pub code: ErrCode, pub http_status: u16, pub message: String, + /// the client-requested model, attached on the engine-call error path for + /// the 424 `resource_name` extra; never set on the success path. + pub resource: Option, + original_status: Option, pub source: Option>, } @@ -26,6 +30,8 @@ impl GatewayError { code, http_status, message: message.into(), + resource: None, + original_status: None, source: None, } } @@ -53,6 +59,18 @@ impl GatewayError { self.source = Some(Box::new(source)); self } + + /// Record a status actually received from the upstream. + pub fn with_original_status(mut self, status: u16) -> Self { + self.original_status = Some(status); + self + } + + /// The real upstream status behind this error, when one was received: + /// synthetic statuses used for failover are never returned. + pub fn original_status(&self) -> Option { + self.original_status + } } impl fmt::Display for GatewayError { diff --git a/crates/models/src/lib.rs b/crates/models/src/lib.rs index 8c52fe7..0c369ef 100644 --- a/crates/models/src/lib.rs +++ b/crates/models/src/lib.rs @@ -24,6 +24,6 @@ pub use params::{ pub use request::domain::{Account, ChatMsg}; pub use request::{BatchItem, GatewayRequest, ModelParamV2}; pub use response::GatewayResponse; -pub use response::StreamChunk; +pub use response::{StreamChunk, StreamError}; pub use token_estimate::{HeuristicEncoder, TokenEncoder, estimate_prompt_tokens}; pub use usage::CommonUsage; diff --git a/crates/models/src/response.rs b/crates/models/src/response.rs index 7c566e5..45f3ee3 100644 --- a/crates/models/src/response.rs +++ b/crates/models/src/response.rs @@ -45,6 +45,40 @@ pub struct GatewayResponse { pub raw_usage_json: Vec, } +/// A mid-stream failure as rendered to the client: the contract +/// classification, a human-readable message (no internal numeric codes), and +/// the upstream's original status when an upstream HTTP response was received. +#[derive(Debug, Clone)] +pub struct StreamError { + pub class: gw_consts::ErrClass, + pub message: String, + pub original_status: Option, +} + +impl StreamError { + /// Build from an internal error. `None` for client-closed (499): the + /// peer is gone and no frame is rendered. + pub fn from_error(e: crate::GatewayError) -> Option { + let class = gw_consts::ErrClass::classify(e.code, e.http_status)?; + let original_status = e.original_status(); + Some(Self { + class, + message: e.message, + original_status, + }) + } + + /// Terminal frame for an already-committed stream: a generic upstream + /// failure becomes ModelStreamError; transient classes stay sharper. + pub fn from_committed_error(e: crate::GatewayError) -> Option { + let mut error = Self::from_error(e)?; + if error.class == gw_consts::ErrClass::ModelError { + error.class = gw_consts::ErrClass::ModelStreamError; + } + Some(error) + } +} + /// One streamed response fragment, forwarded to the client as it arrives. #[derive(Debug, Default, Clone)] pub struct StreamChunk { @@ -57,7 +91,7 @@ pub struct StreamChunk { /// cache/reasoning detail riding with `usage_totals` when the vendor sent it. pub common_usage: Option, /// set when the pipeline failed mid-stream; views emit it as an error frame. - pub error: Option, + pub error: Option>, } #[cfg(test)] @@ -78,4 +112,29 @@ mod tests { let j = serde_json::to_value(&r).unwrap(); assert!(j.get("step").is_none()); } + + #[test] + fn committed_generic_vendor_error_is_stream_specific() { + let e = crate::GatewayError::new( + gw_consts::ErrCode::FED_RESP_STATUS_NOT_ZERO, + 502, + "upstream error", + ); + let e = StreamError::from_committed_error(e).unwrap(); + assert_eq!(e.class, gw_consts::ErrClass::ModelStreamError); + assert_eq!(e.original_status, None); + } + + #[test] + fn committed_known_throttle_stays_actionable() { + let e = crate::GatewayError::new( + gw_consts::ErrCode::FED_RESP_STATUS_NOT_ZERO, + 429, + "rate limited", + ) + .with_original_status(429); + let e = StreamError::from_committed_error(e).unwrap(); + assert_eq!(e.class, gw_consts::ErrClass::Throttling); + assert_eq!(e.original_status, Some(429)); + } } diff --git a/crates/protocol/src/anthropic.rs b/crates/protocol/src/anthropic.rs index 5aa893a..433acc3 100644 --- a/crates/protocol/src/anthropic.rs +++ b/crates/protocol/src/anthropic.rs @@ -59,17 +59,22 @@ impl InMessage { pub fn text(&self) -> String { match &self.content { Value::String(s) => s.clone(), - Value::Array(blocks) => blocks - .iter() - .filter(|b| b["type"] == "text" || b.get("type").is_none()) - .filter_map(|b| b["text"].as_str()) - .collect::>() - .join(""), + Value::Array(blocks) => blocks_text(blocks), _ => String::new(), } } } +/// Joined text of the `text`/untyped blocks in an Anthropic content array. +pub fn blocks_text(blocks: &[Value]) -> String { + blocks + .iter() + .filter(|b| b["type"] == "text" || b.get("type").is_none()) + .filter_map(|b| b["text"].as_str()) + .collect::>() + .join("") +} + /// Output content block: text or tool_use. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] diff --git a/crates/protocol/src/openai.rs b/crates/protocol/src/openai.rs index a951e95..8bfe94f 100644 --- a/crates/protocol/src/openai.rs +++ b/crates/protocol/src/openai.rs @@ -22,13 +22,26 @@ impl MessageContent { pub fn text(&self) -> String { match self { MessageContent::Text(s) => s.clone(), - MessageContent::Parts(parts) => parts - .iter() - .filter(|p| p["type"] == "text") - .filter_map(|p| p["text"].as_str()) - .collect(), + MessageContent::Parts(parts) => parts_text(parts), } } + + /// [`text`](Self::text) plus the surrendered raw parts; the Text form + /// moves its string out instead of cloning. + pub fn into_text_and_parts(self) -> (String, Option>) { + match self { + MessageContent::Text(s) => (s, None), + MessageContent::Parts(parts) => (parts_text(&parts), Some(parts)), + } + } +} + +fn parts_text(parts: &[Value]) -> String { + parts + .iter() + .filter(|p| p["type"] == "text") + .filter_map(|p| p["text"].as_str()) + .collect() } impl Default for MessageContent { diff --git a/crates/server/tests/e2e.rs b/crates/server/tests/e2e.rs index 579f6fe..b48a4bb 100644 --- a/crates/server/tests/e2e.rs +++ b/crates/server/tests/e2e.rs @@ -497,16 +497,16 @@ async fn concurrent_requests_cannot_blow_past_quota() { })); } let mut ok = 0; - let mut limited = 0; + let mut exhausted = 0; for h in handles { match h.await.unwrap() { StatusCode::OK => ok += 1, - StatusCode::TOO_MANY_REQUESTS => limited += 1, + StatusCode::BAD_REQUEST => exhausted += 1, other => panic!("unexpected status {other}"), } } assert_eq!( - (ok, limited), + (ok, exhausted), (1, 9), "reservation admits exactly one in-flight request on a quota of 1" ); @@ -524,7 +524,11 @@ async fn failed_request_refunds_its_reservation() { )) .await .unwrap(); - assert_eq!(r.status(), StatusCode::BAD_REQUEST, "vendor error surfaces"); + assert_eq!( + r.status(), + StatusCode::FAILED_DEPENDENCY, + "vendor error surfaces as 424 ModelError" + ); let r = app .oneshot(post( "/v1/chat/completions", @@ -889,7 +893,7 @@ async fn auth_is_enforced() { } #[tokio::test] -async fn model_failure_modes_404_503_501() { +async fn model_failure_modes_404_503_unsupported() { let app = app(); let resp = app .clone() @@ -921,7 +925,12 @@ async fn model_failure_modes_404_503_501() { )) .await .unwrap(); - assert_eq!(resp.status(), StatusCode::NOT_IMPLEMENTED); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + assert_eq!( + body_json(resp).await["error"]["code"], + "validation_exception", + "wrong-surface model must classify as validation" + ); } #[tokio::test] @@ -1436,7 +1445,7 @@ async fn rate_limit_qps1_second_call_429() { } #[tokio::test] -async fn quota_exhaustion_second_call_429() { +async fn quota_exhaustion_second_call_is_non_retryable_400() { let app = app(); let first = app .clone() @@ -1456,9 +1465,10 @@ async fn quota_exhaustion_second_call_429() { )) .await .unwrap(); - assert_eq!(second.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!(second.status(), StatusCode::BAD_REQUEST); let j = body_json(second).await; assert!(j["error"]["message"].as_str().unwrap().contains("quota")); + assert_eq!(j["error"]["code"], "service_quota_exceeded_exception"); } #[tokio::test] @@ -1971,7 +1981,7 @@ async fn dlp_redacts_streaming_output_from_the_vendor() { &self, _req: gw_engines::transport::UpstreamRequest, ) -> gw_models::GResult { - let frames: Vec> = vec![ + let frames: Vec> = vec![ Ok(bytes::Bytes::from( "data: {\"model\":\"gpt-4o\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"reach me at leak@evil.com now\"},\"finish_reason\":null}]}\n\n", )), @@ -2584,6 +2594,13 @@ async fn realtime_websocket_mock_session() { assert_eq!(v["type"], "session.created"); assert_eq!(v["session"]["account"], "mock-realtime-1"); + ws.send(Message::text("not json")).await.unwrap(); + let err = ws.next().await.unwrap().unwrap(); + let v: Value = serde_json::from_str(err.to_text().unwrap()).unwrap(); + assert_eq!(v["type"], "error"); + assert_eq!(v["error"]["code"], "validation_exception"); + assert_eq!(v["error"]["type"], "invalid_request_error"); + ws.send(Message::text( serde_json::json!({"type":"input_text","text":"realtime hello"}).to_string(), )) @@ -3073,7 +3090,13 @@ async fn realtime_turns_are_rate_limited() { let msg = ws.next().await.unwrap().unwrap(); let v: Value = serde_json::from_str(msg.to_text().unwrap()).unwrap(); assert_eq!(v["type"], "error", "second turn must be rate limited: {v}"); - assert!(v["message"].as_str().unwrap().contains("rate limit")); + assert_eq!(v["error"]["code"], "throttling_exception"); + assert!( + v["error"]["message"] + .as_str() + .unwrap() + .contains("rate limit") + ); } #[tokio::test] @@ -3132,7 +3155,13 @@ async fn vendor_error_envelope_propagates_to_client() { )) .await .unwrap(); - assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + assert_eq!(resp.status(), StatusCode::FAILED_DEPENDENCY); + assert_eq!( + resp.headers() + .get("x-amzn-errortype") + .and_then(|v| v.to_str().ok()), + Some("ModelErrorException") + ); let j = body_json(resp).await; assert!( j["error"]["message"] @@ -3140,6 +3169,94 @@ async fn vendor_error_envelope_propagates_to_client() { .unwrap() .contains("mock vendor rejected") ); + assert_eq!(j["error"]["code"], "model_error_exception"); + assert_eq!(j["error"]["type"], "model_error"); + assert_eq!(j["error"]["original_status_code"], 400); + assert_eq!(j["error"]["resource_name"], "erroring-model"); +} + +#[tokio::test] +async fn error_contract_machine_channel() { + let app = app(); + let r = app + .clone() + .oneshot(post("/v1/chat/completions", None, CHAT_BODY)) + .await + .unwrap(); + assert_eq!(r.status(), StatusCode::UNAUTHORIZED); + assert_eq!( + r.headers() + .get("x-amzn-errortype") + .and_then(|v| v.to_str().ok()), + Some("UnrecognizedClientException") + ); + assert_eq!( + r.headers() + .get("access-control-expose-headers") + .and_then(|v| v.to_str().ok()), + Some("x-amzn-errortype, retry-after") + ); + let j = body_json(r).await; + assert_eq!(j["error"]["code"], "unrecognized_client_exception"); + assert_eq!(j["error"]["type"], "authentication_error"); + + let r = app + .clone() + .oneshot(post( + "/v1/chat/completions", + Some("ak-demo-123"), + "{not json", + )) + .await + .unwrap(); + assert_eq!( + r.status(), + StatusCode::BAD_REQUEST, + "rejection stays in the envelope" + ); + let j = body_json(r).await; + assert_eq!(j["error"]["code"], "validation_exception"); + + let r = app.clone().oneshot(get("/v1/nope")).await.unwrap(); + assert_eq!(r.status(), StatusCode::NOT_FOUND); + let j = body_json(r).await; + assert_eq!(j["error"]["code"], "resource_not_found_exception"); + + let r = app + .clone() + .oneshot(get("/v1/chat/completions")) + .await + .unwrap(); + assert_eq!(r.status(), StatusCode::BAD_REQUEST, "405 normalizes to 400"); + let j = body_json(r).await; + assert_eq!(j["error"]["code"], "validation_exception"); + + let r = app + .clone() + .oneshot(post("/v1/messages", None, r#"{"model":"m","messages":[]}"#)) + .await + .unwrap(); + assert_eq!(r.status(), StatusCode::UNAUTHORIZED); + let j = body_json(r).await; + assert_eq!(j["type"], "error"); + assert_eq!(j["error"]["type"], "authentication_error"); + assert_eq!(j["error"]["code"], "unrecognized_client_exception"); + + let oversized = format!( + r#"{{"model":"m-chat","messages":[{{"role":"user","content":"{}"}}]}}"#, + "x".repeat(3 * 1024 * 1024) + ); + let r = app + .oneshot(post( + "/v1/chat/completions", + Some("ak-demo-123"), + &oversized, + )) + .await + .unwrap(); + assert_eq!(r.status(), StatusCode::PAYLOAD_TOO_LARGE); + let j = body_json(r).await; + assert_eq!(j["error"]["code"], "request_entity_too_large_exception"); } #[tokio::test] diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs index 41a6692..5cf0897 100644 --- a/crates/views/src/lib.rs +++ b/crates/views/src/lib.rs @@ -17,6 +17,7 @@ use axum::routing::{get, post}; use axum::{Json, Router}; use base64::Engine as _; use gw_config::GatewayConfig; +use gw_consts::ErrClass; use gw_dag::DagContext; use gw_engines::SharedTransport; use gw_engines::realtime::{is_response_create, realtime_turn_started, realtime_usage}; @@ -148,10 +149,22 @@ pub fn app(state: AppState) -> Router { "/admin/keys/{ak}", axum::routing::patch(admin_key_patch).delete(admin_key_delete), ) + .fallback(unknown_route) + .method_not_allowed_fallback(wrong_method) .layer(axum::middleware::from_fn(track_requests)) .with_state(state) } +/// Route fallback: the envelope's 404 instead of axum's bare one. +async fn unknown_route() -> Response { + error_response(404, "unknown route") +} + +/// Method fallback: the framework 405 normalized to the contract's 400. +async fn wrong_method() -> Response { + error_response(405, "method not allowed for this route") +} + /// Counts every response with bounded labels: route template and status code. async fn track_requests( matched: Option, @@ -172,6 +185,16 @@ async fn track_requests( resp } +/// In-band realtime error event. Never terminal: the session +/// stays open, and only a Close frame or disconnect ends it. +fn rt_error(class: ErrClass, message: impl Into) -> Value { + json!({"type":"error","error":{ + "type": class.openai_type(), + "code": class.code(), + "message": message.into(), + }}) +} + /// The AK carried as `gw-api-key.` in `Sec-WebSocket-Protocol` — the one /// header a browser WebSocket can set; a query param would leak into LB logs. fn ws_subprotocol_ak(headers: &HeaderMap) -> Option { @@ -337,7 +360,10 @@ impl RealtimeAdmit { /// The REST admission chain applied per realtime generation via the shared /// [`admission`] checks, with the key re-fetched each turn so mid-session -/// bans/de-entitlements take effect. Two deliberate divergences from the DAG: +/// bans/de-entitlements take effect. Denials carry the rendering class +/// directly ((ErrClass, message), no ErrCode): the WS surface is deliberately +/// outside the ErrCode-keyed hooks (abuse noting stays a REST-pipeline +/// concern). Two deliberate divergences from the DAG: /// over-quota denies instead of degrading (a session can't swap models /// mid-stream), and the reserve is a fixed turn estimate. Reserves are taken /// last so a denial never leaves one behind; a failed TPM reserve rolls back @@ -347,19 +373,27 @@ async fn realtime_gate( ak: &AkInfo, m: &RtModel, hint: &str, -) -> Result { +) -> Result { let snap = s.handler.config.load(); let (cfg, state) = (&snap.cfg, &snap.state); let ak = match state.auth.authenticate(&ak.ak).await { Some(fresh) if fresh.status_at(gw_state::epoch_secs()) == gw_state::KeyStatus::Active => { fresh } - _ => return Err(format!("access key {} is no longer valid", ak.ak)), + _ => { + return Err(( + ErrClass::AccessDenied, + format!("access key {} is no longer valid", ak.ak), + )); + } }; if !cfg.tenant_allows_model(&ak.tenant, &m.requested) { - return Err(format!( - "model `{}` is not entitled for tenant `{}`", - m.requested, ak.tenant + return Err(( + ErrClass::AccessDenied, + format!( + "model `{}` is not entitled for tenant `{}`", + m.requested, ak.tenant + ), )); } // the session pinned `served` at handshake; a reload may have removed it, @@ -367,31 +401,48 @@ async fn realtime_gate( // reconnects and picks against the live config. Wire-direct sessions // (`?model=realtime`, never in config) are exempt: nothing to vanish. if m.from_config && cfg.find_model(&m.served).is_none() { - return Err(format!( - "model `{}` is no longer configured; reconnect", - m.served + return Err(( + ErrClass::ResourceNotFound, + format!("model `{}` is no longer configured; reconnect", m.served), )); } let gov = state.governance.as_ref(); - admission::check_tenant_rate(gov, cfg, &ak.tenant).await?; - admission::check_ak_rate(gov, &ak).await?; - admission::check_product_qpm(gov, cfg, &ak.product).await?; - admission::check_model_qpm(gov, cfg, &m.served).await?; - admission::check_user_budget(gov, cfg, &ak.tenant, ak.attributed_user(hint)).await?; + let throttled = |m: String| (ErrClass::Throttling, m); + let quota_exceeded = |m: String| (ErrClass::ServiceQuotaExceeded, m); + admission::check_tenant_rate(gov, cfg, &ak.tenant) + .await + .map_err(throttled)?; + admission::check_ak_rate(gov, &ak) + .await + .map_err(throttled)?; + admission::check_product_qpm(gov, cfg, &ak.product) + .await + .map_err(throttled)?; + admission::check_model_qpm(gov, cfg, &m.served) + .await + .map_err(throttled)?; + admission::check_user_budget(gov, cfg, &ak.tenant, ak.attributed_user(hint)) + .await + .map_err(quota_exceeded)?; if let Some(limit) = admission::model_quota_limit(cfg, &ak, &m.requested) && !gov .quota_check(&admission::model_quota_key(&ak.ak, &m.requested), limit) .await { - return Err(format!("model quota exhausted for `{}`", m.requested)); + return Err(( + ErrClass::ServiceQuotaExceeded, + format!("model quota exhausted for `{}`", m.requested), + )); } let at = gw_state::epoch_secs(); - admission::reserve_daily(gov, &ak, REALTIME_TURN_RESERVE, at).await?; + admission::reserve_daily(gov, &ak, REALTIME_TURN_RESERVE, at) + .await + .map_err(quota_exceeded)?; let tpm_reserved = match admission::reserve_tpm(gov, &ak, REALTIME_TURN_RESERVE).await { Ok(reserved) => reserved, Err(denied) => { gov.quota_settle(&ak.ak, -REALTIME_TURN_RESERVE, at).await; - return Err(denied); + return Err((ErrClass::Throttling, denied)); } }; let user = ak.attributed_user(hint).to_owned(); @@ -500,13 +551,13 @@ async fn realtime_session( }; let Ok(mut ev) = serde_json::from_str::(&text) else { let _ = socket - .send(send(json!({"type":"error","message":"invalid json event"}))) + .send(send(rt_error(ErrClass::Validation, "invalid json event"))) .await; continue; }; if let Err(reason) = rt_inbound_policy(&s, &ak, &hint, &mut ev).await { let _ = socket - .send(send(json!({"type":"error","message": reason}))) + .send(send(rt_error(ErrClass::AccessDenied, reason))) .await; continue; } @@ -514,10 +565,8 @@ async fn realtime_session( "input_text" => { let admit = match realtime_gate(&s, &ak, &rtm, &hint).await { Ok(a) => a, - Err(denied) => { - let _ = socket - .send(send(json!({"type":"error","message": denied}))) - .await; + Err((class, denied)) => { + let _ = socket.send(send(rt_error(class, denied))).await; continue; } }; @@ -550,8 +599,10 @@ async fn realtime_session( } other => { let _ = socket - .send(send(json!({"type":"error", - "message": format!("unsupported event type `{other}`")}))) + .send(send(rt_error( + ErrClass::Validation, + format!("unsupported event type `{other}`"), + ))) .await; } } @@ -603,8 +654,6 @@ async fn realtime_bridge( use tokio_tungstenite::tungstenite::Message as UMsg; use tokio_tungstenite::tungstenite::client::IntoClientRequest; - let send_err = |m: String| json!({"type":"error","error":{"type":"gateway_error","message":m}}); - let base = account.endpoint.trim_end_matches('/'); let ws_base = base .replacen("https://", "wss://", 1) @@ -616,7 +665,7 @@ async fn realtime_bridge( s.handler.state().avail.record(&rtm.requested, false); let _ = client .send(CMsg::Text( - send_err(format!("bad upstream url: {e}")) + rt_error(ErrClass::InternalServer, format!("bad upstream url: {e}")) .to_string() .into(), )) @@ -634,9 +683,12 @@ async fn realtime_bridge( s.handler.state().avail.record(&rtm.requested, false); let _ = client .send(CMsg::Text( - send_err(format!("upstream realtime connect failed: {e}")) - .to_string() - .into(), + rt_error( + ErrClass::ModelError, + format!("upstream realtime connect failed: {e}"), + ) + .to_string() + .into(), )) .await; return; @@ -671,7 +723,7 @@ async fn realtime_bridge( match rt_inbound_policy(&s, &ak, &hint, &mut frame).await { Err(reason) => { if cl_tx - .send(CMsg::Text(send_err(reason).to_string().into())) + .send(CMsg::Text(rt_error(ErrClass::AccessDenied, reason).to_string().into())) .await .is_err() { @@ -695,9 +747,9 @@ async fn realtime_bridge( pending = Some(admit); generations += 1; } - Err(denied) => { + Err((class, denied)) => { if cl_tx - .send(CMsg::Text(send_err(denied).to_string().into())) + .send(CMsg::Text(rt_error(class, denied).to_string().into())) .await .is_err() { @@ -747,12 +799,12 @@ async fn realtime_bridge( else if realtime_turn_started(&account.provider, &v) && pending.is_none() { match realtime_gate(&s, &ak, &rtm, &hint).await { Ok(admit) => pending = Some(admit), - Err(denied) => { + Err((class, denied)) => { let _ = up_tx .send(UMsg::text(json!({"type":"response.cancel"}).to_string())) .await; let _ = cl_tx - .send(CMsg::Text(send_err(denied).to_string().into())) + .send(CMsg::Text(rt_error(class, denied).to_string().into())) .await; suppress = true; relay = false; @@ -1032,15 +1084,113 @@ impl axum::extract::FromRequestParts for Authed { } } -fn error_response(status: u16, message: impl Into) -> Response { - let code = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); - ( - code, - Json(json!({ "error": { "message": message.into(), "type": "gateway_error" } })), - ) +/// The shared body behind [`ApiJson`]/[`AnthJson`]: delegate to `axum::Json` +/// and render the rejection through the surface's own envelope, so a +/// malformed body (400/413/415/422) cannot escape the contract. +async fn extract_json( + req: axum::extract::Request, + state: &S, + render: fn(u16, String) -> Response, +) -> Result +where + axum::Json: + axum::extract::FromRequest, + S: Send + Sync, +{ + match as axum::extract::FromRequest>::from_request(req, state).await { + Ok(axum::Json(v)) => Ok(v), + Err(rej) => Err(render(rej.status().as_u16(), rej.body_text())), + } +} + +/// `axum::Json` with rejections in the OpenAI envelope. +struct ApiJson(T); + +impl axum::extract::FromRequest for ApiJson +where + axum::Json: + axum::extract::FromRequest, + S: Send + Sync, +{ + type Rejection = Response; + + async fn from_request(req: axum::extract::Request, state: &S) -> Result { + extract_json(req, state, error_response).await.map(ApiJson) + } +} + +/// [`ApiJson`] in the Anthropic envelope, for `/v1/messages`. +struct AnthJson(T); + +impl axum::extract::FromRequest for AnthJson +where + axum::Json: + axum::extract::FromRequest, + S: Send + Sync, +{ + type Rejection = Response; + + async fn from_request(req: axum::extract::Request, state: &S) -> Result { + extract_json(req, state, anthropic_error) + .await + .map(AnthJson) + } +} + +/// The contract's machine channel on every HTTP error response: the +/// classification header plus the CORS exposure browser clients need. +fn with_error_headers(class: ErrClass, mut resp: Response) -> Response { + let headers = resp.headers_mut(); + headers.insert( + "x-amzn-errortype", + axum::http::HeaderValue::from_static(class.name()), + ); + headers.insert( + "access-control-expose-headers", + axum::http::HeaderValue::from_static("x-amzn-errortype, retry-after"), + ); + resp +} + +/// OpenAI-shaped error body: coarse `type`, precise `code`. +fn openai_error_body(class: ErrClass, message: String) -> Value { + json!({ "error": { + "message": message, + "type": class.openai_type(), + "param": null, + "code": class.code(), + }}) +} + +/// A body-less status passthrough: the 499 client-closed case, where the +/// peer is gone and nothing is rendered. +fn bare_status(status: u16) -> Response { + StatusCode::from_u16(status) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) .into_response() } +/// Status + machine headers + JSON body: the one seam every envelope +/// renders through. +fn class_response(class: ErrClass, body: Value) -> Response { + let status = StatusCode::from_u16(class.status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); + with_error_headers(class, (status, Json(body)).into_response()) +} + +/// OpenAI-envelope error at the class's external status. +fn class_error(class: ErrClass, message: impl Into) -> Response { + class_response(class, openai_error_body(class, message.into())) +} + +/// OpenAI-envelope error from an ad-hoc (status, message) site; the external +/// status is the classification's, not the caller's literal. +fn error_response(status: u16, message: impl Into) -> Response { + match ErrClass::from_status(status) { + Some(class) => class_error(class, message), + None => bare_status(status), + } +} + /// Who an admin bearer token speaks for: the global operator or one tenant. enum AdminScope { Global, @@ -1415,7 +1565,7 @@ async fn admin_key_create( State(s): State, scope: AdminScope, AuditSourceIp(source): AuditSourceIp, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { let (Some(ak), Some(product)) = (body["ak"].as_str(), body["product"].as_str()) else { return error_response(400, "ak and product are required"); @@ -1496,7 +1646,7 @@ async fn admin_key_patch( scope: AdminScope, AuditSourceIp(source): AuditSourceIp, Path(ak): Path, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { if let Err(r) = scoped_key(&s, &scope, &ak).await { return r; @@ -2088,42 +2238,58 @@ async fn admin_content_erase( } } +/// OpenAI-envelope rendering of an internal error: classified, never a +/// status passthrough — a terminal vendor 401 lands as 424 ModelError, not +/// as the client's own 401. fn gateway_error(e: GatewayError) -> Response { - let code = StatusCode::from_u16(e.http_status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); - // OpenAI's error schema types `code` as string-or-null, never a number. - ( - code, - Json(json!({ "error": { "message": e.message, "code": e.code.value().to_string(), "type": "gateway_error" } })), - ) - .into_response() + let Some(class) = ErrClass::classify(e.code, e.http_status) else { + return bare_status(e.http_status); + }; + let original_status = e.original_status(); + let resource = e.resource; + let mut body = openai_error_body(class, e.message); + if class == ErrClass::ModelError { + if let Some(os) = original_status { + body["error"]["original_status_code"] = os.into(); + } + if let Some(r) = resource { + body["error"]["resource_name"] = r.into(); + } + } + class_response(class, body) } -/// Anthropic's error type string for an HTTP status. -fn anthropic_error_type(status: u16) -> &'static str { - match status { - 400 => "invalid_request_error", - 401 => "authentication_error", - 403 => "permission_error", - 404 => "not_found_error", - 413 => "request_too_large", - 429 => "rate_limit_error", - 529 => "overloaded_error", - _ => "api_error", - } +/// Anthropic-shaped error body — `error.type` is the discriminator the +/// Anthropic SDKs key their exception dispatch on; `code` is additive. +fn anthropic_error_body(class: ErrClass, message: String) -> Value { + json!({ + "type": "error", + "error": { + "type": class.anthropic_type(), + "code": class.code(), + "message": message, + }, + }) +} + +fn anthropic_error_with(class: ErrClass, message: impl Into) -> Response { + class_response(class, anthropic_error_body(class, message.into())) } -/// Anthropic-shaped error body: `{"type":"error","error":{"type","message"}}` — -/// the discriminator the Anthropic SDKs key their exception dispatch on. +/// Anthropic-shaped error from an ad-hoc (status, message) site. fn anthropic_error(status: u16, message: impl Into) -> Response { - let code = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); - ( - code, - Json(json!({ - "type": "error", - "error": { "type": anthropic_error_type(status), "message": message.into() }, - })), - ) - .into_response() + match ErrClass::from_status(status) { + Some(class) => anthropic_error_with(class, message), + None => bare_status(status), + } +} + +/// [`gateway_error`]'s classification, rendered in the Anthropic shape. +fn anthropic_gateway_error(e: GatewayError) -> Response { + match ErrClass::classify(e.code, e.http_status) { + Some(class) => anthropic_error_with(class, e.message), + None => bare_status(e.http_status), + } } /// Run the pipeline on its own task so a client disconnect can't cancel it @@ -2236,7 +2402,7 @@ async fn chat_completions( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { let started = Instant::now(); if body.messages.is_empty() { @@ -2247,14 +2413,14 @@ async fn chat_completions( .messages .into_iter() .map(|m| { - let content = m.content_text(); + let (content, parts) = m + .content + .map(|c| c.into_text_and_parts()) + .unwrap_or_default(); ChatMsg { role: m.role, content, - parts: m.content.and_then(|c| match c { - gw_protocol::openai::MessageContent::Parts(p) => Some(Value::Array(p)), - _ => None, - }), + parts: parts.map(Value::Array), tool_calls: m.tool_calls.and_then(|tc| serde_json::to_value(tc).ok()), tool_call_id: m.tool_call_id, } @@ -2384,12 +2550,16 @@ fn spawn_stream_pipeline( } } Err(e) => { - let _ = tx - .send(gw_engines::StreamChunk { - error: Some(e.to_string()), - ..Default::default() - }) - .await; + // 499 client-closed classifies to None: the peer is gone and + // no frame is rendered. + if let Some(error) = gw_models::StreamError::from_error(e) { + let _ = tx + .send(gw_engines::StreamChunk { + error: Some(Box::new(error)), + ..Default::default() + }) + .await; + } } } }); @@ -2425,6 +2595,16 @@ fn sse_stream( Sse::new(stream) } +/// Chat/legacy SSE error frame: the OpenAI envelope plus the upstream's +/// original status when one was received. +fn stream_error_body(err: gw_models::StreamError) -> Value { + let mut body = openai_error_body(err.class, err.message); + if let Some(os) = err.original_status { + body["error"]["original_status_code"] = os.into(); + } + body +} + /// Streaming chat: pipeline chunks re-emitted as OpenAI SSE. fn chat_stream_response( s: AppState, @@ -2449,11 +2629,10 @@ fn chat_stream_response( fn apply(&mut self, chunk: Option) -> bool { match chunk { Some(gw_engines::StreamChunk { - error: Some(msg), .. + error: Some(err), .. }) => { - self.queue.push_back(Event::default().data( - json!({"error": {"message": msg, "type": "gateway_error"}}).to_string(), - )); + self.queue + .push_back(Event::default().data(stream_error_body(*err).to_string())); self.queue.push_back(Event::default().data("[DONE]")); true } @@ -2584,7 +2763,7 @@ fn redacted_stream_tail(outcome: gw_engines::EngineOutcome) -> Vec, headers: HeaderMap, - Json(body): Json, + AnthJson(body): AnthJson, ) -> Response { let started = Instant::now(); let ak = match authenticate(&s, &headers).await { @@ -2621,13 +2800,15 @@ async fn messages( message: body .messages .into_iter() - .map(|m| { - let text = m.text(); - let mut msg = ChatMsg::text(m.role, text); - if m.content.is_array() { - msg.parts = Some(m.content); + .map(|m| match m.content { + Value::String(s) => ChatMsg::text(m.role, s), + Value::Array(blocks) => { + let text = gw_protocol::anthropic::blocks_text(&blocks); + let mut msg = ChatMsg::text(m.role, text); + msg.parts = Some(Value::Array(blocks)); + msg } - msg + _ => ChatMsg::text(m.role, String::new()), }) .collect(), model_param_v2: Some(param), @@ -2641,7 +2822,7 @@ async fn messages( let ctx = match run_pipeline(&s, request, ak).await { Ok(ctx) => ctx, - Err(e) => return anthropic_error(e.http_status, e.message), + Err(e) => return anthropic_gateway_error(e), }; log_access("messages", &ctx, started); let Some(mut outcome) = ctx.outcome else { @@ -2816,11 +2997,12 @@ fn messages_stream_response( fn apply(&mut self, chunk: Option) -> bool { match chunk { Some(gw_engines::StreamChunk { - error: Some(msg), .. + error: Some(err), .. }) => { + let err = *err; self.queue.push_back(St::ev( "error", - json!({"type":"error","error":{"type":"api_error","message":msg}}), + anthropic_error_body(err.class, err.message), )); true } @@ -2944,7 +3126,7 @@ async fn completions( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(mut body): Json, + ApiJson(mut body): ApiJson, ) -> Response { let started = Instant::now(); let model = body["model"].as_str().unwrap_or_default().to_owned(); @@ -3008,7 +3190,7 @@ async fn responses( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { let started = Instant::now(); let model = body["model"].as_str().unwrap_or_default().to_owned(); @@ -3060,6 +3242,7 @@ fn responses_stream_response( model: String, created: bool, status: String, + seq: u64, } impl St { fn ensure_created(&mut self) { @@ -3067,6 +3250,7 @@ fn responses_stream_response( return; } self.created = true; + self.seq += 1; self.queue.push_back(Event::default().data( json!({"type":"response.created","response":{"model":self.model,"status":"in_progress"}}) .to_string(), @@ -3080,13 +3264,22 @@ fn responses_stream_response( fn apply(&mut self, chunk: Option) -> bool { match chunk { Some(gw_engines::StreamChunk { - error: Some(msg), .. + error: Some(err), .. }) => { + let err = *err; self.ensure_created(); + // the official Responses error event is flat: no nested + // `error` object, sequence_number continues the stream self.queue.push_back( Event::default().data( - json!({"type":"error","error":{"type":"gateway_error","message":msg}}) - .to_string(), + json!({ + "type": "error", + "code": err.class.code(), + "message": err.message, + "param": null, + "sequence_number": self.seq, + }) + .to_string(), ), ); self.queue.push_back(Event::default().data("[DONE]")); @@ -3095,6 +3288,7 @@ fn responses_stream_response( Some(c) => { self.ensure_created(); if !c.delta.is_empty() { + self.seq += 1; self.queue.push_back( Event::default().data( json!({"type":"response.output_text.delta","delta":c.delta}) @@ -3108,6 +3302,7 @@ fn responses_stream_response( let Some((pt, ct, tt)) = c.usage_totals else { return false; }; + self.seq += 1; self.queue.push_back( Event::default().data( json!({"type":"response.completed","response":{ @@ -3134,6 +3329,7 @@ fn responses_stream_response( queue: VecDeque::new(), model, created: false, + seq: 0, status: "completed".to_owned(), }, ) @@ -3144,7 +3340,7 @@ async fn embeddings( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(mut body): Json, + ApiJson(mut body): ApiJson, ) -> Response { let started = Instant::now(); let model = body["model"].as_str().unwrap_or_default().to_owned(); @@ -3179,7 +3375,7 @@ async fn images_generations( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { let started = Instant::now(); let model = body["model"].as_str().unwrap_or_default().to_owned(); @@ -3217,7 +3413,7 @@ async fn images_edits( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { let started = Instant::now(); let model = body["model"].as_str().unwrap_or_default().to_owned(); @@ -3256,7 +3452,7 @@ async fn audio_speech( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { let started = Instant::now(); let model = body["model"].as_str().unwrap_or_default().to_owned(); @@ -3314,7 +3510,7 @@ async fn audio_transcriptions( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { audio_transcribe(s, headers, ak, body, false).await } @@ -3325,7 +3521,7 @@ async fn audio_translations( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { audio_transcribe(s, headers, ak, body, true).await } @@ -3383,7 +3579,7 @@ async fn moderations( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(mut body): Json, + ApiJson(mut body): ApiJson, ) -> Response { let started = Instant::now(); let model = body["model"].as_str().unwrap_or_default().to_owned(); @@ -3415,7 +3611,7 @@ async fn rerank( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(mut body): Json, + ApiJson(mut body): ApiJson, ) -> Response { let started = Instant::now(); let model = body["model"].as_str().unwrap_or_default().to_owned(); @@ -3477,7 +3673,7 @@ async fn batches_submit( State(s): State, headers: HeaderMap, Authed(ak): Authed, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { let mut model = body["model"].as_str().unwrap_or_default().to_owned(); let mut batch_items = Vec::new(); @@ -3550,7 +3746,7 @@ async fn batches_submit( async fn files_upload( State(s): State, Authed(ak): Authed, - Json(body): Json, + ApiJson(body): ApiJson, ) -> Response { let purpose = body["purpose"].as_str().unwrap_or("batch").to_owned(); let Some(content) = body["file"].as_str() else { @@ -4542,6 +4738,13 @@ mod tests { ledger_before, ledger_after, "zero-usage turn writes no ledger row" ); + + gov().quota_consume(&ak.ak, ak.daily_token_quota).await; + let denied = realtime_gate(&s, &ak, &rt("gpt-4o"), "") + .await + .err() + .expect("exhausted daily quota must deny"); + assert_eq!(denied.0, ErrClass::ServiceQuotaExceeded); } #[tokio::test] @@ -4604,8 +4807,9 @@ mod tests { let denied = realtime_gate(&s, &ak, &pinned, "").await.err(); assert!( denied - .as_deref() - .is_some_and(|e| e.contains("no longer configured")), + .as_ref() + .is_some_and(|(class, e)| *class == ErrClass::ResourceNotFound + && e.contains("no longer configured")), "a pinned variant removed by reload must deny the turn, not bill zero: {denied:?}" ); } diff --git a/docs/api.md b/docs/api.md index 59c0b25..e7f7be7 100644 --- a/docs/api.md +++ b/docs/api.md @@ -7,21 +7,31 @@ Authorization: Bearer x-api-key: ``` -A missing or unknown key is `401`. Errors use a consistent envelope; the -numeric `code` appears only on pipeline-originated errors (routing, quota, -upstream) — auth and validation errors carry just `message` and `type`: +A missing or unknown key is `401`. Every error carries one classification +from a closed, Bedrock-derived set, on two machine channels: the +`x-amzn-errortype` response header (PascalCase name, e.g. +`ThrottlingException`) and the body's `code` field (snake_case, e.g. +`throttling_exception`). The envelope shape follows the surface — OpenAI +surfaces use the OpenAI error object: ```json -{"error": {"message": "...", "code": "3002", "type": "gateway_error"}} +{"error": {"message": "...", "type": "rate_limit_error", "param": null, "code": "throttling_exception"}} ``` -The Anthropic-compatible surface (`/v1/messages`) instead emits Anthropic's -error shape, so its SDKs can dispatch on it: +The Anthropic-compatible surface (`/v1/messages`) emits Anthropic's error +shape, so its SDKs can dispatch on it (`code` is additive): ```json -{"type": "error", "error": {"type": "invalid_request_error", "message": "..."}} +{"type": "error", "error": {"type": "rate_limit_error", "code": "throttling_exception", "message": "..."}} ``` +A terminal upstream failure (failover exhausted) is `424` with +`code: "model_error_exception"`, plus `original_status_code` (when the +upstream returned a status) and `resource_name` (the requested model) inside +the error object. Retry on 408/429/500/503 with backoff (honor +`retry-after`); never on the rest. Mid-stream failures arrive as a terminal +SSE error frame carrying the same `code` field. + For per-user attribution on a shared key, send `x-gw-user: ` (it also reads OpenAI's body `user` field and Anthropic's `metadata.user_id`). A key's own `owner` overrides the hint, so a key issued to one user always bills to that