From 5444c1739fe971ff52e206d694713cf904b380a9 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 28 Jul 2026 19:23:20 +0800 Subject: [PATCH 1/6] error contract core: ErrClass taxonomy, structured stream errors, deadline split ErrClass (consts) is the Bedrock-derived closed classification set from the gateway error contract; ErrCode stays internal. StreamChunk.error becomes a structured StreamError (class + message + original upstream status). The pump now delivers a terminal error frame on committed mid-stream aborts instead of ending silently; client disconnects stay silent. Upstream deadlines split off as FED_RESP_TIMEOUT (internal 502 keeps failover, renders as 408). vendor_error accepts the AWS auth layer's capitalized Message field. --- crates/consts/src/error_class.rs | 248 ++++++++++++++++++++ crates/consts/src/error_code.rs | 3 + crates/consts/src/lib.rs | 2 + crates/engines/src/engine.rs | 6 + crates/engines/src/http_transport.rs | 30 ++- crates/engines/src/pump.rs | 75 ++++-- crates/engines/src/transport.rs | 50 +++- crates/engines/tests/http_transport_wire.rs | 27 ++- crates/handler/src/lib.rs | 18 +- crates/models/src/lib.rs | 2 +- crates/models/src/response.rs | 29 ++- crates/server/tests/e2e.rs | 2 +- crates/views/src/lib.rs | 35 +-- 13 files changed, 457 insertions(+), 70 deletions(-) create mode 100644 crates/consts/src/error_class.rs diff --git a/crates/consts/src/error_class.rs b/crates/consts/src/error_class.rs new file mode 100644 index 0000000..85798c8 --- /dev/null +++ b/crates/consts/src/error_class.rs @@ -0,0 +1,248 @@ +//! 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, + 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, + // in-stream only; the HTTP status never reaches a client + ErrClass::ModelError | ErrClass::ModelStreamError => 424, + ErrClass::Throttling => 429, + ErrClass::InternalServer => 500, + ErrClass::ServiceUnavailable => 503, + } + } + + /// Coarse `type` for the OpenAI envelope (contract §4.2). + 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 (contract §4.3). + 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 + | ErrCode::EMPTY_RESP => match status { + 408 => ErrClass::ModelTimeout, + 429 => ErrClass::Throttling, + _ => ErrClass::ModelError, + }, + ErrCode::STOP_LIMIT_MSG => ErrClass::Throttling, + 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 keeps 429 semantics + let c = ErrClass::classify(ErrCode::FED_RESP_STATUS_NOT_ZERO, 429).unwrap(); + assert_eq!(c, ErrClass::Throttling); + // deadline family + let c = ErrClass::classify(ErrCode::FED_RESP_TIMEOUT, 502).unwrap(); + assert_eq!((c, c.status()), (ErrClass::ModelTimeout, 408)); + } + + #[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..f79df13 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); 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/engines/src/engine.rs b/crates/engines/src/engine.rs index 179f2ef..1b875e2 100644 --- a/crates/engines/src/engine.rs +++ b/crates/engines/src/engine.rs @@ -91,9 +91,12 @@ 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, } @@ -146,6 +149,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/http_transport.rs b/crates/engines/src/http_transport.rs index 4dd21b3..4bd3bcd 100644 --- a/crates/engines/src/http_transport.rs +++ b/crates/engines/src/http_transport.rs @@ -11,7 +11,9 @@ 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, +}; const RETRY_BACKOFF: Duration = Duration::from_millis(100); // A hung connect (black-holed SYN) must surface as a connect error — which the @@ -121,7 +123,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 {:?}", @@ -145,8 +147,13 @@ impl Transport for HttpTransport { tokio::time::sleep(RETRY_BACKOFF * attempt).await; } Err(e) => { + let code = if e.is_timeout() { + gw_consts::ErrCode::FED_RESP_TIMEOUT + } else { + gw_consts::ErrCode::FED_RESP_RPC_FAILED + }; return Err(GatewayError::new( - gw_consts::ErrCode::FED_RESP_RPC_FAILED, + code, 502, format!("upstream request failed: {e}"), )); @@ -162,7 +169,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 +184,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), ) @@ -195,7 +205,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 +216,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..c48b054 100644 --- a/crates/engines/src/pump.rs +++ b/crates/engines/src/pump.rs @@ -1,7 +1,7 @@ //! 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; @@ -41,20 +41,25 @@ 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. A buffered +/// run keeps the frame in `chunks` so the replay tail carries it. +async fn abort_frame( + tx: &Option>, + out: &mut PumpResult, + error: StreamError, +) { + let chunk = StreamChunk { + error: Some(error), + ..Default::default() + }; + match tx { + Some(sender) => { + let _ = sender.send(chunk).await; + } + None => out.chunks.push(chunk), } + out.aborted = true; } pub async fn pump_sse( @@ -88,24 +93,38 @@ 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; + Err(e) if sent_any => { + tracing::warn!(vendor, error = %e, "upstream stream failed mid-response"); + let error = StreamError { + class: gw_consts::ErrClass::ModelStreamError, + message: format!("upstream stream failed: {e}"), + original_status: None, + }; + abort_frame(&tx, &mut out, error).await; break; } + Err(e) => { + return Err(GatewayError::new( + gw_consts::ErrCode::FED_RESP_RPC_FAILED, + 502, + format!("upstream stream failed: {e}"), + )); + } }; for data in events { let v: Value = serde_json::from_str(&data).map_err(|e| { @@ -119,7 +138,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_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..6f447bc 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,50 @@ 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 { + let code = if self.timeout { + gw_consts::ErrCode::FED_RESP_TIMEOUT + } else { + gw_consts::ErrCode::FED_RESP_RPC_FAILED + }; + GatewayError::new( + code, + 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, + } + } +} + /// 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 +100,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..f13a630 100644 --- a/crates/engines/tests/http_transport_wire.rs +++ b/crates/engines/tests/http_transport_wire.rs @@ -12,7 +12,7 @@ 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}; @@ -377,11 +377,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 +394,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 +416,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 +436,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", )), diff --git a/crates/handler/src/lib.rs b/crates/handler/src/lib.rs index d10cf88..60618a8 100644 --- a/crates/handler/src/lib.rs +++ b/crates/handler/src/lib.rs @@ -1092,11 +1092,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 +1156,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 +1216,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 +1274,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/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..a2e9511 100644 --- a/crates/models/src/response.rs +++ b/crates/models/src/response.rs @@ -45,6 +45,33 @@ 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)?; + // FED_RESP_STATUS_NOT_ZERO is the one code whose status is the real + // upstream reply; the rest never received one. + let original_status = + (e.code == gw_consts::ErrCode::FED_RESP_STATUS_NOT_ZERO).then_some(e.http_status); + Some(Self { + class, + message: e.message.clone(), + original_status, + }) + } +} + /// One streamed response fragment, forwarded to the client as it arrives. #[derive(Debug, Default, Clone)] pub struct StreamChunk { @@ -57,7 +84,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)] diff --git a/crates/server/tests/e2e.rs b/crates/server/tests/e2e.rs index 579f6fe..23c19e7 100644 --- a/crates/server/tests/e2e.rs +++ b/crates/server/tests/e2e.rs @@ -1971,7 +1971,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", )), diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs index 41a6692..435113b 100644 --- a/crates/views/src/lib.rs +++ b/crates/views/src/lib.rs @@ -2384,12 +2384,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(error), + ..Default::default() + }) + .await; + } } } }); @@ -2449,11 +2453,14 @@ 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( + json!({"error": {"message": err.message, "type": "gateway_error"}}) + .to_string(), + ), + ); self.queue.push_back(Event::default().data("[DONE]")); true } @@ -2816,11 +2823,11 @@ fn messages_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(St::ev( "error", - json!({"type":"error","error":{"type":"api_error","message":msg}}), + json!({"type":"error","error":{"type":"api_error","message":err.message}}), )); true } @@ -3080,12 +3087,12 @@ fn responses_stream_response( fn apply(&mut self, chunk: Option) -> bool { match chunk { Some(gw_engines::StreamChunk { - error: Some(msg), .. + error: Some(err), .. }) => { self.ensure_created(); self.queue.push_back( Event::default().data( - json!({"type":"error","error":{"type":"gateway_error","message":msg}}) + json!({"type":"error","error":{"type":"gateway_error","message":err.message}}) .to_string(), ), ); From 1f7e126a02536712b3d9870a7c53e43cac9fe6c5 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 28 Jul 2026 19:37:23 +0800 Subject: [PATCH 2/6] views: render every client error through the contract classification One classify-then-render path replaces the three ad-hoc serializers: OpenAI and Anthropic envelopes both carry the classification code, every HTTP error carries x-amzn-errortype plus the CORS exposure, and external statuses come from the classification (terminal vendor errors land as 424 ModelError with original_status_code/resource_name instead of leaking the upstream status). Json extractor rejections, unknown routes, and wrong methods render in the envelope instead of axum's bare responses. Realtime WS uses one nested in-band error shape with real classes from the gate; SSE error frames carry class-driven types (Responses uses the official flat error event with sequence_number). Terminal upstream 429/503 keep Throttling/ ServiceUnavailable semantics; the wrong-surface factory error is a 400 validation error, not a 501 internal. --- crates/consts/src/error_class.rs | 7 +- crates/dag/src/nodes.rs | 15 +- crates/engines/src/factory.rs | 8 +- crates/engines/src/pump.rs | 20 +- crates/models/src/error.rs | 4 + crates/server/tests/e2e.rs | 97 +++++++- crates/views/src/lib.rs | 377 +++++++++++++++++++++++-------- 7 files changed, 407 insertions(+), 121 deletions(-) diff --git a/crates/consts/src/error_class.rs b/crates/consts/src/error_class.rs index 85798c8..c369ec2 100644 --- a/crates/consts/src/error_class.rs +++ b/crates/consts/src/error_class.rs @@ -134,9 +134,12 @@ impl ErrClass { | 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, @@ -212,9 +215,11 @@ mod tests { 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 keeps 429 semantics + // 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)); diff --git a/crates/dag/src/nodes.rs b/crates/dag/src/nodes.rs index 7c9d500..c48b708 100644 --- a/crates/dag/src/nodes.rs +++ b/crates/dag/src/nodes.rs @@ -534,7 +534,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 +564,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/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/pump.rs b/crates/engines/src/pump.rs index c48b054..1013d99 100644 --- a/crates/engines/src/pump.rs +++ b/crates/engines/src/pump.rs @@ -42,22 +42,20 @@ pub(crate) fn reject_json_error(what: &str, status: u16, body: &UpstreamBody) -> } /// Deliver the terminal error frame for a committed abort (best effort — the -/// client may already be gone) and mark the pump result aborted. A buffered -/// run keeps the frame in `chunks` so the replay tail carries it. +/// 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, ) { - let chunk = StreamChunk { - error: Some(error), - ..Default::default() - }; - match tx { - Some(sender) => { - let _ = sender.send(chunk).await; - } - None => out.chunks.push(chunk), + if let Some(sender) = tx { + let _ = sender + .send(StreamChunk { + error: Some(error), + ..Default::default() + }) + .await; } out.aborted = true; } diff --git a/crates/models/src/error.rs b/crates/models/src/error.rs index 2dfd567..e0a8037 100644 --- a/crates/models/src/error.rs +++ b/crates/models/src/error.rs @@ -17,6 +17,9 @@ 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, pub source: Option>, } @@ -26,6 +29,7 @@ impl GatewayError { code, http_status, message: message.into(), + resource: None, source: None, } } diff --git a/crates/server/tests/e2e.rs b/crates/server/tests/e2e.rs index 23c19e7..0ce626b 100644 --- a/crates/server/tests/e2e.rs +++ b/crates/server/tests/e2e.rs @@ -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,7 @@ async fn model_failure_modes_404_503_501() { )) .await .unwrap(); - assert_eq!(resp.status(), StatusCode::NOT_IMPLEMENTED); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); } #[tokio::test] @@ -3073,7 +3077,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 +3142,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 +3156,77 @@ 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 + .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"); } #[tokio::test] diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs index 435113b..be6e4ce 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 (spec 4.4). 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 { @@ -347,19 +370,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 +398,47 @@ 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); + 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(throttled)?; 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::Throttling, + 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(throttled)?; 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 +547,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 +561,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 +595,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,7 +650,7 @@ 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 send_err = rt_error; let base = account.endpoint.trim_end_matches('/'); let ws_base = base @@ -616,7 +663,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}")) + send_err(ErrClass::InternalServer, format!("bad upstream url: {e}")) .to_string() .into(), )) @@ -634,9 +681,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(), + send_err( + ErrClass::ModelError, + format!("upstream realtime connect failed: {e}"), + ) + .to_string() + .into(), )) .await; return; @@ -671,7 +721,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(send_err(ErrClass::AccessDenied, reason).to_string().into())) .await .is_err() { @@ -695,9 +745,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(send_err(class, denied).to_string().into())) .await .is_err() { @@ -747,12 +797,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(send_err(class, denied).to_string().into())) .await; suppress = true; relay = false; @@ -1032,13 +1082,89 @@ 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" } })), +/// `axum::Json` with rejections rendered in the OpenAI envelope, so a +/// malformed body (400/413/415/422) cannot escape the contract. +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 { + match axum::Json::::from_request(req, state).await { + Ok(axum::Json(v)) => Ok(ApiJson(v)), + Err(rej) => Err(error_response(rej.status().as_u16(), rej.body_text())), + } + } +} + +/// [`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 { + match axum::Json::::from_request(req, state).await { + Ok(axum::Json(v)) => Ok(AnthJson(v)), + Err(rej) => Err(anthropic_error(rej.status().as_u16(), rej.body_text())), + } + } +} + +/// The contract's machine channel on every HTTP error response (spec 4.1): +/// the classification header plus the CORS exposure both channels 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 (spec 4.2): 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(), + }}) +} + +/// OpenAI-envelope error at the class's external status. +fn class_error(class: ErrClass, message: impl Into) -> Response { + let status = StatusCode::from_u16(class.status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); + with_error_headers( + class, + (status, Json(openai_error_body(class, message.into()))).into_response(), ) - .into_response() +} + +/// 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), + // 499 client-closed: the peer is gone; nothing to render + None => StatusCode::from_u16(status) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) + .into_response(), + } } /// Who an admin bearer token speaks for: the global operator or one tenant. @@ -1415,7 +1541,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 +1622,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 +2214,70 @@ 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" } })), + let Some(class) = ErrClass::classify(e.code, e.http_status) else { + // 499 client-closed: the peer is gone; nothing to render + return StatusCode::from_u16(e.http_status) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) + .into_response(); + }; + let original_status = + (e.code == gw_consts::ErrCode::FED_RESP_STATUS_NOT_ZERO).then_some(e.http_status); + 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) = e.resource { + body["error"]["resource_name"] = r.into(); + } + } + let status = StatusCode::from_u16(class.status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); + with_error_headers(class, (status, Json(body)).into_response()) +} + +/// Anthropic-shaped error body (spec 4.3) — `error.type` is the discriminator +/// the Anthropic SDKs key their exception dispatch on; `code` is additive. +fn anthropic_error_with(class: ErrClass, message: impl Into) -> Response { + let status = StatusCode::from_u16(class.status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); + with_error_headers( + class, + ( + status, + Json(json!({ + "type": "error", + "error": { + "type": class.anthropic_type(), + "code": class.code(), + "message": message.into(), + }, + })), + ) + .into_response(), ) - .into_response() } -/// 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 from an ad-hoc (status, message) site. +fn anthropic_error(status: u16, message: impl Into) -> Response { + match ErrClass::from_status(status) { + Some(class) => anthropic_error_with(class, message), + None => StatusCode::from_u16(status) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) + .into_response(), } } -/// Anthropic-shaped error body: `{"type":"error","error":{"type","message"}}` — -/// the discriminator the Anthropic SDKs key their exception dispatch on. -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() +/// [`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 => StatusCode::from_u16(e.http_status) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) + .into_response(), + } } /// Run the pipeline on its own task so a client disconnect can't cancel it @@ -2236,7 +2390,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() { @@ -2429,6 +2583,21 @@ fn sse_stream( Sse::new(stream) } +/// Chat/legacy SSE error frame (spec 4.5): the OpenAI envelope plus the +/// upstream's original status when one was received. +fn stream_error_body(err: &gw_models::StreamError) -> Value { + let mut e = json!({ + "message": err.message, + "type": err.class.openai_type(), + "param": null, + "code": err.class.code(), + }); + if let Some(os) = err.original_status { + e["original_status_code"] = os.into(); + } + json!({ "error": e }) +} + /// Streaming chat: pipeline chunks re-emitted as OpenAI SSE. fn chat_stream_response( s: AppState, @@ -2455,12 +2624,8 @@ fn chat_stream_response( Some(gw_engines::StreamChunk { error: Some(err), .. }) => { - self.queue.push_back( - Event::default().data( - json!({"error": {"message": err.message, "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 } @@ -2591,7 +2756,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 { @@ -2648,7 +2813,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 { @@ -2827,7 +2992,11 @@ fn messages_stream_response( }) => { self.queue.push_back(St::ev( "error", - json!({"type":"error","error":{"type":"api_error","message":err.message}}), + json!({"type":"error","error":{ + "type": err.class.anthropic_type(), + "code": err.class.code(), + "message": err.message, + }}), )); true } @@ -2951,7 +3120,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(); @@ -3015,7 +3184,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(); @@ -3067,6 +3236,7 @@ fn responses_stream_response( model: String, created: bool, status: String, + seq: u64, } impl St { fn ensure_created(&mut self) { @@ -3074,6 +3244,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(), @@ -3090,10 +3261,18 @@ fn responses_stream_response( error: Some(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":err.message}}) - .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]")); @@ -3102,6 +3281,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}) @@ -3115,6 +3295,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":{ @@ -3141,6 +3322,7 @@ fn responses_stream_response( queue: VecDeque::new(), model, created: false, + seq: 0, status: "completed".to_owned(), }, ) @@ -3151,7 +3333,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(); @@ -3186,7 +3368,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(); @@ -3224,7 +3406,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(); @@ -3263,7 +3445,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(); @@ -3321,7 +3503,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 } @@ -3332,7 +3514,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 } @@ -3390,7 +3572,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(); @@ -3422,7 +3604,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(); @@ -3484,7 +3666,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(); @@ -3557,7 +3739,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 { @@ -4611,8 +4793,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:?}" ); } From d3099448a7bcc357669d9915760722bb1d0d4892 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 28 Jul 2026 19:49:05 +0800 Subject: [PATCH 3/6] review: error-contract quality round Simplify convergence from the four-lens pass: one render core in views (bare_status + class_response shared by every envelope), the 424-extras rule lives once as GatewayError::original_status, StreamError::from_error takes ownership and moves the message, the pump's decode-fault branch reuses StreamFault instead of hand-rolling both error forms, the ApiJson/AnthJson extractors share one body, the chat SSE frame reuses openai_error_body, and the realtime send_err alias is gone. abort_frame's live-channel invariant is now a debug_assert; ServiceQuotaExceeded documents that it is contract- reserved with no construction site yet; the wrong-surface e2e asserts the classification, not just the status. --- crates/consts/src/error_class.rs | 4 +- crates/engines/src/pump.rs | 24 +++--- crates/models/src/error.rs | 6 ++ crates/models/src/response.rs | 9 +- crates/server/tests/e2e.rs | 5 ++ crates/views/src/lib.rs | 137 ++++++++++++++++--------------- docs/api.md | 24 ++++-- 7 files changed, 117 insertions(+), 92 deletions(-) diff --git a/crates/consts/src/error_class.rs b/crates/consts/src/error_class.rs index c369ec2..0327f24 100644 --- a/crates/consts/src/error_class.rs +++ b/crates/consts/src/error_class.rs @@ -9,6 +9,8 @@ use crate::ErrCode; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ErrClass { Validation, + /// Reserved by the contract for hard quota/balance exhaustion; no + /// construction site classifies into it yet (billing wires it later). ServiceQuotaExceeded, UnrecognizedClient, AccessDenied, @@ -72,7 +74,7 @@ impl ErrClass { ErrClass::ModelTimeout => 408, ErrClass::Conflict => 409, ErrClass::RequestEntityTooLarge => 413, - // in-stream only; the HTTP status never reaches a client + // ModelStreamError never renders at the HTTP phase; 424 nominal ErrClass::ModelError | ErrClass::ModelStreamError => 424, ErrClass::Throttling => 429, ErrClass::InternalServer => 500, diff --git a/crates/engines/src/pump.rs b/crates/engines/src/pump.rs index 1013d99..630c014 100644 --- a/crates/engines/src/pump.rs +++ b/crates/engines/src/pump.rs @@ -5,7 +5,7 @@ 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)] @@ -49,6 +49,7 @@ async fn abort_frame( 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 { @@ -108,20 +109,19 @@ where Ok(events) => events, Err(e) if sent_any => { tracing::warn!(vendor, error = %e, "upstream stream failed mid-response"); - let error = StreamError { - class: gw_consts::ErrClass::ModelStreamError, - message: format!("upstream stream failed: {e}"), - original_status: None, + let fault = StreamFault { + timeout: false, + message: e, }; - abort_frame(&tx, &mut out, error).await; + abort_frame(&tx, &mut out, fault.stream_error()).await; break; } Err(e) => { - return Err(GatewayError::new( - gw_consts::ErrCode::FED_RESP_RPC_FAILED, - 502, - format!("upstream stream failed: {e}"), - )); + return Err(StreamFault { + timeout: false, + message: e, + } + .into_error()); } }; for data in events { @@ -136,7 +136,7 @@ where Ok(c) => c, Err(e) if sent_any => { tracing::warn!(vendor, error = %e, "vendor error frame after commit"); - if let Some(error) = StreamError::from_error(&e) { + if let Some(error) = StreamError::from_error(e) { abort_frame(&tx, &mut out, error).await; } else { out.aborted = true; diff --git a/crates/models/src/error.rs b/crates/models/src/error.rs index e0a8037..3046e5b 100644 --- a/crates/models/src/error.rs +++ b/crates/models/src/error.rs @@ -57,6 +57,12 @@ impl GatewayError { self.source = Some(Box::new(source)); self } + + /// The real upstream status behind this error, when one was received: + /// only FED_RESP_STATUS_NOT_ZERO carries the vendor's actual reply status. + pub fn original_status(&self) -> Option { + (self.code == ErrCode::FED_RESP_STATUS_NOT_ZERO).then_some(self.http_status) + } } impl fmt::Display for GatewayError { diff --git a/crates/models/src/response.rs b/crates/models/src/response.rs index a2e9511..4a1be41 100644 --- a/crates/models/src/response.rs +++ b/crates/models/src/response.rs @@ -58,15 +58,12 @@ pub struct StreamError { 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 { + pub fn from_error(e: crate::GatewayError) -> Option { let class = gw_consts::ErrClass::classify(e.code, e.http_status)?; - // FED_RESP_STATUS_NOT_ZERO is the one code whose status is the real - // upstream reply; the rest never received one. - let original_status = - (e.code == gw_consts::ErrCode::FED_RESP_STATUS_NOT_ZERO).then_some(e.http_status); + let original_status = e.original_status(); Some(Self { class, - message: e.message.clone(), + message: e.message, original_status, }) } diff --git a/crates/server/tests/e2e.rs b/crates/server/tests/e2e.rs index 0ce626b..030e34a 100644 --- a/crates/server/tests/e2e.rs +++ b/crates/server/tests/e2e.rs @@ -926,6 +926,11 @@ async fn model_failure_modes_404_503_unsupported() { .await .unwrap(); 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] diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs index be6e4ce..757fe64 100644 --- a/crates/views/src/lib.rs +++ b/crates/views/src/lib.rs @@ -360,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 @@ -650,8 +653,6 @@ async fn realtime_bridge( use tokio_tungstenite::tungstenite::Message as UMsg; use tokio_tungstenite::tungstenite::client::IntoClientRequest; - let send_err = rt_error; - let base = account.endpoint.trim_end_matches('/'); let ws_base = base .replacen("https://", "wss://", 1) @@ -663,7 +664,7 @@ async fn realtime_bridge( s.handler.state().avail.record(&rtm.requested, false); let _ = client .send(CMsg::Text( - send_err(ErrClass::InternalServer, format!("bad upstream url: {e}")) + rt_error(ErrClass::InternalServer, format!("bad upstream url: {e}")) .to_string() .into(), )) @@ -681,7 +682,7 @@ async fn realtime_bridge( s.handler.state().avail.record(&rtm.requested, false); let _ = client .send(CMsg::Text( - send_err( + rt_error( ErrClass::ModelError, format!("upstream realtime connect failed: {e}"), ) @@ -721,7 +722,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(ErrClass::AccessDenied, reason).to_string().into())) + .send(CMsg::Text(rt_error(ErrClass::AccessDenied, reason).to_string().into())) .await .is_err() { @@ -747,7 +748,7 @@ async fn realtime_bridge( } Err((class, denied)) => { if cl_tx - .send(CMsg::Text(send_err(class, denied).to_string().into())) + .send(CMsg::Text(rt_error(class, denied).to_string().into())) .await .is_err() { @@ -802,7 +803,7 @@ async fn realtime_bridge( .send(UMsg::text(json!({"type":"response.cancel"}).to_string())) .await; let _ = cl_tx - .send(CMsg::Text(send_err(class, denied).to_string().into())) + .send(CMsg::Text(rt_error(class, denied).to_string().into())) .await; suppress = true; relay = false; @@ -1082,8 +1083,26 @@ impl axum::extract::FromRequestParts for Authed { } } -/// `axum::Json` with rejections rendered in the OpenAI envelope, so a +/// 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 @@ -1095,10 +1114,7 @@ where type Rejection = Response; async fn from_request(req: axum::extract::Request, state: &S) -> Result { - match axum::Json::::from_request(req, state).await { - Ok(axum::Json(v)) => Ok(ApiJson(v)), - Err(rej) => Err(error_response(rej.status().as_u16(), rej.body_text())), - } + extract_json(req, state, error_response).await.map(ApiJson) } } @@ -1114,10 +1130,9 @@ where type Rejection = Response; async fn from_request(req: axum::extract::Request, state: &S) -> Result { - match axum::Json::::from_request(req, state).await { - Ok(axum::Json(v)) => Ok(AnthJson(v)), - Err(rej) => Err(anthropic_error(rej.status().as_u16(), rej.body_text())), - } + extract_json(req, state, anthropic_error) + .await + .map(AnthJson) } } @@ -1146,13 +1161,24 @@ fn openai_error_body(class: ErrClass, message: String) -> Value { }}) } +/// 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 { - let status = StatusCode::from_u16(class.status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); - with_error_headers( - class, - (status, Json(openai_error_body(class, message.into()))).into_response(), - ) + class_response(class, openai_error_body(class, message.into())) } /// OpenAI-envelope error from an ad-hoc (status, message) site; the external @@ -1160,10 +1186,7 @@ fn class_error(class: ErrClass, message: impl Into) -> Response { fn error_response(status: u16, message: impl Into) -> Response { match ErrClass::from_status(status) { Some(class) => class_error(class, message), - // 499 client-closed: the peer is gone; nothing to render - None => StatusCode::from_u16(status) - .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) - .into_response(), + None => bare_status(status), } } @@ -2219,44 +2242,35 @@ async fn admin_content_erase( /// as the client's own 401. fn gateway_error(e: GatewayError) -> Response { let Some(class) = ErrClass::classify(e.code, e.http_status) else { - // 499 client-closed: the peer is gone; nothing to render - return StatusCode::from_u16(e.http_status) - .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) - .into_response(); + return bare_status(e.http_status); }; - let original_status = - (e.code == gw_consts::ErrCode::FED_RESP_STATUS_NOT_ZERO).then_some(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) = e.resource { + if let Some(r) = resource { body["error"]["resource_name"] = r.into(); } } - let status = StatusCode::from_u16(class.status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); - with_error_headers(class, (status, Json(body)).into_response()) + class_response(class, body) } /// Anthropic-shaped error body (spec 4.3) — `error.type` is the discriminator /// the Anthropic SDKs key their exception dispatch on; `code` is additive. fn anthropic_error_with(class: ErrClass, message: impl Into) -> Response { - let status = StatusCode::from_u16(class.status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); - with_error_headers( + class_response( class, - ( - status, - Json(json!({ - "type": "error", - "error": { - "type": class.anthropic_type(), - "code": class.code(), - "message": message.into(), - }, - })), - ) - .into_response(), + json!({ + "type": "error", + "error": { + "type": class.anthropic_type(), + "code": class.code(), + "message": message.into(), + }, + }), ) } @@ -2264,9 +2278,7 @@ fn anthropic_error_with(class: ErrClass, message: impl Into) -> Response fn anthropic_error(status: u16, message: impl Into) -> Response { match ErrClass::from_status(status) { Some(class) => anthropic_error_with(class, message), - None => StatusCode::from_u16(status) - .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) - .into_response(), + None => bare_status(status), } } @@ -2274,9 +2286,7 @@ fn anthropic_error(status: u16, message: impl Into) -> Response { fn anthropic_gateway_error(e: GatewayError) -> Response { match ErrClass::classify(e.code, e.http_status) { Some(class) => anthropic_error_with(class, e.message), - None => StatusCode::from_u16(e.http_status) - .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) - .into_response(), + None => bare_status(e.http_status), } } @@ -2540,7 +2550,7 @@ fn spawn_stream_pipeline( Err(e) => { // 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) { + if let Some(error) = gw_models::StreamError::from_error(e) { let _ = tx .send(gw_engines::StreamChunk { error: Some(error), @@ -2585,17 +2595,12 @@ fn sse_stream( /// Chat/legacy SSE error frame (spec 4.5): the OpenAI envelope plus the /// upstream's original status when one was received. -fn stream_error_body(err: &gw_models::StreamError) -> Value { - let mut e = json!({ - "message": err.message, - "type": err.class.openai_type(), - "param": null, - "code": err.class.code(), - }); +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 { - e["original_status_code"] = os.into(); + body["error"]["original_status_code"] = os.into(); } - json!({ "error": e }) + body } /// Streaming chat: pipeline chunks re-emitted as OpenAI SSE. @@ -2625,7 +2630,7 @@ fn chat_stream_response( error: Some(err), .. }) => { self.queue - .push_back(Event::default().data(stream_error_body(&err).to_string())); + .push_back(Event::default().data(stream_error_body(err).to_string())); self.queue.push_back(Event::default().data("[DONE]")); true } 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 From 2da6a649c20f29336a60c982c15fc3594bc5cc63 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 28 Jul 2026 20:10:05 +0800 Subject: [PATCH 4/6] test: close error-contract coverage gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Idle-gap timeouts now assert the FED_RESP_TIMEOUT code (the ModelTimeout classification link), oversized bodies assert the 413 envelope, and the realtime mock session asserts the nested validation error frame. Spec section references in code comments replaced with self-contained wording — the numbered contract lives in a private repo public comments must not point at. --- crates/consts/src/error_class.rs | 5 +++-- crates/engines/tests/http_transport_wire.rs | 5 +++++ crates/server/tests/e2e.rs | 24 +++++++++++++++++++++ crates/views/src/lib.rs | 16 +++++++------- 4 files changed, 40 insertions(+), 10 deletions(-) diff --git a/crates/consts/src/error_class.rs b/crates/consts/src/error_class.rs index 0327f24..ebab17d 100644 --- a/crates/consts/src/error_class.rs +++ b/crates/consts/src/error_class.rs @@ -82,7 +82,7 @@ impl ErrClass { } } - /// Coarse `type` for the OpenAI envelope (contract §4.2). + /// Coarse `type` for the OpenAI envelope, for SDKs that key on it. pub const fn openai_type(self) -> &'static str { match self { ErrClass::Validation @@ -99,7 +99,8 @@ impl ErrClass { } } - /// `type` for the Anthropic envelope (contract §4.3). + /// `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 => { diff --git a/crates/engines/tests/http_transport_wire.rs b/crates/engines/tests/http_transport_wire.rs index f13a630..6adfb2e 100644 --- a/crates/engines/tests/http_transport_wire.rs +++ b/crates/engines/tests/http_transport_wire.rs @@ -318,6 +318,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: {:?}", diff --git a/crates/server/tests/e2e.rs b/crates/server/tests/e2e.rs index 030e34a..73acf42 100644 --- a/crates/server/tests/e2e.rs +++ b/crates/server/tests/e2e.rs @@ -2593,6 +2593,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(), )) @@ -3224,6 +3231,7 @@ async fn error_contract_machine_channel() { assert_eq!(j["error"]["code"], "validation_exception"); let r = app + .clone() .oneshot(post("/v1/messages", None, r#"{"model":"m","messages":[]}"#)) .await .unwrap(); @@ -3232,6 +3240,22 @@ async fn error_contract_machine_channel() { 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 757fe64..ca3733f 100644 --- a/crates/views/src/lib.rs +++ b/crates/views/src/lib.rs @@ -185,7 +185,7 @@ async fn track_requests( resp } -/// In-band realtime error event (spec 4.4). Never terminal: the session +/// 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":{ @@ -1136,8 +1136,8 @@ where } } -/// The contract's machine channel on every HTTP error response (spec 4.1): -/// the classification header plus the CORS exposure both channels need. +/// 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( @@ -1151,7 +1151,7 @@ fn with_error_headers(class: ErrClass, mut resp: Response) -> Response { resp } -/// OpenAI-shaped error body (spec 4.2): coarse `type`, precise `code`. +/// OpenAI-shaped error body: coarse `type`, precise `code`. fn openai_error_body(class: ErrClass, message: String) -> Value { json!({ "error": { "message": message, @@ -2258,8 +2258,8 @@ fn gateway_error(e: GatewayError) -> Response { class_response(class, body) } -/// Anthropic-shaped error body (spec 4.3) — `error.type` is the discriminator -/// the Anthropic SDKs key their exception dispatch on; `code` is additive. +/// Anthropic-shaped error body — `error.type` is the discriminator the +/// Anthropic SDKs key their exception dispatch on; `code` is additive. fn anthropic_error_with(class: ErrClass, message: impl Into) -> Response { class_response( class, @@ -2593,8 +2593,8 @@ fn sse_stream( Sse::new(stream) } -/// Chat/legacy SSE error frame (spec 4.5): the OpenAI envelope plus the -/// upstream's original status when one was received. +/// 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 { From c910baa6b20af8c391c4a84641a888f28a3b0f94 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 28 Jul 2026 20:53:15 +0800 Subject: [PATCH 5/6] fix: close gateway error contract gaps --- crates/consts/src/error_class.rs | 6 +- crates/consts/src/error_code.rs | 1 + crates/dag/src/nodes.rs | 8 ++- crates/engines/src/engine.rs | 20 ++++-- crates/engines/src/http_transport.rs | 14 +++- crates/engines/src/pump.rs | 4 +- crates/engines/tests/http_transport_wire.rs | 71 ++++++++++++++++++++- crates/handler/src/lib.rs | 35 ++++++++-- crates/models/src/error.rs | 12 +++- crates/models/src/response.rs | 38 ++++++++++- crates/server/tests/e2e.rs | 11 ++-- crates/views/src/lib.rs | 19 ++++-- 12 files changed, 204 insertions(+), 35 deletions(-) diff --git a/crates/consts/src/error_class.rs b/crates/consts/src/error_class.rs index ebab17d..5f0f68f 100644 --- a/crates/consts/src/error_class.rs +++ b/crates/consts/src/error_class.rs @@ -9,8 +9,7 @@ use crate::ErrCode; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ErrClass { Validation, - /// Reserved by the contract for hard quota/balance exhaustion; no - /// construction site classifies into it yet (billing wires it later). + /// Hard quota or balance exhaustion; retry only after quota becomes available. ServiceQuotaExceeded, UnrecognizedClient, AccessDenied, @@ -146,6 +145,7 @@ impl ErrClass { _ => 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), @@ -226,6 +226,8 @@ mod tests { // 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] diff --git a/crates/consts/src/error_code.rs b/crates/consts/src/error_code.rs index f79df13..67b12b5 100644 --- a/crates/consts/src/error_code.rs +++ b/crates/consts/src/error_code.rs @@ -43,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/dag/src/nodes.rs b/crates/dag/src/nodes.rs index c48b708..c320152 100644 --- a/crates/dag/src/nodes.rs +++ b/crates/dag/src/nodes.rs @@ -20,6 +20,10 @@ 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 +302,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 +465,7 @@ impl DagNode for UserBudgetGate { ctx.effective_user_id(), ) .await - .map_err(limit_denied) + .map_err(quota_denied) } } diff --git a/crates/engines/src/engine.rs b/crates/engines/src/engine.rs index 1b875e2..302730b 100644 --- a/crates/engines/src/engine.rs +++ b/crates/engines/src/engine.rs @@ -101,8 +101,8 @@ pub fn vendor_error(http_status: u16, v: &Value) -> Option { (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"] @@ -112,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. @@ -139,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] diff --git a/crates/engines/src/http_transport.rs b/crates/engines/src/http_transport.rs index 4bd3bcd..6c6fec6 100644 --- a/crates/engines/src/http_transport.rs +++ b/crates/engines/src/http_transport.rs @@ -192,8 +192,18 @@ 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| { + if e.is_timeout() { + GatewayError::new( + gw_consts::ErrCode::FED_RESP_TIMEOUT, + 502, + "read upstream body timed out", + ) + .with_source(e) + } else { + GatewayError::internal("read upstream body").with_source(e) + } + })?; Ok(UpstreamResponse { status, body: UpstreamBody::Json(bytes), diff --git a/crates/engines/src/pump.rs b/crates/engines/src/pump.rs index 630c014..fd91909 100644 --- a/crates/engines/src/pump.rs +++ b/crates/engines/src/pump.rs @@ -53,7 +53,7 @@ async fn abort_frame( if let Some(sender) = tx { let _ = sender .send(StreamChunk { - error: Some(error), + error: Some(Box::new(error)), ..Default::default() }) .await; @@ -136,7 +136,7 @@ where Ok(c) => c, Err(e) if sent_any => { tracing::warn!(vendor, error = %e, "vendor error frame after commit"); - if let Some(error) = StreamError::from_error(e) { + if let Some(error) = StreamError::from_committed_error(e) { abort_frame(&tx, &mut out, error).await; } else { out.aborted = true; diff --git a/crates/engines/tests/http_transport_wire.rs b/crates/engines/tests/http_transport_wire.rs index 6adfb2e..1f14ff4 100644 --- a/crates/engines/tests/http_transport_wire.rs +++ b/crates/engines/tests/http_transport_wire.rs @@ -84,6 +84,61 @@ 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 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(); + }); + 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")); +} + #[tokio::test] async fn http_transport_sse_over_real_socket() { let base = spawn_vendor().await; @@ -457,7 +512,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")), @@ -473,4 +534,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 60618a8..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] diff --git a/crates/models/src/error.rs b/crates/models/src/error.rs index 3046e5b..ec864de 100644 --- a/crates/models/src/error.rs +++ b/crates/models/src/error.rs @@ -20,6 +20,7 @@ pub struct GatewayError { /// 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>, } @@ -30,6 +31,7 @@ impl GatewayError { http_status, message: message.into(), resource: None, + original_status: None, source: None, } } @@ -58,10 +60,16 @@ impl GatewayError { 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: - /// only FED_RESP_STATUS_NOT_ZERO carries the vendor's actual reply status. + /// synthetic statuses used for failover are never returned. pub fn original_status(&self) -> Option { - (self.code == ErrCode::FED_RESP_STATUS_NOT_ZERO).then_some(self.http_status) + self.original_status } } diff --git a/crates/models/src/response.rs b/crates/models/src/response.rs index 4a1be41..3c130af 100644 --- a/crates/models/src/response.rs +++ b/crates/models/src/response.rs @@ -67,6 +67,17 @@ impl StreamError { original_status, }) } + + /// Build the terminal frame for an already-committed stream. A generic + /// upstream failure becomes the stream-specific class; known transient + /// statuses keep their more actionable classification. + 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. @@ -81,7 +92,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)] @@ -102,4 +113,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/server/tests/e2e.rs b/crates/server/tests/e2e.rs index 73acf42..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" ); @@ -1445,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() @@ -1465,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] diff --git a/crates/views/src/lib.rs b/crates/views/src/lib.rs index ca3733f..60cfaeb 100644 --- a/crates/views/src/lib.rs +++ b/crates/views/src/lib.rs @@ -422,21 +422,21 @@ async fn realtime_gate( .map_err(throttled)?; admission::check_user_budget(gov, cfg, &ak.tenant, ak.attributed_user(hint)) .await - .map_err(throttled)?; + .map_err(|m| (ErrClass::ServiceQuotaExceeded, m))?; 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(( - ErrClass::Throttling, + 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 - .map_err(throttled)?; + .map_err(|m| (ErrClass::ServiceQuotaExceeded, m))?; let tpm_reserved = match admission::reserve_tpm(gov, &ak, REALTIME_TURN_RESERVE).await { Ok(reserved) => reserved, Err(denied) => { @@ -2553,7 +2553,7 @@ fn spawn_stream_pipeline( if let Some(error) = gw_models::StreamError::from_error(e) { let _ = tx .send(gw_engines::StreamChunk { - error: Some(error), + error: Some(Box::new(error)), ..Default::default() }) .await; @@ -2630,7 +2630,7 @@ fn chat_stream_response( error: Some(err), .. }) => { self.queue - .push_back(Event::default().data(stream_error_body(err).to_string())); + .push_back(Event::default().data(stream_error_body(*err).to_string())); self.queue.push_back(Event::default().data("[DONE]")); true } @@ -2995,6 +2995,7 @@ fn messages_stream_response( Some(gw_engines::StreamChunk { error: Some(err), .. }) => { + let err = *err; self.queue.push_back(St::ev( "error", json!({"type":"error","error":{ @@ -3265,6 +3266,7 @@ fn responses_stream_response( Some(gw_engines::StreamChunk { 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 @@ -4736,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] From 844375ee523bdadc83fe031540e9df125452f080 Mon Sep 17 00:00:00 2001 From: CMGS Date: Tue, 28 Jul 2026 22:58:55 +0800 Subject: [PATCH 6/6] review: whole-repo code-rs + simplify round Non-timeout body-read failures join the upstream fault family (502 FED_RESP_RPC_FAILED -> external 424) instead of masquerading as internal 500s, with wire tests for both the timeout and mid-body-break paths. The timeout-vs-rpc code split now lives once (transport::upstream_fault_code); the pump's decode-fault arms build one StreamFault; the Anthropic error body has one owner shared by HTTP and SSE like the OpenAI side; realtime quota denials use a named closure beside throttled; wire tests share one serve_router bootstrap. Message ingest stops cloning plain-text content just to drop the original (into_text_and_parts / blocks_text move it). Comment pass: two docs tightened to the one-line budget, limit_denied's stale every-limit claim scoped to throttling now that quota_denied exists. Whole-repo sweep found no layout, unwrap, or comment-budget violations; DAG ordering guards adjudicated keep (runtime-enforced topology). --- crates/dag/src/nodes.rs | 3 +- crates/engines/src/http_transport.rs | 24 +++----- crates/engines/src/pump.rs | 16 ++--- crates/engines/src/transport.rs | 17 ++++-- crates/engines/tests/http_transport_wire.rs | 68 ++++++++++++++++----- crates/models/src/response.rs | 5 +- crates/protocol/src/anthropic.rs | 17 ++++-- crates/protocol/src/openai.rs | 23 +++++-- crates/views/src/lib.rs | 58 +++++++++--------- 9 files changed, 141 insertions(+), 90 deletions(-) diff --git a/crates/dag/src/nodes.rs b/crates/dag/src/nodes.rs index c320152..b99262f 100644 --- a/crates/dag/src/nodes.rs +++ b/crates/dag/src/nodes.rs @@ -15,7 +15,8 @@ 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) } diff --git a/crates/engines/src/http_transport.rs b/crates/engines/src/http_transport.rs index 6c6fec6..a16b7e8 100644 --- a/crates/engines/src/http_transport.rs +++ b/crates/engines/src/http_transport.rs @@ -13,6 +13,7 @@ use gw_models::{GResult, GatewayError}; use crate::transport::{ MockTransport, StreamFault, Transport, UpstreamBody, UpstreamRequest, UpstreamResponse, + upstream_fault_code, }; const RETRY_BACKOFF: Duration = Duration::from_millis(100); @@ -147,13 +148,8 @@ impl Transport for HttpTransport { tokio::time::sleep(RETRY_BACKOFF * attempt).await; } Err(e) => { - let code = if e.is_timeout() { - gw_consts::ErrCode::FED_RESP_TIMEOUT - } else { - gw_consts::ErrCode::FED_RESP_RPC_FAILED - }; return Err(GatewayError::new( - code, + upstream_fault_code(e.is_timeout()), 502, format!("upstream request failed: {e}"), )); @@ -193,16 +189,14 @@ impl Transport for HttpTransport { read.await }; let bytes = bytes.map_err(|e| { - if e.is_timeout() { - GatewayError::new( - gw_consts::ErrCode::FED_RESP_TIMEOUT, - 502, - "read upstream body timed out", - ) - .with_source(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 { - GatewayError::internal("read upstream body").with_source(e) - } + "read upstream body failed" + }; + GatewayError::new(upstream_fault_code(e.is_timeout()), 502, what).with_source(e) })?; Ok(UpstreamResponse { status, diff --git a/crates/engines/src/pump.rs b/crates/engines/src/pump.rs index fd91909..8975189 100644 --- a/crates/engines/src/pump.rs +++ b/crates/engines/src/pump.rs @@ -107,21 +107,17 @@ where }; let events = match dec.feed(&bytes) { Ok(events) => events, - Err(e) if sent_any => { - tracing::warn!(vendor, error = %e, "upstream stream failed mid-response"); + Err(e) => { let fault = StreamFault { timeout: false, message: e, }; - abort_frame(&tx, &mut out, fault.stream_error()).await; - break; - } - Err(e) => { - return Err(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; } - .into_error()); + return Err(fault.into_error()); } }; for data in events { diff --git a/crates/engines/src/transport.rs b/crates/engines/src/transport.rs index 6f447bc..4b9adb9 100644 --- a/crates/engines/src/transport.rs +++ b/crates/engines/src/transport.rs @@ -40,13 +40,8 @@ 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 { - let code = if self.timeout { - gw_consts::ErrCode::FED_RESP_TIMEOUT - } else { - gw_consts::ErrCode::FED_RESP_RPC_FAILED - }; GatewayError::new( - code, + upstream_fault_code(self.timeout), 502, format!("upstream stream failed: {}", self.message), ) @@ -66,6 +61,16 @@ impl StreamFault { } } +/// 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 { diff --git a/crates/engines/tests/http_transport_wire.rs b/crates/engines/tests/http_transport_wire.rs index 1f14ff4..3fa859b 100644 --- a/crates/engines/tests/http_transport_wire.rs +++ b/crates/engines/tests/http_transport_wire.rs @@ -17,6 +17,15 @@ 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}") } @@ -107,11 +112,7 @@ async fn spawn_stalled_json_vendor() -> 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}/slow-json") } @@ -139,6 +140,47 @@ async fn non_stream_body_timeout_classifies_as_model_timeout() { 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; @@ -321,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") } diff --git a/crates/models/src/response.rs b/crates/models/src/response.rs index 3c130af..45f3ee3 100644 --- a/crates/models/src/response.rs +++ b/crates/models/src/response.rs @@ -68,9 +68,8 @@ impl StreamError { }) } - /// Build the terminal frame for an already-committed stream. A generic - /// upstream failure becomes the stream-specific class; known transient - /// statuses keep their more actionable classification. + /// 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 { 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/views/src/lib.rs b/crates/views/src/lib.rs index 60cfaeb..5cf0897 100644 --- a/crates/views/src/lib.rs +++ b/crates/views/src/lib.rs @@ -408,6 +408,7 @@ async fn realtime_gate( } let gov = state.governance.as_ref(); 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)?; @@ -422,7 +423,7 @@ async fn realtime_gate( .map_err(throttled)?; admission::check_user_budget(gov, cfg, &ak.tenant, ak.attributed_user(hint)) .await - .map_err(|m| (ErrClass::ServiceQuotaExceeded, m))?; + .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) @@ -436,7 +437,7 @@ async fn realtime_gate( let at = gw_state::epoch_secs(); admission::reserve_daily(gov, &ak, REALTIME_TURN_RESERVE, at) .await - .map_err(|m| (ErrClass::ServiceQuotaExceeded, m))?; + .map_err(quota_exceeded)?; let tpm_reserved = match admission::reserve_tpm(gov, &ak, REALTIME_TURN_RESERVE).await { Ok(reserved) => reserved, Err(denied) => { @@ -2260,18 +2261,19 @@ fn gateway_error(e: GatewayError) -> Response { /// 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, - json!({ - "type": "error", - "error": { - "type": class.anthropic_type(), - "code": class.code(), - "message": message.into(), - }, - }), - ) + class_response(class, anthropic_error_body(class, message.into())) } /// Anthropic-shaped error from an ad-hoc (status, message) site. @@ -2411,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, } @@ -2798,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), @@ -2998,11 +3002,7 @@ fn messages_stream_response( let err = *err; self.queue.push_back(St::ev( "error", - json!({"type":"error","error":{ - "type": err.class.anthropic_type(), - "code": err.class.code(), - "message": err.message, - }}), + anthropic_error_body(err.class, err.message), )); true }