From 34314a99f099811348c656819841469af506dca9 Mon Sep 17 00:00:00 2001 From: Nate Armstrong Date: Mon, 2 Nov 2026 21:02:52 -0800 Subject: [PATCH] fix(messages): make rust messages copy python messages behaviour (#49611) Closes the Rust vs Python /v1/messages differences the parity suite marks as known gaps, copying Python exactly. --- .../python_contract.json | 18 + .../callbacks-legacy-python/src/adapter.rs | 117 ++++++- .../callbacks-legacy-python/src/callbacks.rs | 51 ++- .../callbacks-legacy-python/src/python.rs | 6 + .../src/test_support.rs | 7 + litellm-rust/crates/host-http/tests/serve.rs | 14 +- .../crates/host-native/tests/driver.rs | 8 +- .../crates/host-python/tests/support/mod.rs | 2 +- litellm-rust/crates/host/src/interceptors.rs | 24 +- litellm-rust/crates/host/tests/observation.rs | 4 +- litellm-rust/crates/http/src/error.rs | 76 ++++- .../crates/inference-chat/src/handler.rs | 2 +- .../crates/inference-messages/Cargo.toml | 2 +- .../inference-messages/src/common_utils.rs | 74 ++++ .../crates/inference-messages/src/handler.rs | 191 ++++++++--- .../crates/inference-messages/src/lib.rs | 1 + .../tests/messages/request.rs | 14 +- .../tests/messages/response.rs | 25 +- .../tests/messages/stream.rs | 43 +++ .../crates/inference-ocr/src/handler.rs | 4 +- .../crates/inference-responses/src/handler.rs | 2 +- litellm-rust/crates/inference/src/context.rs | 8 +- litellm-rust/crates/inference/src/error.rs | 28 ++ .../src/formats/messages/request.rs | 7 +- .../src/formats/messages/response.rs | 36 +- .../crates/llms-types/src/recognized.rs | 10 + .../crates/llms/src/anthropic/common_utils.rs | 57 +++- .../llms/src/anthropic/messages/thinking.rs | 233 +++++++++++-- .../src/anthropic/messages/transformation.rs | 28 +- .../src/azure_ai/messages/transformation.rs | 2 +- .../llms/src/base_llm/messages/context.rs | 19 ++ litellm-rust/crates/llms/src/error.rs | 22 ++ .../src/vertex_ai/messages/transformation.rs | 2 +- .../crates/python-bridge/src/errors.rs | 59 +++- .../python-bridge/src/routes/messages/host.rs | 21 +- .../python-bridge/src/routes/messages/mod.rs | 13 +- .../host/callbacks_legacy_python.py | 45 ++- litellm/rust_bridge/host/routes/messages.py | 200 ++++++++++- .../sdk/messages/test_native_parity.py | 321 ++++++------------ .../cache/test_python_cache.py | 48 +-- .../messages/test_callbacks.py | 25 +- .../messages/test_streaming.py | 13 +- .../support/recording_server.py | 8 +- .../rust_bridge/host/routes/test_messages.py | 107 ++++++ .../host/test_callbacks_legacy_python.py | 54 +++ .../unit/rust_bridge/messages/test_secrets.py | 12 +- 46 files changed, 1595 insertions(+), 468 deletions(-) diff --git a/litellm-rust/crates/callbacks-legacy-python/python_contract.json b/litellm-rust/crates/callbacks-legacy-python/python_contract.json index f65516265c5..ab887bbc014 100644 --- a/litellm-rust/crates/callbacks-legacy-python/python_contract.json +++ b/litellm-rust/crates/callbacks-legacy-python/python_contract.json @@ -33,6 +33,17 @@ "api_key", "additional_args" ], + "record_response": [ + "logger", + "status", + "headers", + "body" + ], + "record_api_window": [ + "logger", + "start", + "end" + ], "defers_async_logging": [ "logger" ], @@ -91,6 +102,13 @@ "logger", "head" ], + "stamp_stream_head": [ + "head", + "logger", + "kwargs", + "start_time", + "end_time" + ], "stream_success": [ "logger", "url_route", diff --git a/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs b/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs index 9431b9d1326..1bb84184e4c 100644 --- a/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs +++ b/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs @@ -57,6 +57,8 @@ pub struct LegacyLogging { asynchronous: bool, internal: bool, cache_key: Option, + /// When the provider request went out, until its window is recorded (Messages only). + api_started: Option, } fn datetime(py: Python<'_>, epoch_seconds: f64) -> PyResult> { @@ -87,6 +89,60 @@ impl LegacyLogging { asynchronous, internal: false, cache_key: None, + api_started: None, + } + } + + /// The kwargs `Logging.update_from_kwargs` sees. Python's Messages handler takes `metadata` as + /// its own argument, so the logger never sees the caller's raw metadata, only the validated + /// request metadata merged in through `litellm_params`. Python updates the logger once, before + /// the call; the refresh once the result is ready leaves out `litellm_metadata` too, so the + /// logged metadata is not rebuilt from what the call wrote into it since. + fn logging_kwargs<'py>(&self, py: Python<'py>, refresh: bool) -> PyResult> { + let kwargs = self.call.kwargs().bind(py).copy()?; + if self.operation != LoggingOperation::Messages { + return Ok(kwargs); + } + let consumed: &[&str] = if refresh { + &["metadata", "litellm_metadata"] + } else { + &["metadata"] + }; + for name in consumed { + if kwargs.contains(name)? { + kwargs.del_item(name)?; + } + } + Ok(kwargs) + } + + /// Python's Messages handler hands the logger its request params, validated `metadata` + /// included, as `litellm_params`, beside the `aanthropic_messages` call flag. + fn logged_request_params( + &self, + context: &RequestContext, + ) -> Option> { + if self.operation != LoggingOperation::Messages { + return None; + } + let mut params = context + .optional_params + .as_object() + .cloned() + .unwrap_or_default(); + params.insert( + "aanthropic_messages".to_owned(), + Value::Bool(self.asynchronous), + ); + Some(params) + } + + /// Python times the provider exchange inside the async `AsyncHTTPHandler.post`; its sync + /// client is untimed. + fn close_api_window(&mut self, py: Python<'_>, end: f64) -> PyResult<()> { + match self.api_started.take() { + Some(start) => self.logger()?.record_api_window(py, start, end), + None => Ok(()), } } @@ -111,6 +167,16 @@ impl LegacyLogging { } } + /// The key `pre_call` logs. Python's Messages handler logs an empty string, never the key. + fn logged_api_key<'a>(&self, context: &'a RequestContext) -> Option<&'a str> { + match self.operation { + LoggingOperation::Messages => Some(""), + LoggingOperation::Completion | LoggingOperation::Responses | LoggingOperation::Ocr => { + context.api_key.as_ref().map(|api_key| api_key.expose()) + } + } + } + fn stream_billing(&self) -> Option { match self.operation { LoggingOperation::Messages => Some(PassThroughStream { @@ -377,7 +443,13 @@ impl LegacyLogging { context: &RequestContext, ) -> PyResult>> { let logger = self.logger()?; - logger.update_from_kwargs(py, self.call.kwargs(), &wire, context)?; + logger.update_from_kwargs( + py, + &self.logging_kwargs(py, false)?, + &wire, + context, + self.logged_request_params(context).as_ref(), + )?; let body = to_py(py, &wire.body)? .into_bound(py) .cast_into::()?; @@ -400,11 +472,14 @@ impl LegacyLogging { self.logger()?.pre_call( py, self.input_description(), - context.api_key.as_ref().map(|api_key| api_key.expose()), + self.logged_api_key(context), &body, &headers, &wire.url, )?; + if self.operation == LoggingOperation::Messages && self.asynchronous { + self.api_started = Some(epoch_seconds()); + } let headers = headers .iter() .map(|(name, value)| Ok((name.extract::()?, value.extract::()?))) @@ -455,7 +530,7 @@ impl LegacyLogging { py, ( &logger, - self.call.kwargs(), + self.logging_kwargs(py, true)?, &facts.provider.model, logger.getattr("optional_params")?, params, @@ -483,6 +558,16 @@ impl LegacyLogging { py: Python<'_>, raw: &RawResponse, ) -> PyResult> { + if self.operation == LoggingOperation::Messages { + // Python's Messages handler never calls `post_call` at receipt: it stores the raw + // response, and logging replays it through `AnthropicConfig.transform_response`. + self.close_api_window(py, epoch_seconds())?; + if let Some(head) = &raw.head { + self.logger()? + .record_response(py, head.status, &head.headers, &raw.body)?; + } + return Ok(HookStep::Ready(())); + } let api_key = self .request .as_ref() @@ -522,6 +607,9 @@ impl LegacyLogging { ) -> PyResult> { self.end = Some(datetime(py, timing.end_time)?); self.error = Some(error.clone_ref(py).into_value(py)); + if self.logger.is_some() { + self.close_api_window(py, timing.end_time)?; + } if self.stream.is_some() { return self.stream_failure(py); } @@ -543,6 +631,20 @@ impl LegacyLogging { head.bind(py).set_item("cache_key", key)?; head.bind(py).set_item("cache_hit", true)?; } + self.close_api_window(py, epoch_seconds())?; + if self.operation == LoggingOperation::Messages { + // Python's `@client` stamps the call metadata onto the stream it hands back. + Streaming::StampHead.call( + py, + ( + head.bind(py), + self.logger()?.object(py), + self.call.kwargs(), + &self.start, + datetime(py, epoch_seconds())?, + ), + )?; + } Streaming::Opened.call(py, (self.logger()?.object(py), head.bind(py)))?; self.stream = Some(DeliveredStream { chunks: PyList::empty(py).unbind(), @@ -1117,9 +1219,7 @@ assert logger.update['custom_llm_provider'] == 'edenai', logger.update let step = logging .before_provider_request(py, Box::new(wire), context) .unwrap(); - let raw = RawResponse { - body: "raw response".into(), - }; + let raw = RawResponse::new("raw response".into()); assert!(matches!( logging .on_event( @@ -1869,9 +1969,10 @@ assert logger.calls[1][1] is response py, &locals, c" -assert logger.names() == ['stream_opened', 'stream_success'], logger.calls +assert logger.names() == ['stamp_stream_head', 'stream_opened', 'stream_success'], logger.calls assert logger.calls[0][1] is head -chunks = logger.calls[1][1] +assert logger.calls[1][1] is head +chunks = logger.calls[2][1] assert len(chunks) == 2 assert chunks[0] is first assert chunks[1] is last diff --git a/litellm-rust/crates/callbacks-legacy-python/src/callbacks.rs b/litellm-rust/crates/callbacks-legacy-python/src/callbacks.rs index f185f21a9c3..1951c9ebfb4 100644 --- a/litellm-rust/crates/callbacks-legacy-python/src/callbacks.rs +++ b/litellm-rust/crates/callbacks-legacy-python/src/callbacks.rs @@ -5,6 +5,7 @@ use litellm_host::interceptors::{RequestContext, WireRequest}; use litellm_host_python::to_py; use pyo3::{exceptions::PyBaseException, prelude::*, types::PyDict}; +use serde_json::Value; use crate::logger::PythonLogger; use crate::python::{Logging, Wrapper}; @@ -15,9 +16,10 @@ pub trait LegacyCallbacks { fn update_from_kwargs( &self, py: Python<'_>, - kwargs: &Py, + kwargs: &Bound<'_, PyDict>, wire: &WireRequest, context: &RequestContext, + request_params: Option<&serde_json::Map>, ) -> PyResult<()>; /// `Logging.pre_call`. @@ -31,6 +33,20 @@ pub trait LegacyCallbacks { url: &str, ) -> PyResult<()>; + /// Records the provider's raw HTTP response on the logger, as Python's Messages handler + /// does with `httpx_response` and `response_headers`. + fn record_response( + &self, + py: Python<'_>, + status: u16, + headers: &[(String, String)], + body: &str, + ) -> PyResult<()>; + + /// Records one provider HTTP exchange's window, as Python's timed `AsyncHTTPHandler.post` + /// does with `llm_api_duration_ms` and the metadata's `llm_api_timing_windows`. + fn record_api_window(&self, py: Python<'_>, start: f64, end: f64) -> PyResult<()>; + /// `Logging.post_call`. fn post_call( &self, @@ -83,12 +99,13 @@ impl LegacyCallbacks for PythonLogger { fn update_from_kwargs( &self, py: Python<'_>, - kwargs: &Py, + kwargs: &Bound<'_, PyDict>, wire: &WireRequest, context: &RequestContext, + request_params: Option<&serde_json::Map>, ) -> PyResult<()> { let secret_fields: Vec<&str> = context.secret_fields.iter().map(String::as_str).collect(); - let redacted_kwargs = redact(py, kwargs.bind(py), &secret_fields)?; + let redacted_kwargs = redact(py, kwargs, &secret_fields)?; let optional_params = redact( py, &to_py(py, &context.optional_params)? @@ -97,23 +114,23 @@ impl LegacyCallbacks for PythonLogger { &secret_fields, )?; let params = PyDict::new(py); - params.set_item( - "litellm_call_id", - kwargs.bind(py).get_item("litellm_call_id")?, - )?; + params.set_item("litellm_call_id", kwargs.get_item("litellm_call_id")?)?; params.set_item("api_base", &wire.url)?; for name in ["logger_fn", "litellm_request_debug"] { - if let Some(value) = kwargs.bind(py).get_item(name)? { + if let Some(value) = kwargs.get_item(name)? { params.set_item(name, value)?; } } for name in custom_pricing_fields(py)? { - if let Some(value) = kwargs.bind(py).get_item(&name)? + if let Some(value) = kwargs.get_item(&name)? && !value.is_none() { params.set_item(name, value)?; } } + for (name, value) in request_params.into_iter().flatten() { + params.set_item(name, to_py(py, value)?)?; + } Logging::Update.call( py, ( @@ -145,6 +162,22 @@ impl LegacyCallbacks for PythonLogger { Ok(()) } + fn record_response( + &self, + py: Python<'_>, + status: u16, + headers: &[(String, String)], + body: &str, + ) -> PyResult<()> { + Logging::RecordResponse.call(py, (self.object(py), status, headers.to_vec(), body))?; + Ok(()) + } + + fn record_api_window(&self, py: Python<'_>, start: f64, end: f64) -> PyResult<()> { + Logging::RecordApiWindow.call(py, (self.object(py), start, end))?; + Ok(()) + } + fn post_call( &self, py: Python<'_>, diff --git a/litellm-rust/crates/callbacks-legacy-python/src/python.rs b/litellm-rust/crates/callbacks-legacy-python/src/python.rs index 6f1ee9dfd78..6d910d35f53 100644 --- a/litellm-rust/crates/callbacks-legacy-python/src/python.rs +++ b/litellm-rust/crates/callbacks-legacy-python/src/python.rs @@ -44,6 +44,10 @@ pub(crate) enum Logging { PreCall, #[strum(serialize = "post_call")] PostCall, + #[strum(serialize = "record_response")] + RecordResponse, + #[strum(serialize = "record_api_window")] + RecordApiWindow, #[strum(serialize = "defers_async_logging")] DefersAsync, #[strum(serialize = "defer_success")] @@ -77,6 +81,8 @@ pub(crate) enum DeploymentHooks { pub(crate) enum Streaming { #[strum(serialize = "stream_opened")] Opened, + #[strum(serialize = "stamp_stream_head")] + StampHead, #[strum(serialize = "stream_success")] Success, #[strum(serialize = "stream_failure")] diff --git a/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs b/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs index 093ce60e510..8acfa7a898b 100644 --- a/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs +++ b/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs @@ -65,6 +65,12 @@ FAKES = { 'post_call': lambda logger, original_response, api_key, additional_args: logger.post_call( original_response, api_key, additional_args ), + 'record_response': lambda logger, status, headers, body: logger.model_call_details.update( + response_headers=dict(headers) + ), + 'record_api_window': lambda logger, start, end: logger.model_call_details.update( + llm_api_duration_ms=(end - start) * 1000 + ), 'defers_async_logging': lambda logger: bool(getattr(logger, '_defer_async_logging', False)), 'defer_success': lambda logger, pending: setattr(logger, '_native_pending_logging', pending), 'sync_success_for_async_call': lambda logger, response, start, end: logger.handle_sync_success_callbacks_for_async_calls( @@ -85,6 +91,7 @@ FAKES = { ), 'after_deployment_failure': lambda kwargs, error, call_type: kwargs['logger'].hook('failure', error, call_type), 'stream_opened': lambda logger, head: logger.record('stream_opened', head), + 'stamp_stream_head': lambda head, logger, kwargs, start, end: logger.record('stamp_stream_head', head), 'stream_success': lambda logger, url_route, endpoint_type, request_body, chunks, start, end, first_chunk: logger.record( 'stream_success', list(chunks) ), diff --git a/litellm-rust/crates/host-http/tests/serve.rs b/litellm-rust/crates/host-http/tests/serve.rs index 2ba39ad4cd2..fd1a8940b33 100644 --- a/litellm-rust/crates/host-http/tests/serve.rs +++ b/litellm-rust/crates/host-http/tests/serve.rs @@ -215,9 +215,7 @@ async fn projection_custom_operations_and_hooks_feed_the_http_response(intercept ) .await?; route_hooks - .after_provider_response(RawResponse { - body: wire.url.clone(), - }) + .after_provider_response(RawResponse::new(wire.url.clone())) .await?; Ok(CallOutput::Complete(Bytes::from(wire.url))) }, @@ -501,9 +499,7 @@ async fn hook_rejection_stops_execution_and_is_reported_once( move |_, _, route_hooks, _observations| async move { if event { route_hooks - .after_provider_response(RawResponse { - body: "response".into(), - }) + .after_provider_response(RawResponse::new("response".into())) .await?; } else { route_hooks @@ -637,9 +633,7 @@ async fn unary_calls_use_into_response_after_hooks_and_before_success(intercepto ) .await?; route_hooks - .after_provider_response(RawResponse { - body: wire.url.clone(), - }) + .after_provider_response(RawResponse::new(wire.url.clone())) .await?; Ok(CallOutput::Complete(json!({"url": wire.url}))) }, @@ -689,7 +683,7 @@ async fn unary_failure_preserves_the_error_without_converting( None, |_, _, route_hooks, _observations| async move { route_hooks - .after_provider_response(RawResponse { body: "raw".into() }) + .after_provider_response(RawResponse::new("raw".into())) .await?; Err(TestError::Provider) }, diff --git a/litellm-rust/crates/host-native/tests/driver.rs b/litellm-rust/crates/host-native/tests/driver.rs index c41c1e5a43b..44c30234328 100644 --- a/litellm-rust/crates/host-native/tests/driver.rs +++ b/litellm-rust/crates/host-native/tests/driver.rs @@ -170,9 +170,7 @@ fn dispatching_call() -> TestMachine { ) .await?; route_hooks - .after_provider_response(RawResponse { - body: wire.url.clone(), - }) + .after_provider_response(RawResponse::new(wire.url.clone())) .await?; Ok(CallOutput::Complete(wire.url)) }, @@ -660,9 +658,7 @@ async fn response_interception_waits_and_can_reject_after_observation(#[case] re "input", Some(sender), |_, _, interceptors, observers| async move { - let raw = RawResponse { - body: "provider response".into(), - }; + let raw = RawResponse::new("provider response".into()); observers.unwrap().emit(CallEvent::Execution( ExecutionEvent::ProviderResponseReceived { raw: raw.clone() }, )); diff --git a/litellm-rust/crates/host-python/tests/support/mod.rs b/litellm-rust/crates/host-python/tests/support/mod.rs index 003fce74309..15e864c0718 100644 --- a/litellm-rust/crates/host-python/tests/support/mod.rs +++ b/litellm-rust/crates/host-python/tests/support/mod.rs @@ -467,7 +467,7 @@ pub fn success_machine() -> MachineFactory { .before_provider_request(wire(), context()) .await?; host.interceptors - .after_provider_response(RawResponse { body: "raw".into() }) + .after_provider_response(RawResponse::new("raw".into())) .await?; Ok(format!("{projected}|{signed}|{}", wire.url)) }) diff --git a/litellm-rust/crates/host/src/interceptors.rs b/litellm-rust/crates/host/src/interceptors.rs index 39af7c410e0..9398ea4bad6 100644 --- a/litellm-rust/crates/host/src/interceptors.rs +++ b/litellm-rust/crates/host/src/interceptors.rs @@ -28,6 +28,28 @@ pub struct RequestContext { #[derive(Clone, Debug, PartialEq, Eq)] pub struct RawResponse { pub body: String, + /// The provider's HTTP status and headers, when the route forwards them to logging. + pub head: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ResponseHead { + pub status: u16, + pub headers: Vec<(String, String)>, +} + +impl RawResponse { + pub fn new(body: String) -> Self { + Self { body, head: None } + } + + #[must_use] + pub fn with_head(self, head: ResponseHead) -> Self { + Self { + head: Some(head), + ..self + } + } } #[derive(Clone, Debug, PartialEq, Eq)] @@ -186,7 +208,7 @@ mod tests { .await?; Interceptors::after_provider_response( &channel.interceptors, - RawResponse { body: "raw".into() }, + RawResponse::new("raw".into()), ) .await?; Ok((sent, ())) diff --git a/litellm-rust/crates/host/tests/observation.rs b/litellm-rust/crates/host/tests/observation.rs index 22872047d81..fba85b63bea 100644 --- a/litellm-rust/crates/host/tests/observation.rs +++ b/litellm-rust/crates/host/tests/observation.rs @@ -74,9 +74,7 @@ fn snapshots_do_not_retain_runtime_objects(capacity: NonZeroUsize, #[case] faile #[rstest] fn provider_snapshots_own_the_response_body(capacity: NonZeroUsize) { - let mut raw = RawResponse { - body: "provider response".into(), - }; + let mut raw = RawResponse::new("provider response".into()); let event: CallEvent<(), (), &RawResponse> = CallEvent::Execution(ExecutionEvent::ProviderResponseReceived { raw: &raw }); let expected = raw.clone(); diff --git a/litellm-rust/crates/http/src/error.rs b/litellm-rust/crates/http/src/error.rs index 29c8b988c0b..7078d4aba2a 100644 --- a/litellm-rust/crates/http/src/error.rs +++ b/litellm-rust/crates/http/src/error.rs @@ -58,8 +58,20 @@ pub(crate) enum Repr { Network(#[source] reqwest::Error), #[error("upstream request timed out")] Timeout(#[source] reqwest::Error), + /// No bytes arrived within the read timeout, which the caller enforces per read; `body` says + /// the response head had already arrived. + #[error("upstream request timed out")] + ReadTimeout { + #[source] + elapsed: Elapsed, + body: bool, + }, #[error("upstream request failed with status {status}: {body}")] - Status { status: u16, body: UpstreamBody }, + Status { + status: u16, + body: UpstreamBody, + headers: Vec<(String, String)>, + }, #[error("invalid WebSocket URL")] WebSocketUrl(#[source] url::ParseError), @@ -136,13 +148,59 @@ impl Repr { impl Error { pub fn upstream(status: u16, body: String) -> Self { + Self::upstream_with_headers(status, body, Vec::new()) + } + + /// An upstream failure that keeps the provider's response headers (`retry-after`, + /// `request-id`) for the caller. + pub fn upstream_with_headers( + status: u16, + body: String, + headers: Vec<(String, String)>, + ) -> Self { Repr::Status { status, body: UpstreamBody(body), + headers, } .into() } + /// The provider's response headers on an upstream failure; empty for every other error. + pub fn upstream_headers(&self) -> &[(String, String)] { + match &*self.0 { + Repr::Status { headers, .. } => headers, + _ => &[], + } + } + + /// A timeout reached before the connection was established, as opposed to while reading. + pub fn is_connect_timeout(&self) -> bool { + matches!(&*self.0, Repr::Timeout(error) if error.is_connect()) + } + + /// A read the caller timed itself (httpx's per-read timeout) ran out. + pub fn read_timeout(elapsed: Elapsed) -> Self { + Repr::ReadTimeout { + elapsed, + body: false, + } + .into() + } + + /// A read of the response body the caller timed itself ran out, after the head arrived. + pub fn body_read_timeout(elapsed: Elapsed) -> Self { + Repr::ReadTimeout { + elapsed, + body: true, + } + .into() + } + + pub fn is_body_read_timeout(&self) -> bool { + matches!(&*self.0, Repr::ReadTimeout { body: true, .. }) + } + pub fn computed_header(name: impl Into) -> Self { Repr::ComputedHeader(name.into()).into() } @@ -169,6 +227,7 @@ impl Error { match &*self.0 { Repr::Network(_) | Repr::Timeout(_) + | Repr::ReadTimeout { .. } | Repr::Status { .. } | Repr::Handshake(_) | Repr::HandshakeTimeout(_) @@ -227,6 +286,7 @@ impl Error { | Repr::Connect(_) | Repr::Network(_) | Repr::Timeout(_) + | Repr::ReadTimeout { .. } | Repr::Status { .. } | Repr::Handshake(_) | Repr::HandshakeTimeout(_) @@ -259,6 +319,7 @@ impl Error { | Repr::Connect(_) | Repr::Network(_) | Repr::Timeout(_) + | Repr::ReadTimeout { .. } | Repr::Status { .. } | Repr::WebSocketUrl(_) | Repr::WebSocketScheme(_) @@ -295,6 +356,7 @@ impl Error { | Repr::Connect(_) | Repr::Network(_) | Repr::Timeout(_) + | Repr::ReadTimeout { .. } | Repr::Status { .. } | Repr::WebSocketUrl(_) | Repr::WebSocketScheme(_) @@ -333,6 +395,7 @@ impl Error { | Repr::Signature(_) | Repr::Network(_) | Repr::Timeout(_) + | Repr::ReadTimeout { .. } | Repr::Status { .. } | Repr::WebSocketUrl(_) | Repr::WebSocketScheme(_) @@ -356,7 +419,10 @@ impl Error { pub fn is_timeout(&self) -> bool { match &*self.0 { - Repr::Timeout(_) | Repr::HandshakeTimeout(_) | Repr::DownloadTimeout(_) => true, + Repr::Timeout(_) + | Repr::ReadTimeout { .. } + | Repr::HandshakeTimeout(_) + | Repr::DownloadTimeout(_) => true, Repr::Read { .. } | Repr::InvalidPem { .. } | Repr::Client(_) @@ -392,7 +458,7 @@ impl Error { pub fn upstream_response(&self) -> Option<(u16, &str)> { match &*self.0 { - Repr::Status { status, body } => Some((*status, &body.0)), + Repr::Status { status, body, .. } => Some((*status, &body.0)), Repr::Read { .. } | Repr::InvalidPem { .. } | Repr::Client(_) @@ -408,6 +474,7 @@ impl Error { | Repr::Connect(_) | Repr::Network(_) | Repr::Timeout(_) + | Repr::ReadTimeout { .. } | Repr::WebSocketUrl(_) | Repr::WebSocketScheme(_) | Repr::WebSocketSchemeRewrite(_) @@ -431,7 +498,7 @@ impl Error { pub fn status(&self) -> Option { match &*self.0 { Repr::Status { status, .. } | Repr::DownloadStatus(status) => Some(*status), - Repr::Timeout(_) | Repr::DownloadTimeout(_) => Some(408), + Repr::Timeout(_) | Repr::ReadTimeout { .. } | Repr::DownloadTimeout(_) => Some(408), Repr::Read { .. } | Repr::InvalidPem { .. } | Repr::Client(_) @@ -482,6 +549,7 @@ impl Error { | Repr::Connect(_) | Repr::Network(_) | Repr::Timeout(_) + | Repr::ReadTimeout { .. } | Repr::Status { .. } | Repr::WebSocketUrl(_) | Repr::WebSocketScheme(_) diff --git a/litellm-rust/crates/inference-chat/src/handler.rs b/litellm-rust/crates/inference-chat/src/handler.rs index f948e8fffaf..cdcc3e911b9 100644 --- a/litellm-rust/crates/inference-chat/src/handler.rs +++ b/litellm-rust/crates/inference-chat/src/handler.rs @@ -108,7 +108,7 @@ pub(super) async fn execute( if !status.is_success() { return Err(litellm_http::Error::upstream(status.as_u16(), text).into()); } - let raw = RawResponse { body: text.clone() }; + let raw = RawResponse::new(text.clone()); if let Some(observers) = observers { observers.emit(litellm_host::lifecycle::CallEvent::Execution( ExecutionEvent::ProviderResponseReceived { raw: raw.clone() }, diff --git a/litellm-rust/crates/inference-messages/Cargo.toml b/litellm-rust/crates/inference-messages/Cargo.toml index ac52cb65a0d..0c5be52550d 100644 --- a/litellm-rust/crates/inference-messages/Cargo.toml +++ b/litellm-rust/crates/inference-messages/Cargo.toml @@ -24,6 +24,7 @@ litellm-tracing.workspace = true reqwest.workspace = true serde.workspace = true serde_json = { workspace = true, features = ["preserve_order"] } +tokio.workspace = true tracing.workspace = true [dev-dependencies] @@ -34,5 +35,4 @@ litellm-host-native.workspace = true litellm-http = { workspace = true, features = ["test-support"] } litellm-inference-testing.workspace = true rstest.workspace = true -tokio.workspace = true wiremock.workspace = true diff --git a/litellm-rust/crates/inference-messages/src/common_utils.rs b/litellm-rust/crates/inference-messages/src/common_utils.rs index a57eca1c6a9..e5ece2837d8 100644 --- a/litellm-rust/crates/inference-messages/src/common_utils.rs +++ b/litellm-rust/crates/inference-messages/src/common_utils.rs @@ -35,6 +35,14 @@ impl MessagesProvider { pub(crate) fn config(self) -> &'static dyn BaseMessagesConfig { self.config } + + /// Python's `MESSAGES_SERVICE_TIER_PROVIDERS`: everyone else has `service_tier` dropped. + pub(crate) fn accepts_service_tier(self) -> bool { + matches!( + self.provider, + LlmProviders::Anthropic | LlmProviders::AzureAi | LlmProviders::BedrockMantle + ) + } } /// Python's `get_provider_anthropic_messages_config`: Vertex AI serves only its Claude @@ -119,3 +127,69 @@ mod tests { ); } } + +/// Python's `_merge_messages_header_sources`: later sources win per header name, except +/// `anthropic-beta` (any casing), whose values from every source are unioned, deduplicated and +/// sorted into one lowercase `anthropic-beta` entry. +pub fn merge_messages_header_sources( + sources: impl IntoIterator>, +) -> Map { + let mut merged = Map::new(); + let mut betas = std::collections::BTreeSet::new(); + for (name, value) in sources.into_iter().flatten() { + if !name.eq_ignore_ascii_case("anthropic-beta") { + merged.insert(name, value); + continue; + } + let values: Vec<&str> = match &value { + Value::String(text) => vec![text.as_str()], + Value::Array(entries) => entries.iter().filter_map(Value::as_str).collect(), + _ => Vec::new(), + }; + betas.extend( + values + .iter() + .flat_map(|value| value.split(',')) + .map(str::trim) + .filter(|beta| !beta.is_empty()) + .map(str::to_owned), + ); + } + if !betas.is_empty() { + merged.insert( + "anthropic-beta".to_owned(), + Value::String(betas.into_iter().collect::>().join(",")), + ); + } + merged +} + +#[cfg(test)] +mod merge_tests { + use serde_json::{Map, Value, json}; + + use super::merge_messages_header_sources; + + fn map(value: Value) -> Map { + value.as_object().cloned().unwrap() + } + + #[test] + fn later_sources_win_except_anthropic_beta_which_is_unioned_and_sorted() { + let merged = merge_messages_header_sources([ + map(json!({"x-a": "1", "anthropic-beta": "b-2, a-1"})), + map(json!({"x-a": "2", "Anthropic-Beta": ["c-3", 7, "a-1"]})), + map(json!({"anthropic-beta": 5})), + ]); + assert_eq!( + Value::Object(merged), + json!({"x-a": "2", "anthropic-beta": "a-1,b-2,c-3"}) + ); + } + + #[test] + fn no_beta_values_leave_no_beta_header() { + let merged = merge_messages_header_sources([map(json!({"anthropic-beta": ""}))]); + assert!(merged.is_empty()); + } +} diff --git a/litellm-rust/crates/inference-messages/src/handler.rs b/litellm-rust/crates/inference-messages/src/handler.rs index 2e647688a1a..e0a7aa5adb8 100644 --- a/litellm-rust/crates/inference-messages/src/handler.rs +++ b/litellm-rust/crates/inference-messages/src/handler.rs @@ -10,9 +10,9 @@ use litellm_core_utils::{ get_provider_specific_headers::get_provider_specific_headers, settings::Lookup, }; use litellm_host::interceptors::{ - Cost, ExecutionFacts, Interceptors, ProviderIdentity, RequestContext, ResultSource, WireRequest, + Cost, ExecutionFacts, Interceptors, ProviderIdentity, RawResponse, RequestContext, + ResponseHead, ResultSource, WireRequest, }; -use litellm_http::request::with_default_headers; use litellm_llms::base_llm::{ auth::{Authenticated, ValidatedEnvironment, resolve_auth}, messages::{ @@ -21,14 +21,16 @@ use litellm_llms::base_llm::{ transformation::BaseMessagesConfig, }, }; -use litellm_llms_types::formats::messages::{MessagesRequest, MessagesResponse}; +use litellm_llms_types::formats::messages::{ + MessagesOptionalParams, MessagesRequest, MessagesResponse, +}; use litellm_secrets::source::Secrets; use litellm_tracing::ByteChunk; use serde_json::Value; use super::{ Error, MessagesCall, MessagesCallResponse, MessagesRoute, - common_utils::{MessagesProvider, string_headers}, + common_utils::{MessagesProvider, merge_messages_header_sources, string_headers}, constants::MESSAGES_TIMEOUT_SECS, route::Messages, types::invalid_request, @@ -126,14 +128,18 @@ async fn provider_call( MessagesRequest { model, ..body }, shaping.settings.reasoning_auto_summary, )?; + let transform_context = + MessagesTransformContext::new(shaping.capabilities, shaping.settings.drop_params); + let filtered = requested_optional_params(sanitized, provider, &transform_context)?; let trimmed = - without_additional_drop_params(sanitized, &shaping.settings.additional_drop_params)?; + without_additional_drop_params(filtered, &shaping.settings.additional_drop_params)?; let scoped = get_provider_specific_headers(provider_specific_header.as_ref(), provider.as_str()); - let forwarded = string_headers(Some( - extra_headers.into_iter().flatten().chain(scoped).collect(), - ))?; + let forwarded = string_headers(Some(merge_messages_header_sources([ + extra_headers.unwrap_or_default(), + scoped, + ])))?; let validated = config.validate_environment( forwarded, api_key.as_deref(), @@ -144,16 +150,13 @@ async fn provider_call( )?; let body = config.request_body( - config.transform_anthropic_messages_request( - trimmed, - &MessagesTransformContext::new(shaping.capabilities, shaping.settings.drop_params), - )?, + config.transform_anthropic_messages_request(trimmed, &transform_context)?, &validated.headers, shaping.settings.drop_params, )?; let environment = ValidatedEnvironment { headers: config.request_headers( - with_default_headers(validated.headers, config.default_headers()), + with_python_default_headers(validated.headers, config.default_headers()), &body, ), auth: validated.auth, @@ -184,7 +187,7 @@ async fn provider_call( .before_provider_request( WireRequest { url, - headers: authenticated.headers, + headers: joined_header_values(authenticated.headers), body: config.wire_body(serde_json::to_value(&body).map_err(serialize_failure)?), }, request_context, @@ -213,6 +216,64 @@ async fn provider_call( }) } +/// Python sends its header dict through `httpx.Headers`, which folds names that differ only in +/// case into one lowercase header whose values are joined with `", "`. +fn joined_header_values(headers: Vec<(String, String)>) -> Vec<(String, String)> { + let mut joined: Vec<(String, String)> = Vec::with_capacity(headers.len()); + let mut positions: std::collections::HashMap = + std::collections::HashMap::with_capacity(headers.len()); + for (name, value) in headers { + match positions.entry(name.to_ascii_lowercase()) { + std::collections::hash_map::Entry::Occupied(position) => { + let present = &mut joined[*position.get()].1; + present.push_str(", "); + present.push_str(&value); + } + std::collections::hash_map::Entry::Vacant(position) => { + position.insert(joined.len()); + joined.push((name, value)); + } + } + } + joined +} + +/// Python's `finalize_messages_headers` adds a default only when the exact lowercase name is +/// absent, so a caller's `Anthropic-Version` is sent alongside the default, not instead of it. +fn with_python_default_headers( + headers: Vec<(String, String)>, + defaults: &[(&str, &str)], +) -> Vec<(String, String)> { + let missing: Vec<(String, String)> = defaults + .iter() + .filter(|(name, _)| !headers.iter().any(|(present, _)| present == name)) + .map(|(name, value)| ((*name).to_string(), (*value).to_string())) + .collect(); + headers.into_iter().chain(missing).collect() +} + +/// Python's `get_requested_anthropic_messages_optional_param`, which gates speed, service tier +/// and sampling params before `additional_drop_params` and the environment check run. +fn requested_optional_params( + request: MessagesRequest, + provider: MessagesProvider, + context: &MessagesTransformContext, +) -> Result { + let request = litellm_llms::anthropic::messages::transformation::drop_unsupported_params( + request, context, + )?; + if provider.accepts_service_tier() { + return Ok(request); + } + Ok(MessagesRequest { + params: MessagesOptionalParams { + service_tier: None, + ..request.params + }, + ..request + }) +} + fn without_additional_drop_params( request: MessagesRequest, paths: &[String], @@ -245,6 +306,7 @@ async fn call_provider( } = request; let provider_name = provider.as_str(); log_request_body(provider_name, stream, &wire.body); + let timeout = timeout.unwrap_or(Duration::from_secs(MESSAGES_TIMEOUT_SECS)); let response = send( http, Authenticated { @@ -254,10 +316,11 @@ async fn call_provider( &wire.url, &wire.body, timeout, + stream, ) .await?; if !response.status().is_success() { - return Err(provider_error(response).await); + return Err(provider_error(response, timeout).await); } let config = provider.config(); if stream { @@ -265,11 +328,24 @@ async fn call_provider( response, config.stream_decoder(), provider_name, + timeout, )); } + let head = ResponseHead { + status: response.status().as_u16(), + headers: litellm_http::request::header_pairs(response.headers()), + }; + let status = head.status; let text = response.text().await.map_err(litellm_http::Error::from)?; log_response_body(&text); - context.response_received(&text).await?; + context + .raw_response_received(RawResponse::new(text.clone()).with_head(head)) + .await?; + if serde_json::from_str::(&text).is_err() { + // Python raises `AnthropicError(status_code=, message=)` for a + // body it cannot parse, and the public mapping classifies it by that status. + return Err(litellm_http::Error::upstream(status, text).into()); + } decode_response(config, &identity.model, &text) .map(|message| MessagesCallResponse::Complete(Box::new(message))) } @@ -281,30 +357,53 @@ fn serialize_failure(err: serde_json::Error) -> Error { )) } +/// Python's httpx timeout bounds each phase and each read, never the whole exchange. A complete +/// response keeps the client's overall deadline; a stream is bounded until its head arrives and +/// then per read (see [`timed_chunks`]), so a long stream that keeps sending is not cut off. async fn send( http: &litellm_http::Client, authenticated: Authenticated, url: &str, body: &Value, - timeout: Option, + timeout: Duration, + stream: bool, ) -> Result { - let request = outbound_request( - authenticated, - url, - body, - Some(timeout.unwrap_or(Duration::from_secs(MESSAGES_TIMEOUT_SECS))), - )?; - Ok(litellm_inference::outbound::send(request, http).await?) + let request = outbound_request(authenticated, url, body, (!stream).then_some(timeout))?; + if !stream { + return Ok(litellm_inference::outbound::send(request, http).await?); + } + tokio::time::timeout(timeout, litellm_inference::outbound::send(request, http)) + .await + .map_err(litellm_http::Error::read_timeout)? + .map_err(Error::from) } -async fn provider_error(response: reqwest::Response) -> Error { +/// The stream's body, each read bounded by `timeout` as httpx bounds each read. +fn timed_chunks( + response: reqwest::Response, + timeout: Duration, +) -> impl futures_util::Stream> + Send + 'static { + futures_util::stream::try_unfold(response, move |mut response| async move { + let chunk = tokio::time::timeout(timeout, response.chunk()) + .await + .map_err(litellm_http::Error::body_read_timeout)? + .map_err(litellm_http::Error::from)?; + Ok(chunk.map(|chunk| (chunk, response))) + }) +} + +/// The provider's error, its body read under `timeout`: a stream has no overall deadline once +/// its head arrives, so an error body that stalls would otherwise hold the call forever. +async fn provider_error(response: reqwest::Response, timeout: Duration) -> Error { let status = response.status().as_u16(); - match response.text().await { - Ok(text) => { + let headers = litellm_http::request::header_pairs(response.headers()); + match tokio::time::timeout(timeout, response.text()).await { + Ok(Ok(text)) => { log_error_body(status, &text); - litellm_http::Error::upstream(status, text).into() + litellm_http::Error::upstream_with_headers(status, text, headers).into() } - Err(error) => litellm_http::Error::from(error).into(), + Ok(Err(error)) => litellm_http::Error::from(error).into(), + Err(elapsed) => litellm_http::Error::read_timeout(elapsed).into(), } } @@ -329,22 +428,22 @@ fn streaming_response( response: reqwest::Response, decoder: Option, provider: &'static str, + timeout: Duration, ) -> MessagesCallResponse { let headers = response .headers() .iter() .filter_map(|(name, value)| Some((name.to_string(), value.to_str().ok()?.to_string()))) .collect(); + let chunks = timed_chunks(response, timeout) + .inspect_ok(move |chunk| log_chunk(provider, "provider_response", chunk)); let chunks = match decoder { - None => futures_util::stream::try_unfold(response, move |mut response| async move { - let chunk = response.chunk().await.map_err(litellm_http::Error::from)?; - Ok(chunk.map(|chunk| { - log_chunk(provider, "provider_response", &chunk); - (chunk, response) - })) - }) - .boxed(), - Some(decode) => decoded_chunks(response, decode, provider), + None => chunks.map_err(Error::from).boxed(), + Some(decode) => decoded_chunks( + chunks.map_err(std::io::Error::other).boxed(), + decode, + provider, + ), }; MessagesCallResponse::Stream { head: super::route::MessagesStreamHead { headers }, @@ -353,15 +452,10 @@ fn streaming_response( } fn decoded_chunks( - response: reqwest::Response, + bytes: ByteStream, decode: StreamDecoder, provider: &'static str, ) -> BoxStream<'static, Result> { - let bytes: ByteStream = response - .bytes_stream() - .inspect_ok(move |chunk| log_chunk(provider, "provider_response", chunk)) - .map_err(|error| std::io::Error::other(litellm_http::Error::from(error))) - .boxed(); futures_util::stream::try_unfold(decode(bytes), move |mut events| async move { let Some(event) = events.try_next().await? else { return Ok(None); @@ -964,9 +1058,12 @@ mod tests { .send() .await .unwrap(); - let MessagesCallResponse::Stream { mut chunks, .. } = - streaming_response(response, Some(streaming::event_stream), "test") - else { + let MessagesCallResponse::Stream { mut chunks, .. } = streaming_response( + response, + Some(streaming::event_stream), + "test", + Duration::from_secs(5), + ) else { panic!("a streaming response returns chunks"); }; diff --git a/litellm-rust/crates/inference-messages/src/lib.rs b/litellm-rust/crates/inference-messages/src/lib.rs index bfefb9c63b0..745442d0314 100644 --- a/litellm-rust/crates/inference-messages/src/lib.rs +++ b/litellm-rust/crates/inference-messages/src/lib.rs @@ -14,6 +14,7 @@ use std::sync::Arc; use crate::common_utils::messages_provider; +pub use common_utils::merge_messages_header_sources; pub use litellm_inference::Error; pub use types::{ MessagesCall, MessagesCallResponse, MessagesSettings, MessagesShaping, litellm_params, diff --git a/litellm-rust/crates/inference-messages/tests/messages/request.rs b/litellm-rust/crates/inference-messages/tests/messages/request.rs index e53bdb8d252..216aa96fbe6 100644 --- a/litellm-rust/crates/inference-messages/tests/messages/request.rs +++ b/litellm-rust/crates/inference-messages/tests/messages/request.rs @@ -440,7 +440,12 @@ async fn an_oauth_key_sends_the_browser_access_header_and_the_oauth_beta(call: M #[case::deepseek("deepseek")] #[case::minimax("minimax")] #[tokio::test] -async fn caller_protocol_headers_win_over_the_defaults(call: MessagesCall, #[case] provider: &str) { +/// Python adds a default unless that exact lowercase name is present, then folds the casings +/// into one header, so a capitalised caller header is sent beside the default. +async fn capitalised_caller_protocol_headers_are_joined_with_the_defaults( + call: MessagesCall, + #[case] provider: &str, +) { let upstream = upstream([message_response()]).await; run_message(MessagesCall { @@ -457,10 +462,13 @@ async fn caller_protocol_headers_win_over_the_defaults(call: MessagesCall, #[cas .await; let request = only_request(&upstream).await; - assert_eq!(request.header_values("anthropic-version"), ["2024-01-01"]); + assert_eq!( + request.header_values("anthropic-version"), + ["2024-01-01, 2023-06-01"] + ); assert_eq!( request.header_values("content-type"), - ["application/json; charset=utf-8"] + ["application/json; charset=utf-8, application/json"] ); } diff --git a/litellm-rust/crates/inference-messages/tests/messages/response.rs b/litellm-rust/crates/inference-messages/tests/messages/response.rs index 7aff1286858..76813fe0a34 100644 --- a/litellm-rust/crates/inference-messages/tests/messages/response.rs +++ b/litellm-rust/crates/inference-messages/tests/messages/response.rs @@ -225,14 +225,25 @@ async fn an_upstream_error_keeps_its_status_and_body(call: MessagesCall, #[case] } #[rstest] -#[case::not_json(ResponseTemplate::new(200).set_body_string("not json"))] -#[case::not_a_message(json_response(json!({"unexpected": true})))] #[tokio::test] -async fn an_unreadable_success_body_is_an_invalid_response( - call: MessagesCall, - #[case] response: ResponseTemplate, -) { - let upstream = upstream([response]).await; +async fn an_unparseable_success_body_is_reported_as_the_upstream_answer(call: MessagesCall) { + let upstream = upstream([ResponseTemplate::new(200).set_body_string("not json")]).await; + + let error = run(MessagesCall { + api_key: Some("sk".into()), + api_base: Some(upstream.uri()), + ..call + }) + .await + .expect_err("an unparseable body fails"); + + assert_eq!(error.upstream_response(), Some((200, "not json"))); +} + +#[rstest] +#[tokio::test] +async fn an_unreadable_success_body_is_an_invalid_response(call: MessagesCall) { + let upstream = upstream([json_response(json!({"unexpected": true}))]).await; let error = run(MessagesCall { api_key: Some("sk".into()), diff --git a/litellm-rust/crates/inference-messages/tests/messages/stream.rs b/litellm-rust/crates/inference-messages/tests/messages/stream.rs index 4a8a061f764..99ba08495aa 100644 --- a/litellm-rust/crates/inference-messages/tests/messages/stream.rs +++ b/litellm-rust/crates/inference-messages/tests/messages/stream.rs @@ -432,6 +432,49 @@ async fn dropping_the_sdk_stream_closes_the_unfinished_upstream( .unwrap(); } +/// Answers with an error head whose promised body never arrives. +async fn stalling_error_upstream() -> (String, JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let connection = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut request = vec![0; 4096]; + let _ = socket.read(&mut request).await; + socket + .write_all(b"HTTP/1.1 429 Too Many Requests\r\ncontent-type: application/json\r\ncontent-length: 100\r\n\r\n") + .await + .unwrap(); + let _ = socket.read_to_end(&mut Vec::new()).await; + }); + (base, connection) +} + +#[rstest] +#[tokio::test] +async fn the_timeout_covers_a_stalled_stream_error_body(call: MessagesCall) { + let (base, connection) = stalling_error_upstream().await; + let host = RecordingStreamHost::new( + MessagesCall { + timeout: Some(Duration::from_millis(300)), + ..streaming(call, base) + }, + usize::MAX, + ); + + let error = tokio::time::timeout(Duration::from_secs(5), stream_through(&host)) + .await + .expect("the stalled error body gives up within the timeout") + .expect_err("a stalled error body fails the call"); + + assert!(error.dispatched(), "{error:?}"); + assert_eq!(error.upstream_response(), None); + assert!(host.seen.into_inner().unwrap().is_empty()); + tokio::time::timeout(Duration::from_secs(5), connection) + .await + .expect("timing out closes the upstream connection") + .unwrap(); +} + #[rstest] #[tokio::test] async fn the_sdk_yields_a_body_error_once_after_delivered_chunks(call: MessagesCall) { diff --git a/litellm-rust/crates/inference-ocr/src/handler.rs b/litellm-rust/crates/inference-ocr/src/handler.rs index c126a192153..be23af60c69 100644 --- a/litellm-rust/crates/inference-ocr/src/handler.rs +++ b/litellm-rust/crates/inference-ocr/src/handler.rs @@ -91,9 +91,7 @@ impl> CallHooks for OcrCallHooks<'_, H> { } fn response_received<'a>(&'a self, body: &'a [u8]) -> BoxFuture<'a, Result<(), Error>> { - let raw = RawResponse { - body: String::from_utf8_lossy(body).into_owned(), - }; + let raw = RawResponse::new(String::from_utf8_lossy(body).into_owned()); if let Some(observers) = self.observers { observers.emit(litellm_host::lifecycle::CallEvent::Execution( ExecutionEvent::ProviderResponseReceived { raw: raw.clone() }, diff --git a/litellm-rust/crates/inference-responses/src/handler.rs b/litellm-rust/crates/inference-responses/src/handler.rs index 19092a63a14..f81a51ad192 100644 --- a/litellm-rust/crates/inference-responses/src/handler.rs +++ b/litellm-rust/crates/inference-responses/src/handler.rs @@ -114,7 +114,7 @@ pub(super) async fn execute( }); } let body = response.text().await.map_err(litellm_http::Error::from)?; - let raw = RawResponse { body: body.clone() }; + let raw = RawResponse::new(body.clone()); if let Some(observers) = observers { observers.emit(litellm_host::lifecycle::CallEvent::Execution( ExecutionEvent::ProviderResponseReceived { raw: raw.clone() }, diff --git a/litellm-rust/crates/inference/src/context.rs b/litellm-rust/crates/inference/src/context.rs index 963aa7b90c5..ef2635d24a1 100644 --- a/litellm-rust/crates/inference/src/context.rs +++ b/litellm-rust/crates/inference/src/context.rs @@ -42,9 +42,11 @@ impl<'a, I: Interceptors> CallContext<'a, I> { } pub async fn response_received(&self, body: &str) -> Result<(), Error> { - let raw = RawResponse { - body: body.to_owned(), - }; + self.raw_response_received(RawResponse::new(body.to_owned())) + .await + } + + pub async fn raw_response_received(&self, raw: RawResponse) -> Result<(), Error> { if let Some(observers) = &self.observers { observers.emit(CallEvent::Execution( ExecutionEvent::ProviderResponseReceived { raw: raw.clone() }, diff --git a/litellm-rust/crates/inference/src/error.rs b/litellm-rust/crates/inference/src/error.rs index a8e6fb4e1b3..841e0c53799 100644 --- a/litellm-rust/crates/inference/src/error.rs +++ b/litellm-rust/crates/inference/src/error.rs @@ -146,6 +146,34 @@ impl Error { } } + /// The provider's response headers on an upstream failure; empty for every other error. + pub fn upstream_headers(&self) -> &[(String, String)] { + match self { + Self::Llm(error) => error.upstream_headers(), + Self::Http(error) => error.upstream_headers(), + _ => &[], + } + } + + /// A read of a streamed body timed out after the response head arrived. + pub fn body_read_timeout(&self) -> bool { + match self { + Self::Llm(error) => error.is_body_read_timeout(), + Self::Http(error) => error.is_body_read_timeout(), + _ => false, + } + } + + /// `Some(connect)` when the call timed out, where `connect` says the connection was never + /// established; `None` for every other error. + pub fn timeout(&self) -> Option { + match self { + Self::Llm(error) => error.is_timeout(), + Self::Http(error) if error.is_timeout() => Some(error.is_connect_timeout()), + _ => None, + } + } + pub fn upstream_response(&self) -> Option<(u16, &str)> { match self { Self::Llm(error) => error.upstream_response(), diff --git a/litellm-rust/crates/llms-types/src/formats/messages/request.rs b/litellm-rust/crates/llms-types/src/formats/messages/request.rs index 9dbfa068025..d30d21e74e7 100644 --- a/litellm-rust/crates/llms-types/src/formats/messages/request.rs +++ b/litellm-rust/crates/llms-types/src/formats/messages/request.rs @@ -264,7 +264,12 @@ pub enum ThinkingDisplay { #[macro_rules_attribute::apply(crate::wire_type)] #[derive(Default)] pub struct EnabledThinking { - #[serde(default, skip_serializing_if = "Option::is_none")] + /// An explicit `null` stays `Unrecognized(Null)`: Python rejects it, but defaults a missing key to 0. + #[serde( + default, + skip_serializing_if = "Option::is_none", + deserialize_with = "crate::recognized::present" + )] pub budget_tokens: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] pub display: Option>, diff --git a/litellm-rust/crates/llms-types/src/formats/messages/response.rs b/litellm-rust/crates/llms-types/src/formats/messages/response.rs index 9b0c5d31b3a..276a30a6cb4 100644 --- a/litellm-rust/crates/llms-types/src/formats/messages/response.rs +++ b/litellm-rust/crates/llms-types/src/formats/messages/response.rs @@ -10,14 +10,29 @@ pub struct MessagesResponse { pub content: Vec, pub stop_reason: Option, pub stop_sequence: Option, - #[serde(skip_serializing_if = "Option::is_none")] + /// `Some(Value::Null)` when the provider sent an explicit `null`, which Python keeps. + #[serde( + default, + deserialize_with = "present_value", + skip_serializing_if = "Option::is_none" + )] pub usage: Option, - #[serde(skip_serializing_if = "Option::is_none")] + #[serde( + default, + deserialize_with = "present_value", + skip_serializing_if = "Option::is_none" + )] pub container: Option, #[serde(flatten)] pub extra: Map, } +fn present_value<'de, D: serde::Deserializer<'de>>( + deserializer: D, +) -> Result, D::Error> { + >::deserialize(deserializer).map(Some) +} + #[cfg(test)] mod tests { use super::*; @@ -73,4 +88,21 @@ mod tests { assert_eq!(body.get("usage").cloned(), usage); assert_eq!(body.get("container").cloned(), container); } + + #[test] + fn explicit_null_usage_and_container_survive_a_round_trip() { + let body = json!({"id": "msg_1", "type": "message", "role": "assistant", "model": "m", + "content": [], "stop_reason": null, "stop_sequence": null, + "usage": null, "container": null}); + let parsed: MessagesResponse = serde_json::from_value(body.clone()).expect("parses"); + assert_eq!(serde_json::to_value(parsed).expect("serializable"), body); + } + + #[test] + fn absent_usage_and_container_stay_absent() { + let body = json!({"id": "msg_1", "type": "message", "role": "assistant", "model": "m", + "content": [], "stop_reason": null, "stop_sequence": null}); + let parsed: MessagesResponse = serde_json::from_value(body.clone()).expect("parses"); + assert_eq!(serde_json::to_value(parsed).expect("serializable"), body); + } } diff --git a/litellm-rust/crates/llms-types/src/recognized.rs b/litellm-rust/crates/llms-types/src/recognized.rs index 58e8dd7a2e7..0545389afbc 100644 --- a/litellm-rust/crates/llms-types/src/recognized.rs +++ b/litellm-rust/crates/llms-types/src/recognized.rs @@ -7,6 +7,16 @@ pub enum Recognized { Unrecognized(Value), } +/// `deserialize_with` for an optional `Recognized` field whose explicit `null` must stay +/// distinguishable from an absent key: `null` becomes `Some(Unrecognized(Null))`. +pub fn present<'de, D, T>(deserializer: D) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, + Recognized: serde::Deserialize<'de>, +{ + as serde::Deserialize<'de>>::deserialize(deserializer).map(Some) +} + impl Recognized { pub fn known(&self) -> Option<&T> { match self { diff --git a/litellm-rust/crates/llms/src/anthropic/common_utils.rs b/litellm-rust/crates/llms/src/anthropic/common_utils.rs index aec5d9ebc27..13653f438f7 100644 --- a/litellm-rust/crates/llms/src/anthropic/common_utils.rs +++ b/litellm-rust/crates/llms/src/anthropic/common_utils.rs @@ -154,11 +154,42 @@ pub fn complete_anthropic_url( ) -> String { let api_base = resolve_anthropic_api_base(api_base, env_lookup); - let api_base = api_base.trim_end_matches('/'); - if api_base.ends_with(MESSAGES_PATH) { + ensure_api_route(&api_base, MESSAGES_PATH) +} + +/// Python's `urlsplit(url).path`, for the URLs an `api_base` holds. +fn url_path(url: &str) -> &str { + let before_fragment = url.split('#').next().unwrap_or_default(); + let before_query = before_fragment.split('?').next().unwrap_or_default(); + match before_query.split_once("://") { + Some((_, rest)) => rest.find('/').map_or("", |slash| &rest[slash..]), + None => before_query, + } +} + +/// Python's `_strip_trailing_slashes_keeping_the_scheme`. +fn strip_trailing_slashes(api_base: &str) -> String { + let stripped = api_base.trim_end_matches('/'); + if stripped.ends_with(':') && !stripped.contains('/') && !stripped.contains('?') { return api_base.to_string(); } - format!("{api_base}{MESSAGES_PATH}") + match stripped.split_once('?') { + Some((before, query)) if url_path(stripped).ends_with('/') => { + format!("{}?{query}", before.trim_end_matches('/')) + } + _ => stripped.to_string(), + } +} + +/// Python's `ensure_api_route`: an `api_base` whose path already ends in the route is kept as +/// it is, query string included; anything else gets the route appended to the whole string. +fn ensure_api_route(api_base: &str, route: &str) -> String { + let base = strip_trailing_slashes(api_base); + let route = route.trim_start_matches('/'); + if url_path(&base).ends_with(&format!("/{route}")) { + return base; + } + format!("{}/{route}", strip_trailing_slashes(&base)) } pub fn existing_betas(headers: &[(String, String)]) -> BetaSet { @@ -1922,4 +1953,24 @@ mod tests { expected ); } + + #[rstest] + #[case::bare_host("https://api.example.com", "https://api.example.com/v1/messages")] + #[case::trailing_slash("https://api.example.com/", "https://api.example.com/v1/messages")] + #[case::route_present( + "https://api.example.com/v1/messages/", + "https://api.example.com/v1/messages" + )] + #[case::route_and_query( + "https://api.example.com/v1/messages?beta=true", + "https://api.example.com/v1/messages?beta=true" + )] + #[case::query_without_route( + "https://api.example.com/?x=1", + "https://api.example.com?x=1/v1/messages" + )] + #[case::scheme_only("https://", "https:///v1/messages")] + fn the_messages_url_follows_python_ensure_api_route(#[case] api_base: &str, #[case] url: &str) { + assert_eq!(complete_anthropic_url(Some(api_base), &|_| None), url); + } } diff --git a/litellm-rust/crates/llms/src/anthropic/messages/thinking.rs b/litellm-rust/crates/llms/src/anthropic/messages/thinking.rs index df1ae13fd28..ae3488a21f9 100644 --- a/litellm-rust/crates/llms/src/anthropic/messages/thinking.rs +++ b/litellm-rust/crates/llms/src/anthropic/messages/thinking.rs @@ -2,8 +2,8 @@ use litellm_llms_types::{ formats::{ chat_completions::ReasoningEffort, messages::{ - EffortLevel, MessagesOptionalParams, MessagesRequest, OutputConfig, ThinkingConfig, - ThinkingDisplay, + AdaptiveThinking, EffortLevel, MessagesOptionalParams, MessagesRequest, OutputConfig, + ThinkingConfig, ThinkingDisplay, }, }, recognized::Recognized, @@ -36,16 +36,17 @@ fn budget_for_effort(budgets: &ThinkingBudgets, effort: ReasoningEffort) -> Opti fn effort_for_budget( budgets: &ThinkingBudgets, - budget_tokens: u64, + budget_tokens: i64, capabilities: &MessagesModelCapabilities, ) -> EffortLevel { - if budget_tokens >= budgets.xhigh && capabilities.effort_tiers.xhigh { + let reaches = |tier: u64| i128::from(budget_tokens) >= i128::from(tier); + if reaches(budgets.xhigh) && capabilities.effort_tiers.xhigh { return EffortLevel::Xhigh; } - if budget_tokens >= budgets.high { + if reaches(budgets.high) { return EffortLevel::High; } - if budget_tokens >= budgets.medium { + if reaches(budgets.medium) { return EffortLevel::Medium; } EffortLevel::Low @@ -231,32 +232,129 @@ fn drop_disabled_thinking(request: MessagesRequest, context: &ThinkingContext) - } } +/// `_numeric_budget_tokens`: Python reads any int, float (truncated) or int-parseable string +/// as a budget, and nothing else. +fn numeric_budget_tokens(budget: &Recognized) -> Option { + let value = match budget { + Recognized::Known(budget) => return Some(i64::try_from(*budget).unwrap_or(i64::MAX)), + Recognized::Unrecognized(value) => value, + }; + match value { + Value::Number(number) => number + .as_i64() + .or_else(|| number.as_f64().map(|float| float.trunc() as i64)), + Value::String(text) => python_int(text), + Value::Null | Value::Bool(_) | Value::Array(_) | Value::Object(_) => None, + } +} + +/// Python's `int(str)`: surrounding whitespace, a sign, and single underscores between digits. +fn python_int(text: &str) -> Option { + let trimmed = text.trim(); + let digits = trimmed.trim_start_matches(['+', '-']); + let well_formed = trimmed.len() - digits.len() <= 1 + && !digits.is_empty() + && !digits.starts_with('_') + && !digits.ends_with('_') + && !digits.contains("__") + && digits.chars().all(|ch| ch.is_ascii_digit() || ch == '_'); + if !well_formed { + return None; + } + let cleaned: String = trimmed.chars().filter(|ch| *ch != '_').collect(); + cleaned.parse::().ok().or_else(|| { + Some(if cleaned.starts_with('-') { + i64::MIN + } else { + i64::MAX + }) + }) +} + +fn python_type_name(value: &Value) -> &'static str { + match value { + Value::Null => "NoneType", + Value::Bool(_) => "bool", + Value::Number(number) if number.is_f64() => "float", + Value::Number(_) => "int", + Value::String(_) => "str", + Value::Array(_) => "list", + Value::Object(_) => "dict", + } +} + +fn kept_display( + display: Option<&Recognized>, + context: &ThinkingContext, +) -> Option> { + match display? { + Recognized::Known(ThinkingDisplay::Summarized | ThinkingDisplay::Omitted) => { + display.cloned() + } + Recognized::Known(ThinkingDisplay::Updates) if context.keeps_display_updates => { + display.cloned() + } + Recognized::Known(ThinkingDisplay::Updates) | Recognized::Unrecognized(_) => None, + } +} + +/// Keys Python's `carried_thinking_keys` keeps on the rebuilt adaptive config: the adaptive +/// keys always, and any other key unless `drop_params` is set. +fn carried_thinking_keys( + extra: &serde_json::Map, + context: &ThinkingContext, +) -> serde_json::Map { + extra + .iter() + .filter(|(key, _)| key.as_str() == "block_binding" || !context.drop_params) + .map(|(key, value)| (key.clone(), value.clone())) + .collect() +} + +/// `AnthropicModelInfo.translate_legacy_thinking_for_adaptive_model_silently`. fn translate_legacy_thinking_for_adaptive_model( request: MessagesRequest, context: &ThinkingContext, -) -> MessagesRequest { +) -> Result { let capabilities = &context.capabilities; if !capabilities.supports_adaptive_thinking || capabilities.supports_legacy_thinking { - return request; + return Ok(request); } let Some(ThinkingConfig::Enabled(enabled)) = known_thinking(&request) else { - return request; + return Ok(request); }; - let budget = enabled - .budget_tokens - .as_ref() - .and_then(Recognized::known) - .copied() - .unwrap_or(0); - let level = effort_for_budget(&context.budgets, budget, capabilities); - MessagesRequest { + let budget = match &enabled.budget_tokens { + None => Some(0), + Some(budget) => numeric_budget_tokens(budget), + }; + if budget.is_none() && !context.drop_params { + let actual = match &enabled.budget_tokens { + Some(Recognized::Unrecognized(value)) => python_type_name(value), + Some(Recognized::Known(_)) | None => "int", + }; + return Err(Error::InvalidRequest( + crate::ErrorDetail::NonNumericThinkingBudget { actual }, + )); + } + let thinking = ThinkingConfig::Adaptive(AdaptiveThinking { + display: kept_display(enabled.display.as_ref(), context), + extra: carried_thinking_keys(&enabled.extra, context), + }); + let output_config = match budget { + Some(budget) => with_default_effort( + request.params.output_config, + effort_for_budget(&context.budgets, budget, capabilities), + ), + None => request.params.output_config, + }; + Ok(MessagesRequest { params: MessagesOptionalParams { - thinking: Some(Recognized::Known(ThinkingConfig::adaptive(None))), - output_config: with_default_effort(request.params.output_config, level), + thinking: Some(Recognized::Known(thinking)), + output_config, ..request.params }, ..request - } + }) } fn translate_adaptive_effort_for_non_adaptive_model( @@ -337,7 +435,7 @@ pub fn translate_thinking( ) -> Result { let request = translate_reasoning_effort(request, context)?; let request = drop_disabled_thinking(request, context); - let request = translate_legacy_thinking_for_adaptive_model(request, context); + let request = translate_legacy_thinking_for_adaptive_model(request, context)?; let request = translate_adaptive_effort_for_non_adaptive_model(request, context)?; Ok(drop_incompatible_temperature_for_thinking(request, context)) } @@ -363,6 +461,7 @@ mod tests { ThinkingContext { capabilities, budgets: ThinkingBudgets::default(), + ..ThinkingContext::default() } } @@ -1239,10 +1338,102 @@ mod tests { let context = ThinkingContext { capabilities, budgets: overridden_budgets(overrides), + ..ThinkingContext::default() }; assert_eq!( translate_thinking(request(input), &context), Ok(request(expected)) ); } + + use serde_json::json; + + fn legacy_context(drop_params: bool, keeps_display_updates: bool) -> ThinkingContext { + ThinkingContext { + drop_params, + keeps_display_updates, + ..context(opus_4_7()) + } + } + + #[rstest] + #[case::float_budget(json!(8192.0), "xhigh")] + #[case::float_below_a_tier(json!(1023.9), "low")] + #[case::string_budget(json!(" 16_000 "), "xhigh")] + #[case::negative_budget(json!(-5), "low")] + fn legacy_budgets_read_like_python_numbers(#[case] budget: Value, #[case] effort: &str) { + let translated = translate_thinking( + request( + json!({"max_tokens": 64000, "thinking": {"type": "enabled", "budget_tokens": budget}}), + ), + &legacy_context(false, false), + ); + assert_eq!( + translated, + Ok(request(json!({ + "max_tokens": 64000, + "thinking": {"type": "adaptive"}, + "output_config": {"effort": effort}, + }))) + ); + } + + #[rstest] + #[case::text(json!("lots"), "str")] + #[case::list(json!([1]), "list")] + #[case::null(Value::Null, "NoneType")] + #[case::boolean(json!(true), "bool")] + fn a_non_numeric_legacy_budget_is_rejected_unless_dropped( + #[case] budget: Value, + #[case] actual: &'static str, + ) { + let fields = + json!({"max_tokens": 8192, "thinking": {"type": "enabled", "budget_tokens": budget}}); + assert_eq!( + translate_thinking(request(fields.clone()), &legacy_context(false, false)), + Err(Error::InvalidRequest( + crate::ErrorDetail::NonNumericThinkingBudget { actual } + )) + ); + assert_eq!( + translate_thinking(request(fields), &legacy_context(true, false)), + Ok(request( + json!({"max_tokens": 8192, "thinking": {"type": "adaptive"}}) + )) + ); + } + + #[rstest] + #[case::summarized("summarized", false, Some("summarized"))] + #[case::updates_kept("updates", true, Some("updates"))] + #[case::updates_dropped("updates", false, None)] + #[case::unknown("verbose", true, None)] + fn legacy_display_survives_only_where_python_keeps_it( + #[case] display: &str, + #[case] keeps_display_updates: bool, + #[case] kept: Option<&str>, + ) { + let translated = translate_thinking( + request(json!({"max_tokens": 4096, "thinking": {"type": "enabled", "budget_tokens": 2048, "display": display}})), + &legacy_context(false, keeps_display_updates), + ) + .unwrap(); + let thinking = serde_json::to_value(translated.params.thinking).unwrap(); + assert_eq!(thinking.get("display").and_then(Value::as_str), kept); + } + + #[rstest] + #[case::kept(false, json!({"type": "adaptive", "block_binding": "strict", "note": "x"}))] + #[case::dropped(true, json!({"type": "adaptive", "block_binding": "strict"}))] + fn legacy_extra_keys_follow_drop_params(#[case] drop_params: bool, #[case] expected: Value) { + let translated = translate_thinking( + request(json!({"max_tokens": 4096, "thinking": {"type": "enabled", "budget_tokens": 2048, "block_binding": "strict", "note": "x"}})), + &legacy_context(drop_params, false), + ) + .unwrap(); + assert_eq!( + serde_json::to_value(translated.params.thinking).unwrap(), + expected + ); + } } diff --git a/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs b/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs index fb6d7259bf9..18789e5b2be 100644 --- a/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs @@ -56,7 +56,7 @@ impl BaseMessagesConfig for AnthropicMessagesConfig { request: MessagesRequest, context: &MessagesTransformContext, ) -> Result { - transform_messages_request(request, context) + transform_messages_request(request, &context.keeping_display_updates()) } fn secret_names(&self) -> &'static [&'static str] { @@ -206,6 +206,8 @@ fn feature_betas(request: &MessagesRequest) -> BetaSet { .then_some(AnthropicBeta::PerTurnControl20260701), uses_thinking_display_updates(params) .then_some(AnthropicBeta::ThinkingDisplayUpdates20260818), + uses_thinking_block_binding(params) + .then_some(AnthropicBeta::ThinkingBindingControls20260801), messages_carry_tool_changes(&request.messages) .then_some(AnthropicBeta::MidConversationToolChanges20260701), has_advisor_tool(tools).then_some(AnthropicBeta::AdvisorTool20260301), @@ -267,6 +269,25 @@ fn uses_thinking_display_updates(params: &MessagesOptionalParams) -> bool { display.and_then(Recognized::known) == Some(&ThinkingDisplay::Updates) } +/// `AnthropicModelInfo.thinking_betas`: any thinking object carrying `block_binding`. +fn uses_thinking_block_binding(params: &MessagesOptionalParams) -> bool { + match params.thinking.as_ref() { + Some(Recognized::Known(ThinkingConfig::Enabled(thinking))) => { + thinking.extra.contains_key("block_binding") + } + Some(Recognized::Known(ThinkingConfig::Adaptive(thinking))) => { + thinking.extra.contains_key("block_binding") + } + Some(Recognized::Known(ThinkingConfig::Disabled(thinking))) => { + thinking.extra.contains_key("block_binding") + } + Some(Recognized::Unrecognized(Value::Object(thinking))) => { + thinking.contains_key("block_binding") + } + Some(Recognized::Unrecognized(_)) | None => false, + } +} + fn is_tool_reference_change(block: &ContentBlock) -> bool { matches!( &block.block_type, @@ -297,7 +318,8 @@ fn unsupported_param(model: &str, param: &str, value: &str, hint: &str) -> Error }) } -fn drop_unsupported_params( +/// The speed and sampling gates of Python's `get_requested_anthropic_messages_optional_param`. +pub fn drop_unsupported_params( request: MessagesRequest, context: &MessagesTransformContext, ) -> Result { @@ -838,6 +860,8 @@ mod tests { low: 2000, ..ThinkingBudgets::default() }, + drop_params: true, + keeps_display_updates: false, }, drop_params: true, } diff --git a/litellm-rust/crates/llms/src/azure_ai/messages/transformation.rs b/litellm-rust/crates/llms/src/azure_ai/messages/transformation.rs index a2ba2cee0d3..d13cfa81a94 100644 --- a/litellm-rust/crates/llms/src/azure_ai/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/azure_ai/messages/transformation.rs @@ -68,7 +68,7 @@ impl BaseMessagesConfig for AzureAnthropicMessagesConfig { .capabilities .supports_mid_conversation_system, )), - context, + &context.keeping_display_updates(), ) } diff --git a/litellm-rust/crates/llms/src/base_llm/messages/context.rs b/litellm-rust/crates/llms/src/base_llm/messages/context.rs index e59a63813b0..b514e975aa4 100644 --- a/litellm-rust/crates/llms/src/base_llm/messages/context.rs +++ b/litellm-rust/crates/llms/src/base_llm/messages/context.rs @@ -110,6 +110,11 @@ impl ThinkingBudgets { pub struct ThinkingContext { pub capabilities: MessagesModelCapabilities, pub budgets: ThinkingBudgets, + /// `drop_params` as Python's thinking translation reads it. + pub drop_params: bool, + /// The provider accepts `display: "updates"` on thinking translated from the legacy shape + /// (Python's `LEGACY_THINKING_DISPLAY_UPDATES_PROVIDERS`). + pub keeps_display_updates: bool, } #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct MessagesTransformContext { @@ -130,8 +135,22 @@ impl MessagesTransformContext { thinking: ThinkingContext { capabilities, budgets: ThinkingBudgets::from_lookup(env), + drop_params, + keeps_display_updates: false, }, drop_params, } } + + /// The same context for a provider in Python's `LEGACY_THINKING_DISPLAY_UPDATES_PROVIDERS`. + #[must_use] + pub fn keeping_display_updates(&self) -> Self { + Self { + thinking: ThinkingContext { + keeps_display_updates: true, + ..self.thinking + }, + ..*self + } + } } diff --git a/litellm-rust/crates/llms/src/error.rs b/litellm-rust/crates/llms/src/error.rs index b3d1421083a..2844fbf3e01 100644 --- a/litellm-rust/crates/llms/src/error.rs +++ b/litellm-rust/crates/llms/src/error.rs @@ -92,6 +92,24 @@ impl Error { } } + pub fn upstream_headers(&self) -> &[(String, String)] { + match self { + Self::Http(error) => error.upstream_headers(), + _ => &[], + } + } + + pub fn is_body_read_timeout(&self) -> bool { + matches!(self, Self::Http(error) if error.is_body_read_timeout()) + } + + pub fn is_timeout(&self) -> Option { + match self { + Self::Http(error) if error.is_timeout() => Some(error.is_connect_timeout()), + _ => None, + } + } + pub fn upstream_response(&self) -> Option<(u16, &str)> { match self { Self::Http(error) => error.upstream_response(), @@ -158,6 +176,10 @@ pub enum ErrorDetail { value: String, hint: String, }, + #[error( + "thinking.budget_tokens must be a number, got {actual}. To drop it, set `litellm.drop_params=True`, or `drop_params: true` under `litellm_settings` on the proxy" + )] + NonNumericThinkingBudget { actual: &'static str }, #[error( "Missing API Base - A call is being made to LLM Provider but no api base is set either in the environment variables ({{LLM_PROVIDER}}_API_KEY) or via params" )] diff --git a/litellm-rust/crates/llms/src/vertex_ai/messages/transformation.rs b/litellm-rust/crates/llms/src/vertex_ai/messages/transformation.rs index e5e3a056fab..038617c2e10 100644 --- a/litellm-rust/crates/llms/src/vertex_ai/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/vertex_ai/messages/transformation.rs @@ -77,7 +77,7 @@ impl BaseMessagesConfig for VertexAiPartnerModelsAnthropicMessagesConfig { request: MessagesRequest, context: &MessagesTransformContext, ) -> Result { - let request = transform_messages_request(request, context)?; + let request = transform_messages_request(request, &context.keeping_display_updates())?; let request = strip_cache_control_scope(normalize_system_role_messages( request, context diff --git a/litellm-rust/crates/python-bridge/src/errors.rs b/litellm-rust/crates/python-bridge/src/errors.rs index b42c08b7e78..426cd861729 100644 --- a/litellm-rust/crates/python-bridge/src/errors.rs +++ b/litellm-rust/crates/python-bridge/src/errors.rs @@ -23,6 +23,17 @@ pyo3::create_exception!( ); const MESSAGES_REQUEST_ERROR: &str = "messages_request_error"; +/// The rejection is a parameter the model does not support, which Python raises as +/// `UnsupportedParamsError`. +const MESSAGES_UNSUPPORTED_PARAM: &str = "messages_unsupported_param"; +/// The phase label (`Connection` or `Read`) of a timed-out call, for Python's timeout text, or +/// `Body` for a streamed body read that timed out after the head arrived. +const MESSAGES_TIMEOUT: &str = "messages_timeout"; +/// No credential reached the provider, which Python raises as `AuthenticationError`. +const MESSAGES_MISSING_CREDENTIAL: &str = "messages_missing_credential"; +/// Python refuses the request only once its logger knows the resolved model and provider (the +/// provider config's request transform), so the failure logs them. +const MESSAGES_LATE_REJECTION: &str = "messages_late_rejection"; pub(crate) fn upstream_error( py: Python<'_>, @@ -78,17 +89,57 @@ pub(crate) fn route_failure( return Ok(raised); } if let Some((status, body)) = error.upstream_response() { - return public(upstream_error(py, status, body, &HeaderMap::new())?); + let headers = error + .upstream_headers() + .iter() + .filter_map(|(name, value)| { + Some(( + reqwest::header::HeaderName::from_bytes(name.as_bytes()).ok()?, + reqwest::header::HeaderValue::from_str(value).ok()?, + )) + }) + .collect::(); + return public(upstream_error(py, status, body, &headers)?); } let native = match (marker, error.rejected_request()) { (RequestMarker::Messages, Some(rejection)) => { let native = PyValueError::new_err(Report::new(rejection).to_string()); - native.value(py).setattr(MESSAGES_REQUEST_ERROR, true)?; + let value = native.value(py); + value.setattr(MESSAGES_REQUEST_ERROR, true)?; + if matches!( + rejection.downcast_ref::(), + Some( + litellm_llms::ErrorDetail::UnsupportedParameter { .. } + | litellm_llms::ErrorDetail::NonNumericThinkingBudget { .. } + ) + ) { + value.setattr(MESSAGES_UNSUPPORTED_PARAM, true)?; + } + if matches!( + rejection.downcast_ref::(), + Some(litellm_llms::ErrorDetail::NonNumericThinkingBudget { .. }) + ) { + value.setattr(MESSAGES_LATE_REJECTION, true)?; + } native } - (RequestMarker::Messages, None) | (RequestMarker::Unmarked, _) => { - by_fault(error.is_request(), &error) + (RequestMarker::Messages, None) => { + let native = by_fault(error.is_request(), &error); + let value = native.value(py); + if let Some(connect) = error.timeout() { + let phase = match (connect, error.body_read_timeout()) { + (true, _) => "Connection", + (false, true) => "Body", + (false, false) => "Read", + }; + value.setattr(MESSAGES_TIMEOUT, phase)?; + } + if error.is_missing_credential() { + value.setattr(MESSAGES_MISSING_CREDENTIAL, true)?; + } + native } + (RequestMarker::Unmarked, _) => by_fault(error.is_request(), &error), }; public(native) } diff --git a/litellm-rust/crates/python-bridge/src/routes/messages/host.rs b/litellm-rust/crates/python-bridge/src/routes/messages/host.rs index 27ce8c943f0..af614f17360 100644 --- a/litellm-rust/crates/python-bridge/src/routes/messages/host.rs +++ b/litellm-rust/crates/python-bridge/src/routes/messages/host.rs @@ -34,11 +34,9 @@ fn merge_headers( forwarded: Option>, extra_headers: Option>, ) -> Option> { - let merged: Map = forwarded - .into_iter() - .flatten() - .chain(extra_headers.into_iter().flatten()) - .collect(); + let merged = litellm_inference_messages::merge_messages_header_sources( + forwarded.into_iter().chain(extra_headers), + ); (!merged.is_empty()).then_some(merged) } @@ -107,10 +105,19 @@ impl MessagesPythonHost { .map(|value| python_timeout_seconds(py, value.unbind())) .transpose()? .flatten(); - let custom_llm_provider = string("custom_llm_provider")?; - let shaping = self.shaping(py, &model, custom_llm_provider.as_deref(), arguments)?; let api_key = string("api_key")?; let api_base = string("api_base")?; + // Python's handler resolves a bare model name (`claude-sonnet-4-5`) through + // `get_llm_provider`'s model lists, which the native prefix lookup does not carry. + let custom_llm_provider = match string("custom_llm_provider")? { + Some(provider) => Some(provider), + None => py + .import(ROUTE_HOST_MODULE)? + .getattr("inferred_provider")? + .call1((&model, api_base.as_deref()))? + .extract()?, + }; + let shaping = self.shaping(py, &model, custom_llm_provider.as_deref(), arguments)?; let extra_headers = self.merged_headers(py, arguments)?; let provider_specific_header = self.provider_specific_header(py, arguments)?; let litellm_params = project_litellm_params(argument, |name| module_global(py, name))?; diff --git a/litellm-rust/crates/python-bridge/src/routes/messages/mod.rs b/litellm-rust/crates/python-bridge/src/routes/messages/mod.rs index c788760b268..a9ae01bd7ee 100644 --- a/litellm-rust/crates/python-bridge/src/routes/messages/mod.rs +++ b/litellm-rust/crates/python-bridge/src/routes/messages/mod.rs @@ -50,9 +50,18 @@ fn run_messages(py: Python<'_>, call: NativeCall<'_>, asynchronous: bool) -> PyR ) } +/// Python's sync Messages handler raises for every native provider before it sends anything +/// or opens a logging object (`BaseLLMHTTPHandler.anthropic_messages_handler`), and Rust copies +/// that until both sides serve sync calls. +const SYNC_NOT_IMPLEMENTED: &str = "anthropic_messages_handler is not implemented for sync calls"; + #[pyfunction] -pub(crate) fn messages(py: Python<'_>, call: NativeCall<'_>) -> PyResult> { - run_messages(py, call, false) +#[pyo3(signature = (call))] +pub(crate) fn messages(call: NativeCall<'_>) -> PyResult> { + drop(call); + Err(pyo3::exceptions::PyValueError::new_err( + SYNC_NOT_IMPLEMENTED, + )) } #[pyfunction] diff --git a/litellm/rust_bridge/host/callbacks_legacy_python.py b/litellm/rust_bridge/host/callbacks_legacy_python.py index e155680bd30..92590531cdc 100644 --- a/litellm/rust_bridge/host/callbacks_legacy_python.py +++ b/litellm/rust_bridge/host/callbacks_legacy_python.py @@ -12,7 +12,7 @@ import datetime import traceback import uuid import weakref -from collections.abc import Awaitable, Coroutine, Mapping +from collections.abc import Awaitable, Coroutine, Mapping, Sequence from dataclasses import dataclass from typing import ( TYPE_CHECKING, @@ -22,6 +22,7 @@ from typing import ( cast, # noqa: TID251 # bounded compatibility calls into legacy Python integrations ) +import httpx from pydantic import TypeAdapter if TYPE_CHECKING: @@ -243,6 +244,27 @@ def post_call( logger.post_call(original_response=original_response, api_key=api_key, additional_args=additional_args) +def record_response(logger: Logging, status: int, headers: Sequence[tuple[str, str]], body: str) -> None: + """Stores the provider response the way Python's Messages handler does, for logging to replay.""" + response: Final = httpx.Response(status_code=status, headers=list(headers), content=body.encode()) + details: Final = logger.model_call_details # the call's own logger: recording on it is the point + details["httpx_response"] = response + details["response_headers"] = dict(response.headers) + + +def record_api_window(logger: Logging, start: float, end: float) -> None: + """Records the provider exchange's window the way Python's timed `AsyncHTTPHandler.post` does.""" + from litellm.litellm_core_utils.logging_utils import ( + _set_duration_in_model_call_details, # pyright: ignore[reportPrivateUsage] # the timed post's own recorder + ) + + _set_duration_in_model_call_details( + logging_obj=logger, + start_time=datetime.datetime.fromtimestamp(start), # noqa: DTZ006 # the timed post records naive local times + end_time=datetime.datetime.fromtimestamp(end), # noqa: DTZ006 # the timed post records naive local times + ) + + def defers_async_logging(logger: LoggingSurface) -> bool: return bool(getattr(logger, "defer_async_logging", False)) @@ -359,6 +381,27 @@ def after_deployment_failure(kwargs: dict[str, object], error: Exception, call_t _STREAM_HEADERS: Final = TypeAdapter(Mapping[str, str]) +class _HiddenParamsCarrier: + """Stands in for the stream `@client` stamps, so the head becomes its `_hidden_params`.""" + + def __init__(self, hidden_params: dict[str, object]) -> None: # mutable-ok: the stamps land in this dict + self._hidden_params: object = hidden_params + + +def stamp_stream_head( + head: dict[str, object], # mutable-ok: the stream's hidden params, stamped in place + logger: Logging, + kwargs: dict[str, object], # mutable-ok: `finalize` takes the call kwargs as a dict + start_time: datetime.datetime, + end_time: datetime.datetime, +) -> None: + carrier: Final = _HiddenParamsCarrier(head) + finalize(carrier, logger, kwargs, start_time, end_time) + stamped: Final = carrier._hidden_params # pyright: ignore[reportPrivateUsage] # the carrier's own field + if stamped is not head and isinstance(stamped, dict): + head.update(cast("dict[str, object]", stamped)) # cast-ok: update_response_metadata writes a str-keyed dict + + def stream_opened(logger: Logging, head: Mapping[str, object]) -> None: logger.stream = True logger.model_call_details["stream"] = True diff --git a/litellm/rust_bridge/host/routes/messages.py b/litellm/rust_bridge/host/routes/messages.py index 017ff65e2a7..c8b53097f46 100644 --- a/litellm/rust_bridge/host/routes/messages.py +++ b/litellm/rust_bridge/host/routes/messages.py @@ -1,21 +1,35 @@ from __future__ import annotations -from collections.abc import Mapping, Sequence +import datetime +from collections.abc import Callable, Mapping, Sequence from dataclasses import asdict, dataclass -from typing import Final, cast # noqa: TID251 # narrows the normalized native payload to the public TypedDict +from typing import Final, TypeAlias, cast # noqa: TID251 # narrows native payloads to public TypedDicts import httpx from pydantic import TypeAdapter, ValidationError import litellm +from litellm.litellm_core_utils import exception_mapping_utils from litellm.litellm_core_utils.core_helpers import normalize_drop_params +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.litellm_core_utils.reasoning_effort_capability import deployment_reasoning_effort_levels from litellm.litellm_core_utils.request_timeout_resolver import get_configured_request_timeout +from litellm.llms.anthropic.common_utils import AnthropicError +from litellm.llms.anthropic.messages.native import handler +from litellm.llms.anthropic.messages.native.handler import ( + _native_messages_config as native_messages_config, # pyright: ignore[reportPrivateUsage] # the handler's own config lookup +) +from litellm.llms.anthropic.messages.native.handler import ( + _with_deployment_output_effort as _deployment_output_effort, # pyright: ignore[reportPrivateUsage] # the handler's own effort rounding +) from litellm.llms.anthropic.messages.utils import is_reasoning_auto_summary_enabled from litellm.rust_bridge import failures +from litellm.rust_bridge.host import callbacks_legacy_python from litellm.rust_bridge.public_call import known_params, optional_str from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse _DROP_PATHS: Final = TypeAdapter(list[object]) +_METADATA: Final = TypeAdapter(dict[str, object]) _BODY_FIELDS: Final = frozenset( { "max_tokens", @@ -40,6 +54,7 @@ _BODY_FIELDS: Final = frozenset( "cache_control", "reasoning_effort", "safeguards", + "service_tier", } ) @@ -59,8 +74,40 @@ def response(value: Mapping[str, object]) -> AnthropicMessagesResponse: ) +def inferred_provider(model: str, api_base: str | None) -> str | None: + """The provider Python's handler resolves for a call that names none, or None when the model + carries no provider `get_llm_provider` knows.""" + try: + _, provider, _, _ = litellm.get_llm_provider(model=model, api_base=api_base) + except litellm.BadRequestError: + return None + return provider + + def body_params(request: Mapping[str, object]) -> Mapping[str, object]: - return known_params(_BODY_FIELDS, request) + params: Final = known_params(_BODY_FIELDS, request) + if deployment_reasoning_effort_levels(request.get("model_info")) is None: + return params + return _with_deployment_effort(params, request) + + +def _with_deployment_effort(params: Mapping[str, object], request: Mapping[str, object]) -> Mapping[str, object]: + """Python's handler rounds the requested effort to the levels the deployment declares in + `model_info.reasoning_effort_levels`, the one place a deployment, not the model, decides.""" + explicit_provider: Final = optional_str(request.get("custom_llm_provider")) + try: + model, provider, _, _ = litellm.get_llm_provider( + model=str(request["model"]), custom_llm_provider=explicit_provider + ) + except litellm.BadRequestError: + return params + return _deployment_output_effort( + params, + model=model, + custom_llm_provider=provider, + kwargs=request, + drop_params=litellm.drop_params is True or request.get("drop_params") is True, + ) def stream_hidden_params(headers: Sequence[tuple[str, str]]) -> Mapping[str, object]: @@ -71,18 +118,147 @@ def stream_hidden_params(headers: Sequence[tuple[str, str]]) -> Mapping[str, obj return anthropic_messages_stream_hidden_params(httpx.Headers(list(headers))) -def map_failure(error: Exception, request: Mapping[str, object], request_provider: str) -> Exception: - if getattr(error, "messages_request_error", False): - return litellm.BadRequestError( - message=str(error), - model=str(request["model"]).removeprefix(f"{request_provider}/"), - llm_provider=request_provider, - ) - return failures.map_native_failure( - error, str(request["model"]), request_provider, request, optional_str(request.get("api_base")) +_UPSTREAM_ARGS: Final = TypeAdapter(tuple[int, str]) +_UPSTREAM_HEADERS: Final = TypeAdapter(list[tuple[str, str]]) +_ERROR_CLASS: TypeAlias = Callable[..., Exception] +_validate_metadata: Final = cast( # cast-ok: the handler's untyped metadata validator + Callable[[dict[str, object]], object], handler.validate_anthropic_api_metadata +) +_exception_type: Final = cast( # cast-ok: the legacy public exception mapper is untyped + Callable[..., Exception], exception_mapping_utils.exception_type +) +_BODY_READ_TIMEOUT: Final = "Timeout on reading data from socket" # aiohttp's SocketTimeoutError text +_MISSING_KEY: Final = ( + "Missing Anthropic API Key - A call is being made to anthropic but no key is set " + "either in the environment variables or via params. Please set `ANTHROPIC_API_KEY` " + "or `ANTHROPIC_AUTH_TOKEN` in your environment vars" +) + + +def _rejection(error: Exception, request: Mapping[str, object], request_provider: str) -> Exception: + """The error Python's handler raises for a request it refuses before sending it.""" + metadata: Final = request.get("metadata") + if isinstance(metadata, dict): + try: + _validate_metadata(_METADATA.validate_python(metadata)) + except ValidationError as invalid: + return invalid + if getattr(error, "messages_unsupported_param", False): + return litellm.UnsupportedParamsError(message=str(error)) + return litellm.BadRequestError( + message=str(error), + model=str(request["model"]).removeprefix(f"{request_provider}/"), + llm_provider=request_provider, ) +def _timeout_text(phase: str, request: Mapping[str, object]) -> str: + """`litellm.Timeout`'s text as Python's HTTP client raises it for the same timeout.""" + seconds: Final = request.get("timeout") + logging_obj: Final = request.get("litellm_logging_obj") + start_time: Final = getattr(logging_obj, "start_time", None) + time_taken: Final = ( + round( + (datetime.datetime.now() - start_time).total_seconds(), # noqa: DTZ005 # the logger's start_time is naive local time + 3, + ) + if isinstance(start_time, datetime.datetime) + else None + ) + return str( + litellm.Timeout( + message=f"{phase} timed out. Timeout passed={seconds}, time taken={time_taken} seconds", + model="default-model-name", + llm_provider="litellm-httpx-handler", + ) + ) + + +def _upstream_failure( + error: Exception, request: Mapping[str, object] +) -> tuple[int, str, Sequence[tuple[str, str]]] | None: + """The status, text and headers Python's HTTP client reports for the same failure.""" + timeout_phase: Final = getattr(error, "messages_timeout", None) + if isinstance(timeout_phase, str): + return 408, _timeout_text(timeout_phase, request), () + try: + status, text = _UPSTREAM_ARGS.validate_python(error.args) + return status, text, _UPSTREAM_HEADERS.validate_python(getattr(error, "headers", None)) + except ValidationError: + return None + + +def _provider_error(error: Exception, request: Mapping[str, object], request_provider: str) -> Exception | None: + """The provider exception Python's `_handle_error` builds for the same failure.""" + failure: Final = _upstream_failure(error, request) + if failure is None: + return None + status, text, headers = failure + api_base: Final = request.get("api_base") + config: Final = native_messages_config( + str(request["model"]).removeprefix(f"{request_provider}/"), + request_provider, + request.get("model_info"), + api_base if isinstance(api_base, str) else None, + ) + return ( + cast(_ERROR_CLASS, config.get_error_class)( # cast-ok: the config's untyped headers parameter + error_message=text, status_code=status, headers=dict(headers) + ) + if config is not None + else AnthropicError(status_code=status, message=text, headers=httpx.Headers(headers)) + ) + + +def _log_resolved_call(request: Mapping[str, object], request_provider: str) -> None: + """Tells the logger the resolved call, as Python's handler does before its provider config + transforms the request.""" + logger: Final = request.get("litellm_logging_obj") + if not isinstance(logger, LiteLLMLoggingObj): + return + callbacks_legacy_python.update_logging( + logger, + kwargs={name: value for name, value in request.items() if name != "metadata"}, + model=str(request["model"]).removeprefix(f"{request_provider}/"), + optional_params=dict(body_params(request)), + litellm_params={"model_info": request.get("model_info"), **body_params(request)}, + custom_llm_provider=request_provider, + ) + + +def map_failure(error: Exception, request: Mapping[str, object], request_provider: str) -> Exception: + """Raises what Python's Messages handler raises for the same failure.""" + if getattr(error, "messages_timeout", None) == "Body": + # Python reads a stream's body outside its HTTP client's error mapping, so the transport's + # own timeout reaches the caller unmapped. + return httpx.ReadTimeout(_BODY_READ_TIMEOUT) + if getattr(error, "messages_late_rejection", False): + _log_resolved_call(request, request_provider) + if getattr(error, "messages_request_error", False): + return _rejection(error, request, request_provider) + if getattr(error, "messages_missing_credential", False): + return litellm.AuthenticationError( + message=_MISSING_KEY, + llm_provider=request_provider, + model=str(request["model"]).removeprefix(f"{request_provider}/"), + ) + provider_error: Final = _provider_error(error, request, request_provider) + if provider_error is None: + return failures.map_native_failure( + error, str(request["model"]), request_provider, request, optional_str(request.get("api_base")) + ) + try: + return _exception_type( + model=str(request["model"]), + custom_llm_provider=request_provider, + original_exception=provider_error, + extra_kwargs=dict(request), + ) + except Exception as mapped: # noqa: BLE001 # exception_type raises the mapped public error + mapped.__context__ = error + return mapped + + def _drop_params(kwargs: Mapping[str, object]) -> bool: return bool(litellm.drop_params) or normalize_drop_params(kwargs.get("drop_params")) is True diff --git a/tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py b/tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py index cdfba773cf6..41008f0d57f 100644 --- a/tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py +++ b/tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py @@ -20,6 +20,7 @@ R request, S response, C callbacks, F covered by a fallback detector today) """ import asyncio +import copy import json import os import re @@ -335,38 +336,12 @@ def _kw(**kwargs: object) -> Mapping[str, object]: return MappingProxyType(kwargs) -A2_BARE_MODEL: Final = "A2: native resolves the provider from the model prefix only, so a bare Claude model name fails" -A10_NO_MAX_TOKENS: Final = "A10: native admission declines a call without an int max_tokens and Python serves it" -R3_LEGACY_THINKING: Final = "R3: native maps a float legacy thinking budget to effort low instead of Python's tier" -R4_BINDING_BETA: Final = "R4: native omits the thinking-binding-controls beta for block_binding" -R5_DEPLOYMENT_EFFORT: Final = "R5: native ignores the deployment's reasoning_effort_levels from model_info" -R7_DROP_ORDER: Final = "R7: native applies additional_drop_params before Python's unsupported-param check" -R8_VERSION_HEADER: Final = "R8: Python sends anthropic-version twice when the caller capitalises it; native once" -R9_QUERY_PATH: Final = "R9: native appends /v1/messages again to an api_base that ends in /v1/messages?query" -R11_PARAM_ERROR: Final = "R11: native raises BadRequestError where Python raises UnsupportedParamsError" -R12_METADATA_ERROR: Final = ( - "R12: Python raises a pydantic ValidationError for a non-string metadata.user_id; native a BadRequestError" -) -S1_S2_ERROR_HEADERS: Final = ( - "S1/S2: native drops upstream error headers and wraps the error body in a (status, body) tuple" -) -S2_ERROR_MESSAGE: Final = ( - "S2: native wraps the upstream error body in a (status, body) tuple and strips the provider from model" -) -S2_S12_ERROR_STATUS: Final = "S2/S12: native keeps the upstream status where Python maps it (413 to 400, 529 to 500)" -S3_TIMEOUT: Final = "S3: native raises APIConnectionError (500) for a timeout where Python raises Timeout (408)" -S5_STREAM_DEADLINE: Final = "S5: native applies the timeout to the whole stream instead of to each read" +A10_NO_MAX_TOKENS: Final = "A10: native admission hands a call without an int max_tokens to Python, whose signature raises the TypeError; only dispatch differs" S6_PRIVATE_KEYS: Final = "S6: native writes _hidden_params into the response dict" -S8_TYPED_RESPONSE: Final = "S8: native re-serializes the response and drops null usage and container" -S9_SYNC: Final = "S9: Python's sync handler raises ValueError for native providers; native serves the call" -S13_INVALID_JSON: Final = "S13: native reports an unparseable success body as an internal error, not AnthropicException" -S14_MISSING_KEY: Final = ( - "S14: native raises APIConnectionError for a missing API key where Python raises AuthenticationError" +S15_COMPLETE_DEADLINE: Final = ( + "S15: by design, native bounds a complete response by one overall deadline where Python bounds each read, " + "so a body trickling in slower than the timeout succeeds in Python and times out natively" ) -C4_METADATA: Final = "C4: native callbacks get the raw Anthropic metadata, not Python's merged litellm metadata" -C6_RESPONSE_HEADERS: Final = "C6: native success logs carry no response_headers" -C7_API_KEY: Final = "C7: native callbacks see the caller's api_key where Python logs an empty string" -C8_STREAM_HIDDEN: Final = "C8: native streams lack the call metadata @client stamps into _hidden_params" F_MOCK: Final = "F: mock_response is Python-only (detector mock_response)" F_THINKING_RETRY: Final = ( "F: retry without thinking after a rejected signature is Python-only (detector thinking_signature_retry)" @@ -377,176 +352,10 @@ F_STREAM_COST: Final = "F: cost injection into streamed usage is Python-only (de KNOWN_GAPS: Final[Mapping[str, Mapping[str, str | None]]] = MappingProxyType( { - "plain_text": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "system_string": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "system_blocks_with_cache_control": _gaps( - logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "multi_turn_tool_result": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "tools_with_tool_choice": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "tool_choice_none": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "tool_input_examples": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "sampling_params": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "stop_sequences": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "metadata_user_id": _gaps( - logged_metadata=C4_METADATA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "metadata_extra_keys_are_filtered": _gaps( - logged_metadata=C4_METADATA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "metadata_user_id_not_a_string": _gaps(outcome=R12_METADATA_ERROR, logged=R12_METADATA_ERROR), - "service_tier": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "thinking_enabled": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "thinking_disabled": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "legacy_thinking_on_adaptive_model_keeps_display": _gaps( - logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "legacy_thinking_float_budget_on_adaptive_model": _gaps( - request=R3_LEGACY_THINKING, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "adaptive_thinking_with_block_binding": _gaps( - request=R4_BINDING_BETA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "reasoning_effort": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "reasoning_effort_on_adaptive_model": _gaps( - logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "output_config_effort": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "output_format_json_schema": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "speed_on_model_without_speed": _gaps(outcome=R11_PARAM_ERROR, logged=R11_PARAM_ERROR), - "speed_on_model_without_speed_drop_params": _gaps( - logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "speed_on_fast_model": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "temperature_on_adaptive_model": _gaps(outcome=R11_PARAM_ERROR, logged=R11_PARAM_ERROR), - "top_k_on_adaptive_model": _gaps(outcome=R11_PARAM_ERROR, logged=R11_PARAM_ERROR), - "top_k_on_adaptive_model_drop_params": _gaps( - logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "additional_drop_params_hides_rejected_temperature": _gaps( - request=R7_DROP_ORDER, - outcome=R7_DROP_ORDER, - hidden_params=R7_DROP_ORDER, - logged=R7_DROP_ORDER, - logged_metadata=R7_DROP_ORDER, - logged_api_key=R7_DROP_ORDER, - logged_call=R7_DROP_ORDER, - ), - "additional_drop_params_nested_path": _gaps( - logged_metadata=C4_METADATA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "context_management_edits": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "container_skills": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "remote_mcp_servers": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "inference_geo": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "top_level_cache_control": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "null_safeguards_are_stripped": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "empty_text_blocks_are_stripped": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "tool_use_ids_are_sanitized": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "provider_specific_fields_are_stripped": _gaps( - logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "image_block": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "document_block": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "server_web_search_tool": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "extra_headers": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "anthropic_beta_from_headers_and_extra_headers": _gaps( - logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "caller_anthropic_version_capitalised": _gaps( - request=R8_VERSION_HEADER, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "provider_specific_header": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "explicit_custom_llm_provider": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "bare_model_name": _gaps(**dict.fromkeys(ASPECTS[1:], A2_BARE_MODEL) | {"logged_model_info": None}), - "api_base_with_messages_path": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "api_base_with_messages_path_and_query": _gaps( - request=R9_QUERY_PATH, - logged=R9_QUERY_PATH, - logged_response_headers=C6_RESPONSE_HEADERS, - logged_api_key=C7_API_KEY, - ), - "api_key_from_environment": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "auth_token_from_environment": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "reasoning_auto_summary": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "global_drop_params": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "request_timeout_setting": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "litellm_metadata_and_metadata": _gaps( - logged_metadata=C4_METADATA, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "model_info_with_id": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "deployment_effort_levels": _gaps( - request=R5_DEPLOYMENT_EFFORT, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "response_with_tool_use": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "response_with_thinking": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "response_with_cache_usage": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "response_with_null_usage_and_container": _gaps(outcome=S8_TYPED_RESPONSE), - "response_without_stop_sequence": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "response_with_unknown_field": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "response_key_order": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "response_upstream_headers": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "response_private_keys": _gaps( - outcome=S6_PRIVATE_KEYS, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "upstream_400": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY), - "upstream_401": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY), - "upstream_403": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY), - "upstream_404": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY), - "upstream_413": _gaps(outcome=S2_S12_ERROR_STATUS, logged_api_key=C7_API_KEY), - "upstream_429_with_retry_after": _gaps(outcome=S1_S2_ERROR_HEADERS, logged_api_key=C7_API_KEY), - "upstream_500": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY), - "upstream_529_overloaded": _gaps(outcome=S2_S12_ERROR_STATUS, logged_api_key=C7_API_KEY), - "upstream_400_context_window": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY), - "upstream_non_json_error": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY), - "upstream_invalid_json_success": _gaps( - outcome=S13_INVALID_JSON, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "upstream_timeout": _gaps(outcome=S3_TIMEOUT, logged=S3_TIMEOUT, logged_api_key=C7_API_KEY), + "response_private_keys": _gaps(outcome=S6_PRIVATE_KEYS), "missing_max_tokens": _gaps(dispatch=A10_NO_MAX_TOKENS), - "missing_api_key": _gaps(outcome=S14_MISSING_KEY, logged=S14_MISSING_KEY), - "stream_text": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY), - "stream_tool_use": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY), - "stream_thinking": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY), - "stream_error_event_mid_stream": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY), - "stream_longer_than_timeout": _gaps( - outcome=S5_STREAM_DEADLINE, - hidden_params=S5_STREAM_DEADLINE, - logged=S5_STREAM_DEADLINE, - logged_metadata=S5_STREAM_DEADLINE, - logged_api_key=S5_STREAM_DEADLINE, - ), - "stream_upstream_429": _gaps(outcome=S1_S2_ERROR_HEADERS, logged_api_key=C7_API_KEY), - "stream_false_explicit": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "sync_text": _gaps( - request=S9_SYNC, - outcome=S9_SYNC, - hidden_params=S9_SYNC, - logged=S9_SYNC, - logged_metadata=S9_SYNC, - logged_response_headers=S9_SYNC, - logged_api_key=S9_SYNC, - logged_call=S9_SYNC, - ), - "sync_stream": _gaps( - request=S9_SYNC, - outcome=S9_SYNC, - hidden_params=S9_SYNC, - logged=S9_SYNC, - logged_metadata=S9_SYNC, - logged_response_headers=S9_SYNC, - logged_api_key=S9_SYNC, - logged_call=S9_SYNC, - ), - "sync_upstream_429": _gaps( - request=S9_SYNC, - outcome=S9_SYNC, - logged=S9_SYNC, - logged_response_headers=S9_SYNC, - logged_api_key=S9_SYNC, - logged_call=S9_SYNC, - ), "mock_response": _gaps( + logged_response_headers=F_MOCK, request=F_MOCK, outcome=F_MOCK, hidden_params=F_MOCK, @@ -556,32 +365,25 @@ KNOWN_GAPS: Final[Mapping[str, Mapping[str, str | None]]] = MappingProxyType( logged_call=F_MOCK, ), "mock_response_rate_limit": _gaps( - request=F_MOCK, outcome=F_MOCK, logged=F_MOCK, logged_api_key=F_MOCK, logged_call=F_MOCK + logged_response_headers=F_MOCK, + request=F_MOCK, + outcome=F_MOCK, + logged=F_MOCK, + logged_api_key=F_MOCK, + logged_call=F_MOCK, ), - "thinking_replay": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), "thinking_signature_rejected_then_retried": _gaps( request=F_THINKING_RETRY, outcome=F_THINKING_RETRY, hidden_params=F_THINKING_RETRY, logged=F_THINKING_RETRY, - logged_metadata=C4_METADATA, - logged_response_headers=C6_RESPONSE_HEADERS, - logged_api_key=C7_API_KEY, + logged_metadata=F_THINKING_RETRY, + logged_response_headers=F_THINKING_RETRY, ), - "cache_control_injection_points": _gaps( - request=F_CACHE_INJECTION, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "tool_custom_field": _gaps( - request=F_TOOL_CUSTOM, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "strict_tools_rejected_then_retried": _gaps(outcome=S2_ERROR_MESSAGE, logged_api_key=C7_API_KEY), - "modify_params": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "stream_cost_injection": _gaps(outcome=F_STREAM_COST, hidden_params=F_STREAM_COST, logged_api_key=C7_API_KEY), - "enable_prompt_caching": _gaps( - request=F_CACHE_INJECTION, logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY - ), - "request_timeout_kwarg": _gaps(logged_response_headers=C6_RESPONSE_HEADERS, logged_api_key=C7_API_KEY), - "stream_timeout_kwarg": _gaps(hidden_params=C8_STREAM_HIDDEN, logged_api_key=C7_API_KEY), + "cache_control_injection_points": _gaps(request=F_CACHE_INJECTION), + "tool_custom_field": _gaps(request=F_TOOL_CUSTOM), + "stream_cost_injection": _gaps(outcome=F_STREAM_COST), + "enable_prompt_caching": _gaps(request=F_CACHE_INJECTION), } ) @@ -606,14 +408,55 @@ REQUEST_CASES: Final = ( Case("metadata_user_id", _kw(metadata={"user_id": "user-123"})), Case("metadata_extra_keys_are_filtered", _kw(metadata={"user_id": "user-123", "session": "s-1"})), Case("metadata_user_id_not_a_string", _kw(metadata={"user_id": 123})), - Case("service_tier", _kw(service_tier="auto"), gaps=_gaps(request="R1: native drops service_tier")), + Case("service_tier", _kw(service_tier="auto")), Case("thinking_enabled", _kw(thinking={"type": "enabled", "budget_tokens": 2048}, max_tokens=4096)), Case("thinking_disabled", _kw(thinking={"type": "disabled"})), Case( "legacy_thinking_on_adaptive_model_keeps_display", _kw(thinking={"type": "enabled", "budget_tokens": 2048, "display": "summarized"}, max_tokens=4096), model=OPUS_ADAPTIVE, - gaps=_gaps(request="R3: native replaces legacy thinking with a bare adaptive block and drops display"), + ), + Case( + "legacy_thinking_display_updates_on_adaptive_model", + _kw(thinking={"type": "enabled", "budget_tokens": 2048, "display": "updates"}, max_tokens=4096), + model=OPUS_ADAPTIVE, + ), + Case( + "legacy_thinking_string_budget_on_adaptive_model", + _kw(thinking={"type": "enabled", "budget_tokens": " 4_096 "}, max_tokens=8192), + model=OPUS_ADAPTIVE, + ), + Case( + "legacy_thinking_non_numeric_budget_on_adaptive_model", + _kw(thinking={"type": "enabled", "budget_tokens": "lots"}, max_tokens=8192), + model=OPUS_ADAPTIVE, + ), + Case( + "legacy_thinking_null_budget_on_adaptive_model", + _kw(thinking={"type": "enabled", "budget_tokens": None}, max_tokens=8192), + model=OPUS_ADAPTIVE, + ), + Case( + "legacy_thinking_non_numeric_budget_dropped", + _kw(thinking={"type": "enabled", "budget_tokens": [1]}, max_tokens=8192, drop_params=True), + model=OPUS_ADAPTIVE, + ), + Case( + "legacy_thinking_carries_extra_keys", + _kw( + thinking={"type": "enabled", "budget_tokens": 2048, "block_binding": "strict", "note": "kept"}, + max_tokens=4096, + ), + model=OPUS_ADAPTIVE, + ), + Case( + "legacy_thinking_drops_extra_keys_with_drop_params", + _kw( + thinking={"type": "enabled", "budget_tokens": 2048, "block_binding": "strict", "note": "dropped"}, + max_tokens=4096, + drop_params=True, + ), + model=OPUS_ADAPTIVE, ), Case( "legacy_thinking_float_budget_on_adaptive_model", @@ -731,7 +574,6 @@ REQUEST_CASES: Final = ( Case( "anthropic_beta_from_headers_and_extra_headers", _kw(headers={"anthropic-beta": "client-beta-1"}, extra_headers={"anthropic-beta": "deploy-beta-2"}), - gaps=_gaps(request="R2: native keeps the last anthropic-beta source instead of the union"), ), Case("caller_anthropic_version_capitalised", _kw(extra_headers={"Anthropic-Version": "2023-06-01"})), Case( @@ -774,6 +616,24 @@ REQUEST_CASES: Final = ( ), model=OPUS_ADAPTIVE, ), + Case( + "deployment_effort_levels_from_legacy_thinking", + _kw( + thinking={"type": "enabled", "budget_tokens": 30000}, + max_tokens=40000, + model_info={"id": "deployment-1", "reasoning_effort_levels": ["low", "medium"]}, + ), + model=OPUS_ADAPTIVE, + ), + Case( + "deployment_effort_levels_below_request_with_drop_params", + _kw( + output_config={"effort": "high"}, + drop_params=True, + model_info={"id": "deployment-1", "reasoning_effort_levels": ["low"]}, + ), + model=OPUS_ADAPTIVE, + ), ) RESPONSE_CASES: Final = ( @@ -820,6 +680,18 @@ FAILURE_CASES: Final = ( Case("upstream_non_json_error", upstream=(ResponseSpec(body=b"bad gateway", status=502),)), Case("upstream_invalid_json_success", upstream=(ResponseSpec(body=b"not json", status=200),)), Case("upstream_timeout", _kw(timeout=0.5), upstream=(ResponseSpec(body=dict(TEXT_RESPONSE), delay=1.5),)), + Case( + "response_trickles_slower_than_timeout", + _kw(timeout=1), + upstream=(ResponseSpec(body=dict(TEXT_RESPONSE), chunk_delay=0.5, body_chunks=4),), + gaps=_gaps( + outcome=S15_COMPLETE_DEADLINE, + hidden_params=S15_COMPLETE_DEADLINE, + logged=S15_COMPLETE_DEADLINE, + logged_metadata=S15_COMPLETE_DEADLINE, + logged_response_headers=S15_COMPLETE_DEADLINE, + ), + ), Case("missing_max_tokens", _kw(max_tokens=None), upstream=()), Case("missing_api_key", _kw(api_key=None), upstream=()), ) @@ -838,6 +710,11 @@ STREAM_CASES: Final = ( _kw(stream=True, timeout=1), upstream=(_stream(SLOW_STREAM, chunk_delay=0.25),), ), + Case( + "stream_read_stalls_past_timeout", + _kw(stream=True, timeout=0.5), + upstream=(_stream(SLOW_STREAM, chunk_delay=1.0),), + ), Case( "stream_upstream_429", _kw(stream=True), @@ -967,7 +844,9 @@ def _settings(settings: Mapping[str, object]) -> Generator[None]: def _kwargs(case: Case, server: RecordingServer) -> dict[str, object]: - supplied: Final = dict(case.kwargs) + # Deep: Python writes into the caller's own `litellm_metadata`, which must not leak into the + # other leg. + supplied: Final = copy.deepcopy(dict(case.kwargs)) suffix: Final = supplied.pop("api_base_suffix", "") kwargs: Final[dict[str, object]] = { "model": case.model, diff --git a/tests/test_litellm_rust/cache/test_python_cache.py b/tests/test_litellm_rust/cache/test_python_cache.py index 722cdbcbaca..beb818c7162 100644 --- a/tests/test_litellm_rust/cache/test_python_cache.py +++ b/tests/test_litellm_rust/cache/test_python_cache.py @@ -18,10 +18,7 @@ from litellm.proxy.hooks.model_max_budget_limiter import ( ) from litellm.proxy.hooks.parallel_request_limiter_v3 import PROXY_MaxParallelRequestsHandler_v3 from litellm.proxy.utils import InternalUsageCache -from litellm.rust_bridge import runtime -from litellm.rust_bridge.catalog import Route, RouteContext, RouteRule -from litellm.rust_bridge.configuration import Rollout -from litellm.rust_bridge.public_call import NativeCall +from litellm.rust_bridge.catalog import Route from litellm.types.caching import CachingSupportedCallTypes from tests.test_litellm_rust.cache.support import cache_key, collect, invoke, payload from tests.test_litellm_rust.messages.data import MESSAGES, MESSAGES_EVENTS, MESSAGES_MODEL, MESSAGES_RESPONSE @@ -430,49 +427,6 @@ async def test_python_cache_operations_stay_in_the_rust_callers_task( assert len(recording_server.requests) == 1 -def test_sync_rust_messages_calls_python_cache(recording_server: RecordingServer) -> None: - from litellm.rust_bridge.messages.entrypoints import NATIVE_MESSAGES - - litellm.cache = Cache() - recording_server.default_response = ResponseSpec(body=MESSAGES_RESPONSE) - arguments: Final = { - "model": MESSAGES_MODEL, - "messages": list(MESSAGES), - "max_tokens": 32, - "api_key": "test-key", - "api_base": recording_server.base_url, - } - request: Final = NativeCall( - args=(), - kwargs=arguments, - base={ - "model": MESSAGES_MODEL, - "messages": list(MESSAGES), - "max_tokens": 32, - "stream": None, - "api_key": "test-key", - "api_base": recording_server.base_url, - "custom_llm_provider": "anthropic", - **arguments, - }, - ) - - def call() -> object: - return runtime.run( - RouteContext(Route.MESSAGES), - binding=NATIVE_MESSAGES, - native=lambda hook: hook(request), - python=runtime.NO_PYTHON, - rules=(RouteRule(Route.MESSAGES, Rollout.RUST_REQUIRED),), - ) - - first: Final = call() - second: Final = call() - assert payload(first) == payload(second) - assert cache_key(second) - assert len(recording_server.requests) == 1 - - @pytest.mark.asyncio @pytest.mark.parametrize("namespace_source", ("cache", "metadata")) async def test_rust_messages_legacy_cache_honors_request_namespaces( diff --git a/tests/test_litellm_rust/messages/test_callbacks.py b/tests/test_litellm_rust/messages/test_callbacks.py index cd5478dfc3a..a3c03898a12 100644 --- a/tests/test_litellm_rust/messages/test_callbacks.py +++ b/tests/test_litellm_rust/messages/test_callbacks.py @@ -14,7 +14,6 @@ from litellm.types.integrations.argilla import ArgillaItem from litellm.types.utils import ModelResponse, StandardLoggingPayload from tests.test_litellm_rust.messages.data import BAD_REQUEST, MESSAGES_RESPONSE, RATE_LIMITED, call from tests.test_litellm_rust.support.callback_recorder import HookRecorder, RecordingLogger, request_body -from tests.test_litellm_rust.support.gaps import gap from tests.test_litellm_rust.support.isolation import rebound from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec, relaxed from tests.test_litellm_rust.support.results import response_id, with_id @@ -22,7 +21,6 @@ from tests.test_litellm_rust.support.routing import assert_served_natively pytestmark = pytest.mark.requires_rust_extension -SYNC_HANDLERS_ON_ASYNC_CALLS: Final = gap("native async calls also run the sync success and failure handlers") _OBJECT: Final = TypeAdapter(dict[str, object]) _HEADERS: Final = TypeAdapter(dict[str, str]) @@ -211,7 +209,7 @@ async def test_pre_call_deployment_rejection_logs_failure_without_provider_io(se failures: Final = await recorder.wait_for_async("async_log_failure_event") assert raised.value is rejection - assert server.requests == [] + assert not server.requests assert len(failures) == 1 assert _OBJECT.validate_python(failures[0].kwargs)["exception"] is raised.value assert not {"log_pre_api_call", "log_success_event", "async_log_success_event"} & set(recorder.names) @@ -341,8 +339,8 @@ async def test_a_failing_logger_does_not_fail_the_call(server: RecordingServer) pytest.param("input", False, False, 1, id="global-input"), pytest.param("success", True, False, 1, id="global-async-success"), pytest.param("success", True, True, 1, id="request-async-success"), - pytest.param("success", False, False, 0, id="global-sync-success", marks=SYNC_HANDLERS_ON_ASYNC_CALLS), - pytest.param("success", False, True, 0, id="request-sync-success", marks=SYNC_HANDLERS_ON_ASYNC_CALLS), + pytest.param("success", False, False, 0, id="global-sync-success"), + pytest.param("success", False, True, 0, id="request-sync-success"), pytest.param("failure", True, False, 1, id="global-async-failure"), pytest.param("failure", False, False, 1, id="global-sync-failure"), pytest.param("failure", False, True, 1, id="request-sync-failure"), @@ -384,12 +382,11 @@ async def test_pre_call_rule_rejects_before_provider_io(server: RecordingServer) checked.append(text) return False - with rebound(litellm, "pre_call_rules", [reject]): - with pytest.raises(litellm.APIResponseValidationError): - await call(server) + with rebound(litellm, "pre_call_rules", [reject]), pytest.raises(litellm.APIResponseValidationError): + await call(server) assert checked == ["Hello"] - assert server.requests == [] + assert not server.requests @pytest.mark.asyncio @@ -419,11 +416,11 @@ async def test_dataset_hook_returning_none_drops_the_record_from_the_success_sin @pytest.mark.asyncio -async def test_sync_call_returns_the_provider_message(server: RecordingServer) -> None: +async def test_sync_call_raises_like_python_without_logging(server: RecordingServer) -> None: recorder: Final = RecordingLogger() - response: Final = await call(server, asynchronous=False, callbacks=[recorder]) + with pytest.raises(ValueError, match="anthropic_messages_handler is not implemented for sync calls"): + await call(server, asynchronous=False, callbacks=[recorder]) - assert_served_natively(server, response) - assert response_id(response) == MESSAGES_RESPONSE["id"] - assert len(await recorder.wait_for_async("log_success_event")) == 1 + assert not server.requests + assert not recorder.names diff --git a/tests/test_litellm_rust/messages/test_streaming.py b/tests/test_litellm_rust/messages/test_streaming.py index c700c093d1a..e57631f48b9 100644 --- a/tests/test_litellm_rust/messages/test_streaming.py +++ b/tests/test_litellm_rust/messages/test_streaming.py @@ -6,7 +6,6 @@ import pytest from tests.test_litellm_rust.messages.data import MESSAGES_EVENTS, MESSAGES_RESPONSE, call from tests.test_litellm_rust.support.callback_recorder import RecordingLogger, drain_logging from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec, relaxed -from tests.test_litellm_rust.support.results import consume from tests.test_litellm_rust.support.routing import assert_served_natively pytestmark = pytest.mark.requires_rust_extension @@ -46,16 +45,14 @@ async def test_stream_relays_provider_events_and_logs_success_once_after_the_las @pytest.mark.asyncio -async def test_sync_stream_relays_provider_events_and_logs_success_once(server: RecordingServer) -> None: - server.enqueue(STREAM) +async def test_sync_stream_raises_like_python_before_sending(server: RecordingServer) -> None: recorder: Final = RecordingLogger() - stream: Final = await call(server, asynchronous=False, stream=True, callbacks=[recorder]) - chunks: Final = await consume(stream) + with pytest.raises(ValueError, match="anthropic_messages_handler is not implemented for sync calls"): + await call(server, asynchronous=False, stream=True, callbacks=[recorder]) - assert_served_natively(server) - assert b"".join(chunks) == b"".join(STREAM.payloads()) - assert len(await recorder.wait_for_async("async_log_success_event")) == 1 + assert not server.requests + assert not recorder.names @pytest.mark.asyncio diff --git a/tests/test_litellm_rust/support/recording_server.py b/tests/test_litellm_rust/support/recording_server.py index 7beeb20f109..5743658abdf 100644 --- a/tests/test_litellm_rust/support/recording_server.py +++ b/tests/test_litellm_rust/support/recording_server.py @@ -27,12 +27,14 @@ class ResponseSpec: delay: float = 0 chunk_delay: float = 0 events: tuple[tuple[str, object], ...] = () + # Splits a non-stream body into this many writes, `chunk_delay` apart, so it trickles in. + body_chunks: int = 1 def payloads(self) -> tuple[bytes, ...]: - if isinstance(self.body, bytes): - return (self.body,) if not self.events: - return (json.dumps(self.body).encode(),) + body: Final = self.body if isinstance(self.body, bytes) else json.dumps(self.body).encode() + size: Final = max(-(-len(body) // max(self.body_chunks, 1)), 1) + return tuple(body[start : start + size] for start in range(0, len(body), size)) or (body,) return tuple(f"event: {event}\ndata: {json.dumps(data)}\n\n".encode() for event, data in self.events) diff --git a/tests/unit/rust_bridge/host/routes/test_messages.py b/tests/unit/rust_bridge/host/routes/test_messages.py index abf0e055e73..9d8b09dd1c4 100644 --- a/tests/unit/rust_bridge/host/routes/test_messages.py +++ b/tests/unit/rust_bridge/host/routes/test_messages.py @@ -138,3 +138,110 @@ def test_stream_hidden_params_projects_upstream_headers_the_way_the_python_handl assert additional["llm_provider-request-id"] == "req_upstream_123" assert additional["x-ratelimit-remaining-requests"] == "41" assert additional["request-id"] == additional["llm_provider-request-id"] + + +_ADAPTIVE_OPUS: Final = "anthropic/claude-opus-4-7" + + +@pytest.mark.parametrize( + ("model", "expected"), + [("claude-sonnet-4-5", "anthropic"), ("gpt-4o", "openai"), ("not-a-known-model", None)], +) +def test_inferred_provider_is_the_one_the_python_handler_resolves(model: str, expected: str | None) -> None: + assert messages_host.inferred_provider(model, None) == expected + + +@pytest.mark.parametrize( + ("request_effort", "levels", "drop_params", "expected"), + [ + ("max", ["low", "high"], False, "high"), + ("high", ["low"], True, "low"), + ], +) +def test_body_params_round_effort_to_the_levels_the_deployment_declares( + request_effort: str, levels: list[str], drop_params: bool, expected: str +) -> None: + request: Final[dict[str, object]] = { + "model": _ADAPTIVE_OPUS, + "messages": [], + "max_tokens": 8, + "output_config": {"effort": request_effort}, + "drop_params": drop_params, + "model_info": {"id": "deployment-1", "reasoning_effort_levels": levels}, + } + + assert messages_host.body_params(request)["output_config"] == {"effort": expected} + + +def test_body_params_keep_the_requested_effort_without_deployment_levels() -> None: + request: Final[dict[str, object]] = { + "model": _ADAPTIVE_OPUS, + "messages": [], + "max_tokens": 8, + "output_config": {"effort": "max"}, + } + + assert messages_host.body_params(request)["output_config"] == {"effort": "max"} + + +def _marked(phase: str | None = None, *args: object) -> ValueError: + error: Final = ValueError(*args) + if phase is not None: + error.messages_timeout = phase # pyright: ignore[reportAttributeAccessIssue] # marker the native host sets + return error + + +def test_a_stalled_stream_body_reaches_the_caller_as_the_transport_timeout() -> None: + import httpx + + mapped: Final = messages_host.map_failure(_marked("Body"), {"model": _ADAPTIVE_OPUS}, "anthropic") + + assert isinstance(mapped, httpx.ReadTimeout) + assert str(mapped) == "Timeout on reading data from socket" + + +def test_a_native_read_timeout_maps_to_the_public_timeout_with_python_text() -> None: + mapped: Final = messages_host.map_failure(_marked("Read"), {"model": _ADAPTIVE_OPUS, "timeout": 1.0}, "anthropic") + + assert isinstance(mapped, litellm.Timeout) + assert "Read timed out. Timeout passed=1.0" in str(mapped) + + +def test_an_upstream_status_maps_through_the_provider_error_class() -> None: + upstream: Final = ValueError(429, "slow down") + upstream.headers = [("retry-after", "3")] # pyright: ignore[reportAttributeAccessIssue] # set by the native host + + mapped: Final = messages_host.map_failure(upstream, {"model": _ADAPTIVE_OPUS}, "anthropic") + + assert isinstance(mapped, litellm.RateLimitError) + assert "slow down" in str(mapped) + assert mapped.__context__ is upstream + + +def test_a_late_rejection_tells_the_logger_the_resolved_call_first() -> None: + import datetime + + from litellm.litellm_core_utils.litellm_logging import Logging + + logger: Final = Logging( + model=_ADAPTIVE_OPUS, + messages=[], + stream=False, + call_type="anthropic_messages", + start_time=datetime.datetime.now(), + litellm_call_id="late", + function_id="late", + ) + rejected: Final = ValueError("budget_tokens must be a number") + rejected.messages_late_rejection = True # pyright: ignore[reportAttributeAccessIssue] # marker the native host sets + rejected.messages_request_error = True # pyright: ignore[reportAttributeAccessIssue] # marker the native host sets + + mapped: Final = messages_host.map_failure( + rejected, + {"model": _ADAPTIVE_OPUS, "messages": [], "max_tokens": 8, "litellm_logging_obj": logger}, + "anthropic", + ) + + assert isinstance(mapped, litellm.BadRequestError) + assert logger.model_call_details["custom_llm_provider"] == "anthropic" + assert logger.model_call_details["optional_params"] == {"max_tokens": 8} diff --git a/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py b/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py index a40f8a77105..1ba0188e76d 100644 --- a/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py +++ b/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py @@ -5,6 +5,7 @@ from pathlib import Path from types import MappingProxyType from typing import Final +import httpx import pytest from pydantic import TypeAdapter @@ -244,3 +245,56 @@ def test_the_rust_contract_matches_the_shim_signatures() -> None: contract: Final = TypeAdapter(dict[str, list[str]]).validate_json(CONTRACT_PATH.read_text()) assert contract == {name: list(inspect.signature(getattr(legacy, name)).parameters) for name in contract} + + +def _messages_logger() -> Logging: + return Logging( + model="claude-sonnet-4-5", + messages=[], + stream=False, + call_type="anthropic_messages", + start_time=datetime.datetime.now(), + litellm_call_id="messages", + function_id="messages", + ) + + +def test_native_provider_response_is_recorded_for_logging_replay() -> None: + logger: Final = _messages_logger() + + legacy.record_response(logger, 200, (("request-id", "req_native"),), '{"id": "msg_native"}') + + recorded: Final[object] = logger.model_call_details["httpx_response"] + assert isinstance(recorded, httpx.Response) + assert recorded.status_code == 200 + assert recorded.content == b'{"id": "msg_native"}' + assert logger.model_call_details["response_headers"]["request-id"] == "req_native" + + +def test_native_provider_exchange_window_is_recorded_like_the_timed_post() -> None: + logger: Final = _messages_logger() + logger.model_call_details["litellm_params"] = {"litellm_metadata": {"user_api_key": "hashed"}} + + legacy.record_api_window(logger, 1_700_000_000.0, 1_700_000_000.25) + + assert logger.model_call_details["llm_api_duration_ms"] == 250.0 + metadata: Final = logger.model_call_details["litellm_params"]["litellm_metadata"] + assert metadata["llm_api_timing_windows"] == ((1_700_000_000.0, 1_700_000_000.25),) + + +def test_stream_head_is_stamped_in_place_with_the_call_hidden_params() -> None: + logger: Final = _messages_logger() + head: Final[dict[str, object]] = {"additional_headers": {"request-id": "req_native"}} + started: Final = datetime.datetime.now() + + legacy.stamp_stream_head( + head, + logger, + {"model": "anthropic/claude-sonnet-4-5", "litellm_call_id": "messages"}, + started, + started + datetime.timedelta(milliseconds=5), + ) + + assert head["additional_headers"] == {"request-id": "req_native", "llm_provider-request-id": "req_native"} + assert head["litellm_call_id"] == "messages" + assert "_response_ms" in head diff --git a/tests/unit/rust_bridge/messages/test_secrets.py b/tests/unit/rust_bridge/messages/test_secrets.py index 64559d63193..5ee89fa7794 100644 --- a/tests/unit/rust_bridge/messages/test_secrets.py +++ b/tests/unit/rust_bridge/messages/test_secrets.py @@ -11,7 +11,7 @@ import litellm from litellm.integrations.custom_secret_manager import CustomSecretManager from litellm.llms.anthropic.messages.native.handler import anthropic_messages from litellm.rust_bridge.host import settings -from litellm.rust_bridge.messages.entrypoints import NATIVE_AMESSAGES, NATIVE_MESSAGES +from litellm.rust_bridge.messages.entrypoints import NATIVE_AMESSAGES from litellm.rust_bridge.public_call import NativeCall from litellm.types.secret_managers.main import KeyManagementSettings, KeyManagementSystem from tests.test_litellm_rust.messages.data import MESSAGES, MESSAGES_MODEL, MESSAGES_RESPONSE @@ -72,21 +72,13 @@ async def _python_messages() -> object: return await anthropic_messages(**_public_kwargs()) -async def _rust_messages() -> object: - route: Final = NATIVE_MESSAGES.load() - assert route is not None - return route(_native_request()) - - async def _rust_amessages() -> object: route: Final = NATIVE_AMESSAGES.load() assert route is not None return await route(_native_request()) -@pytest.fixture( - params=(_python_messages, _rust_messages, _rust_amessages), ids=("python-async", "rust-sync", "rust-async") -) +@pytest.fixture(params=(_python_messages, _rust_amessages), ids=("python-async", "rust-async")) def messages(request: pytest.FixtureRequest) -> Messages: return cast(Messages, request.param)