diff --git a/litellm-rust/crates/callbacks-legacy-python/python_contract.json b/litellm-rust/crates/callbacks-legacy-python/python_contract.json index 8a212223523..0378e3621c0 100644 --- a/litellm-rust/crates/callbacks-legacy-python/python_contract.json +++ b/litellm-rust/crates/callbacks-legacy-python/python_contract.json @@ -43,6 +43,12 @@ "headers", "body" ], + "retried_request": [ + "logger", + "logged_body", + "body", + "reasons" + ], "record_api_window": [ "logger", "start", diff --git a/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs b/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs index f68541c5c7b..2c30becab97 100644 --- a/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs +++ b/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs @@ -6,7 +6,7 @@ use crate::LoggingOperation; use litellm_host_python::PythonOwned; use litellm_host::{ - interceptors::{RawResponse, RequestContext, WireRequest}, + interceptors::{RawResponse, RequestContext, RetriedRequest, RetryReason, WireRequest}, lifecycle::{FailureOrigin, Timing, epoch_seconds}, }; use litellm_host_python::{Error as HostError, HookStep, from_py, to_py}; @@ -61,6 +61,15 @@ pub struct LegacyLogging { api_started: Option, } +fn retry_reason(reason: &RetryReason) -> Value { + match reason { + RetryReason::ThinkingSignature => serde_json::json!({"kind": "thinking_signature"}), + RetryReason::StrictTools { dropped } => { + serde_json::json!({"kind": "strict_tools", "dropped": dropped}) + } + } +} + fn datetime(py: Python<'_>, epoch_seconds: f64) -> PyResult> { PyDateTime::from_timestamp(py, epoch_seconds, None).map(|value| value.into_any().unbind()) } @@ -664,6 +673,40 @@ impl LegacyLogging { Ok(HookStep::Ready(())) } + pub(crate) fn request_retried( + &mut self, + py: Python<'_>, + retry: &RetriedRequest, + ) -> PyResult> { + let Some(request) = &self.request else { + return Ok(HookStep::Ready(())); + }; + let reasons: Vec = retry.reasons.iter().map(retry_reason).collect(); + let retried = Logging::RetriedRequest.call( + py, + ( + self.logger()?.object(py), + request.body.bind(py), + to_py(py, &retry.body)?, + to_py(py, &reasons)?, + ), + )?; + let now = epoch_seconds(); + if self.api_started.is_some() { + self.close_api_window(py, now)?; + self.api_started = Some(now); + } + Ok(HookStep::Await(retried.unbind(), Self::resume_retried)) + } + + fn resume_retried( + &mut self, + _py: Python<'_>, + result: PyResult>, + ) -> PyResult> { + result.map(|_| HookStep::Ready(())) + } + pub(crate) fn succeeded( &mut self, py: Python<'_>, @@ -1129,7 +1172,9 @@ mod payload_tests { use std::ffi::CStr; use litellm_host::hooks::CallHooks; - use litellm_host::interceptors::{RawResponse, RequestContext, WireRequest}; + use litellm_host::interceptors::{ + RawResponse, RequestContext, RetriedRequest, RetryReason, WireRequest, + }; use litellm_host::lifecycle::ExecutionEvent; use litellm_host_python::{HookStep, PythonCallEvent, PythonOwned, to_py}; use litellm_secret_value::SecretValue; @@ -1296,6 +1341,61 @@ assert ('custom_llm_provider' in logger.update['litellm_params']) is in_params, json!({"type": "document_url", "document_url": source}) } + #[test] + fn a_retried_request_hands_the_logged_body_the_retried_body_and_the_reasons_to_logging() { + Python::initialize(); + Python::attach(|py| { + let locals = namespace(py, PAYLOAD_LOGGER); + let mut logging = LegacyLogging { + logger: Some(PythonLogger::new(local(&locals, "logger").unbind())), + ..legacy_call(py, &locals, false) + }; + let context = RequestContext { + model: "model".into(), + custom_llm_provider: "provider".into(), + optional_params: json!({}), + secret_fields: Vec::new(), + api_key: None, + }; + let wire = WireRequest { + url: "https://provider.invalid/v1/messages".into(), + headers: Vec::new(), + body: json!({"messages": [{"role": "user", "content": "hi"}], "thinking": {"type": "enabled"}}), + }; + logging + .before_provider_request(py, Box::new(wire.clone()), &context) + .unwrap(); + let retried = RetriedRequest { + reasons: vec![ + RetryReason::ThinkingSignature, + RetryReason::StrictTools { dropped: 2 }, + ], + body: json!({"messages": [{"role": "user", "content": "hi"}]}), + }; + + let step = logging + .on_event( + py, + PythonCallEvent::Execution(ExecutionEvent::ProviderRequestRetried { + retry: retried, + }), + ) + .unwrap(); + + assert!(matches!(step, HookStep::Await(_, _))); + run( + py, + &locals, + c" +[(logged, body, reasons)] = [value for name, value in logger.calls if name == 'retried_request'] +assert logged is logger.pre['complete_input_dict'], logged +assert body == {'messages': [{'role': 'user', 'content': 'hi'}]}, body +assert reasons == [{'kind': 'thinking_signature'}, {'kind': 'strict_tools', 'dropped': 2}], reasons +", + ); + }); + } + fn before_provider_request(script: &CStr, body: Value) -> WireRequest { before_send_with_secrets(script, json!({}), body, &[]) } diff --git a/litellm-rust/crates/callbacks-legacy-python/src/mapping.rs b/litellm-rust/crates/callbacks-legacy-python/src/mapping.rs index cfa9e53aa7b..e321b93b03d 100644 --- a/litellm-rust/crates/callbacks-legacy-python/src/mapping.rs +++ b/litellm-rust/crates/callbacks-legacy-python/src/mapping.rs @@ -179,6 +179,9 @@ impl CallHooks for LegacyLogging { PythonCallEvent::Execution(ExecutionEvent::ProviderResponseReceived { raw }) => { (AFTER.invoke)(self, py, raw) } + PythonCallEvent::Execution(ExecutionEvent::ProviderRequestRetried { retry }) => { + self.request_retried(py, &retry) + } PythonCallEvent::Succeeded { timing, response } => { (SUCCESS.invoke)(self, py, timing, response) } diff --git a/litellm-rust/crates/callbacks-legacy-python/src/python.rs b/litellm-rust/crates/callbacks-legacy-python/src/python.rs index 5b6faa3dbc2..7b96301832e 100644 --- a/litellm-rust/crates/callbacks-legacy-python/src/python.rs +++ b/litellm-rust/crates/callbacks-legacy-python/src/python.rs @@ -50,6 +50,8 @@ pub(crate) enum Logging { RecordResponse, #[strum(serialize = "record_api_window")] RecordApiWindow, + #[strum(serialize = "retried_request")] + RetriedRequest, #[strum(serialize = "defers_async_logging")] DefersAsync, #[strum(serialize = "claim_async_success_log")] 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 ee16e0c0059..f54fd88aefe 100644 --- a/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs +++ b/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs @@ -69,6 +69,10 @@ FAKES = { 'record_response': lambda logger, status, headers, body: logger.model_call_details.update( response_headers=dict(headers) ), + 'retried_request': lambda logger, logged_body, body, reasons: ( + logger.record('retried_request', (logged_body, body, reasons)), + 'awaitable', + )[1], 'record_api_window': lambda logger, start, end: logger.model_call_details.update( llm_api_duration_ms=(end - start) * 1000 ), diff --git a/litellm-rust/crates/host-native/src/driver.rs b/litellm-rust/crates/host-native/src/driver.rs index af3248848a2..bb1a7025e10 100644 --- a/litellm-rust/crates/host-native/src/driver.rs +++ b/litellm-rust/crates/host-native/src/driver.rs @@ -84,6 +84,13 @@ where reply.send(self.interceptors.after_provider_response(raw).await); Ok(()) } + HostRequest::Intercept(InterceptRequest::ProviderRequestRetried { + retry, + reply, + }) => { + reply.send(self.interceptors.provider_request_retried(retry).await); + Ok(()) + } HostRequest::Stream(StreamDelivery::Open(head, reply)) => { self.demand = Some(reply); return Ok(Boundary::Open(head)); diff --git a/litellm-rust/crates/host-python/src/driver.rs b/litellm-rust/crates/host-python/src/driver.rs index 052156cbc5d..03e38a4312a 100644 --- a/litellm-rust/crates/host-python/src/driver.rs +++ b/litellm-rust/crates/host-python/src/driver.rs @@ -310,6 +310,14 @@ where reply.send(Ok(())); Ok(()) } + HostRequest::Intercept(InterceptRequest::ProviderRequestRetried { retry, reply }) => { + self.notify(PythonCallEvent::Execution( + ExecutionEvent::ProviderRequestRetried { retry }, + )) + .await?; + reply.send(Ok(())); + Ok(()) + } HostRequest::Stream(StreamDelivery::Open(head, reply)) => { self.streaming = true; let head = { diff --git a/litellm-rust/crates/host-python/src/hooks/chain/dispatch.rs b/litellm-rust/crates/host-python/src/hooks/chain/dispatch.rs index 4939995cefa..26d5a5692d2 100644 --- a/litellm-rust/crates/host-python/src/hooks/chain/dispatch.rs +++ b/litellm-rust/crates/host-python/src/hooks/chain/dispatch.rs @@ -241,6 +241,9 @@ fn retain_event(py: Python<'_>, event: PythonCallEvent<'_>) -> OwnedEvent { CallEvent::Execution(ExecutionEvent::ProviderResponseReceived { raw }) => { CallEvent::Execution(ExecutionEvent::ProviderResponseReceived { raw: raw.clone() }) } + CallEvent::Execution(ExecutionEvent::ProviderRequestRetried { retry }) => { + CallEvent::Execution(ExecutionEvent::ProviderRequestRetried { retry }) + } CallEvent::Succeeded { timing, response } => CallEvent::Succeeded { timing, response: response.clone_ref(py), @@ -280,6 +283,12 @@ fn dispatch( py, CallEvent::Execution(ExecutionEvent::ProviderResponseReceived { raw }), ), + CallEvent::Execution(ExecutionEvent::ProviderRequestRetried { retry }) => hooks.on_event( + py, + CallEvent::Execution(ExecutionEvent::ProviderRequestRetried { + retry: retry.clone(), + }), + ), CallEvent::Succeeded { timing, response } => hooks.on_event( py, CallEvent::Succeeded { diff --git a/litellm-rust/crates/host-python/tests/support/mod.rs b/litellm-rust/crates/host-python/tests/support/mod.rs index 15e864c0718..f4f0903cc45 100644 --- a/litellm-rust/crates/host-python/tests/support/mod.rs +++ b/litellm-rust/crates/host-python/tests/support/mod.rs @@ -422,6 +422,9 @@ impl CallHooks for SyntheticHooks { PythonCallEvent::Execution(ExecutionEvent::ProviderResponseReceived { raw }) => { format!("response:{}", raw.body) } + PythonCallEvent::Execution(ExecutionEvent::ProviderRequestRetried { retry }) => { + format!("retried:{}", retry.body) + } PythonCallEvent::Succeeded { response, .. } => { format!("succeeded:{}", response.bind(py)) } diff --git a/litellm-rust/crates/host/src/interceptors.rs b/litellm-rust/crates/host/src/interceptors.rs index 9398ea4bad6..a3adf4a3b39 100644 --- a/litellm-rust/crates/host/src/interceptors.rs +++ b/litellm-rust/crates/host/src/interceptors.rs @@ -10,6 +10,18 @@ pub struct WireRequest { pub body: Value, } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RetryReason { + ThinkingSignature, + StrictTools { dropped: usize }, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct RetriedRequest { + pub reasons: Vec, + pub body: Value, +} + /// What the route knows about the request it is sending, for a host that logs it. The /// route owns these facts; a host reads them beside the wire request and never rewrites /// them. @@ -107,6 +119,13 @@ pub trait Interceptors: Send + Sync { &self, raw: RawResponse, ) -> impl Future> + Send; + + fn provider_request_retried( + &self, + _retry: RetriedRequest, + ) -> impl Future> + Send { + async { Ok(()) } + } } impl + ?Sized> Interceptors for &T { @@ -128,6 +147,13 @@ impl + ?Sized> Interceptors for &T { ) -> impl Future> + Send { (**self).after_provider_response(raw) } + + fn provider_request_retried( + &self, + retry: RetriedRequest, + ) -> impl Future> + Send { + (**self).provider_request_retried(retry) + } } impl Interceptors for () { diff --git a/litellm-rust/crates/host/src/lifecycle.rs b/litellm-rust/crates/host/src/lifecycle.rs index e6b31381494..16456ea52d0 100644 --- a/litellm-rust/crates/host/src/lifecycle.rs +++ b/litellm-rust/crates/host/src/lifecycle.rs @@ -6,7 +6,10 @@ use std::{ use futures_util::TryStreamExt; -use crate::{call::CallOutput, interceptors::RawResponse}; +use crate::{ + call::CallOutput, + interceptors::{RawResponse, RetriedRequest}, +}; /// Seconds since the Unix epoch, on one clock for every host. pub fn epoch_seconds() -> f64 { @@ -50,7 +53,7 @@ pub enum CallEvent { }, } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq)] pub enum ExecutionEvent { ResultReady { facts: crate::interceptors::ExecutionFacts, @@ -58,6 +61,9 @@ pub enum ExecutionEvent { ProviderResponseReceived { raw: Raw, }, + ProviderRequestRetried { + retry: RetriedRequest, + }, } impl> CallEvent { @@ -76,6 +82,11 @@ impl> CallEvent { + CallEvent::Execution(ExecutionEvent::ProviderRequestRetried { + retry: retry.clone(), + }) + } Self::Succeeded { timing, .. } => CallEvent::Succeeded { timing: *timing, response: (), diff --git a/litellm-rust/crates/host/src/machine/context.rs b/litellm-rust/crates/host/src/machine/context.rs index 1e53605ed7b..dacb470e97f 100644 --- a/litellm-rust/crates/host/src/machine/context.rs +++ b/litellm-rust/crates/host/src/machine/context.rs @@ -5,7 +5,7 @@ use litellm_coroutine::Co; use super::coroutine::MachineFault; use crate::{ - interceptors::{RawResponse, RequestContext, WireRequest}, + interceptors::{RawResponse, RequestContext, RetriedRequest, WireRequest}, protocol::{HostRequest, InterceptRequest, Protocol, Reply, StreamDelivery}, }; @@ -119,6 +119,14 @@ where }) .await? } + + async fn provider_request_retried(&self, retry: RetriedRequest) -> Result<(), P::Error> { + self.0 + .request_reply(|reply| { + HostRequest::Intercept(InterceptRequest::ProviderRequestRetried { retry, reply }) + }) + .await? + } } pub struct StreamSender(Channel

); diff --git a/litellm-rust/crates/host/src/protocol.rs b/litellm-rust/crates/host/src/protocol.rs index c24557b0348..5ec566b6a65 100644 --- a/litellm-rust/crates/host/src/protocol.rs +++ b/litellm-rust/crates/host/src/protocol.rs @@ -2,7 +2,7 @@ use std::ops::ControlFlow; pub use litellm_coroutine::{Abandoned, Answer, Reply, reply}; -use crate::interceptors::{RawResponse, RequestContext, WireRequest}; +use crate::interceptors::{RawResponse, RequestContext, RetriedRequest, WireRequest}; pub trait Protocol: Send + Sync + 'static { type Request: Send + 'static; @@ -33,6 +33,10 @@ pub enum InterceptRequest { raw: RawResponse, reply: Reply>, }, + ProviderRequestRetried { + retry: RetriedRequest, + reply: Reply>, + }, } pub enum StreamDelivery { diff --git a/litellm-rust/crates/inference-messages/src/constants.rs b/litellm-rust/crates/inference-messages/src/constants.rs index c0b023be7c7..867b2b50da4 100644 --- a/litellm-rust/crates/inference-messages/src/constants.rs +++ b/litellm-rust/crates/inference-messages/src/constants.rs @@ -2,3 +2,5 @@ /// seconds. Mirrors the Python Anthropic Messages default. The per-request /// timeout from the caller still overrides this on the request builder. pub(crate) const MESSAGES_TIMEOUT_SECS: u64 = 600; + +pub(crate) const MESSAGES_HTTP_ERROR_ATTEMPTS: usize = 2; diff --git a/litellm-rust/crates/inference-messages/src/handler.rs b/litellm-rust/crates/inference-messages/src/handler.rs index ebf4118015a..1694aa6a9a3 100644 --- a/litellm-rust/crates/inference-messages/src/handler.rs +++ b/litellm-rust/crates/inference-messages/src/handler.rs @@ -31,7 +31,7 @@ use serde_json::{Map, Value}; use super::{ Error, MessagesBody, MessagesCall, MessagesCallResponse, MessagesRoute, common_utils::{MessagesProvider, merge_messages_header_sources, string_headers}, - constants::MESSAGES_TIMEOUT_SECS, + constants::{MESSAGES_HTTP_ERROR_ATTEMPTS, MESSAGES_TIMEOUT_SECS}, route::Messages, types::invalid_request, }; @@ -45,6 +45,7 @@ struct ProviderCall { timeout: Option, stream: bool, whole_sse_frames: bool, + transform_context: MessagesTransformContext, } impl ProviderCall { @@ -225,6 +226,7 @@ async fn provider_call( timeout: timeout.or_else(|| shaping.settings.request_timeout()), stream, whole_sse_frames: shaping.settings.whole_sse_frames, + transform_context, }) } @@ -308,33 +310,18 @@ async fn call_provider( request: ProviderCall, context: &CallContext<'_, impl Interceptors>, ) -> Result { + let timeout = request + .timeout + .unwrap_or(Duration::from_secs(MESSAGES_TIMEOUT_SECS)); + let response = send_recovering(http, &request, timeout, context).await?; let ProviderCall { identity, - wire, provider, - signer, - timeout, stream, whole_sse_frames, + .. } = 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 { - headers: wire.headers, - signer, - }, - &wire.url, - &wire.body, - timeout, - stream, - ) - .await?; - if !response.status().is_success() { - return Err(provider_error(response, timeout).await); - } let config = provider.config(); if stream { let streamed = @@ -363,6 +350,55 @@ async fn call_provider( decode_response(config, &identity.model, &text).map(MessagesCallResponse::Complete) } +async fn send_recovering( + http: &litellm_http::Client, + request: &ProviderCall, + timeout: Duration, + context: &CallContext<'_, impl Interceptors>, +) -> Result { + let config = request.provider.config(); + let mut body = std::borrow::Cow::Borrowed(&request.wire.body); + let mut attempt = 1; + loop { + log_request_body(request.provider.as_str(), request.stream, &body); + let response = send( + http, + Authenticated { + headers: request.wire.headers.clone(), + signer: request.signer.clone(), + }, + &request.wire.url, + &body, + timeout, + request.stream, + ) + .await?; + if response.status().is_success() { + return Ok(response); + } + let failure = provider_failure(response, timeout).await?; + let retried = if attempt == MESSAGES_HTTP_ERROR_ATTEMPTS { + None + } else { + config.retry_after_http_error( + failure.status, + &failure.text, + &body, + &request.transform_context, + ) + }; + let Some(retried) = retried else { + return Err(failure.into()); + }; + attempt += 1; + tracing::debug!( + "Anthropic /v1/messages: retrying after a recoverable HTTP error (attempt {attempt}/{MESSAGES_HTTP_ERROR_ATTEMPTS})." + ); + body = std::borrow::Cow::Owned(retried.body.clone()); + context.provider_request_retried(retried).await?; + } +} + fn serialize_failure(err: serde_json::Error) -> Error { Error::invalid_request(litellm_llms::ErrorDetail::failed( "Anthropic messages request serialization", @@ -412,18 +448,38 @@ fn timed_chunks( ) } +struct ProviderFailure { + status: u16, + text: String, + headers: Vec<(String, String)>, +} + +impl From for Error { + fn from(failure: ProviderFailure) -> Self { + litellm_http::Error::upstream_with_headers(failure.status, failure.text, failure.headers) + .into() + } +} + /// 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 { +async fn provider_failure( + response: reqwest::Response, + timeout: Duration, +) -> Result { let status = response.status().as_u16(); 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_with_headers(status, text, headers).into() + Ok(ProviderFailure { + status, + text, + headers, + }) } - Ok(Err(error)) => litellm_http::Error::from(error).into(), - Err(elapsed) => litellm_http::Error::read_timeout(elapsed).into(), + Ok(Err(error)) => Err(litellm_http::Error::from(error).into()), + Err(elapsed) => Err(litellm_http::Error::read_timeout(elapsed).into()), } } diff --git a/litellm-rust/crates/inference-messages/tests/messages/main.rs b/litellm-rust/crates/inference-messages/tests/messages/main.rs index fe203287766..e67556af5f3 100644 --- a/litellm-rust/crates/inference-messages/tests/messages/main.rs +++ b/litellm-rust/crates/inference-messages/tests/messages/main.rs @@ -22,6 +22,7 @@ use support::*; mod host; mod request; mod response; +mod retry; mod secrets; mod stream; mod tencent; diff --git a/litellm-rust/crates/inference-messages/tests/messages/retry.rs b/litellm-rust/crates/inference-messages/tests/messages/retry.rs new file mode 100644 index 00000000000..a7df57551a9 --- /dev/null +++ b/litellm-rust/crates/inference-messages/tests/messages/retry.rs @@ -0,0 +1,218 @@ +use futures_util::TryStreamExt; +use litellm_host::{ + interceptors::{RetriedRequest, RetryReason}, + lifecycle::{CallEvent, ExecutionEvent}, +}; +use litellm_inference_messages::MessagesCallResponse; +use litellm_inference_testing::RecordingSecrets; +use litellm_llms::base_llm::messages::context::MessagesModelCapabilities; +use rstest::rstest; + +use super::*; + +const INVALID_SIGNATURE: &str = "messages.1.content.0: Invalid `signature` in `thinking` block"; +const BOUND_ELSEWHERE: &str = "messages.1.content.0: Invalid `signature` in `thinking` block. The block is bound to a different conversation."; + +fn rejection(status: u16, message: &str) -> ResponseTemplate { + status_response( + status, + json!({"type": "error", "error": {"type": "invalid_request_error", "message": message}}), + ) +} + +fn sse_response() -> ResponseTemplate { + ResponseTemplate::new(200).set_body_raw( + "event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n", + "text/event-stream", + ) +} + +fn replay(call: MessagesCall, api_base: String, stream: bool, thinking: Value) -> MessagesCall { + MessagesCall { + api_key: Some("sk-ant".into()), + api_base: Some(api_base), + ..with_fields( + call, + json!({ + "max_tokens": 4096, + "stream": stream, + "thinking": thinking, + "messages": [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": [ + {"type": "thinking", "thinking": "plan", "signature": "sig"}, + {"type": "text", "text": "hello"} + ]}, + {"role": "assistant", "content": [{"type": "redacted_thinking", "data": "opaque"}]}, + {"role": "user", "content": "again"} + ] + }), + ) + } +} + +fn enabled() -> Value { + json!({"type": "enabled", "budget_tokens": 2048}) +} + +async fn execute( + host: &RecordingCall, +) -> Result<(MessagesCallResponse, Vec), Error> { + let route = messages_route(Arc::new(RecordingSecrets::empty())); + let response = route + .execute(host.request()?, host, Some(host.events.0.sender.clone())) + .await?; + match response { + MessagesCallResponse::Stream { head, chunks } => Ok(( + MessagesCallResponse::Stream { + head, + chunks: Box::pin(futures_util::stream::empty()), + }, + chunks.try_collect().await?, + )), + complete => Ok((complete, Vec::new())), + } +} + +fn observed_retries(host: &RecordingCall) -> Vec { + host.events + .0 + .lock() + .unwrap() + .iter() + .filter_map(|event| match event { + CallEvent::Execution(ExecutionEvent::ProviderRequestRetried { retry }) => { + Some(retry.clone()) + } + _ => None, + }) + .collect() +} + +#[rstest] +#[case::invalid_signature(INVALID_SIGNATURE)] +#[case::missing_signature( + "messages.1.content.0.thinking.signature.str: Input should be a valid string" +)] +#[case::empty_thinking("messages.1.content.0.thinking: each thinking block must contain thinking")] +#[case::unpinned_binding_mismatch(BOUND_ELSEWHERE)] +#[tokio::test] +async fn a_rejected_thinking_replay_is_sent_once_more_without_thinking( + call: MessagesCall, + #[case] message: &str, + #[values(false, true)] stream: bool, + #[values("anthropic", "azure_ai", "deepseek", "minimax")] provider: &str, +) { + let success = if stream { + sse_response() + } else { + message_response() + }; + let upstream = upstream([rejection(400, message), success]).await; + let host = RecordingCall::::new(replay( + MessagesCall { + custom_llm_provider: Some(provider.into()), + ..call + }, + upstream.uri(), + stream, + enabled(), + )); + + let (response, chunks) = execute(&host).await.expect("the retry succeeds"); + + let [first, second] = <[_; 2]>::try_from(received(&upstream).await).unwrap(); + assert_eq!(first.json()["thinking"], enabled()); + assert_eq!( + second.json()["messages"], + json!([ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": [{"type": "text", "text": "hello"}]}, + {"role": "user", "content": "again"} + ]) + ); + let mut expected = first.json(); + expected.as_object_mut().unwrap().shift_remove("thinking"); + expected["messages"] = second.json()["messages"].clone(); + assert_eq!(second.json(), expected); + assert_eq!(first.header("x-hook"), second.header("x-hook")); + let expected_retry = RetriedRequest { + reasons: vec![RetryReason::ThinkingSignature], + body: second.json(), + }; + assert_eq!(*host.retried.lock().unwrap(), vec![expected_retry.clone()]); + assert_eq!(observed_retries(&host), vec![expected_retry]); + assert_eq!( + matches!(response, MessagesCallResponse::Stream { .. }), + stream + ); + assert_eq!(chunks.is_empty(), !stream); +} + +#[rstest] +#[tokio::test] +async fn a_second_rejection_is_the_error_the_caller_sees( + call: MessagesCall, + #[values(false, true)] stream: bool, +) { + let upstream = upstream([ + rejection(400, INVALID_SIGNATURE), + rejection( + 400, + "messages.0.content.0: Invalid `signature` in `thinking` block, again", + ), + ]) + .await; + let host = RecordingCall::::new(replay(call, upstream.uri(), stream, enabled())); + + let error = execute(&host).await.err().expect("the retry fails"); + + assert_eq!(received(&upstream).await.len(), 2); + let (status, body) = error.upstream_response().unwrap(); + assert_eq!(status, 400); + assert!(body.contains("again"), "{body}"); + assert_eq!(host.retried.lock().unwrap().len(), 1); +} + +#[rstest] +#[case::not_a_400(429, INVALID_SIGNATURE, json!({"type": "adaptive"}))] +#[case::unrelated_400(400, "max_tokens: 4096 > 64000", json!({"type": "adaptive"}))] +#[case::signature_without_thinking(400, "Invalid `signature` in `tool_use` block", json!({"type": "adaptive"}))] +#[case::caller_pinned_binding( + 400, + BOUND_ELSEWHERE, + json!({"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}) +)] +#[tokio::test] +async fn other_failures_reach_the_caller_without_a_retry( + call: MessagesCall, + #[case] status: u16, + #[case] message: &str, + #[case] thinking: Value, +) { + let upstream = upstream([rejection(status, message), message_response()]).await; + let adaptive = MessagesShaping { + capabilities: MessagesModelCapabilities { + supports_adaptive_thinking: true, + ..MessagesModelCapabilities::default() + }, + ..MessagesShaping::default() + }; + let host = RecordingCall::::new(replay( + MessagesCall { + shaping: adaptive, + ..call + }, + upstream.uri(), + false, + thinking.clone(), + )); + + let error = execute(&host).await.err().expect("the call fails"); + + let sent = only_request(&upstream).await; + assert_eq!(sent.json()["thinking"], thinking); + assert_eq!(error.upstream_response().unwrap().0, status); + assert!(host.retried.lock().unwrap().is_empty()); + assert!(observed_retries(&host).is_empty()); +} diff --git a/litellm-rust/crates/inference-messages/tests/support/mod.rs b/litellm-rust/crates/inference-messages/tests/support/mod.rs index 3cecc865070..9115b6e89bb 100644 --- a/litellm-rust/crates/inference-messages/tests/support/mod.rs +++ b/litellm-rust/crates/inference-messages/tests/support/mod.rs @@ -117,6 +117,7 @@ pub struct RecordingCall { pub events: Arc, pub chunks: Mutex>, pub head: Mutex>, + pub retried: Mutex>, } #[derive(Default)] @@ -167,6 +168,7 @@ impl RecordingCall

{ events: Arc::new(CallEvents::default()), chunks: Mutex::new(Vec::new()), head: Mutex::new(None), + retried: Mutex::new(Vec::new()), } } } @@ -195,6 +197,14 @@ impl litellm_host::interceptors::Intercepto ) -> Result<(), P::Error> { Ok(()) } + + async fn provider_request_retried( + &self, + retry: litellm_host::interceptors::RetriedRequest, + ) -> Result<(), P::Error> { + self.retried.lock().unwrap().push(retry); + Ok(()) + } } impl

