Merge remote-tracking branch 'origin/nate_messages_native_stream_cut_error' into nate_messages_integration_local

litellm-rust/crates/http/src/error.rs

@@ -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,39 @@ impl Error {
         matches!(&*self.0, Repr::ReadTimeout { body: true, .. })
     }
 
+    // A connection that ended or was reset mid-body is a cut; any other failure stays a network
+    // error.
+    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 +274,7 @@ impl Error {
     pub fn dispatched(&self) -> bool {
         match &*self.0 {
             Repr::Network(_)
+            | Repr::BodyCut { .. }
             | Repr::Timeout(_)
             | Repr::ReadTimeout { .. }
             | Repr::Status { .. }
@@ -285,6 +334,7 @@ impl Error {
             | Repr::Signature(_)
             | Repr::Connect(_)
             | Repr::Network(_)
+            | Repr::BodyCut { .. }
             | Repr::Timeout(_)
             | Repr::ReadTimeout { .. }
             | Repr::Status { .. }
@@ -318,6 +368,7 @@ impl Error {
             | Repr::Signature(_)
             | Repr::Connect(_)
             | Repr::Network(_)
+            | Repr::BodyCut { .. }
             | Repr::Timeout(_)
             | Repr::ReadTimeout { .. }
             | Repr::Status { .. }
@@ -355,6 +406,7 @@ impl Error {
             | Repr::Signature(_)
             | Repr::Connect(_)
             | Repr::Network(_)
+            | Repr::BodyCut { .. }
             | Repr::Timeout(_)
             | Repr::ReadTimeout { .. }
             | Repr::Status { .. }
@@ -394,6 +446,7 @@ impl Error {
             | Repr::UnsignableHeader(_)
             | Repr::Signature(_)
             | Repr::Network(_)
+            | Repr::BodyCut { .. }
             | Repr::Timeout(_)
             | Repr::ReadTimeout { .. }
             | Repr::Status { .. }
@@ -437,6 +490,7 @@ impl Error {
             | Repr::Signature(_)
             | Repr::Connect(_)
             | Repr::Network(_)
+            | Repr::BodyCut { .. }
             | Repr::Status { .. }
             | Repr::WebSocketUrl(_)
             | Repr::WebSocketScheme(_)
@@ -473,6 +527,7 @@ impl Error {
             | Repr::Signature(_)
             | Repr::Connect(_)
             | Repr::Network(_)
+            | Repr::BodyCut { .. }
             | Repr::Timeout(_)
             | Repr::ReadTimeout { .. }
             | Repr::WebSocketUrl(_)
@@ -513,6 +568,7 @@ impl Error {
             | Repr::Signature(_)
             | Repr::Connect(_)
             | Repr::Network(_)
+            | Repr::BodyCut { .. }
             | Repr::WebSocketUrl(_)
             | Repr::WebSocketScheme(_)
             | Repr::WebSocketSchemeRewrite(_)
@@ -548,6 +604,7 @@ impl Error {
             | Repr::Signature(_)
             | Repr::Connect(_)
             | Repr::Network(_)
+            | Repr::BodyCut { .. }
             | Repr::Timeout(_)
             | Repr::ReadTimeout { .. }
             | Repr::Status { .. }
@@ -742,4 +799,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");
+    }
 }

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

litellm-rust/crates/inference-messages/src/handler.rs

@@ -383,13 +383,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

litellm-rust/crates/inference/src/error.rs

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

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

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)?;
             }

litellm/rust_bridge/host/callbacks_legacy_python.py

@@ -463,9 +463,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),

litellm/rust_bridge/host/routes/messages.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 import datetime
+import errno
+import os
 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
@@ -24,7 +27,7 @@ from litellm.llms.anthropic.messages.native.handler import (
 )
 from litellm.llms.anthropic.messages.utils import is_reasoning_auto_summary_enabled
 from litellm.rust_bridge import failures
-from litellm.rust_bridge.host import callbacks_legacy_python
+from litellm.rust_bridge.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 +123,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
@@ -174,6 +178,27 @@ 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 _upstream_failure(
     error: Exception, request: Mapping[str, object]
 ) -> tuple[int, str, Sequence[tuple[str, str]]] | None:
@@ -232,6 +257,11 @@ 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))
     if getattr(error, "messages_late_rejection", False):
         _log_resolved_call(request, request_provider)
     if getattr(error, "messages_request_error", False):

tests/unit/rust_bridge/host/routes/test_messages.py

@@ -245,3 +245,57 @@ 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

tests/unit/rust_bridge/host/test_callbacks_legacy_python.py

@@ -298,3 +298,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