refactor(rust): centralize provider debug logging

Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>

litellm-rust/Cargo.lock

@@ -556,6 +556,17 @@ version = "0.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a"
 
+[[package]]
+name = "colored_json"
+version = "5.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e35980a1b846f8e3e359fd18099172a0857140ba9230affc4f71348081e039b6"
+dependencies = [
+ "serde",
+ "serde_json",
+ "yansi",
+]
+
 [[package]]
 name = "const-oid"
 version = "0.10.2"
@@ -1218,6 +1229,7 @@ version = "0.1.0"
 dependencies = [
  "axum",
  "base64",
+ "colored_json",
  "futures-channel",
  "futures-util",
  "litellm-core",
@@ -1242,6 +1254,8 @@ dependencies = [
  "aws-sigv4",
  "aws-smithy-runtime-api",
  "aws-types",
+ "bytes",
+ "futures-util",
  "rand 0.8.7",
  "reqwest",
  "serde",
@@ -1249,6 +1263,7 @@ dependencies = [
  "sha2 0.10.9",
  "thiserror 2.0.19",
  "tokio",
+ "url",
 ]
 
 [[package]]
@@ -2575,6 +2590,12 @@ version = "0.13.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4"
 
+[[package]]
+name = "yansi"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049"
+
 [[package]]
 name = "yoke"
 version = "0.8.3"

litellm-rust/Cargo.toml

@@ -29,3 +29,6 @@ tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "net"]
 tokio-tungstenite = { version = "0.24", default-features = false, features = ["connect", "rustls-tls-native-roots"] }
 futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
 base64 = "0.22"
+bytes = "1"
+colored_json = "5.0"
+url = "2.5"

litellm-rust/README.md

@@ -45,6 +45,15 @@ The folder shape follows the Python provider tree:
 `core/src/providers/<provider>/<route>/transformation.rs`. The bridge exposes one
 function per top-level route, mirroring the core entrypoints.
 
+## Provider debug logging
+
+The typed provider debug contract and `CallLogger` live in
+`crates/core/src/logging/`; the gateway renderer and activation live in
+`crates/ai-gateway/src/integrations/logging/`. Python enables the injected sink
+with `litellm._turn_on_debug()`, while the standalone gateway uses
+`LITELLM_LOG=DEBUG`. `JSON_LOGS=true` selects compact JSON; terminal pretty
+output honors `NO_COLOR`.
+
 ## Checks
 
 Run these before pushing Rust changes. GitHub Actions runs the same checks for

litellm-rust/crates/ai-gateway/Cargo.toml

@@ -24,6 +24,7 @@ tokio-tungstenite.workspace = true
 futures-util.workspace = true
 serde_json.workspace = true
 base64.workspace = true
+colored_json.workspace = true
 axum = { workspace = true, features = ["ws"], optional = true }
 serde.workspace = true
 subtle = { workspace = true, optional = true }

litellm-rust/crates/ai-gateway/src/integrations/logging/console.rs

@@ -0,0 +1,221 @@
+use std::io::{IsTerminal, Write};
+use std::sync::OnceLock;
+use std::sync::{Arc, Mutex};
+
+use colored_json::{ColorMode, ColoredFormatter, Output, PrettyFormatter};
+
+use super::{LogSink, ProviderDebugEvent};
+
+#[derive(Clone, Copy)]
+enum RenderMode {
+    Compact,
+    Pretty,
+}
+
+pub struct ConsoleDebugHook {
+    mode: RenderMode,
+    color_mode: ColorMode,
+    output: Mutex<Box<dyn Write + Send>>,
+}
+
+impl ConsoleDebugHook {
+    pub fn from_env() -> Self {
+        Self::with_writer(Box::new(std::io::stderr()))
+    }
+
+    pub fn with_writer(writer: Box<dyn Write + Send>) -> Self {
+        Self::with_writer_and_mode(writer, matches!(*render_mode(), RenderMode::Pretty))
+    }
+
+    pub fn with_writer_and_mode(writer: Box<dyn Write + Send>, pretty: bool) -> Self {
+        Self {
+            mode: if pretty {
+                RenderMode::Pretty
+            } else {
+                RenderMode::Compact
+            },
+            color_mode: ColorMode::Auto(Output::StdErr).eval(),
+            output: Mutex::new(writer),
+        }
+    }
+}
+
+pub fn hook_from_env() -> Option<Arc<dyn LogSink>> {
+    std::env::var("LITELLM_LOG")
+        .ok()
+        .filter(|value| value.eq_ignore_ascii_case("DEBUG"))
+        .map(|_| Arc::new(ConsoleDebugHook::from_env()) as Arc<dyn LogSink>)
+}
+
+pub fn hook(enabled: bool) -> Option<Arc<dyn LogSink>> {
+    enabled.then(|| Arc::new(ConsoleDebugHook::from_env()) as Arc<dyn LogSink>)
+}
+
+fn render_mode() -> &'static RenderMode {
+    static MODE: OnceLock<RenderMode> = OnceLock::new();
+    MODE.get_or_init(|| {
+        if std::env::var("JSON_LOGS")
+            .map(|value| value.eq_ignore_ascii_case("true"))
+            .unwrap_or(false)
+            || !std::io::stderr().is_terminal()
+        {
+            RenderMode::Compact
+        } else {
+            RenderMode::Pretty
+        }
+    })
+}
+
+fn header(event: &ProviderDebugEvent) -> String {
+    match event {
+        ProviderDebugEvent::Request(value) => {
+            format!("provider.request {} {}", value.call_id, value.provider)
+        }
+        ProviderDebugEvent::Response(value) => format!(
+            "provider.response {} {} status={} duration_ms={}",
+            value.call_id, value.provider, value.status, value.duration_ms
+        ),
+        ProviderDebugEvent::StreamStarted(value) => format!(
+            "provider.stream.started {} {} status={}",
+            value.call_id, value.provider, value.status
+        ),
+        ProviderDebugEvent::StreamCompleted(value) => format!(
+            "provider.stream.completed {} {} duration_ms={}",
+            value.call_id, value.provider, value.duration_ms
+        ),
+        ProviderDebugEvent::Error(value) => format!(
+            "provider.error {} {}{} duration_ms={}",
+            value.call_id,
+            value.provider,
+            value
+                .status
+                .map_or(String::new(), |status| format!(" status={status}")),
+            value.duration_ms
+        ),
+    }
+}
+
+fn decorate(value: &str, color_mode: ColorMode, code: &str) -> String {
+    if color_mode == ColorMode::On {
+        format!("\x1b[{code}m{value}\x1b[0m")
+    } else {
+        value.to_string()
+    }
+}
+
+impl LogSink for ConsoleDebugHook {
+    fn emit(&self, event: &ProviderDebugEvent) {
+        let Ok(mut output) = self.output.lock() else {
+            return;
+        };
+        let Ok(json) = serde_json::to_string(event) else {
+            return;
+        };
+        match self.mode {
+            RenderMode::Compact => {
+                let _ = writeln!(output, "{json}");
+            }
+            RenderMode::Pretty => {
+                let pretty = serde_json::to_string_pretty(event).unwrap_or(json);
+                let _ = writeln!(
+                    output,
+                    "{}",
+                    decorate(&header(event), self.color_mode, "36")
+                );
+                let rendered = if self.color_mode == ColorMode::Off {
+                    pretty
+                } else {
+                    ColoredFormatter::new(PrettyFormatter::new())
+                        .to_colored_json(event, self.color_mode)
+                        .unwrap_or(pretty)
+                };
+                let _ = writeln!(output, "{rendered}");
+                let _ = writeln!(
+                    output,
+                    "{}",
+                    decorate(
+                        "────────────────────────────────────────",
+                        self.color_mode,
+                        "2"
+                    )
+                );
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Arc, Mutex};
+
+    use serde_json::json;
+
+    use super::*;
+    use litellm_core::logging::{RequestEventInput, request_event};
+
+    struct Buffer(Arc<Mutex<Vec<u8>>>);
+
+    impl Write for Buffer {
+        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+            self.0.lock().expect("buffer lock").extend_from_slice(bytes);
+            Ok(bytes.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn compact_output_is_canonical_json() {
+        let buffer = Arc::new(Mutex::new(Vec::new()));
+        let hook = ConsoleDebugHook::with_writer_and_mode(Box::new(Buffer(buffer.clone())), false);
+        let event = request_event(RequestEventInput {
+            call_id: "call_01".to_string(),
+            provider: "anthropic".to_string(),
+            model: "claude".to_string(),
+            stream: false,
+            url: "https://example.test".to_string(),
+            headers: Vec::new(),
+            body: json!({"prompt": "visible"}),
+        });
+        let expected = serde_json::to_value(&event).expect("event serializes");
+        hook.emit(&event);
+        let output =
+            String::from_utf8(buffer.lock().expect("buffer lock").clone()).expect("output is utf8");
+        assert_eq!(
+            serde_json::from_str::<serde_json::Value>(output.trim()).expect("output is JSON"),
+            expected
+        );
+    }
+
+    #[test]
+    fn pretty_output_has_header_separator_and_indented_payload() {
+        let buffer = Arc::new(Mutex::new(Vec::new()));
+        let hook = ConsoleDebugHook::with_writer_and_mode(Box::new(Buffer(buffer.clone())), true);
+        let event = request_event(RequestEventInput {
+            call_id: "call_01".to_string(),
+            provider: "anthropic".to_string(),
+            model: "claude".to_string(),
+            stream: false,
+            url: "https://example.test".to_string(),
+            headers: Vec::new(),
+            body: json!({"prompt": "visible"}),
+        });
+        hook.emit(&event);
+        let output =
+            String::from_utf8(buffer.lock().expect("buffer lock").clone()).expect("output is utf8");
+        assert!(!output.contains('\x1b'));
+        let payload = &output
+            [output.find('{').expect("payload starts")..=output.rfind('}').expect("payload ends")];
+        let expected = serde_json::to_string_pretty(&event).expect("event pretty serializes");
+        assert_eq!(payload, expected);
+        assert!(
+            payload.find("\"event\"").expect("event key")
+                < payload.find("\"body\"").expect("body key")
+        );
+        assert!(output.contains("provider.request call_01 anthropic"));
+        assert!(output.contains("────────────────"));
+        assert!(output.contains("\n  \"event\""));
+    }
+}

litellm-rust/crates/ai-gateway/src/integrations/logging/mod.rs

@@ -0,0 +1,8 @@
+pub mod console;
+
+pub use litellm_core::logging::{
+    BodySnapshot, ErrorEventInput, LogSink, ProviderDebugEvent, ProviderErrorEvent,
+    ProviderRequestEvent, ProviderResponseEvent, ProviderStreamCompletedEvent,
+    ProviderStreamStartedEvent, RequestEventInput, ResponseBody, ResponseEventInput, error_event,
+    request_event, response_event, stream_completed, stream_started,
+};

litellm-rust/crates/ai-gateway/src/integrations/mod.rs

@@ -9,4 +9,5 @@
 pub mod custom_guardrail;
 pub mod custom_logger;
 pub mod litellm_python_proxy_api;
+pub mod logging;
 pub mod types;

litellm-rust/crates/ai-gateway/src/main.rs

@@ -18,6 +18,7 @@ use litellm_core::router::{Deployment, LiteLLMParams, Router};
 
 use litellm_ai_gateway::integrations::custom_logger::CustomLogger;
 use litellm_ai_gateway::integrations::litellm_python_proxy_api::LiteLLMPythonProxyAPILogger;
+use litellm_ai_gateway::integrations::logging::console::hook_from_env;
 #[cfg(feature = "python-config")]
 use litellm_ai_gateway::python;
 
@@ -72,6 +73,7 @@ async fn main() {
         master_key,
         loggers: Arc::new(loggers),
         realtime_pool,
+        logging_sink: hook_from_env(),
     };
 
     let host = std::env::var("HOST").unwrap_or_else(|_| DEFAULT_HOST.to_string());

litellm-rust/crates/ai-gateway/src/ocr/handler.rs

@@ -1,5 +1,6 @@
 use litellm_core::CoreResult;
 use litellm_core::error::CoreError;
+use litellm_core::logging::http::{JsonRequest, execute_json};
 use litellm_core::ocr::transformation::OcrResponseHandling;
 use serde_json::Value;
 
@@ -8,6 +9,26 @@ use super::types::ProviderOcrRequest;
 use crate::client::http_client;
 
 pub(crate) async fn execute_ocr_provider_call(request: ProviderOcrRequest) -> CoreResult<Value> {
+    if request.config.response_handling() != OcrResponseHandling::AzureDocumentIntelligencePoll {
+        let response = execute_json::<Value>(
+            http_client(),
+            JsonRequest {
+                logger: request.logger,
+                model: request.model.clone(),
+                stream: false,
+                url: request.url,
+                headers: request.upstream_headers,
+                body: request.body,
+                timeout: request.timeout,
+            },
+        )
+        .await?;
+        return Ok(request
+            .config
+            .transform_ocr_response(&request.model, response)?
+            .into_json());
+    }
+
     let mut request_builder = http_client().post(&request.url).json(&request.body);
     for (key, value) in &request.upstream_headers {
         request_builder = request_builder.header(key, value);

litellm-rust/crates/ai-gateway/src/ocr/hooks.rs

@@ -111,6 +111,7 @@ impl OcrLifecycleHooks {
             body,
             upstream_headers,
             timeout: request.timeout,
+            logger: request.logger,
         })
     }
 

litellm-rust/crates/ai-gateway/src/ocr/prepare.rs

@@ -1,3 +1,4 @@
+use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -7,6 +8,8 @@ use super::hooks::OcrLifecycleHooks;
 use super::types::{OcrRequest, PreparedOcrRequest};
 use crate::integrations::custom_guardrail::CustomGuardrailRunner;
 use crate::integrations::custom_logger::CustomLoggerRunner;
+use litellm_core::call_lifecycle::CallLifecycleContext;
+use litellm_core::logging::CallLogger;
 
 pub(crate) struct PreparedOcrCall {
     pub(crate) request: PreparedOcrRequest,
@@ -26,6 +29,12 @@ pub(crate) fn prepare_ocr_call(request: OcrRequest<'_>) -> PreparedOcrCall {
     let model = provider_info.model.to_string();
     let custom_llm_provider = provider_info.custom_llm_provider.to_string();
 
+    let context = CallLifecycleContext::new(
+        "ocr",
+        request.model,
+        custom_llm_provider.clone(),
+        call_id.clone(),
+    );
     PreparedOcrCall {
         request: PreparedOcrRequest {
             model,
@@ -37,6 +46,7 @@ pub(crate) fn prepare_ocr_call(request: OcrRequest<'_>) -> PreparedOcrCall {
             extra_headers: request.extra_headers,
             optional_params: request.optional_params,
             timeout: request.timeout,
+            logger: Arc::new(CallLogger::new(&context, request.logging_sink)),
         },
         hooks: OcrLifecycleHooks::new(
             CustomLoggerRunner::new(request.callbacks),

litellm-rust/crates/ai-gateway/src/ocr/tests.rs

@@ -326,6 +326,7 @@ async fn ocr_lifecycle_runs_pre_during_and_success_hooks() {
             ..Default::default()
         },
         litellm_call_id: Some("ocr-call-1"),
+        logging_sink: None,
     })
     .await
     .expect("ocr request succeeds");
@@ -391,6 +392,7 @@ async fn ocr_lifecycle_runs_failure_hook_on_provider_error() {
         guardrails: Vec::new(),
         request_metadata: RequestMetadata::default(),
         litellm_call_id: Some("ocr-call-2"),
+        logging_sink: None,
     })
     .await
     .expect_err("provider error propagates");
@@ -435,6 +437,7 @@ async fn ocr_lifecycle_pre_call_block_skips_provider_socket() {
         guardrails: vec![guardrail.clone()],
         request_metadata: RequestMetadata::default(),
         litellm_call_id: Some("ocr-call-3"),
+        logging_sink: None,
     })
     .await
     .expect_err("guardrail blocks request");
@@ -505,6 +508,7 @@ async fn ocr_does_not_duplicate_authorization_header_when_header_is_supplied() {
         guardrails: Vec::new(),
         request_metadata: RequestMetadata::default(),
         litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect("ocr request succeeds");
@@ -574,6 +578,7 @@ async fn document_intelligence_poll_uses_resolved_subscription_key() {
         guardrails: Vec::new(),
         request_metadata: RequestMetadata::default(),
         litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect("document intelligence request succeeds");

litellm-rust/crates/ai-gateway/src/ocr/types.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use litellm_core::call_lifecycle::{CallLifecycleContext, CallLifecycleRequest};
+use litellm_core::logging::{CallLogger, LogSink};
 use litellm_core::ocr::transformation::OcrProviderConfig;
 use serde_json::{Map, Value};
 
@@ -22,6 +23,7 @@ pub struct OcrRequest<'a> {
     pub guardrails: Vec<Arc<dyn CustomGuardrail>>,
     pub request_metadata: RequestMetadata,
     pub litellm_call_id: Option<&'a str>,
+    pub logging_sink: Option<Arc<dyn LogSink>>,
 }
 
 pub(crate) struct PreparedOcrRequest {
@@ -34,6 +36,7 @@ pub(crate) struct PreparedOcrRequest {
     pub(crate) extra_headers: Option<Map<String, Value>>,
     pub(crate) optional_params: Map<String, Value>,
     pub(crate) timeout: Option<Duration>,
+    pub(crate) logger: Arc<CallLogger>,
 }
 
 impl CallLifecycleRequest for PreparedOcrRequest {
@@ -54,4 +57,5 @@ pub(crate) struct ProviderOcrRequest {
     pub(crate) body: Value,
     pub(crate) upstream_headers: Vec<(String, String)>,
     pub(crate) timeout: Option<Duration>,
+    pub(crate) logger: Arc<CallLogger>,
 }

litellm-rust/crates/ai-gateway/src/realtime/streaming.rs

@@ -106,16 +106,16 @@ impl RealTimeStreaming {
     /// `litellm_call_id`, replacing the gateway-generated fallback.
     fn on_session(&mut self, event: &RealtimeEvent) {
         let session = event.data.get("session").and_then(Value::as_object);
-        if let Some(id) = session.and_then(|s| s.get("id")).and_then(Value::as_str) {
-            if !id.is_empty() {
-                self.id = id.to_string();
-                self.litellm_call_id = id.to_string();
-            }
+        if let Some(id) = session.and_then(|s| s.get("id")).and_then(Value::as_str)
+            && !id.is_empty()
+        {
+            self.id = id.to_string();
+            self.litellm_call_id = id.to_string();
         }
-        if let Some(model) = session.and_then(|s| s.get("model")).and_then(Value::as_str) {
-            if !model.is_empty() {
-                self.model = model.to_string();
-            }
+        if let Some(model) = session.and_then(|s| s.get("model")).and_then(Value::as_str)
+            && !model.is_empty()
+        {
+            self.model = model.to_string();
         }
     }
 

litellm-rust/crates/ai-gateway/src/routes/messages/mod.rs

@@ -10,6 +10,7 @@ use axum::http::header::{CACHE_CONTROL, CONTENT_TYPE, HeaderMap, HeaderValue};
 use axum::response::{IntoResponse, Response};
 use axum::routing::post;
 use litellm_core::CoreError;
+use litellm_core::logging::stream::count_forwarded_stream;
 use serde_json::{Map, Value};
 
 use crate::auth::RequireMasterKey;
@@ -28,7 +29,7 @@ async fn handle(
     Json(body): Json<Value>,
 ) -> Result<Response, MessagesRouteError> {
     let extra_headers = forwarded_headers(&headers)?;
-    match service::run(&state.router, body, extra_headers)
+    match service::run(&state.router, body, extra_headers, state.logging_sink)
         .await
         .map_err(MessagesRouteError::from)?
     {
@@ -37,31 +38,33 @@ async fn handle(
     }
 }
 
-fn stream_response(upstream: reqwest::Response) -> Result<Response, MessagesRouteError> {
+fn stream_response(
+    upstream: litellm_core::messages::types::MessagesStreamResponse,
+) -> Result<Response, MessagesRouteError> {
     let content_type = upstream
+        .response
         .headers()
         .get(CONTENT_TYPE)
         .cloned()
         .unwrap_or_else(|| HeaderValue::from_static("text/event-stream"));
     let mut response = Response::builder()
         .status(
-            StatusCode::from_u16(upstream.status().as_u16()).map_err(|error| {
+            StatusCode::from_u16(upstream.response.status().as_u16()).map_err(|error| {
                 MessagesRouteError(CoreError::InvalidResponse(format!(
                     "invalid upstream response status: {error}"
                 )))
             })?,
         )
         .header(CONTENT_TYPE, content_type);
-    if let Some(value) = upstream.headers().get(CACHE_CONTROL) {
+    if let Some(value) = upstream.response.headers().get(CACHE_CONTROL) {
         response = response.header(CACHE_CONTROL, value);
     }
-    response
-        .body(Body::from_stream(upstream.bytes_stream()))
-        .map_err(|error| {
-            MessagesRouteError(CoreError::InvalidResponse(format!(
-                "failed to build streaming response: {error}"
-            )))
-        })
+    let content = count_forwarded_stream(upstream.response.bytes_stream(), upstream.logger);
+    response.body(Body::from_stream(content)).map_err(|error| {
+        MessagesRouteError(CoreError::InvalidResponse(format!(
+            "failed to build streaming response: {error}"
+        )))
+    })
 }
 
 fn forwarded_headers(headers: &HeaderMap) -> Result<Option<Map<String, Value>>, CoreError> {
@@ -160,6 +163,7 @@ mod tests {
             master_key: master_key.map(Arc::from),
             loggers: Arc::new(Vec::new()),
             realtime_pool: RealtimePool::disabled(),
+            logging_sink: None,
         }
     }
 

litellm-rust/crates/ai-gateway/src/routes/messages/service.rs

@@ -1,3 +1,4 @@
+use litellm_core::logging::LogSink;
 use std::sync::Arc;
 
 use litellm_core::constants::ANTHROPIC_MESSAGES_PROVIDER;
@@ -9,13 +10,14 @@ use serde_json::{Map, Value};
 
 pub(crate) enum MessagesResponse {
     Json(Value),
-    Stream(reqwest::Response),
+    Stream(litellm_core::messages::types::MessagesStreamResponse),
 }
 
 pub async fn run(
     router: &Arc<Router>,
     body: Value,
     extra_headers: Option<Map<String, Value>>,
+    logging_sink: Option<Arc<dyn LogSink>>,
 ) -> CoreResult<MessagesResponse> {
     let model = body
         .get("model")
@@ -51,6 +53,8 @@ pub async fn run(
         custom_llm_provider,
         extra_headers,
         timeout: None,
+        litellm_call_id: None,
+        logging_sink,
     };
     if request.body.get("stream").and_then(Value::as_bool) == Some(true) {
         return messages_stream(request).await.map(MessagesResponse::Stream);

litellm-rust/crates/ai-gateway/src/routes/realtime/service.rs

@@ -51,18 +51,17 @@ where
         provider_model,
         params.api_key.as_deref(),
         params.api_base.as_deref(),
-    ) {
-        if let Some(handoff) = pool.take(&key) {
-            return crate::io::realtime::realtime_warm(
-                provider_model,
-                handoff,
-                idle_timeout,
-                observe,
-                client_in,
-                client_out,
-            )
-            .await;
-        }
+    ) && let Some(handoff) = pool.take(&key)
+    {
+        return crate::io::realtime::realtime_warm(
+            provider_model,
+            handoff,
+            idle_timeout,
+            observe,
+            client_in,
+            client_out,
+        )
+        .await;
     }
 
     // Cold path: fresh dial (the original behavior).

litellm-rust/crates/ai-gateway/src/routes/responses/mod.rs

@@ -318,6 +318,7 @@ mod tests {
             master_key: Some(Arc::from("master-key")),
             loggers: Arc::new(Vec::new()),
             realtime_pool: RealtimePool::disabled(),
+            logging_sink: None,
         }
     }
 

litellm-rust/crates/ai-gateway/src/state.rs

@@ -4,6 +4,7 @@ use crate::io::realtime_pool::RealtimePool;
 use litellm_core::router::Router;
 
 use crate::integrations::custom_logger::CustomLogger;
+use litellm_core::logging::LogSink;
 
 /// Shared application state handed to every route handler.
 #[derive(Clone)]
@@ -18,4 +19,5 @@ pub struct AppState {
     /// (`RealtimePool::disabled()`) when `REALTIME_POOL_SIZE=0`, in which case
     /// every realtime connect fresh-dials exactly as before.
     pub realtime_pool: Arc<RealtimePool>,
+    pub logging_sink: Option<Arc<dyn LogSink>>,
 }

litellm-rust/crates/core/CLAUDE.md

@@ -27,6 +27,8 @@ Not allowed:
 - Config file reading or rollout state; the host resolves those and passes them
   in. Env reads are limited to credential fallback in a route's `prepare.rs`.
 - Logging callbacks, tracing spans, spend writes, or customer callbacks.
+- The host-injected debug log sink is the exception; it receives redacted events
+  without env reads, I/O, or callback dispatch in core.
 - Provider-specific branching that belongs in `providers`.
 - Panics for user/provider-controlled input.
 

litellm-rust/crates/core/Cargo.toml

@@ -8,6 +8,9 @@ repository.workspace = true
 [dependencies]
 rand.workspace = true
 reqwest.workspace = true
+bytes.workspace = true
+futures-util.workspace = true
+url.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true

litellm-rust/crates/core/src/lib.rs

@@ -3,6 +3,7 @@ pub mod caching;
 pub mod call_lifecycle;
 pub mod constants;
 pub mod error;
+pub mod logging;
 pub mod messages;
 pub mod ocr;
 pub mod providers;

litellm-rust/crates/core/src/logging/events.rs

@@ -0,0 +1,256 @@
+use serde::Serialize;
+use serde_json::Value;
+use std::collections::BTreeMap;
+
+pub use super::redaction::BodySnapshot;
+use super::redaction::{redact_headers, redact_url, snapshot_json};
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(tag = "event")]
+pub enum ProviderDebugEvent {
+    #[serde(rename = "provider.request")]
+    Request(ProviderRequestEvent),
+    #[serde(rename = "provider.response")]
+    Response(ProviderResponseEvent),
+    #[serde(rename = "provider.stream.started")]
+    StreamStarted(ProviderStreamStartedEvent),
+    #[serde(rename = "provider.stream.completed")]
+    StreamCompleted(ProviderStreamCompletedEvent),
+    #[serde(rename = "provider.error")]
+    Error(ProviderErrorEvent),
+}
+
+pub struct RequestEventInput {
+    pub call_id: String,
+    pub provider: String,
+    pub model: String,
+    pub stream: bool,
+    pub url: String,
+    pub headers: Vec<(String, String)>,
+    pub body: Value,
+}
+
+pub struct ResponseEventInput {
+    pub call_id: String,
+    pub provider: String,
+    pub status: u16,
+    pub duration_ms: u128,
+    pub headers: Vec<(String, String)>,
+    pub body: ResponseBody,
+}
+
+pub struct ErrorEventInput {
+    pub call_id: String,
+    pub provider: String,
+    pub duration_ms: u128,
+    pub status: Option<u16>,
+    pub kind: &'static str,
+    pub message: String,
+    pub body: Option<ResponseBody>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct ProviderRequestEvent {
+    pub source: &'static str,
+    pub call_id: String,
+    pub provider: String,
+    pub model: String,
+    pub stream: bool,
+    pub method: &'static str,
+    pub url: String,
+    pub headers: BTreeMap<String, String>,
+    pub body: Value,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub body_truncated: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub body_original_bytes: Option<usize>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct ProviderResponseEvent {
+    pub source: &'static str,
+    pub call_id: String,
+    pub provider: String,
+    pub status: u16,
+    pub duration_ms: u128,
+    pub headers: BTreeMap<String, String>,
+    pub body: Value,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub body_truncated: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub body_original_bytes: Option<usize>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct ProviderStreamStartedEvent {
+    pub source: &'static str,
+    pub call_id: String,
+    pub provider: String,
+    pub status: u16,
+    pub content_type: Option<String>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct ProviderStreamCompletedEvent {
+    pub source: &'static str,
+    pub call_id: String,
+    pub provider: String,
+    pub duration_ms: u128,
+    pub bytes_received: usize,
+    pub frames_received: usize,
+    pub events_decoded: usize,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct ProviderErrorEvent {
+    pub source: &'static str,
+    pub call_id: String,
+    pub provider: String,
+    pub duration_ms: u128,
+    pub status: Option<u16>,
+    pub kind: &'static str,
+    pub message: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub body: Option<Value>,
+}
+
+pub fn request_event(input: RequestEventInput) -> ProviderDebugEvent {
+    let snapshot = snapshot_json(input.body);
+    ProviderDebugEvent::Request(ProviderRequestEvent {
+        source: "litellm-rust",
+        call_id: input.call_id,
+        provider: input.provider,
+        model: input.model,
+        stream: input.stream,
+        method: "POST",
+        url: redact_url(&input.url),
+        headers: redact_headers(&input.headers),
+        body: snapshot.body,
+        body_truncated: snapshot.body_truncated,
+        body_original_bytes: snapshot.body_original_bytes,
+    })
+}
+
+pub fn response_event(input: ResponseEventInput) -> ProviderDebugEvent {
+    let snapshot = input.body.snapshot();
+    ProviderDebugEvent::Response(ProviderResponseEvent {
+        source: "litellm-rust",
+        call_id: input.call_id,
+        provider: input.provider,
+        status: input.status,
+        duration_ms: input.duration_ms,
+        headers: redact_headers(&input.headers),
+        body: snapshot.body,
+        body_truncated: snapshot.body_truncated,
+        body_original_bytes: snapshot.body_original_bytes,
+    })
+}
+
+pub fn error_event(input: ErrorEventInput) -> ProviderDebugEvent {
+    ProviderDebugEvent::Error(ProviderErrorEvent {
+        source: "litellm-rust",
+        call_id: input.call_id,
+        provider: input.provider,
+        duration_ms: input.duration_ms,
+        status: input.status,
+        kind: input.kind,
+        message: input.message,
+        body: input.body.map(|body| body.snapshot().body),
+    })
+}
+
+pub fn stream_started(
+    call_id: String,
+    provider: String,
+    status: u16,
+    content_type: Option<String>,
+) -> ProviderDebugEvent {
+    ProviderDebugEvent::StreamStarted(ProviderStreamStartedEvent {
+        source: "litellm-rust",
+        call_id,
+        provider,
+        status,
+        content_type,
+    })
+}
+
+pub fn stream_completed(
+    call_id: String,
+    provider: String,
+    duration_ms: u128,
+    bytes_received: usize,
+    frames_received: usize,
+    events_decoded: usize,
+) -> ProviderDebugEvent {
+    ProviderDebugEvent::StreamCompleted(ProviderStreamCompletedEvent {
+        source: "litellm-rust",
+        call_id,
+        provider,
+        duration_ms,
+        bytes_received,
+        frames_received,
+        events_decoded,
+    })
+}
+
+#[derive(Clone, Debug)]
+pub enum ResponseBody {
+    Json(Value),
+    Binary {
+        media_type: Option<String>,
+        bytes: usize,
+    },
+}
+
+impl ResponseBody {
+    fn snapshot(self) -> BodySnapshot {
+        match self {
+            Self::Json(value) => snapshot_json(value),
+            Self::Binary { media_type, bytes } => BodySnapshot {
+                body: serde_json::json!({"media_type": media_type, "bytes": bytes}),
+                body_truncated: None,
+                body_original_bytes: None,
+            },
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn redacts_credentials_recursively() {
+        let event = request_event(RequestEventInput {
+            call_id: "call_01".to_string(),
+            provider: "anthropic".to_string(),
+            model: "claude".to_string(),
+            stream: false,
+            url: "https://example.test?signature=secret&x=ok".to_string(),
+            headers: vec![("Authorization".to_string(), "Bearer secret".to_string())],
+            body: serde_json::json!({"nested": {"token": "secret"}, "prompt": "visible"}),
+        });
+        let json = serde_json::to_string(&event).expect("serializes");
+        assert!(!json.contains("secret"));
+        assert!(json.contains("visible"));
+        assert!(json.contains("[REDACTED]"));
+    }
+
+    #[test]
+    fn redact_url_preserves_queryless_encoded_paths() {
+        assert_eq!(
+            redact_url("https://example.test/v1%3A0/invoke"),
+            "https://example.test/v1%3A0/invoke"
+        );
+    }
+
+    #[test]
+    fn redact_url_preserves_non_secret_query_params() {
+        let redacted = redact_url(
+            "https://example.test/invoke?X-Amz-Signature=sig&X-Amz-Credential=cred&foo=bar",
+        );
+        assert!(redacted.contains("X-Amz-Signature=%5BREDACTED%5D"));
+        assert!(redacted.contains("X-Amz-Credential=%5BREDACTED%5D"));
+        assert!(redacted.contains("foo=bar"));
+    }
+}

litellm-rust/crates/core/src/logging/http.rs

@@ -0,0 +1,202 @@
+use std::time::Duration;
+
+use serde::de::DeserializeOwned;
+use serde_json::Value;
+
+use crate::CoreResult;
+use crate::error::CoreError;
+
+use super::{CallLogger, ResponseBody};
+
+pub struct JsonRequest {
+    pub logger: std::sync::Arc<CallLogger>,
+    pub model: String,
+    pub stream: bool,
+    pub url: String,
+    pub headers: Vec<(String, String)>,
+    pub body: Value,
+    pub timeout: Option<Duration>,
+}
+
+pub async fn execute_json<T: DeserializeOwned>(
+    client: &reqwest::Client,
+    request: JsonRequest,
+) -> CoreResult<T> {
+    let body_bytes = serde_json::to_vec(&request.body)
+        .map_err(|error| CoreError::InvalidRequest(error.to_string()))?;
+    request.logger.request_about_to_be_sent(
+        request.model,
+        request.stream,
+        request.url.clone(),
+        request.headers.clone(),
+        request.body,
+    );
+    let builder = request.headers.iter().fold(
+        client.post(&request.url).body(body_bytes),
+        |builder, (name, value)| builder.header(name, value),
+    );
+    let builder = match request.timeout {
+        Some(timeout) => builder.timeout(timeout),
+        None => builder,
+    };
+    let response = builder.send().await.map_err(|error| {
+        request
+            .logger
+            .failure(None, "network_error", error.to_string(), None);
+        CoreError::Network(error.to_string())
+    })?;
+    let status = response.status();
+    let headers = response
+        .headers()
+        .iter()
+        .filter_map(|(name, value)| {
+            value
+                .to_str()
+                .ok()
+                .map(|value| (name.to_string(), value.to_string()))
+        })
+        .collect::<Vec<_>>();
+    let media_type = headers
+        .iter()
+        .find(|(name, _)| name.eq_ignore_ascii_case("content-type"))
+        .map(|(_, value)| value.clone());
+    let text = response.text().await.map_err(|error| {
+        request.logger.failure(
+            Some(status.as_u16()),
+            "network_error",
+            error.to_string(),
+            None,
+        );
+        CoreError::Network(error.to_string())
+    })?;
+    if !status.is_success() {
+        let body = serde_json::from_str(&text)
+            .map(ResponseBody::Json)
+            .unwrap_or(ResponseBody::Binary {
+                media_type,
+                bytes: text.len(),
+            });
+        request.logger.failure(
+            Some(status.as_u16()),
+            "http_error",
+            format!("provider returned HTTP {}", status.as_u16()),
+            Some(body),
+        );
+        return Err(CoreError::Http {
+            status: status.as_u16(),
+            body: crate::messages::common_utils::truncate_error_body(&text),
+        });
+    }
+    if !media_type
+        .as_deref()
+        .is_some_and(|value| value.to_ascii_lowercase().contains("json"))
+    {
+        request.logger.response_received(
+            status.as_u16(),
+            headers,
+            ResponseBody::Binary {
+                media_type,
+                bytes: text.len(),
+            },
+        );
+        return Err(CoreError::InvalidResponse(
+            "provider response was not JSON".to_string(),
+        ));
+    }
+    let value = serde_json::from_str::<Value>(&text).map_err(|error| {
+        request.logger.failure(
+            Some(status.as_u16()),
+            "invalid_json",
+            error.to_string(),
+            Some(ResponseBody::Binary {
+                media_type,
+                bytes: text.len(),
+            }),
+        );
+        CoreError::InvalidResponse(format!("invalid provider response JSON: {error}"))
+    })?;
+    let typed = T::deserialize(&value).map_err(|error| {
+        CoreError::InvalidResponse(format!("invalid provider response: {error}"))
+    })?;
+    request
+        .logger
+        .response_received(status.as_u16(), headers, ResponseBody::Json(value));
+    Ok(typed)
+}
+
+pub async fn execute_stream(
+    client: &reqwest::Client,
+    request: JsonRequest,
+) -> CoreResult<reqwest::Response> {
+    let body_bytes = serde_json::to_vec(&request.body)
+        .map_err(|error| CoreError::InvalidRequest(error.to_string()))?;
+    request.logger.request_about_to_be_sent(
+        request.model,
+        true,
+        request.url.clone(),
+        request.headers.clone(),
+        request.body,
+    );
+    let builder = request.headers.iter().fold(
+        client.post(&request.url).body(body_bytes),
+        |builder, (name, value)| builder.header(name, value),
+    );
+    let builder = match request.timeout {
+        Some(timeout) => builder.timeout(timeout),
+        None => builder,
+    };
+    let response = builder.send().await.map_err(|error| {
+        request
+            .logger
+            .failure(None, "network_error", error.to_string(), None);
+        CoreError::Network(error.to_string())
+    })?;
+    let status = response.status();
+    if !status.is_success() {
+        let headers = response
+            .headers()
+            .iter()
+            .filter_map(|(name, value)| {
+                value
+                    .to_str()
+                    .ok()
+                    .map(|value| (name.to_string(), value.to_string()))
+            })
+            .collect::<Vec<_>>();
+        let text = response.text().await.map_err(|error| {
+            request.logger.failure(
+                Some(status.as_u16()),
+                "network_error",
+                error.to_string(),
+                None,
+            );
+            CoreError::Network(error.to_string())
+        })?;
+        let body = serde_json::from_str(&text)
+            .map(ResponseBody::Json)
+            .unwrap_or(ResponseBody::Binary {
+                media_type: headers
+                    .iter()
+                    .find(|(name, _)| name.eq_ignore_ascii_case("content-type"))
+                    .map(|(_, value)| value.clone()),
+                bytes: text.len(),
+            });
+        request.logger.failure(
+            Some(status.as_u16()),
+            "http_error",
+            format!("provider returned HTTP {}", status.as_u16()),
+            Some(body),
+        );
+        return Err(CoreError::Http {
+            status: status.as_u16(),
+            body: crate::messages::common_utils::truncate_error_body(&text),
+        });
+    }
+    let content_type = response
+        .headers()
+        .get("content-type")
+        .and_then(|value| value.to_str().ok())
+        .map(str::to_string);
+    request.logger.stream_started(status.as_u16(), content_type);
+    Ok(response)
+}

litellm-rust/crates/core/src/logging/mod.rs

@@ -0,0 +1,165 @@
+mod redaction;
+
+pub mod events;
+pub mod http;
+pub mod stream;
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Instant;
+
+use crate::call_lifecycle::CallLifecycleContext;
+
+pub trait LogSink: Send + Sync {
+    fn emit(&self, event: &ProviderDebugEvent);
+}
+
+pub struct CallLogger {
+    context: CallLifecycleContext,
+    sink: Option<Arc<dyn LogSink>>,
+    started: Instant,
+    bytes_received: AtomicUsize,
+    frames_received: AtomicUsize,
+    events_decoded: AtomicUsize,
+}
+
+impl CallLogger {
+    pub fn new(context: &CallLifecycleContext, sink: Option<Arc<dyn LogSink>>) -> Self {
+        Self {
+            context: context.clone(),
+            sink,
+            started: Instant::now(),
+            bytes_received: AtomicUsize::new(0),
+            frames_received: AtomicUsize::new(0),
+            events_decoded: AtomicUsize::new(0),
+        }
+    }
+
+    pub fn request_about_to_be_sent(
+        &self,
+        model: String,
+        stream: bool,
+        url: String,
+        headers: Vec<(String, String)>,
+        body: serde_json::Value,
+    ) {
+        self.emit(events::request_event(events::RequestEventInput {
+            call_id: self.context.litellm_call_id.clone(),
+            provider: self.context.custom_llm_provider.clone(),
+            model,
+            stream,
+            url,
+            headers,
+            body,
+        }));
+    }
+
+    pub fn response_received(
+        &self,
+        status: u16,
+        headers: Vec<(String, String)>,
+        body: ResponseBody,
+    ) {
+        self.emit(events::response_event(events::ResponseEventInput {
+            call_id: self.context.litellm_call_id.clone(),
+            provider: self.context.custom_llm_provider.clone(),
+            status,
+            duration_ms: self.started.elapsed().as_millis(),
+            headers,
+            body,
+        }));
+    }
+
+    pub fn failure(
+        &self,
+        status: Option<u16>,
+        kind: &'static str,
+        message: String,
+        body: Option<ResponseBody>,
+    ) {
+        self.emit(events::error_event(events::ErrorEventInput {
+            call_id: self.context.litellm_call_id.clone(),
+            provider: self.context.custom_llm_provider.clone(),
+            duration_ms: self.started.elapsed().as_millis(),
+            status,
+            kind,
+            message,
+            body,
+        }));
+    }
+
+    pub fn stream_started(&self, status: u16, content_type: Option<String>) {
+        self.emit(events::stream_started(
+            self.context.litellm_call_id.clone(),
+            self.context.custom_llm_provider.clone(),
+            status,
+            content_type,
+        ));
+    }
+
+    pub fn stream_finished(&self) {
+        self.emit(events::stream_completed(
+            self.context.litellm_call_id.clone(),
+            self.context.custom_llm_provider.clone(),
+            self.started.elapsed().as_millis(),
+            self.bytes_received.load(Ordering::Relaxed),
+            self.frames_received.load(Ordering::Relaxed),
+            self.events_decoded.load(Ordering::Relaxed),
+        ));
+    }
+
+    pub fn stream_chunk_observed(&self, bytes: usize, events: usize) {
+        self.bytes_received.fetch_add(bytes, Ordering::Relaxed);
+        self.frames_received.fetch_add(1, Ordering::Relaxed);
+        self.events_decoded.fetch_add(events, Ordering::Relaxed);
+    }
+
+    fn emit(&self, event: ProviderDebugEvent) {
+        if let Some(sink) = &self.sink {
+            sink.emit(&event);
+        }
+    }
+}
+
+pub use events::{
+    BodySnapshot, ErrorEventInput, ProviderDebugEvent, ProviderErrorEvent, ProviderRequestEvent,
+    ProviderResponseEvent, ProviderStreamCompletedEvent, ProviderStreamStartedEvent,
+    RequestEventInput, ResponseBody, ResponseEventInput, error_event, request_event,
+    response_event, stream_completed, stream_started,
+};
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Arc, Mutex};
+
+    use super::*;
+
+    #[derive(Clone, Default)]
+    struct RecordingSink(Arc<Mutex<Vec<ProviderDebugEvent>>>);
+
+    impl LogSink for RecordingSink {
+        fn emit(&self, event: &ProviderDebugEvent) {
+            self.0.lock().expect("recording lock").push(event.clone());
+        }
+    }
+
+    #[test]
+    fn logger_uses_lifecycle_context_and_redacts_events() {
+        let sink = RecordingSink::default();
+        let context = CallLifecycleContext::new("messages", "claude", "anthropic", "req_123");
+        let logger = CallLogger::new(&context, Some(Arc::new(sink.clone())));
+        logger.request_about_to_be_sent(
+            "claude".to_string(),
+            false,
+            "https://example.test/v1/messages".to_string(),
+            vec![("authorization".to_string(), "Bearer secret".to_string())],
+            serde_json::json!({"token": "secret", "prompt": "visible"}),
+        );
+
+        let events = sink.0.lock().expect("recording lock");
+        let serialized = serde_json::to_string(&events[0]).expect("event serializes");
+        assert!(serialized.contains("\"call_id\":\"req_123\""));
+        assert!(!serialized.contains("secret"));
+        assert!(serialized.contains("visible"));
+    }
+}

litellm-rust/crates/core/src/logging/redaction.rs

@@ -0,0 +1,119 @@
+use std::collections::BTreeMap;
+
+use serde_json::{Map, Value};
+
+pub const PROVIDER_DEBUG_BODY_MAX_BYTES: usize = 64 * 1024;
+
+pub struct BodySnapshot {
+    pub body: Value,
+    pub body_truncated: Option<bool>,
+    pub body_original_bytes: Option<usize>,
+}
+
+pub fn snapshot_json(value: Value) -> BodySnapshot {
+    let redacted = redact_value(value);
+    let serialized = serde_json::to_vec(&redacted).unwrap_or_default();
+    if serialized.len() <= PROVIDER_DEBUG_BODY_MAX_BYTES {
+        return BodySnapshot {
+            body: redacted,
+            body_truncated: None,
+            body_original_bytes: None,
+        };
+    }
+    BodySnapshot {
+        body: Value::String(
+            String::from_utf8_lossy(&serialized[..PROVIDER_DEBUG_BODY_MAX_BYTES]).into_owned(),
+        ),
+        body_truncated: Some(true),
+        body_original_bytes: Some(serialized.len()),
+    }
+}
+
+pub fn redact_headers(headers: &[(String, String)]) -> BTreeMap<String, String> {
+    headers
+        .iter()
+        .map(|(name, value)| {
+            let value = if matches!(
+                name.to_ascii_lowercase().as_str(),
+                "authorization"
+                    | "proxy-authorization"
+                    | "x-api-key"
+                    | "api-key"
+                    | "x-amz-security-token"
+                    | "cookie"
+                    | "set-cookie"
+            ) {
+                "[REDACTED]".to_string()
+            } else {
+                value.clone()
+            };
+            (name.clone(), value)
+        })
+        .collect()
+}
+
+pub fn redact_url(url: &str) -> String {
+    let Ok(mut parsed) = url::Url::parse(url) else {
+        return url.to_string();
+    };
+    let Some(_) = parsed.query() else {
+        return parsed.to_string();
+    };
+    let pairs = parsed
+        .query_pairs()
+        .map(|(key, value)| {
+            let value = if matches!(
+                key.to_ascii_lowercase().as_str(),
+                "x-amz-signature"
+                    | "x-amz-credential"
+                    | "x-amz-security-token"
+                    | "api-key"
+                    | "key"
+                    | "access_token"
+                    | "signature"
+            ) {
+                "[REDACTED]"
+            } else {
+                value.as_ref()
+            };
+            (key.into_owned(), value.to_string())
+        })
+        .collect::<Vec<_>>();
+    parsed.query_pairs_mut().clear().extend_pairs(pairs);
+    parsed.to_string()
+}
+
+fn redact_value(value: Value) -> Value {
+    match value {
+        Value::Object(map) => Value::Object(
+            map.into_iter()
+                .map(|(key, value)| {
+                    if is_secret_key(&key) {
+                        (key, Value::String("[REDACTED]".to_string()))
+                    } else {
+                        (key, redact_value(value))
+                    }
+                })
+                .collect::<Map<_, _>>(),
+        ),
+        Value::Array(values) => Value::Array(values.into_iter().map(redact_value).collect()),
+        other => other,
+    }
+}
+
+fn is_secret_key(key: &str) -> bool {
+    matches!(
+        key.to_ascii_lowercase().as_str(),
+        "api_key"
+            | "apikey"
+            | "secret"
+            | "password"
+            | "token"
+            | "access_token"
+            | "client_secret"
+            | "aws_secret_access_key"
+            | "aws_access_key_id"
+            | "aws_session_token"
+            | "x-amz-security-token"
+    )
+}

litellm-rust/crates/core/src/logging/stream.rs

@@ -0,0 +1,51 @@
+use std::sync::Arc;
+
+use bytes::Bytes;
+use futures_util::Stream;
+use futures_util::StreamExt;
+
+use super::CallLogger;
+
+pub fn count_forwarded_stream<S, E>(
+    stream: S,
+    logger: Arc<CallLogger>,
+) -> impl Stream<Item = Result<Bytes, E>>
+where
+    S: Stream<Item = Result<Bytes, E>>,
+{
+    futures_util::stream::unfold(
+        (Box::pin(stream), logger, false, false),
+        |(mut stream, logger, trailing_newline, failed)| async move {
+            match stream.next().await {
+                None => {
+                    if !failed {
+                        logger.stream_finished();
+                    }
+                    None
+                }
+                Some(Ok(bytes)) => {
+                    let events = bytes
+                        .iter()
+                        .enumerate()
+                        .filter(|(index, byte)| {
+                            **byte == b'\n'
+                                && ((*index > 0 && bytes[*index - 1] == b'\n') || trailing_newline)
+                        })
+                        .count();
+                    let trailing_newline = bytes.last().copied() == Some(b'\n');
+                    logger.stream_chunk_observed(bytes.len(), events);
+                    Some((Ok(bytes), (stream, logger, trailing_newline, failed)))
+                }
+                Some(Err(error)) => {
+                    logger.failure(
+                        None,
+                        "stream_error",
+                        "provider stream failed".to_string(),
+                        None,
+                    );
+                    Some((Err(error), (stream, logger, trailing_newline, true)))
+                }
+            }
+        },
+    )
+}

litellm-rust/crates/core/src/messages/client.rs

@@ -3,7 +3,7 @@ use std::time::Duration;
 
 use crate::constants::{MESSAGES_CONNECT_TIMEOUT_SECS, MESSAGES_TIMEOUT_SECS};
 
-pub(super) fn http_client() -> &'static reqwest::Client {
+pub(crate) fn http_client() -> &'static reqwest::Client {
     static CLIENT: OnceLock<reqwest::Client> = OnceLock::new();
     CLIENT.get_or_init(|| {
         reqwest::Client::builder()

litellm-rust/crates/core/src/messages/common_utils.rs

@@ -7,7 +7,7 @@ use crate::providers::azure_ai::messages::transformation::AZURE_ANTHROPIC_MESSAG
 
 use super::transformation::AnthropicMessagesProviderConfig;
 
-pub(super) fn truncate_error_body(body: &str) -> String {
+pub(crate) fn truncate_error_body(body: &str) -> String {
     if body.chars().count() <= MESSAGES_ERROR_BODY_MAX_CHARS {
         return body.to_string();
     }

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

@@ -1,76 +1,51 @@
 use crate::constants::ANTHROPIC_MESSAGES_PROVIDER;
 use crate::error::{CoreError, CoreResult};
+use crate::logging::http::{JsonRequest, execute_json, execute_stream};
 
 use super::client::http_client;
-use super::common_utils::truncate_error_body;
 use super::types::{AnthropicMessagesResponse, ProviderMessagesRequest};
 
 pub(super) async fn execute_messages_provider_call(
     request: ProviderMessagesRequest,
 ) -> CoreResult<AnthropicMessagesResponse> {
-    let mut request_builder = http_client().post(&request.url).json(&request.body);
-    for (key, value) in &request.upstream_headers {
-        request_builder = request_builder.header(key, value);
-    }
-    if let Some(duration) = request.timeout {
-        request_builder = request_builder.timeout(duration);
-    }
-
-    let response = request_builder
-        .send()
-        .await
-        .map_err(|err| CoreError::Network(err.to_string()))?;
-
-    let status = response.status();
-    let text = response
-        .text()
-        .await
-        .map_err(|err| CoreError::Network(err.to_string()))?;
-
-    if !status.is_success() {
-        return Err(CoreError::Http {
-            status: status.as_u16(),
-            body: truncate_error_body(&text),
-        });
-    }
-
-    let response = serde_json::from_str(&text).map_err(|err| {
-        CoreError::InvalidResponse(format!("invalid messages response JSON: {err}"))
-    })?;
+    let response = execute_json::<AnthropicMessagesResponse>(
+        http_client(),
+        JsonRequest {
+            logger: request.logger,
+            model: request.model.clone(),
+            stream: false,
+            url: request.url,
+            headers: request.upstream_headers,
+            body: request.body,
+            timeout: request.timeout,
+        },
+    )
+    .await?;
     request.config.transform_response(&request.model, response)
 }
 
 pub(super) async fn execute_messages_provider_stream(
     request: ProviderMessagesRequest,
-) -> CoreResult<reqwest::Response> {
+) -> CoreResult<super::types::MessagesStreamResponse> {
     if request.provider != ANTHROPIC_MESSAGES_PROVIDER {
         return Err(CoreError::InvalidRequest(
             "streaming messages is not supported for this provider".to_string(),
         ));
     }
 
-    let mut request_builder = http_client().post(&request.url).json(&request.body);
-    for (key, value) in &request.upstream_headers {
-        request_builder = request_builder.header(key, value);
-    }
-    if let Some(duration) = request.timeout {
-        request_builder = request_builder.timeout(duration);
-    }
-
-    let response = request_builder
-        .send()
-        .await
-        .map_err(|err| CoreError::Network(err.to_string()))?;
-    let status = response.status();
-    if !status.is_success() {
-        let text = response
-            .text()
-            .await
-            .map_err(|err| CoreError::Network(err.to_string()))?;
-        return Err(CoreError::Http {
-            status: status.as_u16(),
-            body: truncate_error_body(&text),
-        });
-    }
-    Ok(response)
+    let logger = request.logger.clone();
+    let response = execute_stream(
+        http_client(),
+        JsonRequest {
+            logger: logger.clone(),
+            model: request.model,
+            stream: true,
+            url: request.url,
+            headers: request.upstream_headers,
+            body: request.body,
+            timeout: request.timeout,
+        },
+    )
+    .await?;
+    Ok(super::types::MessagesStreamResponse { response, logger })
 }

litellm-rust/crates/core/src/messages/mod.rs

@@ -8,7 +8,7 @@
 //! can splice the event stream to its own caller.
 
 mod client;
-mod common_utils;
+pub(crate) mod common_utils;
 mod handler;
 mod prepare;
 pub mod transformation;
@@ -18,13 +18,13 @@ use crate::error::CoreResult;
 
 use handler::{execute_messages_provider_call, execute_messages_provider_stream};
 use prepare::prepare_messages_call;
-use types::{AnthropicMessagesResponse, MessagesRequest};
+use types::{AnthropicMessagesResponse, MessagesRequest, MessagesStreamResponse};
 
 pub async fn messages(request: MessagesRequest<'_>) -> CoreResult<AnthropicMessagesResponse> {
     execute_messages_provider_call(prepare_messages_call(request)?).await
 }
 
-pub async fn messages_stream(request: MessagesRequest<'_>) -> CoreResult<reqwest::Response> {
+pub async fn messages_stream(request: MessagesRequest<'_>) -> CoreResult<MessagesStreamResponse> {
     execute_messages_provider_stream(prepare_messages_call(request)?).await
 }
 

litellm-rust/crates/core/src/messages/prepare.rs

@@ -1,4 +1,8 @@
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use crate::call_lifecycle::CallLifecycleContext;
 use crate::error::{CoreError, CoreResult};
+use crate::logging::CallLogger;
 use crate::routing_utils::provider::{CustomLlmProvider, get_custom_llm_provider};
 
 use super::common_utils::{has_bearer_auth, has_header, messages_provider_config, string_headers};
@@ -50,6 +54,9 @@ pub(super) fn prepare_messages_call(
             headers.push((name.to_string(), value.to_string()));
         }
     }
+    if !has_header(&headers, "content-type") {
+        headers.push(("content-type".to_string(), "application/json".to_string()));
+    }
 
     let url = config.complete_url(request.api_base, &model, &env_lookup)?;
     let typed_request = serde_json::from_value(request.body).map_err(|err| {
@@ -62,6 +69,17 @@ pub(super) fn prepare_messages_call(
         ))
     })?;
 
+    let call_id = request
+        .litellm_call_id
+        .map(str::to_owned)
+        .unwrap_or_else(|| {
+            let nanos = SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .map_or(0, |duration| duration.as_nanos());
+            format!("messages-{nanos}")
+        });
+    let context =
+        CallLifecycleContext::new("messages", model.clone(), provider.to_string(), call_id);
     Ok(ProviderMessagesRequest {
         provider: provider.to_string(),
         model,
@@ -70,5 +88,6 @@ pub(super) fn prepare_messages_call(
         body,
         upstream_headers: headers,
         timeout: request.timeout,
+        logger: std::sync::Arc::new(CallLogger::new(&context, request.logging_sink)),
     })
 }

litellm-rust/crates/core/src/messages/tests.rs

@@ -150,6 +150,8 @@ async fn messages_round_trip_builds_azure_request_and_passes_response_through()
         custom_llm_provider: Some("azure_ai"),
         extra_headers: None,
         timeout: Some(Duration::from_secs(5)),
+        litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect("messages request succeeds");
@@ -206,6 +208,8 @@ async fn messages_round_trip_builds_native_anthropic_request() {
         custom_llm_provider: Some("anthropic"),
         extra_headers: None,
         timeout: Some(Duration::from_secs(5)),
+        litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect("messages request succeeds");
@@ -259,6 +263,8 @@ async fn messages_does_not_duplicate_auth_when_x_api_key_supplied() {
         custom_llm_provider: Some("azure_ai"),
         extra_headers: Some(headers),
         timeout: Some(Duration::from_secs(5)),
+        litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect("messages request succeeds");
@@ -313,6 +319,8 @@ async fn messages_forwards_entra_id_bearer_without_requiring_api_key() {
         custom_llm_provider: Some("azure_ai"),
         extra_headers: Some(headers),
         timeout: Some(Duration::from_secs(5)),
+        litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect("entra id request succeeds without api key");
@@ -337,6 +345,8 @@ async fn messages_requires_auth_when_no_key_and_no_header() {
         custom_llm_provider: Some("azure_ai"),
         extra_headers: None,
         timeout: Some(Duration::from_millis(50)),
+        litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect_err("missing auth errors");
@@ -375,6 +385,8 @@ async fn messages_ignores_malformed_authorization_and_uses_api_key() {
         custom_llm_provider: Some("azure_ai"),
         extra_headers: Some(headers),
         timeout: Some(Duration::from_secs(5)),
+        litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect("falls back to api key");
@@ -416,6 +428,8 @@ async fn messages_maps_provider_error_status_to_http_error() {
         custom_llm_provider: Some("azure_ai"),
         extra_headers: None,
         timeout: Some(Duration::from_secs(5)),
+        litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect_err("provider error propagates");
@@ -433,6 +447,8 @@ async fn messages_rejects_unsupported_provider() {
         custom_llm_provider: Some("openai"),
         extra_headers: None,
         timeout: Some(Duration::from_millis(50)),
+        litellm_call_id: None,
+        logging_sink: None,
     })
     .await
     .expect_err("unsupported provider errors");

litellm-rust/crates/core/src/messages/types.rs

@@ -1,5 +1,7 @@
+use std::sync::Arc;
 use std::time::Duration;
 
+use crate::logging::{CallLogger, LogSink};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
 
@@ -13,6 +15,8 @@ pub struct MessagesRequest<'a> {
     pub custom_llm_provider: Option<&'a str>,
     pub extra_headers: Option<Map<String, Value>>,
     pub timeout: Option<Duration>,
+    pub litellm_call_id: Option<&'a str>,
+    pub logging_sink: Option<Arc<dyn LogSink>>,
 }
 
 pub(super) struct ProviderMessagesRequest {
@@ -23,6 +27,12 @@ pub(super) struct ProviderMessagesRequest {
     pub(super) body: Value,
     pub(super) upstream_headers: Vec<(String, String)>,
     pub(super) timeout: Option<Duration>,
+    pub(super) logger: Arc<CallLogger>,
+}
+
+pub struct MessagesStreamResponse {
+    pub response: reqwest::Response,
+    pub logger: Arc<CallLogger>,
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]

litellm-rust/crates/python-bridge/src/lib.rs

@@ -15,6 +15,7 @@ use pyo3::types::{PyAny, PyDict};
 use serde_json::{Map, Value};
 
 mod gil;
+mod logging;
 
 type MarshaledOcrInputs = (
     Value,
@@ -204,6 +205,7 @@ fn ocr(
             guardrails: Vec::new(),
             request_metadata: Default::default(),
             litellm_call_id: None,
+            logging_sink: None,
         }))
     });
 
@@ -249,6 +251,7 @@ fn aocr(
             guardrails: Vec::new(),
             request_metadata: Default::default(),
             litellm_call_id: None,
+            logging_sink: None,
         })
         .await
         .map_err(core_error_to_pyerr)?;
@@ -364,7 +367,7 @@ fn marshal_messages_inputs(
 }
 
 #[pyfunction]
-#[pyo3(signature = (model, body, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None))]
+#[pyo3(signature = (model, body, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None, debug=false))]
 #[allow(clippy::too_many_arguments)]
 fn messages(
     py: Python<'_>,
@@ -375,6 +378,7 @@ fn messages(
     custom_llm_provider: Option<String>,
     extra_headers: Option<Py<PyAny>>,
     timeout_seconds: Option<f64>,
+    debug: bool,
 ) -> PyResult<Py<PyAny>> {
     let (body, extra_headers, timeout) =
         marshal_messages_inputs(py, body, extra_headers, timeout_seconds)?;
@@ -388,6 +392,8 @@ fn messages(
             custom_llm_provider: custom_llm_provider.as_deref(),
             extra_headers,
             timeout,
+            litellm_call_id: None,
+            logging_sink: logging::hook(debug),
         }))
     });
 
@@ -398,7 +404,7 @@ fn messages(
 }
 
 #[pyfunction]
-#[pyo3(signature = (model, body, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None))]
+#[pyo3(signature = (model, body, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None, debug=false))]
 #[allow(clippy::too_many_arguments)]
 fn amessages(
     py: Python<'_>,
@@ -409,6 +415,7 @@ fn amessages(
     custom_llm_provider: Option<String>,
     extra_headers: Option<Py<PyAny>>,
     timeout_seconds: Option<f64>,
+    debug: bool,
 ) -> PyResult<Bound<'_, PyAny>> {
     let (body, extra_headers, timeout) =
         marshal_messages_inputs(py, body, extra_headers, timeout_seconds)?;
@@ -422,6 +429,8 @@ fn amessages(
             custom_llm_provider: custom_llm_provider.as_deref(),
             extra_headers,
             timeout,
+            litellm_call_id: None,
+            logging_sink: logging::hook(debug),
         })
         .await
         .map_err(core_error_to_pyerr)?;

litellm-rust/crates/python-bridge/src/logging.rs

@@ -0,0 +1,8 @@
+use std::sync::Arc;
+
+use litellm_ai_gateway::integrations::logging::console::hook as console_hook;
+use litellm_core::logging::LogSink;
+
+pub fn hook(enabled: bool) -> Option<Arc<dyn LogSink>> {
+    console_hook(enabled)
+}

litellm/rust_bridge/messages.py

@@ -6,6 +6,8 @@ from dataclasses import dataclass
 from typing import Awaitable, Final, Protocol, Union, cast
 
 import httpx
+import inspect
+from litellm._logging import _is_debugging_on
 
 from litellm.rust_bridge.timeouts import timeout_to_seconds
 
@@ -20,6 +22,7 @@ class RustMessages(Protocol):
         custom_llm_provider: str | None,
         extra_headers: dict[str, object] | None,
         timeout_seconds: float | None,
+        debug: bool,
     ) -> dict[str, object]:
         raise NotImplementedError
 
@@ -34,6 +37,7 @@ class RustAmessages(Protocol):
         custom_llm_provider: str | None,
         extra_headers: dict[str, object] | None,
         timeout_seconds: float | None,
+        debug: bool,
     ) -> Awaitable[dict[str, object]]:
         raise NotImplementedError
 
@@ -100,6 +104,17 @@ def messages(
     rust_messages = load_rust_messages()
     if rust_messages is None:
         return None
+    if "debug" in inspect.signature(rust_messages).parameters:
+        return rust_messages(
+            model=model,
+            body=body,
+            api_key=api_key,
+            api_base=api_base,
+            custom_llm_provider=custom_llm_provider,
+            extra_headers=extra_headers,
+            timeout_seconds=timeout_to_seconds(timeout),
+            debug=_is_debugging_on(),
+        )
     return rust_messages(
         model=model,
         body=body,
@@ -124,6 +139,17 @@ async def amessages(
     rust_amessages = load_rust_amessages()
     if rust_amessages is None:
         return None
+    if "debug" in inspect.signature(rust_amessages).parameters:
+        return await rust_amessages(
+            model=model,
+            body=body,
+            api_key=api_key,
+            api_base=api_base,
+            custom_llm_provider=custom_llm_provider,
+            extra_headers=extra_headers,
+            timeout_seconds=timeout_to_seconds(timeout),
+            debug=_is_debugging_on(),
+        )
     return await rust_amessages(
         model=model,
         body=body,