scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-12 08:42:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 207

feat(messages): retry rejected thinking signatures on rust

Browse source
This commit is contained in:
narmstrong-berry 2026-11-10 20:43:24 +00:00
parent 43939661b8
commit 7728b426ce
No known key found for this signature in database
43 changed files with 1589 additions and 98 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
litellm-rust/crates/callbacks-legacy-python/python_contract.json
View file

@ -43,6 +43,12 @@
"headers",
"body"
],
"retried_request": [
"logger",
"logged_body",
"body",
"reasons"
],
"record_api_window": [
"logger",
"start",

104
litellm-rust/crates/callbacks-legacy-python/src/adapter.rs
View file

@ -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<f64>,
}
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<Py<PyAny>> {
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<HookStep<Self, ()>> {
let Some(request) = &self.request else {
return Ok(HookStep::Ready(()));
};
let reasons: Vec<Value> = 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<Py<PyAny>>,
) -> PyResult<HookStep<Self, ()>> {
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, &[])
}

3
litellm-rust/crates/callbacks-legacy-python/src/mapping.rs
View file

@ -179,6 +179,9 @@ impl CallHooks<PythonRuntime> 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)
}

2
litellm-rust/crates/callbacks-legacy-python/src/python.rs
View file

@ -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")]

4
litellm-rust/crates/callbacks-legacy-python/src/test_support.rs
View file

@ -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
),

7
litellm-rust/crates/host-native/src/driver.rs
View file

@ -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));

8
litellm-rust/crates/host-python/src/driver.rs
View file

@ -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 = {

9
litellm-rust/crates/host-python/src/hooks/chain/dispatch.rs
View file

@ -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 {

3
litellm-rust/crates/host-python/tests/support/mod.rs
View file

@ -422,6 +422,9 @@ impl CallHooks<PythonRuntime> 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))
}

26
litellm-rust/crates/host/src/interceptors.rs
View file

@ -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<RetryReason>,
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<E>: Send + Sync {
&self,
raw: RawResponse,
) -> impl Future<Output = Result<(), E>> + Send;
fn provider_request_retried(
&self,
_retry: RetriedRequest,
) -> impl Future<Output = Result<(), E>> + Send {
async { Ok(()) }
}
}
impl<E, T: Interceptors<E> + ?Sized> Interceptors<E> for &T {
@ -128,6 +147,13 @@ impl<E, T: Interceptors<E> + ?Sized> Interceptors<E> for &T {
) -> impl Future<Output = Result<(), E>> + Send {
(**self).after_provider_response(raw)
}
fn provider_request_retried(
&self,
retry: RetriedRequest,
) -> impl Future<Output = Result<(), E>> + Send {
(**self).provider_request_retried(retry)
}
}
impl<E> Interceptors<E> for () {

15
litellm-rust/crates/host/src/lifecycle.rs
View file

@ -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<Response = (), Error = (), Raw = RawResponse> {
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
#[derive(Clone, Debug, PartialEq)]
pub enum ExecutionEvent<Raw = RawResponse> {
ResultReady {
facts: crate::interceptors::ExecutionFacts,
@ -58,6 +61,9 @@ pub enum ExecutionEvent<Raw = RawResponse> {
ProviderResponseReceived {
raw: Raw,
},
ProviderRequestRetried {
retry: RetriedRequest,
},
}
impl<Response, Error, Raw: std::borrow::Borrow<RawResponse>> CallEvent<Response, Error, Raw> {
@ -76,6 +82,11 @@ impl<Response, Error, Raw: std::borrow::Borrow<RawResponse>> CallEvent<Response,
facts: facts.clone(),
})
}
Self::Execution(ExecutionEvent::ProviderRequestRetried { retry }) => {
CallEvent::Execution(ExecutionEvent::ProviderRequestRetried {
retry: retry.clone(),
})
}
Self::Succeeded { timing, .. } => CallEvent::Succeeded {
timing: *timing,
response: (),

10
litellm-rust/crates/host/src/machine/context.rs
View file

@ -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<P: Protocol>(Channel<P>);

6
litellm-rust/crates/host/src/protocol.rs
View file

@ -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<E> {
raw: RawResponse,
reply: Reply<Result<(), E>>,
},
ProviderRequestRetried {
retry: RetriedRequest,
reply: Reply<Result<(), E>>,
},
}
pub enum StreamDelivery<P: Protocol> {

2
litellm-rust/crates/inference-messages/src/constants.rs
View file

@ -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;

106
litellm-rust/crates/inference-messages/src/handler.rs
View file

@ -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<Duration>,
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<Error>>,
) -> Result<MessagesCallResponse, Error> {
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<Error>>,
) -> Result<reqwest::Response, Error> {
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<ProviderFailure> 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<ProviderFailure, Error> {
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()),
}
}

1
litellm-rust/crates/inference-messages/tests/messages/main.rs
View file

@ -22,6 +22,7 @@ use support::*;
mod host;
mod request;
mod response;
mod retry;
mod secrets;
mod stream;
mod tencent;

218
litellm-rust/crates/inference-messages/tests/messages/retry.rs Normal file
View file

@ -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<Messages>,
) -> Result<(MessagesCallResponse, Vec<bytes::Bytes>), 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<Messages>) -> Vec<RetriedRequest> {
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::<Messages>::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::<Messages>::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::<Messages>::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());
}