RecordingCall

diff --git a/litellm-rust/crates/inference/src/context.rs b/litellm-rust/crates/inference/src/context.rs index ef2635d24a1..39b259a4c97 100644 --- a/litellm-rust/crates/inference/src/context.rs +++ b/litellm-rust/crates/inference/src/context.rs @@ -1,6 +1,6 @@ use litellm_cache_response::CachePolicy; use litellm_host::{ - interceptors::{ExecutionFacts, Interceptors, RawResponse, ResultSource}, + interceptors::{ExecutionFacts, Interceptors, RawResponse, ResultSource, RetriedRequest}, lifecycle::{CallEvent, ExecutionEvent}, observation::ObservationSender, }; @@ -57,4 +57,15 @@ impl<'a, I: Interceptors> CallContext<'a, I> { .await .map_err(Error::post_call) } + + pub async fn provider_request_retried(&self, retry: RetriedRequest) -> Result<(), Error> { + if let Some(observers) = &self.observers { + observers.emit(CallEvent::Execution( + ExecutionEvent::ProviderRequestRetried { + retry: retry.clone(), + }, + )); + } + self.interceptors.provider_request_retried(retry).await + } } diff --git a/litellm-rust/crates/llms/src/anthropic/common_utils.rs b/litellm-rust/crates/llms/src/anthropic/common_utils.rs index 0854820358b..f11416ea720 100644 --- a/litellm-rust/crates/llms/src/anthropic/common_utils.rs +++ b/litellm-rust/crates/llms/src/anthropic/common_utils.rs @@ -650,6 +650,62 @@ pub fn flatten_unencrypted_web_search_results(messages: Vec) -> Vec bool { + let lower = error_text.to_lowercase(); + if !lower.contains("thinking") { + return false; + } + (lower.contains("signature") && (lower.contains("invalid") || lower.contains("valid string"))) + || lower.contains("must contain thinking") +} + +pub fn is_anthropic_thinking_binding_mismatch_error(error_text: &str) -> bool { + error_text + .to_lowercase() + .contains("bound to a different conversation") +} + +pub fn anthropic_request_errors_on_thinking_binding_mismatch(request_body: &Value) -> bool { + request_body + .pointer("/thinking/block_binding/prefix_mismatch_behavior") + .and_then(Value::as_str) + == Some("error") +} + +fn is_thinking_value_block(block: &Value) -> bool { + matches!( + block.get("type").and_then(Value::as_str), + Some("thinking" | "redacted_thinking") + ) +} + +fn without_thinking_value_blocks(message: Value) -> Option { + let Value::Object(mut message) = message else { + return Some(message); + }; + if let Some(Value::Array(content)) = message.get_mut("content") { + content.retain(|block| !is_thinking_value_block(block)); + if content.is_empty() { + return None; + } + } + Some(Value::Object(message)) +} + +pub fn strip_thinking_blocks_from_anthropic_messages_request(request_body: Value) -> Value { + let Value::Object(mut body) = request_body else { + return request_body; + }; + if let Some(Value::Array(messages)) = body.get_mut("messages") { + *messages = std::mem::take(messages) + .into_iter() + .filter_map(without_thinking_value_blocks) + .collect(); + } + body.shift_remove("thinking"); + Value::Object(body) +} + #[cfg(test)] mod tests { use crate::base_llm::messages::context::SupportedEffortTiers; @@ -1974,4 +2030,115 @@ mod tests { 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); } + + fn shared_cases(table: &str) -> Vec<(String, String, bool)> { + let fixture: Value = serde_json::from_str(include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/anthropic_messages_http_error_retry.json" + ))) + .unwrap(); + fixture[table] + .as_array() + .unwrap() + .iter() + .map(|case| { + ( + case["name"].as_str().unwrap().to_string(), + case["error_text"].as_str().unwrap().to_string(), + case["expected"].as_bool().unwrap(), + ) + }) + .collect() + } + + #[rstest] + #[case::invalid_thinking_block( + "invalid_thinking_block_error", + is_anthropic_invalid_thinking_block_error + )] + #[case::thinking_binding_mismatch( + "thinking_binding_mismatch_error", + is_anthropic_thinking_binding_mismatch_error + )] + fn error_texts_match_the_cases_shared_with_python( + #[case] table: &str, + #[case] recognizes: fn(&str) -> bool, + ) { + let cases = shared_cases(table); + assert!(!cases.is_empty()); + let mismatched: Vec<&String> = cases + .iter() + .filter(|(_, error_text, expected)| recognizes(error_text) != *expected) + .map(|(name, _, _)| name) + .collect(); + assert_eq!(mismatched, Vec::<&String>::new()); + } + + #[rstest] + #[case::pinned(json!({"thinking": {"block_binding": {"prefix_mismatch_behavior": "error"}}}), true)] + #[case::other_behavior(json!({"thinking": {"block_binding": {"prefix_mismatch_behavior": "drop"}}}), false)] + #[case::string_binding(json!({"thinking": {"block_binding": "strict"}}), false)] + #[case::no_thinking(json!({"messages": []}), false)] + #[case::not_an_object(json!(["thinking"]), false)] + fn a_pinned_binding_is_the_caller_asking_for_the_error( + #[case] body: Value, + #[case] expected: bool, + ) { + assert_eq!( + anthropic_request_errors_on_thinking_binding_mismatch(&body), + expected + ); + } + + #[rstest] + #[case::mixed_turn( + json!({"model": "m", "thinking": {"type": "enabled"}, "messages": [ + {"role": "assistant", "content": [ + {"type": "thinking", "thinking": "t", "signature": "s"}, + {"type": "text", "text": "a"}, + {"type": "redacted_thinking", "data": "d"} + ]} + ]}), + json!({"model": "m", "messages": [{"role": "assistant", "content": [{"type": "text", "text": "a"}]}]}) + )] + #[case::thinking_only_and_empty_turns_are_dropped( + json!({"messages": [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": [{"type": "redacted_thinking", "data": "d"}]}, + {"role": "assistant", "content": []}, + {"role": "user", "content": "again"} + ]}), + json!({"messages": [{"role": "user", "content": "hi"}, {"role": "user", "content": "again"}]}) + )] + #[case::non_object_entries_are_kept( + json!({"messages": ["raw", {"role": "user", "content": ["raw", {"type": "thinking"}]}]}), + json!({"messages": ["raw", {"role": "user", "content": ["raw"]}]}) + )] + #[case::messages_not_a_list( + json!({"messages": "hi", "thinking": {"type": "adaptive"}}), + json!({"messages": "hi"}) + )] + fn retrying_strips_thinking_from_the_replay_and_the_request( + #[case] body: Value, + #[case] expected: Value, + ) { + assert_eq!( + strip_thinking_blocks_from_anthropic_messages_request(body), + expected + ); + } + + #[test] + fn stripping_keeps_the_order_of_the_remaining_fields() { + let stripped = strip_thinking_blocks_from_anthropic_messages_request( + json!({"model": "m", "thinking": {}, "messages": [], "max_tokens": 1}), + ); + let keys: Vec<&str> = stripped + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect(); + assert_eq!(keys, ["model", "messages", "max_tokens"]); + } } diff --git a/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs b/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs index fb7b782e599..bb1f63bc029 100644 --- a/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/anthropic/messages/transformation.rs @@ -1,4 +1,5 @@ use litellm_auth::CredentialPlacement; +use litellm_host::interceptors::{RetriedRequest, RetryReason}; use litellm_http::request::{with_header, without_headers}; use litellm_llms_types::{ formats::messages::{ @@ -18,10 +19,13 @@ use crate::{ Error, anthropic::common_utils::{ ANTHROPIC_API_BASE_ENV, ANTHROPIC_API_KEY_ENV, ANTHROPIC_AUTH_TOKEN_ENV, - ANTHROPIC_BASE_URL_ENV, OauthHandling, complete_anthropic_url, existing_betas, - get_auth_header, has_advisor_tool, has_anthropic_credential, is_tool_search_used, - merge_beta_headers, optionally_handle_anthropic_oauth, requires_native_compaction_beta, - strip_advisor_blocks, strip_encrypted_reasoning_blocks, + ANTHROPIC_BASE_URL_ENV, OauthHandling, + anthropic_request_errors_on_thinking_binding_mismatch, complete_anthropic_url, + existing_betas, get_auth_header, has_advisor_tool, has_anthropic_credential, + is_anthropic_invalid_thinking_block_error, is_anthropic_thinking_binding_mismatch_error, + is_tool_search_used, merge_beta_headers, optionally_handle_anthropic_oauth, + requires_native_compaction_beta, strip_advisor_blocks, strip_encrypted_reasoning_blocks, + strip_thinking_blocks_from_anthropic_messages_request, }, base_llm::{ auth::AuthScheme, @@ -125,6 +129,16 @@ impl BaseMessagesConfig for AnthropicMessagesConfig { ) } + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn headers_for_url(&self, headers: Headers, url: &str) -> Headers { if !is_foundry_url(url) { return headers; @@ -167,6 +181,55 @@ pub(crate) fn is_claude_model(model: &str) -> bool { model.to_lowercase().contains("claude") } +pub fn should_retry_anthropic_messages_on_http_error( + status: u16, + error_text: &str, + request_body: &Value, + _context: &MessagesTransformContext, +) -> bool { + if status != 400 { + return false; + } + let caller_asked_for_this_error = + anthropic_request_errors_on_thinking_binding_mismatch(request_body) + && is_anthropic_thinking_binding_mismatch_error(error_text); + !caller_asked_for_this_error && is_anthropic_invalid_thinking_block_error(error_text) +} + +pub fn transform_anthropic_messages_request_on_http_error( + status: u16, + error_text: &str, + request_body: Value, +) -> RetriedRequest { + if status == 400 && is_anthropic_invalid_thinking_block_error(error_text) { + return RetriedRequest { + reasons: vec![RetryReason::ThinkingSignature], + body: strip_thinking_blocks_from_anthropic_messages_request(request_body), + }; + } + RetriedRequest { + reasons: Vec::new(), + body: request_body, + } +} + +pub fn retry_anthropic_messages_after_http_error( + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, +) -> Option { + should_retry_anthropic_messages_on_http_error(status, error_text, request_body, context).then( + || { + transform_anthropic_messages_request_on_http_error( + status, + error_text, + request_body.clone(), + ) + }, + ) +} + pub(crate) fn transform_messages_request( request: MessagesRequest, context: &MessagesTransformContext, @@ -1454,4 +1517,51 @@ mod tests { sent ); } + + #[test] + fn the_http_error_retry_matches_the_cases_shared_with_python() { + let fixture: Value = serde_json::from_str(include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/tests/fixtures/anthropic_messages_http_error_retry.json" + ))) + .unwrap(); + let table = &fixture["http_error_retry"]; + let cases = table["cases"].as_array().unwrap(); + assert!(!cases.is_empty()); + let mismatched: Vec<&str> = cases + .iter() + .filter(|case| { + let mut body = table["request"].clone(); + body["thinking"] = case["thinking"].clone(); + let status = u16::try_from(case["status"].as_u64().unwrap()).unwrap(); + let error_text = case["error_text"].as_str().unwrap(); + let retried = should_retry_anthropic_messages_on_http_error( + status, + error_text, + &body, + &MessagesTransformContext::default(), + ); + let stripped = case["stripped"].as_bool().unwrap(); + let transformed = transform_anthropic_messages_request_on_http_error( + status, + error_text, + body.clone(), + ); + let expected = if stripped { + RetriedRequest { + reasons: vec![RetryReason::ThinkingSignature], + body: table["stripped_request"].clone(), + } + } else { + RetriedRequest { + reasons: Vec::new(), + body, + } + }; + retried != case["retried"].as_bool().unwrap() || transformed != expected + }) + .map(|case| case["name"].as_str().unwrap()) + .collect(); + assert_eq!(mismatched, Vec::<&str>::new()); + } } 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 43fd3c897cd..037d1cdb722 100644 --- a/litellm-rust/crates/llms/src/azure_ai/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/azure_ai/messages/transformation.rs @@ -1,3 +1,5 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; +use litellm_host::interceptors::RetriedRequest; use litellm_http::request::{has_bearer_auth, has_header}; use litellm_llms_types::{ formats::messages::{MessagesOptionalParams, MessagesRequest}, @@ -5,6 +7,7 @@ use litellm_llms_types::{ }; use litellm_router_types::LitellmParams; use litellm_secret_value::SecretValue; +use serde_json::Value; use crate::{ Error, @@ -42,6 +45,16 @@ impl BaseMessagesConfig for AzureAnthropicMessagesConfig { } } + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn shape_request( &self, request: MessagesRequest, diff --git a/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs b/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs index 8d0b377ca85..cdb9c0fb8d5 100644 --- a/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/base_llm/messages/transformation.rs @@ -1,3 +1,4 @@ +use litellm_host::interceptors::RetriedRequest; use litellm_llms_types::formats::messages::{ MessagesOptionalParams, MessagesRequest, MessagesResponse, }; @@ -99,6 +100,16 @@ pub trait BaseMessagesConfig: Sync { fn wire_body(&self, body: Value) -> Value { body } + + fn retry_after_http_error( + &self, + _status: u16, + _error_text: &str, + _request_body: &Value, + _context: &MessagesTransformContext, + ) -> Option { + None + } } #[cfg(test)] diff --git a/litellm-rust/crates/llms/src/bedrock/claude_platform/messages/transformation.rs b/litellm-rust/crates/llms/src/bedrock/claude_platform/messages/transformation.rs index 0567178d5ed..aedf72e5615 100644 --- a/litellm-rust/crates/llms/src/bedrock/claude_platform/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/bedrock/claude_platform/messages/transformation.rs @@ -1,3 +1,5 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; +use litellm_host::interceptors::RetriedRequest; use std::sync::LazyLock; use litellm_auth::{AwsParams, CredentialPlacement, ParamKind, ParamSpec}; @@ -56,6 +58,16 @@ impl BaseMessagesConfig for BedrockClaudePlatformMessagesConfig { } } + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn shape_request( &self, request: MessagesRequest, diff --git a/litellm-rust/crates/llms/src/bedrock/messages/invoke_transformations/anthropic_claude_transformation.rs b/litellm-rust/crates/llms/src/bedrock/messages/invoke_transformations/anthropic_claude_transformation.rs index 1d380c917f6..77014d068d1 100644 --- a/litellm-rust/crates/llms/src/bedrock/messages/invoke_transformations/anthropic_claude_transformation.rs +++ b/litellm-rust/crates/llms/src/bedrock/messages/invoke_transformations/anthropic_claude_transformation.rs @@ -1,4 +1,6 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; use litellm_core_utils::decline::Decline; +use litellm_host::interceptors::RetriedRequest; use std::convert::Infallible; use crate::{ @@ -103,6 +105,16 @@ fn invoke_url( } impl BaseMessagesConfig for AmazonAnthropicClaudeMessagesConfig { + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn shape_request( &self, request: MessagesRequest, diff --git a/litellm-rust/crates/llms/src/bedrock_mantle/messages/transformation.rs b/litellm-rust/crates/llms/src/bedrock_mantle/messages/transformation.rs index 2e0121c9571..58d76662977 100644 --- a/litellm-rust/crates/llms/src/bedrock_mantle/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/bedrock_mantle/messages/transformation.rs @@ -1,3 +1,5 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; +use litellm_host::interceptors::RetriedRequest; use std::sync::LazyLock; use litellm_auth::{AwsParams, CredentialPlacement}; @@ -57,6 +59,16 @@ pub const BEDROCK_MANTLE_MESSAGES_CONFIG: BedrockMantleAnthropicMessagesConfig = BedrockMantleAnthropicMessagesConfig; impl BaseMessagesConfig for BedrockMantleAnthropicMessagesConfig { + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn validate_environment( &self, headers: Headers, diff --git a/litellm-rust/crates/llms/src/deepseek/messages/transformation.rs b/litellm-rust/crates/llms/src/deepseek/messages/transformation.rs index dce9178e7b9..2c4945e41bf 100644 --- a/litellm-rust/crates/llms/src/deepseek/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/deepseek/messages/transformation.rs @@ -1,4 +1,6 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; use litellm_auth::CredentialPlacement; +use litellm_host::interceptors::RetriedRequest; use litellm_secret_value::SecretValue; use litellm_core_utils::settings::resolve_non_empty; @@ -48,6 +50,16 @@ impl BaseMessagesConfig for DeepSeekAnthropicMessagesConfig { } } + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn shape_request( &self, request: MessagesRequest, diff --git a/litellm-rust/crates/llms/src/edenai/messages/transformation.rs b/litellm-rust/crates/llms/src/edenai/messages/transformation.rs index 75a8b16d1eb..d4f577e09a9 100644 --- a/litellm-rust/crates/llms/src/edenai/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/edenai/messages/transformation.rs @@ -1,4 +1,6 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; use litellm_auth::CredentialPlacement; +use litellm_host::interceptors::RetriedRequest; use litellm_secret_value::SecretValue; use litellm_core_utils::settings::resolve_non_empty; @@ -39,6 +41,16 @@ impl BaseMessagesConfig for EdenAIAnthropicMessagesConfig { } } + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn validate_environment( &self, headers: Headers, diff --git a/litellm-rust/crates/llms/src/minimax/messages/transformation.rs b/litellm-rust/crates/llms/src/minimax/messages/transformation.rs index aa467619131..b82442d9294 100644 --- a/litellm-rust/crates/llms/src/minimax/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/minimax/messages/transformation.rs @@ -1,5 +1,8 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; use litellm_auth::CredentialPlacement; +use litellm_host::interceptors::RetriedRequest; use litellm_secret_value::SecretValue; +use serde_json::Value; use litellm_core_utils::settings::resolve_non_empty; use litellm_llms_types::{ @@ -37,6 +40,16 @@ impl BaseMessagesConfig for MinimaxMessagesConfig { } } + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn shape_request( &self, request: MessagesRequest, diff --git a/litellm-rust/crates/llms/src/openrouter/messages/transformation.rs b/litellm-rust/crates/llms/src/openrouter/messages/transformation.rs index eeb3d1cb7f8..dc09f4e4149 100644 --- a/litellm-rust/crates/llms/src/openrouter/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/openrouter/messages/transformation.rs @@ -1,5 +1,8 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; use litellm_auth::CredentialPlacement; +use litellm_host::interceptors::RetriedRequest; use litellm_secret_value::SecretValue; +use serde_json::Value; use litellm_core_utils::settings::resolve_non_empty; use litellm_http::request::{has_bearer_auth, without_headers}; @@ -31,6 +34,16 @@ pub const OPENROUTER_MESSAGES_CONFIG: OpenRouterAnthropicMessagesConfig = OpenRouterAnthropicMessagesConfig; impl BaseMessagesConfig for OpenRouterAnthropicMessagesConfig { + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn validate_environment( &self, headers: Headers, diff --git a/litellm-rust/crates/llms/src/tencent/messages/transformation.rs b/litellm-rust/crates/llms/src/tencent/messages/transformation.rs index 97a359826a7..9aef16728e0 100644 --- a/litellm-rust/crates/llms/src/tencent/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/tencent/messages/transformation.rs @@ -1,5 +1,8 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; use litellm_auth::CredentialPlacement; +use litellm_host::interceptors::RetriedRequest; use litellm_secret_value::SecretValue; +use serde_json::Value; use litellm_core_utils::settings::resolve_non_empty; use litellm_llms_types::{ @@ -38,6 +41,16 @@ impl BaseMessagesConfig for TencentAnthropicMessagesConfig { } } + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn shape_request( &self, request: MessagesRequest, 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 0f3ed086eaf..18e66c5a77e 100644 --- a/litellm-rust/crates/llms/src/vertex_ai/messages/transformation.rs +++ b/litellm-rust/crates/llms/src/vertex_ai/messages/transformation.rs @@ -1,8 +1,10 @@ +use crate::anthropic::messages::transformation::retry_anthropic_messages_after_http_error; use litellm_auth::VertexParams; use litellm_auth_gcp::{ VertexConfig, get_vertex_ai_location, get_vertex_ai_project, get_vertex_ai_project_from_credentials, secret_names, }; +use litellm_host::interceptors::RetriedRequest; use litellm_http::request::{has_header, header_value, without_headers}; use litellm_llms_types::{ formats::messages::{MessagesOptionalParams, MessagesRequest, MessagesTool, OutputConfig}, @@ -53,6 +55,16 @@ impl BaseMessagesConfig for VertexAiPartnerModelsAnthropicMessagesConfig { } } + fn retry_after_http_error( + &self, + status: u16, + error_text: &str, + request_body: &Value, + context: &MessagesTransformContext, + ) -> Option { + retry_anthropic_messages_after_http_error(status, error_text, request_body, context) + } + fn shape_request( &self, request: MessagesRequest, diff --git a/litellm-rust/crates/llms/tests/fixtures/anthropic_messages_http_error_retry.json b/litellm-rust/crates/llms/tests/fixtures/anthropic_messages_http_error_retry.json new file mode 100644 index 00000000000..06fb06f316e --- /dev/null +++ b/litellm-rust/crates/llms/tests/fixtures/anthropic_messages_http_error_retry.json @@ -0,0 +1,252 @@ +{ + "invalid_thinking_block_error": [ + { + "name": "invalid_signature", + "error_text": "messages.1.content.0: Invalid `signature` in `thinking` block", + "expected": true + }, + { + "name": "invalid_signature_in_an_error_envelope", + "error_text": "{\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"messages.3.content.3: Invalid `signature` in `thinking` block\"},\"request_id\":\"req_011Ca2EtQDxp7x6RGUY2jVn9\"}", + "expected": true + }, + { + "name": "signature_not_a_string", + "error_text": "{\"message\":\"messages.2.content.0.thinking.signature.str: Input should be a valid string\"}", + "expected": true + }, + { + "name": "signature_not_a_string_bare", + "error_text": "messages.4.content.1.thinking.signature.str: Input should be a valid string", + "expected": true + }, + { + "name": "empty_thinking", + "error_text": "messages.1.content.0.thinking: each thinking block must contain thinking", + "expected": true + }, + { + "name": "bound_to_a_different_conversation", + "error_text": "messages.1.content.0: Invalid `signature` in `thinking` block. The block is bound to a different conversation.", + "expected": true + }, + { + "name": "uppercase", + "error_text": "MESSAGES.1: INVALID SIGNATURE IN THINKING BLOCK", + "expected": true + }, + { + "name": "signature_without_a_verdict", + "error_text": "thinking signature is malformed", + "expected": false + }, + { + "name": "invalid_without_thinking", + "error_text": "Invalid `signature` in `tool_use` block", + "expected": false + }, + { + "name": "model_not_found", + "error_text": "invalid_request_error: model not found", + "expected": false + }, + { + "name": "rate_limited", + "error_text": "rate limit exceeded", + "expected": false + }, + { + "name": "unrelated", + "error_text": "max_tokens: 4096 > 64000", + "expected": false + }, + { + "name": "empty", + "error_text": "", + "expected": false + } + ], + "thinking_binding_mismatch_error": [ + { + "name": "bound_elsewhere", + "error_text": "messages.1.content.0: Invalid `signature` in `thinking` block. The block is bound to a different conversation.", + "expected": true + }, + { + "name": "mixed_case", + "error_text": "The block is Bound To A Different Conversation.", + "expected": true + }, + { + "name": "plain_signature", + "error_text": "messages.1.content.0: Invalid `signature` in `thinking` block", + "expected": false + } + ], + "http_error_retry": { + "request": { + "model": "claude-sonnet-4-5", + "max_tokens": 4096, + "messages": [ + { + "role": "user", + "content": "hi" + }, + { + "role": "assistant", + "content": [ + { + "type": "thinking", + "thinking": "plan", + "signature": "sig" + }, + { + "type": "text", + "text": "hello" + } + ] + }, + { + "role": "assistant", + "content": [ + { + "type": "redacted_thinking", + "data": "opaque" + } + ] + }, + { + "role": "user", + "content": "again" + } + ] + }, + "stripped_request": { + "model": "claude-sonnet-4-5", + "max_tokens": 4096, + "messages": [ + { + "role": "user", + "content": "hi" + }, + { + "role": "assistant", + "content": [ + { + "type": "text", + "text": "hello" + } + ] + }, + { + "role": "user", + "content": "again" + } + ] + }, + "cases": [ + { + "name": "invalid_signature", + "status": 400, + "error_text": "messages.1.content.0: Invalid `signature` in `thinking` block", + "thinking": { + "type": "enabled", + "budget_tokens": 2048 + }, + "retried": true, + "stripped": true + }, + { + "name": "missing_signature", + "status": 400, + "error_text": "messages.1.content.0.thinking.signature.str: Input should be a valid string", + "thinking": { + "type": "enabled", + "budget_tokens": 2048 + }, + "retried": true, + "stripped": true + }, + { + "name": "empty_thinking", + "status": 400, + "error_text": "messages.1.content.0.thinking: each thinking block must contain thinking", + "thinking": { + "type": "enabled", + "budget_tokens": 2048 + }, + "retried": true, + "stripped": true + }, + { + "name": "unpinned_binding_mismatch", + "status": 400, + "error_text": "messages.1.content.0: Invalid `signature` in `thinking` block. The block is bound to a different conversation.", + "thinking": { + "type": "adaptive" + }, + "retried": true, + "stripped": true + }, + { + "name": "caller_pinned_binding", + "status": 400, + "error_text": "messages.1.content.0: Invalid `signature` in `thinking` block. The block is bound to a different conversation.", + "thinking": { + "type": "adaptive", + "block_binding": { + "prefix_mismatch_behavior": "error" + } + }, + "retried": false, + "stripped": true + }, + { + "name": "caller_pinned_binding_other_signature_error", + "status": 400, + "error_text": "messages.1.content.0: Invalid `signature` in `thinking` block", + "thinking": { + "type": "adaptive", + "block_binding": { + "prefix_mismatch_behavior": "error" + } + }, + "retried": true, + "stripped": true + }, + { + "name": "not_a_400", + "status": 429, + "error_text": "messages.1.content.0: Invalid `signature` in `thinking` block", + "thinking": { + "type": "enabled", + "budget_tokens": 2048 + }, + "retried": false, + "stripped": false + }, + { + "name": "unrelated_400", + "status": 400, + "error_text": "max_tokens: 4096 > 64000", + "thinking": { + "type": "enabled", + "budget_tokens": 2048 + }, + "retried": false, + "stripped": false + }, + { + "name": "signature_without_thinking", + "status": 400, + "error_text": "Invalid `signature` in `tool_use` block", + "thinking": { + "type": "enabled", + "budget_tokens": 2048 + }, + "retried": false, + "stripped": false + } + ] + } +} diff --git a/litellm-rust/crates/llms/tests/messages_http_error_retry.rs b/litellm-rust/crates/llms/tests/messages_http_error_retry.rs new file mode 100644 index 00000000000..1a2635f701f --- /dev/null +++ b/litellm-rust/crates/llms/tests/messages_http_error_retry.rs @@ -0,0 +1,72 @@ +use litellm_host::interceptors::{RetriedRequest, RetryReason}; +use litellm_llms::{ + anthropic::messages::transformation::ANTHROPIC_MESSAGES_CONFIG, + azure_ai::messages::transformation::AZURE_ANTHROPIC_MESSAGES_CONFIG, + base_llm::messages::{context::MessagesTransformContext, transformation::BaseMessagesConfig}, + bedrock::{ + claude_platform::messages::transformation::BEDROCK_CLAUDE_PLATFORM_MESSAGES_CONFIG, + messages::invoke_transformations::anthropic_claude_transformation::BEDROCK_ANTHROPIC_MESSAGES_CONFIG, + }, + bedrock_mantle::messages::transformation::BEDROCK_MANTLE_MESSAGES_CONFIG, + deepseek::messages::transformation::DEEPSEEK_ANTHROPIC_MESSAGES_CONFIG, + edenai::messages::transformation::EDENAI_MESSAGES_CONFIG, + minimax::messages::transformation::MINIMAX_MESSAGES_CONFIG, + openrouter::messages::transformation::OPENROUTER_MESSAGES_CONFIG, + tencent::messages::transformation::TENCENT_MESSAGES_CONFIG, + vertex_ai::messages::transformation::VERTEX_ANTHROPIC_MESSAGES_CONFIG, +}; +use rstest::rstest; +use serde_json::Value; + +#[rstest] +#[case::anthropic(&ANTHROPIC_MESSAGES_CONFIG)] +#[case::azure_ai(&AZURE_ANTHROPIC_MESSAGES_CONFIG)] +#[case::bedrock_invoke(&BEDROCK_ANTHROPIC_MESSAGES_CONFIG)] +#[case::bedrock_claude_platform(&BEDROCK_CLAUDE_PLATFORM_MESSAGES_CONFIG)] +#[case::bedrock_mantle(&BEDROCK_MANTLE_MESSAGES_CONFIG)] +#[case::deepseek(&DEEPSEEK_ANTHROPIC_MESSAGES_CONFIG)] +#[case::edenai(&EDENAI_MESSAGES_CONFIG)] +#[case::minimax(&MINIMAX_MESSAGES_CONFIG)] +#[case::openrouter(&OPENROUTER_MESSAGES_CONFIG)] +#[case::tencent(&TENCENT_MESSAGES_CONFIG)] +#[case::vertex_ai(&VERTEX_ANTHROPIC_MESSAGES_CONFIG)] +fn every_messages_provider_retries_like_the_python_base_config( + #[case] config: &dyn BaseMessagesConfig, +) { + let fixture: Value = serde_json::from_str(include_str!( + "fixtures/anthropic_messages_http_error_retry.json" + )) + .unwrap(); + let table = &fixture["http_error_retry"]; + let cases = table["cases"].as_array().unwrap(); + assert!(!cases.is_empty()); + let mismatched: Vec<&str> = cases + .iter() + .filter(|case| { + let mut body = table["request"].clone(); + body["thinking"] = case["thinking"].clone(); + let status = u16::try_from(case["status"].as_u64().unwrap()).unwrap(); + let expected = case["retried"].as_bool().unwrap().then(|| { + if case["stripped"].as_bool().unwrap() { + RetriedRequest { + reasons: vec![RetryReason::ThinkingSignature], + body: table["stripped_request"].clone(), + } + } else { + RetriedRequest { + reasons: Vec::new(), + body: body.clone(), + } + } + }); + config.retry_after_http_error( + status, + case["error_text"].as_str().unwrap(), + &body, + &MessagesTransformContext::default(), + ) != expected + }) + .map(|case| case["name"].as_str().unwrap()) + .collect(); + assert_eq!(mismatched, Vec::<&str>::new()); +} diff --git a/litellm/messages/dispatch.py b/litellm/messages/dispatch.py index 642ae57c03d..fa393850e76 100644 --- a/litellm/messages/dispatch.py +++ b/litellm/messages/dispatch.py @@ -143,7 +143,6 @@ _FEATURES: Final[detectors.Detectors] = MappingProxyType( Feature.MODIFY_PARAMS: detectors.modify_params, Feature.WORKLOAD_IDENTITY: route_detectors.anthropic_workload_identity, Feature.CACHE_CONTROL_INJECTION: detectors.cache_control_injection, - Feature.THINKING_SIGNATURE_RETRY: route_detectors.thinking_signature_retry, } ) diff --git a/litellm/rust_bridge/catalog.py b/litellm/rust_bridge/catalog.py index 210df7b0c10..31277713ad5 100644 --- a/litellm/rust_bridge/catalog.py +++ b/litellm/rust_bridge/catalog.py @@ -92,8 +92,6 @@ class Feature(str, Enum): """No static key, so Python mints a workload identity federation token.""" CACHE_CONTROL_INJECTION = "cache_control_injection" """Prompt caching injection that Python applies to messages and system.""" - THINKING_SIGNATURE_RETRY = "thinking_signature_retry" - """Replayed thinking blocks, which Python strips and retries once if Anthropic rejects their signature.""" STRICT_TOOLS_RETRY = "strict_tools_retry" """Strict tools under drop_params, which Python sends again without strict if Claude rejects their limits.""" CLIENT_HEADERS = "client_headers" @@ -175,7 +173,6 @@ MESSAGES_PYTHON_FEATURES: Final = (COMMON_PYTHON_FEATURES - {Feature.TIMEOUT_VAR Feature.MODIFY_PARAMS, Feature.WORKLOAD_IDENTITY, Feature.CACHE_CONTROL_INJECTION, - Feature.THINKING_SIGNATURE_RETRY, Feature.STRICT_TOOLS_RETRY, } diff --git a/litellm/rust_bridge/host/callbacks_legacy_python.py b/litellm/rust_bridge/host/callbacks_legacy_python.py index ba557f769eb..f2c3bb63ec0 100644 --- a/litellm/rust_bridge/host/callbacks_legacy_python.py +++ b/litellm/rust_bridge/host/callbacks_legacy_python.py @@ -27,6 +27,7 @@ from typing import ( import httpx from pydantic import TypeAdapter +from typing_extensions import ReadOnly, TypedDict if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging @@ -285,6 +286,37 @@ def record_response(logger: Logging, status: int, headers: Sequence[tuple[str, s details["response_headers"] = dict(response.headers) +class _ThinkingSignatureRetry(TypedDict): + kind: ReadOnly[Literal["thinking_signature"]] + + +class _StrictToolsRetry(TypedDict): + kind: ReadOnly[Literal["strict_tools"]] + dropped: ReadOnly[int] + + +_RETRY_REASONS: Final = TypeAdapter(tuple[_ThinkingSignatureRetry | _StrictToolsRetry, ...]) + + +async def retried_request( + logger: Logging, + logged_body: dict[str, object], # mutable-ok: the retry rewrites the body pre_call logged, as Python's loop does + body: Mapping[str, object], + reasons: Sequence[Mapping[str, object]], +) -> None: + from litellm.llms.anthropic.common_utils import warn_claude_strict_tools_dropped + + if logger.baseline_cache_context is not None: + await logger.invalidate_baseline_cache_estimate("retried_request") + for reason in _RETRY_REASONS.validate_python(reasons): + if reason["kind"] == "strict_tools": + warn_claude_strict_tools_dropped(reason["dropped"]) + for name in logged_body.keys() - body.keys(): + del logged_body[name] # rebind-ok: the logged body is the one logging already holds + logged_body.update({name: value for name, value in body.items() if logged_body.get(name) != value}) + logger.model_call_details.update(logged_body) + + 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 ( diff --git a/litellm/rust_bridge/messages/detectors.py b/litellm/rust_bridge/messages/detectors.py index f2129fe62d1..346ad8721cd 100644 --- a/litellm/rust_bridge/messages/detectors.py +++ b/litellm/rust_bridge/messages/detectors.py @@ -57,17 +57,6 @@ def anthropic_workload_identity(request: NativeCall) -> bool: ) -_THINKING_BLOCKS: Final = frozenset({"thinking", "redacted_thinking"}) - - -def _replays_thinking(message: object) -> bool: - content: Final = optional_sequence((optional_mapping(message) or {}).get("content")) or () - return any((optional_mapping(block) or {}).get("type") in _THINKING_BLOCKS for block in content) - - -def thinking_signature_retry(request: NativeCall) -> bool: - """Python may strip replayed thinking blocks and retry after a rejected signature.""" - return any(_replays_thinking(message) for message in optional_sequence(request.resolved.get("messages")) or ()) def strict_tools_retry(request: NativeCall) -> bool: 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 6239ce7bbc0..11622233551 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 @@ -307,6 +307,7 @@ CALLBACK_ASPECTS: Final = ( "logged_response_headers", "logged_api_key", "logged_call", + "logged_request", ) ASPECTS: Final = ("dispatch", "request", "outcome", "hidden_params", "logged", *CALLBACK_ASPECTS) @@ -344,9 +345,6 @@ 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" ) -F_THINKING_RETRY: Final = ( - "F: retry without thinking after a rejected signature is Python-only (detector thinking_signature_retry)" -) F_CACHE_INJECTION: Final = "F: prompt cache_control injection is Python-only (detector cache_control_injection)" F_STREAMED_MOCK: Final = ( "F: a mock with stream=True returns the mock dict and logs no success in Python (detector streamed_mock_response)" @@ -356,21 +354,14 @@ KNOWN_GAPS: Final[Mapping[str, Mapping[str, str | None]]] = MappingProxyType( { "response_private_keys": _gaps(outcome=S6_PRIVATE_KEYS), "missing_max_tokens": _gaps(dispatch=A10_NO_MAX_TOKENS), - "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=F_THINKING_RETRY, - logged_response_headers=F_THINKING_RETRY, - ), - "cache_control_injection_points": _gaps(request=F_CACHE_INJECTION), - "enable_prompt_caching": _gaps(request=F_CACHE_INJECTION), + "cache_control_injection_points": _gaps(request=F_CACHE_INJECTION, logged_request=F_CACHE_INJECTION), + "enable_prompt_caching": _gaps(request=F_CACHE_INJECTION, logged_request=F_CACHE_INJECTION), "mock_response_streamed": _gaps( logged=F_STREAMED_MOCK, logged_metadata=F_STREAMED_MOCK, logged_response_headers=F_STREAMED_MOCK, logged_call=F_STREAMED_MOCK, + logged_request=F_STREAMED_MOCK, ), } ) @@ -673,6 +664,7 @@ FAILURE_CASES: Final = ( _kw(timeout=1), upstream=(ResponseSpec(body=dict(TEXT_RESPONSE), chunk_delay=0.5, body_chunks=4),), gaps=_gaps( + logged_request=S15_COMPLETE_DEADLINE, outcome=S15_COMPLETE_DEADLINE, hidden_params=S15_COMPLETE_DEADLINE, logged=S15_COMPLETE_DEADLINE, @@ -774,18 +766,6 @@ SERVED_NATIVELY_CASES: Final = ( FALLBACK_CASES: Final = ( Case("mock_response_streamed", _kw(stream=True, mock_response="Hi from the mock"), upstream=()), - Case( - "thinking_replay", - _kw(messages=list(THINKING_REPLAY), thinking={"type": "enabled", "budget_tokens": 2048}, max_tokens=4096), - ), - Case( - "thinking_signature_rejected_then_retried", - _kw(messages=list(THINKING_REPLAY), thinking={"type": "enabled", "budget_tokens": 2048}, max_tokens=4096), - upstream=( - _failure(400, "invalid_request_error", "messages.1.content.0: Invalid `signature` in `thinking` block"), - _ok(), - ), - ), Case( "cache_control_injection_points", _kw(system="Answer in one word.", cache_control_injection_points=[{"location": "message", "role": "system"}]), @@ -804,6 +784,84 @@ FALLBACK_CASES: Final = ( Case("stream_timeout_kwarg", _kw(stream=True, stream_timeout=30), upstream=(_stream(TEXT_STREAM),)), ) +INVALID_SIGNATURE: Final = "messages.1.content.0: Invalid `signature` in `thinking` block" +BOUND_ELSEWHERE: Final = f"{INVALID_SIGNATURE}. The block is bound to a different conversation." +THINKING: Final[Mapping[str, JsonValue]] = MappingProxyType({"type": "enabled", "budget_tokens": 2048}) +BINDING_ERRORS: Final[Mapping[str, JsonValue]] = MappingProxyType( + {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}} +) +THINKING_ONLY_REPLAY: Final[tuple[JsonValue, ...]] = ( + {"role": "user", "content": "What is the capital of France?"}, + { + "role": "assistant", + "content": [ + {"type": "redacted_thinking", "data": "EmwKAhgBEgy"}, + {"type": "thinking", "thinking": "Paris.", "signature": "EqQBCkYIARgCKkB"}, + ], + }, + {"role": "assistant", "content": []}, + {"role": "user", "content": "And Germany?"}, +) + + +def _replay(**kwargs: object) -> Mapping[str, object]: + return _kw(messages=list(THINKING_REPLAY), thinking=dict(THINKING), max_tokens=4096, **kwargs) + + +def _rejected(message: str = INVALID_SIGNATURE, status: int = 400) -> ResponseSpec: + return _failure(status, "invalid_request_error", message) + + +THINKING_RETRY_CASES: Final = ( + Case("thinking_replay", _replay()), + Case("thinking_signature_rejected_then_retried", _replay(), upstream=(_rejected(), _ok())), + Case( + "thinking_signature_rejected_then_retried_stream", + _replay(stream=True), + upstream=(_rejected(), _stream(TEXT_STREAM)), + ), + Case( + "thinking_signature_missing_then_retried", + _replay(), + upstream=(_rejected("messages.1.content.0.thinking.signature.str: Input should be a valid string"), _ok()), + ), + Case( + "empty_thinking_rejected_then_retried_without_thinking_only_turns", + _kw(messages=list(THINKING_ONLY_REPLAY), thinking=dict(THINKING), max_tokens=4096), + upstream=(_rejected("messages.1.content.1.thinking: each thinking block must contain thinking"), _ok()), + ), + Case("thinking_signature_rejected_twice", _replay(), upstream=(_rejected(), _rejected())), + Case( + "thinking_signature_rejected_twice_stream", + _replay(stream=True), + upstream=(_rejected(), _rejected()), + ), + Case("thinking_signature_text_on_a_429_is_not_retried", _replay(), upstream=(_rejected(status=429),)), + Case( + "unrelated_400_is_not_retried", + _replay(), + upstream=(_rejected("max_tokens: 4096 > 64000, which is the maximum allowed"),), + ), + Case( + "binding_mismatch_retried_when_the_caller_did_not_pin_it", + _kw(messages=list(THINKING_REPLAY), thinking={"type": "adaptive"}, max_tokens=4096), + model=OPUS_ADAPTIVE, + upstream=(_rejected(BOUND_ELSEWHERE), _ok()), + ), + Case( + "binding_mismatch_raised_when_the_caller_pinned_it", + _kw(messages=list(THINKING_REPLAY), thinking=dict(BINDING_ERRORS), max_tokens=4096), + model=OPUS_ADAPTIVE, + upstream=(_rejected(BOUND_ELSEWHERE),), + ), + Case( + "invalid_signature_retried_even_when_the_caller_pinned_binding", + _kw(messages=list(THINKING_REPLAY), thinking=dict(BINDING_ERRORS), max_tokens=4096), + model=OPUS_ADAPTIVE, + upstream=(_rejected(), _ok()), + ), +) + CASES: Final = ( *REQUEST_CASES, *RESPONSE_CASES, @@ -811,6 +869,7 @@ CASES: Final = ( *STREAM_CASES, *SYNC_CASES, *SERVED_NATIVELY_CASES, + *THINKING_RETRY_CASES, *FALLBACK_CASES, ) @@ -1039,6 +1098,11 @@ def _logged(recorder: RecordingLogger, events: tuple[str, ...]) -> tuple[Mapping "model": details.get("model"), "stream": details.get("stream"), }, + "logged_request": { + "messages": details.get("messages"), + "thinking": details.get("thinking"), + "complete_input_dict": _mapping(details.get("additional_args")).get("complete_input_dict"), + }, } return logged, callback_kwargs diff --git a/tests/test_litellm_rust/messages/test_dispatch.py b/tests/test_litellm_rust/messages/test_dispatch.py index 309ea01fb8a..6ce91df957e 100644 --- a/tests/test_litellm_rust/messages/test_dispatch.py +++ b/tests/test_litellm_rust/messages/test_dispatch.py @@ -18,7 +18,7 @@ from litellm.messages.dispatch import ( ) from litellm.proxy import proxy_server from litellm.rust_bridge.bindings import NativeBinding -from litellm.rust_bridge.catalog import Feature, Route, RouteRule, Rules, Target +from litellm.rust_bridge.catalog import RULES, Feature, Route, RouteRule, Rules, Target from litellm.rust_bridge.configuration import Rollout from litellm.rust_bridge.messages.entrypoints import NATIVE_AMESSAGES, NATIVE_MESSAGES, NativeAmessages, NativeMessages from litellm.rust_bridge.public_call import NativeCall @@ -495,12 +495,6 @@ def _handler_cache() -> Cache: {}, ), (Feature.CACHE_CONTROL_INJECTION, "anthropic/claude-sonnet-4-5", {"enable_prompt_caching": True}, {}), - ( - Feature.THINKING_SIGNATURE_RETRY, - "anthropic/claude-sonnet-4-5", - {"messages": [{"role": "assistant", "content": [{"type": "redacted_thinking", "data": "x"}]}]}, - {}, - ), ), ) def test_each_detected_feature_reaches_its_rule( @@ -575,6 +569,46 @@ def test_proxy_request_shape_stays_native(kwargs: Mapping[str, object]) -> None: assert served is native_result +@pytest.mark.asyncio +@pytest.mark.parametrize( + "block", + ( + pytest.param({"type": "thinking", "thinking": "x", "signature": "s"}, id="thinking"), + pytest.param({"type": "redacted_thinking", "data": "x"}, id="redacted-thinking"), + ), +) +async def test_replayed_thinking_is_served_natively_under_the_shipped_rules( + monkeypatch: pytest.MonkeyPatch, block: Mapping[str, object] +) -> None: + monkeypatch.setenv("LITELLM_RUST", "1") + monkeypatch.setattr(litellm, "callbacks", []) + monkeypatch.setattr(litellm, "cache", None) + replay: Final = [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": [block, {"type": "text", "text": "hello"}]}, + {"role": "user", "content": "again"}, + ] + native_result: Final = response() + python_result: Final = response() + + async def python(*call_args: object, **python_kwargs: object) -> AnthropicMessagesResponse: # kwargs-ok: stub + return python_result + + async def native(request: NativeCall) -> AnthropicMessagesResponse: + return native_result + + served: Final = await _ADISPATCH.arun( + (16, replay, "anthropic/claude-sonnet-4-5"), + {"api_key": "sk-ant-test"}, + python=python, + binding=amessages_binding(native), + native=lambda hook, request, call_args, native_kwargs: hook(request), + rules=RULES, + ) + + assert served is native_result + + @pytest.mark.asyncio @pytest.mark.parametrize( "kwargs", diff --git a/tests/unit/llms/anthropic/test_messages_http_error_retry_cases.py b/tests/unit/llms/anthropic/test_messages_http_error_retry_cases.py new file mode 100644 index 00000000000..58f7012921f --- /dev/null +++ b/tests/unit/llms/anthropic/test_messages_http_error_retry_cases.py @@ -0,0 +1,95 @@ +import copy +from collections.abc import Mapping +from pathlib import Path +from typing import Final + +import httpx +import pytest +from pydantic import TypeAdapter +from typing_extensions import ReadOnly, TypedDict + +from litellm.llms.anthropic.common_utils import ( + is_anthropic_invalid_thinking_block_error, + is_anthropic_thinking_binding_mismatch_error, +) +from litellm.llms.anthropic.messages.native.transformation import AnthropicMessagesConfig +from litellm.llms.deepseek.messages.transformation import DeepSeekAnthropicMessagesConfig +from litellm.llms.base_llm.anthropic_messages.transformation import BaseAnthropicMessagesConfig + +FIXTURE_PATH: Final = ( + Path(__file__).parents[4] / "litellm-rust/crates/llms/tests/fixtures/anthropic_messages_http_error_retry.json" +) + + +class _TextCase(TypedDict): + name: ReadOnly[str] + error_text: ReadOnly[str] + expected: ReadOnly[bool] + + +class _RetryCase(TypedDict): + name: ReadOnly[str] + status: ReadOnly[int] + error_text: ReadOnly[str] + thinking: ReadOnly[Mapping[str, object]] + retried: ReadOnly[bool] + stripped: ReadOnly[bool] + + +class _RetryTable(TypedDict): + request: ReadOnly[Mapping[str, object]] + stripped_request: ReadOnly[Mapping[str, object]] + cases: ReadOnly[tuple[_RetryCase, ...]] + + +class _Fixture(TypedDict): + invalid_thinking_block_error: ReadOnly[tuple[_TextCase, ...]] + thinking_binding_mismatch_error: ReadOnly[tuple[_TextCase, ...]] + http_error_retry: ReadOnly[_RetryTable] + + +FIXTURE: Final = TypeAdapter(_Fixture).validate_json(FIXTURE_PATH.read_text()) +RETRY: Final = FIXTURE["http_error_retry"] + + +@pytest.mark.parametrize( + "case", [pytest.param(case, id=case["name"]) for case in FIXTURE["invalid_thinking_block_error"]] +) +def test_invalid_thinking_block_errors_match_the_shared_cases(case: _TextCase) -> None: + assert is_anthropic_invalid_thinking_block_error(case["error_text"]) is case["expected"] + + +@pytest.mark.parametrize( + "case", [pytest.param(case, id=case["name"]) for case in FIXTURE["thinking_binding_mismatch_error"]] +) +def test_binding_mismatch_errors_match_the_shared_cases(case: _TextCase) -> None: + assert is_anthropic_thinking_binding_mismatch_error(case["error_text"]) is case["expected"] + + +def _rejection(case: _RetryCase) -> httpx.HTTPStatusError: + request: Final = httpx.Request("POST", "https://api.anthropic.com/v1/messages") + return httpx.HTTPStatusError( + "rejected", + request=request, + response=httpx.Response(case["status"], text=case["error_text"], request=request), + ) + + +@pytest.mark.parametrize( + "config", + [AnthropicMessagesConfig(), DeepSeekAnthropicMessagesConfig()], + ids=["anthropic", "deepseek"], +) +@pytest.mark.parametrize("case", [pytest.param(case, id=case["name"]) for case in RETRY["cases"]]) +def test_the_http_error_retry_matches_the_shared_cases(config: BaseAnthropicMessagesConfig, case: _RetryCase) -> None: + body: Final = {**copy.deepcopy(RETRY["request"]), "thinking": case["thinking"]} + + retried: Final = config.should_retry_anthropic_messages_on_http_error( + e=_rejection(case), litellm_params={}, request_body=body + ) + transformed: Final = config.transform_anthropic_messages_request_on_http_error( + e=_rejection(case), request_data=copy.deepcopy(body) + ) + + assert retried is case["retried"] + assert transformed == (RETRY["stripped_request"] if case["stripped"] else body) 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 cb594767908..78648842271 100644 --- a/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py +++ b/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py @@ -4,7 +4,7 @@ import inspect from collections.abc import Mapping from pathlib import Path from types import MappingProxyType -from typing import Final +from typing import Final, cast # noqa: TID251 # a duck-typed logger double stands in for Logging import httpx import pytest @@ -323,6 +323,76 @@ def test_native_provider_exchange_window_is_recorded_like_the_timed_post() -> No assert metadata["llm_api_timing_windows"] == ((1_700_000_000.0, 1_700_000_000.25),) +@pytest.mark.asyncio +async def test_a_retried_request_rewrites_the_logged_body_in_place_like_the_python_retry_loop() -> None: + logger: Final = _messages_logger() + metadata: Final = {"user_id": "u-1"} + replayed: Final = [ + {"role": "assistant", "content": [{"type": "thinking", "thinking": "x", "signature": "s"}]}, + {"role": "user", "content": "again"}, + ] + logged: Final[dict[str, object]] = { + "model": "claude-sonnet-4-5", + "messages": replayed, + "metadata": metadata, + "thinking": {"type": "enabled", "budget_tokens": 1024}, + } + logger.model_call_details.update(logged) + retried: Final = { + "model": "claude-sonnet-4-5", + "messages": [{"role": "user", "content": "again"}], + "metadata": {"user_id": "u-1"}, + } + + await legacy.retried_request(logger, logged, retried, ({"kind": "thinking_signature"},)) + + assert logged == retried + assert logged["metadata"] is metadata + assert logger.model_call_details["messages"] == [{"role": "user", "content": "again"}] + assert logger.model_call_details["thinking"] == {"type": "enabled", "budget_tokens": 1024} + + +class _BaselineLogger: + def __init__(self) -> None: + self.baseline_cache_context: object = object() + self.model_call_details: dict[str, object] = {} + self.invalidated: list[str] = [] + + async def invalidate_baseline_cache_estimate(self, reason: str) -> None: + self.invalidated.append(reason) + + +@pytest.mark.asyncio +async def test_a_retried_request_invalidates_the_baseline_cache_estimate() -> None: + logger: Final = _BaselineLogger() + + await legacy.retried_request(cast(Logging, logger), {"messages": []}, {"messages": []}, ()) # cast-ok: duck-typed + + assert logger.invalidated == ["retried_request"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("reasons", "warned"), + [ + pytest.param(({"kind": "thinking_signature"},), [], id="thinking_signature"), + pytest.param(({"kind": "strict_tools", "dropped": 2},), [2], id="strict_tools"), + pytest.param(({"kind": "thinking_signature"}, {"kind": "strict_tools", "dropped": 3}), [3], id="both_reasons"), + ], +) +async def test_a_retry_that_dropped_strict_tools_warns_with_the_dropped_count( + monkeypatch: pytest.MonkeyPatch, reasons: tuple[Mapping[str, object], ...], warned: list[int] +) -> None: + from litellm.llms.anthropic import common_utils + + warnings: Final[list[int]] = [] + monkeypatch.setattr(common_utils, "warn_claude_strict_tools_dropped", warnings.append) + + await legacy.retried_request(_messages_logger(), {"messages": []}, {"messages": []}, reasons) + + assert warnings == warned + + 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"}} diff --git a/tests/unit/rust_bridge/test_detectors.py b/tests/unit/rust_bridge/test_detectors.py index f1869eb8b5c..000141f5714 100644 --- a/tests/unit/rust_bridge/test_detectors.py +++ b/tests/unit/rust_bridge/test_detectors.py @@ -32,8 +32,6 @@ MAPPED_FILE: Final = { } MAPPED_FILE_INPUT: Final = [{"role": "user", "content": [{"type": "input_file", "file_id": "litellm-file"}]}] CACHED_TEXT: Final = {"type": "input_text", "text": "hi", "cache_control": {"type": "ephemeral"}} -THINKING_TURN: Final = {"role": "assistant", "content": [{"type": "thinking", "thinking": "x", "signature": "s"}]} -REDACTED: Final = {"type": "redacted_thinking", "data": "x"} STRICT_TOOL: Final = {"name": "b", "input_schema": {"type": "object"}, "strict": True} TOOL_SEARCH_TYPE: Final = sorted(ANTHROPIC_TOOL_SEARCH_TOOL_TYPES)[0] @@ -195,18 +193,6 @@ def _wrap_nested_level(value: object, depth: int) -> object: (detectors.cache_control_injection, {"enable_prompt_caching": True}, True), (detectors.cache_control_injection, {"cache_control_injection_points": [{"location": "message"}]}, True), (detectors.cache_control_injection, {"cache_control_injection_points": []}, False), - (messages_detectors.thinking_signature_retry, {"messages": [THINKING_TURN]}, True), - ( - messages_detectors.thinking_signature_retry, - {"messages": [{"role": "assistant", "content": [REDACTED]}]}, - True, - ), - (messages_detectors.thinking_signature_retry, {"messages": [{"role": "assistant", "content": "done"}]}, False), - ( - messages_detectors.thinking_signature_retry, - {"messages": [{"role": "user", "content": [{"type": "text"}]}]}, - False, - ), (messages_detectors.strict_tools_retry, {"drop_params": True, "tools": [{"name": "a"}, STRICT_TOOL]}, True), ( messages_detectors.strict_tools_retry,