diff --git a/litellm-rust/crates/http/src/error.rs b/litellm-rust/crates/http/src/error.rs index 7078d4aba2a..9505aacf44b 100644 --- a/litellm-rust/crates/http/src/error.rs +++ b/litellm-rust/crates/http/src/error.rs @@ -16,6 +16,15 @@ pub enum TlsSource { #[error(transparent)] pub struct Error(Arc); +// 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, + 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) -> 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::().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 { + match &*self.0 { + Repr::BodyCut { cut, .. } => Some(*cut), + _ => None, + } + } + pub fn computed_header(name: impl Into) -> 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, + ) { + 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"); + } } diff --git a/litellm-rust/crates/http/src/lib.rs b/litellm-rust/crates/http/src/lib.rs index 02129c62083..1aa11843b46 100644 --- a/litellm-rust/crates/http/src/lib.rs +++ b/litellm-rust/crates/http/src/lib.rs @@ -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}; diff --git a/litellm-rust/crates/inference-messages/src/handler.rs b/litellm-rust/crates/inference-messages/src/handler.rs index 5eb017ddfc5..6a8ed7fff8e 100644 --- a/litellm-rust/crates/inference-messages/src/handler.rs +++ b/litellm-rust/crates/inference-messages/src/handler.rs @@ -386,13 +386,20 @@ fn timed_chunks( response: reqwest::Response, timeout: Duration, ) -> impl futures_util::Stream> + Send + 'static { - futures_util::stream::try_unfold(response, move |mut response| async move { - let chunk = tokio::time::timeout(timeout, response.chunk()) - .await - .map_err(litellm_http::Error::body_read_timeout)? - .map_err(litellm_http::Error::from)?; - Ok(chunk.map(|chunk| (chunk, response))) - }) + 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 diff --git a/litellm-rust/crates/inference/src/error.rs b/litellm-rust/crates/inference/src/error.rs index 841e0c53799..4135424d6ef 100644 --- a/litellm-rust/crates/inference/src/error.rs +++ b/litellm-rust/crates/inference/src/error.rs @@ -164,6 +164,14 @@ impl Error { } } + pub fn body_cut(&self) -> Option { + 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 { diff --git a/litellm-rust/crates/llms/src/error.rs b/litellm-rust/crates/llms/src/error.rs index 2844fbf3e01..2975ceee3cb 100644 --- a/litellm-rust/crates/llms/src/error.rs +++ b/litellm-rust/crates/llms/src/error.rs @@ -103,6 +103,13 @@ impl Error { matches!(self, Self::Http(error) if error.is_body_read_timeout()) } + pub fn body_cut(&self) -> Option { + match self { + Self::Http(error) => error.body_cut(), + _ => None, + } + } + pub fn is_timeout(&self) -> Option { match self { Self::Http(error) if error.is_timeout() => Some(error.is_connect_timeout()), diff --git a/litellm-rust/crates/python-bridge/src/errors.rs b/litellm-rust/crates/python-bridge/src/errors.rs index 426cd861729..db5696d52ce 100644 --- a/litellm-rust/crates/python-bridge/src/errors.rs +++ b/litellm-rust/crates/python-bridge/src/errors.rs @@ -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)?; } diff --git a/litellm/rust_bridge/host/callbacks_legacy_python.py b/litellm/rust_bridge/host/callbacks_legacy_python.py index 8c281c73842..5a368573f3d 100644 --- a/litellm/rust_bridge/host/callbacks_legacy_python.py +++ b/litellm/rust_bridge/host/callbacks_legacy_python.py @@ -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), diff --git a/litellm/rust_bridge/host/routes/messages.py b/litellm/rust_bridge/host/routes/messages.py index c8b53097f46..c50791f8a9c 100644 --- a/litellm/rust_bridge/host/routes/messages.py +++ b/litellm/rust_bridge/host/routes/messages.py @@ -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): diff --git a/tests/unit/rust_bridge/host/routes/test_messages.py b/tests/unit/rust_bridge/host/routes/test_messages.py index 9d8b09dd1c4..704d52386a8 100644 --- a/tests/unit/rust_bridge/host/routes/test_messages.py +++ b/tests/unit/rust_bridge/host/routes/test_messages.py @@ -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: ", + ), + ( + True, + (19, 100, True), + "ReadError", + "Response payload is not completed: . " + "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: [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: [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: [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: "") + + 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) diff --git a/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py b/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py index 87b9e70612c..b00df6547d2 100644 --- a/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py +++ b/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py @@ -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