10
litellm-rust/crates/inference-messages/tests/support/mod.rs
View file

@ -117,6 +117,7 @@ pub struct RecordingCall<P: litellm_host::protocol::Protocol> {
pub events: Arc<CallEvents>,
pub chunks: Mutex<Vec<P::Chunk>>,
pub head: Mutex<Option<P::StreamHead>>,
pub retried: Mutex<Vec<litellm_host::interceptors::RetriedRequest>>,
}
#[derive(Default)]
@ -167,6 +168,7 @@ impl<P: litellm_host::protocol::Protocol> RecordingCall<P> {
events: Arc::new(CallEvents::default()),
chunks: Mutex::new(Vec::new()),
head: Mutex::new(None),
retried: Mutex::new(Vec::new()),
}
}
}
@ -195,6 +197,14 @@ impl<P: litellm_host::protocol::Protocol> 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<P> RecordingCall<P>

13
litellm-rust/crates/inference/src/context.rs
View file

@ -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<Error>> 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
}
}

167
litellm-rust/crates/llms/src/anthropic/common_utils.rs
View file

@ -650,6 +650,62 @@ pub fn flatten_unencrypted_web_search_results(messages: Vec<Message>) -> Vec<Mes
.collect()
}
pub fn is_anthropic_invalid_thinking_block_error(error_text: &str) -> 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<Value> {
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"]);
}
}

118
litellm-rust/crates/llms/src/anthropic/messages/transformation.rs
View file

@ -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<RetriedRequest> {
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<RetriedRequest> {
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());
}
}

13
litellm-rust/crates/llms/src/azure_ai/messages/transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn shape_request(
&self,
request: MessagesRequest,

11
litellm-rust/crates/llms/src/base_llm/messages/transformation.rs
View file

@ -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<RetriedRequest> {
None
}
}
#[cfg(test)]

12
litellm-rust/crates/llms/src/bedrock/claude_platform/messages/transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn shape_request(
&self,
request: MessagesRequest,

12
litellm-rust/crates/llms/src/bedrock/messages/invoke_transformations/anthropic_claude_transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn shape_request(
&self,
request: MessagesRequest,

12
litellm-rust/crates/llms/src/bedrock_mantle/messages/transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn validate_environment(
&self,
headers: Headers,

12
litellm-rust/crates/llms/src/deepseek/messages/transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn shape_request(
&self,
request: MessagesRequest,

12
litellm-rust/crates/llms/src/edenai/messages/transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn validate_environment(
&self,
headers: Headers,

13
litellm-rust/crates/llms/src/minimax/messages/transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn shape_request(
&self,
request: MessagesRequest,

13
litellm-rust/crates/llms/src/openrouter/messages/transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn validate_environment(
&self,
headers: Headers,

13
litellm-rust/crates/llms/src/tencent/messages/transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn shape_request(
&self,
request: MessagesRequest,

12
litellm-rust/crates/llms/src/vertex_ai/messages/transformation.rs
View file

@ -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<RetriedRequest> {
retry_anthropic_messages_after_http_error(status, error_text, request_body, context)
}
fn shape_request(
&self,
request: MessagesRequest,

252
litellm-rust/crates/llms/tests/fixtures/anthropic_messages_http_error_retry.json vendored Normal file
View file

@ -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
}
]
}
}

72
litellm-rust/crates/llms/tests/messages_http_error_retry.rs Normal file
View file

@ -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());
}

1
litellm/messages/dispatch.py
View file

@ -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,
}
)

3
litellm/rust_bridge/catalog.py
View file

@ -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,
}

32
litellm/rust_bridge/host/callbacks_legacy_python.py
View file

@ -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 (

11
litellm/rust_bridge/messages/detectors.py
View file

@ -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:

114
tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_native_parity.py
View file

@ -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

48
tests/test_litellm_rust/messages/test_dispatch.py
View file

@ -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",

95
tests/unit/llms/anthropic/test_messages_http_error_retry_cases.py Normal file
View file

@ -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)

72
tests/unit/rust_bridge/host/test_callbacks_legacy_python.py
View file

@ -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"}}

14
tests/unit/rust_bridge/test_detectors.py
View file

@ -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,