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

fix(messages): raise native messages failures as python's errors (#50424)

Browse source 
A stream whose upstream body ends mid-frame reached proxy callers as a mapped APIConnectionError with status 500. Python's transport raises an unmapped httpx error at that point. The native body read now records a body cut and the host raises the same exception the active transport would

Python wraps a connect, DNS, TLS or reset failure in a synthesized 500 AnthropicError, which maps to InternalServerError. The native path surfaced these as APIConnectionError with the bug-report footer, which also changed cooldown decisions

The proxy bills the estimated prompt of a failed call only when the Timeout was caused by an httpx ReadTimeout. The native timeout now chains the same httpx timeout Python's client raises, so its failure row bills the prompt

Python rebuilds an upstream error response from the decoded body without content-encoding and content-length. The native failure now drops those wire headers, and a failure raised without an HTTP response states no headers at all, like Python
This commit is contained in:
Nate Armstrong 2026-11-09 16:24:20 -08:00 • committed by GitHub
parent 1467b13fd2
commit ee34a3c2d4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 475 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

104
litellm-rust/crates/http/src/error.rs
View file

@ -16,6 +16,15 @@ pub enum TlsSource {
#[error(transparent)]
pub struct Error(Arc<Repr>);
// A streamed body the connection ended early, after the response head arrived; `expected` is
// the declared `content-length`, `None` for a chunked body.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct BodyCut {
pub received: u64,
pub expected: Option<u64>,
pub reset: bool,
}
#[derive(Debug, thiserror::Error)]
pub(crate) enum Repr {
#[error("could not read {}", path.display())]
@ -56,6 +65,12 @@ pub(crate) enum Repr {
Connect(#[source] reqwest::Error),
#[error("upstream network error")]
Network(#[source] reqwest::Error),
#[error("upstream network error")]
BodyCut {
#[source]
source: reqwest::Error,
cut: BodyCut,
},
#[error("upstream request timed out")]
Timeout(#[source] reqwest::Error),
/// No bytes arrived within the read timeout, which the caller enforces per read; `body` says
@ -201,6 +216,37 @@ impl Error {
matches!(&*self.0, Repr::ReadTimeout { body: true, .. })
}
pub fn body_read(error: reqwest::Error, received: u64, expected: Option<u64>) -> Self {
let mut source: Option<&(dyn std::error::Error + 'static)> = Some(&error);
let kind = std::iter::from_fn(|| {
let current = source?;
source = current.source();
Some(current)
})
.find_map(|cause| cause.downcast_ref::<io::Error>().map(io::Error::kind));
let reset = match kind {
Some(io::ErrorKind::UnexpectedEof) => false,
Some(io::ErrorKind::ConnectionReset) => true,
_ => return error.into(),
};
Repr::BodyCut {
source: error.without_url(),
cut: BodyCut {
received,
expected,
reset,
},
}
.into()
}
pub fn body_cut(&self) -> Option<BodyCut> {
match &*self.0 {
Repr::BodyCut { cut, .. } => Some(*cut),
_ => None,
}
}
pub fn computed_header(name: impl Into<String>) -> Self {
Repr::ComputedHeader(name.into()).into()
}
@ -226,6 +272,7 @@ impl Error {
pub fn dispatched(&self) -> bool {
match &*self.0 {
Repr::Network(_)
| Repr::BodyCut { .. }
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
@ -285,6 +332,7 @@ impl Error {
| Repr::Signature(_)
| Repr::Connect(_)
| Repr::Network(_)
| Repr::BodyCut { .. }
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
@ -318,6 +366,7 @@ impl Error {
| Repr::Signature(_)
| Repr::Connect(_)
| Repr::Network(_)
| Repr::BodyCut { .. }
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
@ -355,6 +404,7 @@ impl Error {
| Repr::Signature(_)
| Repr::Connect(_)
| Repr::Network(_)
| Repr::BodyCut { .. }
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
@ -394,6 +444,7 @@ impl Error {
| Repr::UnsignableHeader(_)
| Repr::Signature(_)
| Repr::Network(_)
| Repr::BodyCut { .. }
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
@ -437,6 +488,7 @@ impl Error {
| Repr::Signature(_)
| Repr::Connect(_)
| Repr::Network(_)
| Repr::BodyCut { .. }
| Repr::Status { .. }
| Repr::WebSocketUrl(_)
| Repr::WebSocketScheme(_)
@ -473,6 +525,7 @@ impl Error {
| Repr::Signature(_)
| Repr::Connect(_)
| Repr::Network(_)
| Repr::BodyCut { .. }
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::WebSocketUrl(_)
@ -513,6 +566,7 @@ impl Error {
| Repr::Signature(_)
| Repr::Connect(_)
| Repr::Network(_)
| Repr::BodyCut { .. }
| Repr::WebSocketUrl(_)
| Repr::WebSocketScheme(_)
| Repr::WebSocketSchemeRewrite(_)
@ -548,6 +602,7 @@ impl Error {
| Repr::Signature(_)
| Repr::Connect(_)
| Repr::Network(_)
| Repr::BodyCut { .. }
| Repr::Timeout(_)
| Repr::ReadTimeout { .. }
| Repr::Status { .. }
@ -742,4 +797,53 @@ mod tests {
assert_eq!(error.upstream_response(), Some((429, body.as_str())));
assert_eq!(error.status(), Some(429));
}
#[rstest]
#[case::chunked("transfer-encoding: chunked\r\n\r\n5\r\nhello\r\n", None)]
#[case::content_length("content-length: 100\r\n\r\nhello", Some(100))]
#[tokio::test]
async fn a_connection_that_ends_mid_body_is_a_body_cut(
#[case] tail: &'static str,
#[case] expected: Option<u64>,
) {
use std::io::{Read as _, Write as _};
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
let address = listener.local_addr().expect("address");
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("accept");
let mut request = [0_u8; 1024];
let _ = stream.read(&mut request).expect("request");
stream
.write_all(format!("HTTP/1.1 200 OK\r\n{tail}").as_bytes())
.expect("response");
});
let mut response = reqwest::Client::builder()
.no_proxy()
.build()
.expect("client")
.get(format!("http://{address}/"))
.send()
.await
.expect("head");
server.join().expect("server");
let mut received = 0_u64;
let error = loop {
match response.chunk().await {
Ok(Some(chunk)) => received += chunk.len() as u64,
Ok(None) => panic!("the body must end with an error"),
Err(error) => break crate::Error::body_read(error, received, expected),
}
};
assert_eq!(
error.body_cut(),
Some(super::BodyCut {
received: 5,
expected,
reset: false,
})
);
assert!(error.dispatched());
assert_eq!(error.to_string(), "upstream network error");
}
}

2
litellm-rust/crates/http/src/lib.rs
View file

@ -18,7 +18,7 @@ pub mod websocket;
pub use client::Client;
pub use config::{ClientIdentity, HttpClientConfig, Resolution, Verify};
pub use error::{Error, TlsSource};
pub use error::{BodyCut, Error, TlsSource};
pub use pool::{ClientVariant, HttpClientPool};
pub use proxy::EnvironmentProxies;
pub use settings::{HttpSettings, HttpSettingsLayer, SslVerify, TcpKeepalive};

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

@ -386,13 +386,20 @@ fn timed_chunks(
response: reqwest::Response,
timeout: Duration,
) -> impl futures_util::Stream<Item = Result<Bytes, litellm_http::Error>> + Send + 'static {
futures_util::stream::try_unfold(response, move |mut response| async move {
let chunk = tokio::time::timeout(timeout, response.chunk())
.await
.map_err(litellm_http::Error::body_read_timeout)?
.map_err(litellm_http::Error::from)?;
Ok(chunk.map(|chunk| (chunk, response)))
})
let expected = response.content_length();
futures_util::stream::try_unfold(
(response, 0_u64),
move |(mut response, received)| async move {
let chunk = tokio::time::timeout(timeout, response.chunk())
.await
.map_err(litellm_http::Error::body_read_timeout)?
.map_err(|error| litellm_http::Error::body_read(error, received, expected))?;
Ok(chunk.map(|chunk| {
let received = received + chunk.len() as u64;
(chunk, (response, received))
}))
},
)
}
/// The provider's error, its body read under `timeout`: a stream has no overall deadline once

8
litellm-rust/crates/inference/src/error.rs
View file

@ -164,6 +164,14 @@ impl Error {
}
}
pub fn body_cut(&self) -> Option<litellm_http::BodyCut> {
match self {
Self::Llm(error) => error.body_cut(),
Self::Http(error) => error.body_cut(),
_ => None,
}
}
/// `Some(connect)` when the call timed out, where `connect` says the connection was never
/// established; `None` for every other error.
pub fn timeout(&self) -> Option<bool> {

7
litellm-rust/crates/llms/src/error.rs
View file

@ -103,6 +103,13 @@ impl Error {
matches!(self, Self::Http(error) if error.is_body_read_timeout())
}
pub fn body_cut(&self) -> Option<litellm_http::BodyCut> {
match self {
Self::Http(error) => error.body_cut(),
_ => None,
}
}
pub fn is_timeout(&self) -> Option<bool> {
match self {
Self::Http(error) if error.is_timeout() => Some(error.is_connect_timeout()),

5
litellm-rust/crates/python-bridge/src/errors.rs
View file

@ -29,6 +29,8 @@ const MESSAGES_UNSUPPORTED_PARAM: &str = "messages_unsupported_param";
/// The phase label (`Connection` or `Read`) of a timed-out call, for Python's timeout text, or
/// `Body` for a streamed body read that timed out after the head arrived.
const MESSAGES_TIMEOUT: &str = "messages_timeout";
// `(received, expected, reset)` of a streamed body the connection ended early.
const MESSAGES_BODY_CUT: &str = "messages_body_cut";
/// No credential reached the provider, which Python raises as `AuthenticationError`.
const MESSAGES_MISSING_CREDENTIAL: &str = "messages_missing_credential";
/// Python refuses the request only once its logger knows the resolved model and provider (the
@ -134,6 +136,9 @@ pub(crate) fn route_failure(
};
value.setattr(MESSAGES_TIMEOUT, phase)?;
}
if let Some(cut) = error.body_cut() {
value.setattr(MESSAGES_BODY_CUT, (cut.received, cut.expected, cut.reset))?;
}
if error.is_missing_credential() {
value.setattr(MESSAGES_MISSING_CREDENTIAL, true)?;
}

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

@ -510,9 +510,15 @@ def stream_failure(
chunks: list[bytes],
error: Exception,
) -> Coroutine[object, object, None]:
from litellm._logging import verbose_proxy_logger
from litellm.proxy.pass_through_endpoints.streaming_handler import PassThroughStreamingHandler
from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
verbose_proxy_logger.error("Error in chunk_processor: %s", error)
# Python's stream fails before its success handler records `cache_hit`; only a cache hit
# stamps it earlier.
if logger.model_call_details.get("cache_hit") is not True:
logger.model_call_details.pop("cache_hit", None)
return PassThroughStreamingHandler.schedule_stream_failure_logging(
litellm_logging_obj=logger,
endpoint_type=EndpointType(endpoint_type),

133
litellm/rust_bridge/host/routes/messages.py
View file

@ -1,11 +1,16 @@
from __future__ import annotations
import datetime
import errno
import ipaddress
import os
import socket
from collections.abc import Callable, Mapping, Sequence
from dataclasses import asdict, dataclass
from typing import Final, TypeAlias, cast # noqa: TID251 # narrows native payloads to public TypedDicts
import httpx
from aiohttp.http_exceptions import ContentLengthError, TransferEncodingError
from pydantic import TypeAdapter, ValidationError
import litellm
@ -14,7 +19,7 @@ from litellm.litellm_core_utils.core_helpers import normalize_drop_params
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.reasoning_effort_capability import deployment_reasoning_effort_levels
from litellm.litellm_core_utils.request_timeout_resolver import get_configured_request_timeout
from litellm.llms.anthropic.common_utils import AnthropicError
from litellm.llms.anthropic.common_utils import AnthropicError, anthropic_pass_through_api_base
from litellm.llms.anthropic.messages.native import handler
from litellm.llms.anthropic.messages.native.handler import (
_native_messages_config as native_messages_config, # pyright: ignore[reportPrivateUsage] # the handler's own config lookup
@ -23,8 +28,10 @@ from litellm.llms.anthropic.messages.native.handler import (
_with_deployment_output_effort as _deployment_output_effort, # pyright: ignore[reportPrivateUsage] # the handler's own effort rounding
)
from litellm.llms.anthropic.messages.utils import is_reasoning_auto_summary_enabled
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.custom_httpx.http_handler import get_ssl_configuration
from litellm.rust_bridge import failures
from litellm.rust_bridge.host import callbacks_legacy_python
from litellm.rust_bridge.host import callbacks_legacy_python, http_transport
from litellm.rust_bridge.public_call import known_params, optional_str
from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse
@ -120,6 +127,7 @@ def stream_hidden_params(headers: Sequence[tuple[str, str]]) -> Mapping[str, obj
_UPSTREAM_ARGS: Final = TypeAdapter(tuple[int, str])
_UPSTREAM_HEADERS: Final = TypeAdapter(list[tuple[str, str]])
_BODY_CUT: Final = TypeAdapter(tuple[int, int | None, bool])
_ERROR_CLASS: TypeAlias = Callable[..., Exception]
_validate_metadata: Final = cast( # cast-ok: the handler's untyped metadata validator
Callable[[dict[str, object]], object], handler.validate_anthropic_api_metadata
@ -128,6 +136,17 @@ _exception_type: Final = cast( # cast-ok: the legacy public exception mapper is
Callable[..., Exception], exception_mapping_utils.exception_type
)
_BODY_READ_TIMEOUT: Final = "Timeout on reading data from socket" # aiohttp's SocketTimeoutError text
_UNREACHABLE: Final = "could not reach the provider: "
_SEND_FAILED: Final = "upstream network error: error sending request: client error (SendRequest): "
_DNS_FAILURE: Final = "failed to lookup address information: "
_HTTPX_DNS_ERRNO: Final = {
"Name or service not known": socket.EAI_NONAME,
"Temporary failure in name resolution": socket.EAI_AGAIN,
}
_TRANSPORT_TIMEOUTS: Final[Mapping[object, type[httpx.TimeoutException]]] = {
"Connection": httpx.ConnectTimeout,
"Read": httpx.ReadTimeout,
}
_MISSING_KEY: Final = (
"Missing Anthropic API Key - A call is being made to anthropic but no key is set "
"either in the environment variables or via params. Please set `ANTHROPIC_API_KEY` "
@ -174,6 +193,82 @@ def _timeout_text(phase: str, request: Mapping[str, object]) -> str:
)
def _body_cut(received: int, expected: int | None, reset: bool) -> httpx.TransportError:
if not http_transport.uses_aiohttp_transport():
if reset:
return httpx.ReadError("")
detail: Final = (
"incomplete chunked read" if expected is None else f"received {received} bytes, expected {expected}"
)
return httpx.RemoteProtocolError(f"peer closed connection without sending complete message body ({detail})")
payload: Final = (
TransferEncodingError("Not enough data to satisfy transfer length header.")
if expected is None
else ContentLengthError(
f"Not enough data to satisfy content length header (received {received} of {expected} bytes)."
)
)
text: Final = f"Response payload is not completed: {payload!r}"
if not reset:
return httpx.ReadError(text)
return httpx.ReadError(f"{text}. {ConnectionResetError(errno.ECONNRESET, os.strerror(errno.ECONNRESET))!r}")
def _connect_failure(native: str, request: Mapping[str, object], request_provider: str) -> str | None:
api_base: Final = request.get("api_base")
if not isinstance(api_base, str) and request_provider != "anthropic":
return None
refused: Final = "tcp connect error: Connection refused" in native
dns_reason: Final = native.rsplit(_DNS_FAILURE, 1)[1] if _DNS_FAILURE in native else None
tls_eof: Final = native.endswith("tls handshake eof")
if not http_transport.uses_aiohttp_transport():
if refused:
return "All connection attempts failed"
dns_errno: Final = None if dns_reason is None else _HTTPX_DNS_ERRNO.get(dns_reason)
if dns_errno is not None:
return f"[Errno {dns_errno}] {dns_reason}"
return "" if tls_eof else None
url: Final = httpx.URL(api_base if isinstance(api_base, str) else anthropic_pass_through_api_base())
port: Final = url.port or (443 if url.scheme == "https" else 80)
if refused:
try:
address: Final = ipaddress.ip_address(url.host)
except ValueError:
return None
peer: Final = (str(address), port) if address.version == 4 else (str(address), port, 0, 0)
return _aiohttp_connect_failure(url, port, request, f"Connect call failed {peer}")
if dns_reason is not None:
return _aiohttp_connect_failure(url, port, request, dns_reason)
if tls_eof:
return _aiohttp_connect_failure(url, port, request, None)
return None
def _aiohttp_connect_failure(url: httpx.URL, port: int, request: Mapping[str, object], reason: str | None) -> str:
ssl_verify: Final = request.get("ssl_verify")
ssl_config: Final = get_ssl_configuration(ssl_verify if isinstance(ssl_verify, bool | str) else None)
return f"Cannot connect to host {url.host}:{port} ssl:{'default' if ssl_config is True else ssl_config} [{reason}]"
def _transport_failure(error: Exception, request: Mapping[str, object], request_provider: str) -> str | None:
"""The text Python's HTTP client raises for a call that failed before any response arrived."""
native: Final = str(error)
if native.startswith(_SEND_FAILED):
reset: Final = native.endswith(f"Connection reset by peer (os error {errno.ECONNRESET})")
closed: Final = native.endswith("connection closed before message completed")
if http_transport.uses_aiohttp_transport():
if reset:
return str(ConnectionResetError(errno.ECONNRESET, os.strerror(errno.ECONNRESET)))
return "Server disconnected" if closed else native
if reset:
return ""
return "Server disconnected without sending a response." if closed else native
if not native.startswith(_UNREACHABLE):
return None
connect: Final = _connect_failure(native, request, request_provider)
return native if connect is None else connect
def _upstream_failure(
error: Exception, request: Mapping[str, object]
) -> tuple[int, str, Sequence[tuple[str, str]]] | None:
@ -183,14 +278,27 @@ def _upstream_failure(
return 408, _timeout_text(timeout_phase, request), ()
try:
status, text = _UPSTREAM_ARGS.validate_python(error.args)
return status, text, _UPSTREAM_HEADERS.validate_python(getattr(error, "headers", None))
headers: Final = _UPSTREAM_HEADERS.validate_python(getattr(error, "headers", None))
except ValidationError:
return None
if not headers:
return status, text, headers
# Python's `MaskedHTTPStatusError` rebuilds the response from its decoded body without the
# wire's content-encoding and content-length, so httpx states a length only for a non-empty body.
rebuilt: Final = httpx.Response(
status,
content=text.encode(),
headers=[
(name, value) for name, value in headers if name.lower() not in ("content-encoding", "content-length")
],
)
return status, text, tuple(rebuilt.headers.items())
def _provider_error(error: Exception, request: Mapping[str, object], request_provider: str) -> Exception | None:
"""The provider exception Python's `_handle_error` builds for the same failure."""
failure: Final = _upstream_failure(error, request)
transport: Final = _transport_failure(error, request, request_provider)
failure: Final = _upstream_failure(error, request) if transport is None else (500, transport, ())
if failure is None:
return None
status, text, headers = failure
@ -201,13 +309,17 @@ def _provider_error(error: Exception, request: Mapping[str, object], request_pro
request.get("model_info"),
api_base if isinstance(api_base, str) else None,
)
return (
provider_error: Final = (
cast(_ERROR_CLASS, config.get_error_class)( # cast-ok: the config's untyped headers parameter
error_message=text, status_code=status, headers=dict(headers)
)
if config is not None
else AnthropicError(status_code=status, message=text, headers=httpx.Headers(headers))
)
if transport is not None and isinstance(provider_error, BaseLLMException):
# Python's `_handle_error` gives a failure without a response the status 500.
provider_error.status_code_is_synthesized = True
return provider_error
def _log_resolved_call(request: Mapping[str, object], request_provider: str) -> None:
@ -232,6 +344,17 @@ def map_failure(error: Exception, request: Mapping[str, object], request_provide
# Python reads a stream's body outside its HTTP client's error mapping, so the transport's
# own timeout reaches the caller unmapped.
return httpx.ReadTimeout(_BODY_READ_TIMEOUT)
body_cut: Final = getattr(error, "messages_body_cut", None)
if body_cut is not None:
# Python reads a stream's body outside its HTTP client's error mapping, so a body the
# connection cut short reaches the caller as its transport raised it.
return _body_cut(*_BODY_CUT.validate_python(body_cut))
transport_timeout: Final = _TRANSPORT_TIMEOUTS.get(getattr(error, "messages_timeout", None))
if transport_timeout is not None:
# Python's `litellm.Timeout` is raised while handling httpx's own timeout, which the proxy's
# failure spend reads to bill a read timeout's prompt.
chained: Final = transport_timeout(str(error))
error.__context__ = chained # rebind-ok: Python chains the transport timeout as the exception context
if getattr(error, "messages_late_rejection", False):
_log_resolved_call(request, request_provider)
if getattr(error, "messages_request_error", False):

178
tests/unit/rust_bridge/host/routes/test_messages.py
View file

@ -207,6 +207,16 @@ def test_a_native_read_timeout_maps_to_the_public_timeout_with_python_text() ->
assert "Read timed out. Timeout passed=1.0" in str(mapped)
@pytest.mark.parametrize(("phase", "billed"), (("Read", True), ("Connection", False)))
def test_a_native_timeout_is_billed_on_the_failure_row_as_python_bills_it(phase: str, billed: bool) -> None:
from litellm.proxy.utils import _provider_billed_the_failure
mapped: Final = messages_host.map_failure(_marked(phase), {"model": _ADAPTIVE_OPUS, "timeout": 1.0}, "anthropic")
assert isinstance(mapped, litellm.Timeout)
assert _provider_billed_the_failure(mapped) is billed
def test_an_upstream_status_maps_through_the_provider_error_class() -> None:
upstream: Final = ValueError(429, "slow down")
upstream.headers = [("retry-after", "3")] # pyright: ignore[reportAttributeAccessIssue] # set by the native host
@ -218,6 +228,44 @@ def test_an_upstream_status_maps_through_the_provider_error_class() -> None:
assert mapped.__context__ is upstream
@pytest.mark.parametrize(
("body", "expected"),
[
("", {"content-type": "text/plain"}),
("gateway", {"content-type": "text/plain", "content-length": "7"}),
],
)
def test_upstream_failure_headers_match_the_python_clients_rebuilt_response(
body: str, expected: dict[str, str]
) -> None:
from litellm.litellm_core_utils.exception_mapping_utils import _get_response_headers
upstream: Final = ValueError(504, body)
upstream.headers = [ # pyright: ignore[reportAttributeAccessIssue] # set by the native host
("content-type", "text/plain"),
("content-encoding", "gzip"),
("content-length", "42"),
]
mapped: Final = messages_host.map_failure(upstream, {"model": _ADAPTIVE_OPUS}, "anthropic")
assert isinstance(mapped, litellm.Timeout)
assert dict(_get_response_headers(mapped) or {}) == expected
@pytest.mark.parametrize("status", [200, 500])
def test_upstream_failure_without_wire_headers_states_no_headers(status: int) -> None:
from litellm.litellm_core_utils.exception_mapping_utils import _get_response_headers
upstream: Final = ValueError(status, "not a messages response")
upstream.headers = [] # pyright: ignore[reportAttributeAccessIssue] # set by the native host
mapped: Final = messages_host.map_failure(upstream, {"model": _ADAPTIVE_OPUS}, "anthropic")
assert isinstance(mapped, Exception)
assert not dict(_get_response_headers(mapped) or {})
def test_a_late_rejection_tells_the_logger_the_resolved_call_first() -> None:
import datetime
@ -245,3 +293,133 @@ def test_a_late_rejection_tells_the_logger_the_resolved_call_first() -> None:
assert isinstance(mapped, litellm.BadRequestError)
assert logger.model_call_details["custom_llm_provider"] == "anthropic"
assert logger.model_call_details["optional_params"] == {"max_tokens": 8}
@pytest.mark.parametrize(
("aiohttp", "cut", "error_class", "text"),
(
(
True,
(60, None, False),
"ReadError",
"Response payload is not completed: <TransferEncodingError: 400, "
"message='Not enough data to satisfy transfer length header.'>",
),
(
True,
(19, 100, True),
"ReadError",
"Response payload is not completed: <ContentLengthError: 400, "
"message='Not enough data to satisfy content length header (received 19 of 100 bytes).'>. "
"ConnectionResetError(104, 'Connection reset by peer')",
),
(
False,
(60, None, False),
"RemoteProtocolError",
"peer closed connection without sending complete message body (incomplete chunked read)",
),
(
False,
(19, 100, False),
"RemoteProtocolError",
"peer closed connection without sending complete message body (received 19 bytes, expected 100)",
),
(False, (19, 100, True), "ReadError", ""),
),
)
def test_a_stream_body_cut_reaches_the_caller_as_the_transport_raised_it(
monkeypatch: pytest.MonkeyPatch,
aiohttp: bool,
cut: tuple[int, int | None, bool],
error_class: str,
text: str,
) -> None:
import httpx
from litellm.rust_bridge.host import http_transport
monkeypatch.setattr(http_transport, "uses_aiohttp_transport", lambda: aiohttp)
native: Final = RuntimeError("upstream network error")
native.messages_body_cut = cut # pyright: ignore[reportAttributeAccessIssue] # marker the native host sets
mapped: Final = messages_host.map_failure(native, {"model": _ADAPTIVE_OPUS}, "anthropic")
assert type(mapped) is getattr(httpx, error_class)
assert str(mapped) == text
_TRANSPORT: Final = "error sending request: client error"
@pytest.mark.parametrize(
("aiohttp", "native", "text"),
(
(
True,
f"could not reach the provider: {_TRANSPORT} (Connect): tcp connect error: Connection refused (os error 111)",
"Cannot connect to host 127.0.0.1:9 ssl:<ssl> [Connect call failed ('127.0.0.1', 9)]",
),
(
True,
f"could not reach the provider: {_TRANSPORT} (Connect): dns error: error resolving DNS: "
"failed to lookup address information: Name or service not known",
"Cannot connect to host 127.0.0.1:9 ssl:<ssl> [Name or service not known]",
),
(
True,
f"could not reach the provider: {_TRANSPORT} (Connect): tls handshake eof",
"Cannot connect to host 127.0.0.1:9 ssl:<ssl> [None]",
),
(
True,
f"upstream network error: {_TRANSPORT} (SendRequest): connection error: Connection reset by peer (os error 104)",
"[Errno 104] Connection reset by peer",
),
(
True,
f"upstream network error: {_TRANSPORT} (SendRequest): connection closed before message completed",
"Server disconnected",
),
(
False,
f"could not reach the provider: {_TRANSPORT} (Connect): tcp connect error: Connection refused (os error 111)",
"All connection attempts failed",
),
(
False,
f"could not reach the provider: {_TRANSPORT} (Connect): dns error: error resolving DNS: "
"failed to lookup address information: Name or service not known",
"[Errno -2] Name or service not known",
),
(False, f"could not reach the provider: {_TRANSPORT} (Connect): tls handshake eof", ""),
(
False,
f"upstream network error: {_TRANSPORT} (SendRequest): connection error: Connection reset by peer (os error 104)",
"",
),
(
False,
f"upstream network error: {_TRANSPORT} (SendRequest): connection closed before message completed",
"Server disconnected without sending a response.",
),
),
)
def test_a_failure_before_any_response_maps_like_the_python_transport_error(
monkeypatch: pytest.MonkeyPatch, aiohttp: bool, native: str, text: str
) -> None:
from litellm.rust_bridge.host import http_transport
monkeypatch.setattr(http_transport, "uses_aiohttp_transport", lambda: aiohttp)
monkeypatch.setattr(messages_host, "get_ssl_configuration", lambda _ssl_verify: "<ssl>")
mapped: Final = messages_host.map_failure(
RuntimeError(native), {"model": _ADAPTIVE_OPUS, "api_base": "http://127.0.0.1:9"}, "anthropic"
)
assert type(mapped) is litellm.InternalServerError
assert mapped.status_code == 500
assert mapped.message == (
f"litellm.InternalServerError: AnthropicException - {text}. Handle with `litellm.InternalServerError`."
)
assert "This looks like a bug in LiteLLM" not in str(mapped)

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

@ -338,3 +338,27 @@ def test_stream_head_is_stamped_in_place_with_the_call_hidden_params() -> None:
assert head["additional_headers"] == {"request-id": "req_native", "llm_provider-request-id": "req_native"}
assert head["litellm_call_id"] == "messages"
assert "_response_ms" in head
@pytest.mark.asyncio
async def test_a_native_stream_failure_logs_like_the_python_chunk_processor(monkeypatch: pytest.MonkeyPatch) -> None:
from litellm._logging import verbose_proxy_logger
from litellm.proxy.pass_through_endpoints.streaming_handler import PassThroughStreamingHandler
logged: Final[list[tuple[object, ...]]] = []
scheduled: Final[list[Mapping[str, object]]] = []
async def schedule(**kwargs: object) -> None:
scheduled.append(MappingProxyType(dict(kwargs)))
monkeypatch.setattr(verbose_proxy_logger, "error", lambda *args: logged.append(args))
monkeypatch.setattr(PassThroughStreamingHandler, "schedule_stream_failure_logging", schedule)
logger: Final = _supplied_logger()
logger.model_call_details["cache_hit"] = False
cut: Final = httpx.ReadError("cut")
await legacy.stream_failure(logger, "anthropic", {}, [b"event: ping\n\n"], cut)
assert logged == [("Error in chunk_processor: %s", cut)]
assert "cache_hit" not in logger.model_call_details
assert scheduled[0]["exception"] is cut