diff --git a/litellm-rust/crates/callbacks-legacy-python/python_contract.json b/litellm-rust/crates/callbacks-legacy-python/python_contract.json
index 1d4a2f7f7c2..c0fe1c7b52f 100644
--- a/litellm-rust/crates/callbacks-legacy-python/python_contract.json
+++ b/litellm-rust/crates/callbacks-legacy-python/python_contract.json
@@ -27,6 +27,10 @@
     "api_key",
     "additional_args"
   ],
+  "messages_input": [
+    "body",
+    "kwargs"
+  ],
   "post_call": [
     "logger",
     "original_response",
diff --git a/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs b/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs
index 8d1946861aa..28bd5f2cefb 100644
--- a/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs
+++ b/litellm-rust/crates/callbacks-legacy-python/src/adapter.rs
@@ -14,7 +14,7 @@ use pyo3::{
     exceptions::{PyBaseException, PyException},
     gc::{PyTraverseError, PyVisit},
     prelude::*,
-    types::{PyDateTime, PyDict, PyList},
+    types::{PyDateTime, PyDict, PyList, PyString},
 };
 use serde_json::Value;
 
@@ -22,7 +22,7 @@ use crate::{
     DeploymentHooks, LegacyCallbacks, PublicCall, PythonLogger,
     deferred::{PendingLogging, PendingSuccess},
     finalize, is_internal_call,
-    python::Streaming,
+    python::{Logging, Streaming},
     setup,
 };
 
@@ -137,6 +137,40 @@ impl LegacyLogging {
         Some(params)
     }
 
+    // Compared as Python compares them, so a caller's `0` is kept over a typed float's `0.0`.
+    fn caller_value<'py>(
+        &self,
+        py: Python<'py>,
+        name: &str,
+        sent: &Value,
+    ) -> PyResult<Option<Bound<'py, PyAny>>> {
+        match self.call.lookup(py, name)? {
+            Some(value) if from_py::<Value>(&value).is_ok() && value.eq(to_py(py, sent)?)? => {
+                Ok(Some(value))
+            }
+            _ => Ok(None),
+        }
+    }
+
+    // Python's Messages handler logs the caller's own optional params, not their typed copies.
+    fn as_called(&self, py: Python<'_>, context: &RequestContext) -> PyResult<RequestContext> {
+        let mut context = context.clone();
+        if self.operation != LoggingOperation::Messages {
+            return Ok(context);
+        }
+        for (name, sent) in context
+            .optional_params
+            .as_object_mut()
+            .into_iter()
+            .flatten()
+        {
+            if let Some(value) = self.caller_value(py, name, sent)? {
+                *sent = from_py(&value)?;
+            }
+        }
+        Ok(context)
+    }
+
     /// Python times the provider exchange inside the async `AsyncHTTPHandler.post`; its sync
     /// client is untimed.
     fn close_api_window(&mut self, py: Python<'_>, end: f64) -> PyResult<()> {
@@ -158,13 +192,20 @@ impl LegacyLogging {
         }
     }
 
-    fn input_description(&self) -> &'static str {
-        match self.operation {
+    fn logged_input<'py>(
+        &self,
+        py: Python<'py>,
+        body: &Bound<'py, PyDict>,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let description = match self.operation {
             LoggingOperation::Completion => "Chat completions",
             LoggingOperation::Responses => "Responses",
-            LoggingOperation::Messages => "Messages",
+            LoggingOperation::Messages => {
+                return Logging::MessagesInput.call(py, (body, self.call.kwargs().bind(py)));
+            }
             LoggingOperation::Ocr => "OCR document processing",
-        }
+        };
+        Ok(PyString::new(py, description).into_any())
     }
 
     /// The key `pre_call` logs. Python's Messages handler logs an empty string, never the key.
@@ -456,6 +497,7 @@ impl LegacyLogging {
         wire: Box<WireRequest>,
         context: &RequestContext,
     ) -> PyResult<HookStep<Self, Box<WireRequest>>> {
+        let context = &self.as_called(py, context)?;
         let logger = self.logger()?;
         logger.update_from_kwargs(
             py,
@@ -468,12 +510,17 @@ impl LegacyLogging {
             .into_bound(py)
             .cast_into::<PyDict>()?;
         for (name, sent) in wire.body.as_object().into_iter().flatten() {
-            if let Some(value) = self.call.lookup(py, name)?
-                && from_py::<Value>(&value).is_ok_and(|caller| caller == *sent)
-            {
+            if let Some(value) = self.caller_value(py, name, sent)? {
                 body.set_item(name, value)?;
             }
         }
+        if self.operation == LoggingOperation::Messages {
+            // Python's Messages handler merges its transformed request body into the call details.
+            logger
+                .object(py)
+                .getattr("model_call_details")?
+                .call_method1("update", (&body,))?;
+        }
         let headers = PyDict::new(py);
         for (name, value) in &wire.headers {
             headers.set_item(name, value)?;
@@ -483,9 +530,10 @@ impl LegacyLogging {
             headers: headers.clone().unbind(),
             context: context.clone(),
         });
+        let input = self.logged_input(py, &body)?;
         self.logger()?.pre_call(
             py,
-            self.input_description(),
+            &input,
             self.logged_api_key(context),
             &body,
             &headers,
@@ -539,7 +587,11 @@ impl LegacyLogging {
             .getattr("litellm_params")?
             .cast_into::<PyDict>()?
             .copy()?;
-        params.set_item("custom_llm_provider", &facts.provider.provider)?;
+        if self.operation != LoggingOperation::Messages {
+            // Python's Messages handler hands the provider to the logger beside its params, never
+            // in them, and the proxy's logger carries over to the next fallback hop.
+            params.set_item("custom_llm_provider", &facts.provider.provider)?;
+        }
         crate::python::Logging::Update.call(
             py,
             (
@@ -562,8 +614,17 @@ impl LegacyLogging {
             ResultSource::Provider => None,
             ResultSource::Cache { key } => Some(key.clone()),
         };
-        details.set_item("cache_hit", self.cache_key.is_some())?;
+        if self.cache_key.is_some() {
+            details.set_item("cache_hit", true)?;
+        } else {
+            details.call_method1("pop", ("cache_hit", py.None()))?;
+        }
         details.set_item("cache_key", self.cache_key.as_deref())?;
+        if let (LoggingOperation::Messages, ResultSource::Provider, Some(request)) =
+            (self.operation, &facts.source, &self.request)
+        {
+            details.call_method1("update", (request.body.bind(py),))?;
+        }
         Ok(HookStep::Ready(()))
     }
 
@@ -1128,10 +1189,10 @@ logger._provider_reported_cost = None
             };
             let previous = litellm_host::interceptors::ExecutionFacts {
                 provider: facts.provider.clone(),
-                cost: litellm_host::interceptors::Cost::Reported {
-                    amount: serde_json::Number::from_f64(9.0).unwrap(),
+                cost: litellm_host::interceptors::Cost::CacheHit,
+                source: litellm_host::interceptors::ResultSource::Cache {
+                    key: "earlier-hit".into(),
                 },
-                source: litellm_host::interceptors::ResultSource::Provider,
             };
             logging.result_ready(py, &previous).unwrap();
             local(&locals, "logger")
@@ -1149,13 +1210,59 @@ logger._provider_reported_cost = None
                 c"
 assert logger.model_call_details.get('response_cost') is None, logger.model_call_details
 assert logger._provider_reported_cost == reported, logger._provider_reported_cost
-assert logger.model_call_details['cache_hit'] is cache_hit, logger.model_call_details
+assert logger.model_call_details.get('cache_hit') is (True if cache_hit else None), logger.model_call_details
 assert logger.update['custom_llm_provider'] == 'edenai', logger.update
 ",
             );
         });
     }
 
+    #[rstest]
+    #[case::messages(crate::LoggingOperation::Messages, false)]
+    #[case::ocr(crate::LoggingOperation::Ocr, true)]
+    fn result_ready_puts_the_provider_in_the_logged_params_except_for_messages(
+        #[case] operation: crate::LoggingOperation,
+        #[case] in_params: bool,
+    ) {
+        Python::initialize();
+        Python::attach(|py| {
+            let locals = namespace(py, PAYLOAD_LOGGER);
+            run(
+                py,
+                &locals,
+                c"
+logger.litellm_params = {}
+logger.optional_params = {}
+logger.model_call_details = {}
+logger._provider_reported_cost = None
+",
+            );
+            let mut logging = LegacyLogging {
+                logger: Some(PythonLogger::new(local(&locals, "logger").unbind())),
+                operation,
+                ..legacy_call(py, &locals, false)
+            };
+            let facts = litellm_host::interceptors::ExecutionFacts {
+                provider: litellm_host::interceptors::ProviderIdentity {
+                    model: "native-test".into(),
+                    provider: "anthropic".into(),
+                },
+                cost: litellm_host::interceptors::Cost::Deferred,
+                source: litellm_host::interceptors::ResultSource::Provider,
+            };
+            logging.result_ready(py, &facts).unwrap();
+            locals.set_item("in_params", in_params).unwrap();
+            run(
+                py,
+                &locals,
+                c"
+assert logger.update['custom_llm_provider'] == 'anthropic', logger.update
+assert ('custom_llm_provider' in logger.update['litellm_params']) is in_params, logger.update
+",
+            );
+        });
+    }
+
     const DOCUMENT: &str = "data:application/pdf;base64,YWJj";
     const EDITED: &str = "data:application/pdf;base64,ZWRpdGVk";
 
@@ -1290,7 +1397,10 @@ assert logger.update['custom_llm_provider'] == 'edenai', logger.update
     #[rstest]
     #[case::completion(crate::LoggingOperation::Completion, "Chat completions")]
     #[case::responses(crate::LoggingOperation::Responses, "Responses")]
-    #[case::messages(crate::LoggingOperation::Messages, "Messages")]
+    #[case::messages(
+        crate::LoggingOperation::Messages,
+        r#"[{'role': 'user', 'content': '{"pages": [1]}'}]"#
+    )]
     #[case::ocr(crate::LoggingOperation::Ocr, "OCR document processing")]
     fn prepared_arguments_replace_the_legacy_view_without_losing_callback_aliases(
         #[case] operation: crate::LoggingOperation,
@@ -1331,7 +1441,7 @@ prepared = {'pages': replacement}
                 &locals,
                 c"
 assert logger.pre['complete_input_dict']['pages'] is replacement
-assert logger.pre_input == description
+assert str(logger.pre_input) == description
 assert original == [0]
 ",
             );
diff --git a/litellm-rust/crates/callbacks-legacy-python/src/callbacks.rs b/litellm-rust/crates/callbacks-legacy-python/src/callbacks.rs
index a428afa9ece..093e86c43dc 100644
--- a/litellm-rust/crates/callbacks-legacy-python/src/callbacks.rs
+++ b/litellm-rust/crates/callbacks-legacy-python/src/callbacks.rs
@@ -26,7 +26,7 @@ pub trait LegacyCallbacks {
     fn pre_call(
         &self,
         py: Python<'_>,
-        input: &str,
+        input: &Bound<'_, PyAny>,
         api_key: Option<&str>,
         body: &Bound<'_, PyDict>,
         headers: &Bound<'_, PyDict>,
@@ -157,7 +157,7 @@ impl LegacyCallbacks for PythonLogger {
     fn pre_call(
         &self,
         py: Python<'_>,
-        input: &str,
+        input: &Bound<'_, PyAny>,
         api_key: Option<&str>,
         body: &Bound<'_, PyDict>,
         headers: &Bound<'_, PyDict>,
diff --git a/litellm-rust/crates/callbacks-legacy-python/src/python.rs b/litellm-rust/crates/callbacks-legacy-python/src/python.rs
index 32b11988703..5b6faa3dbc2 100644
--- a/litellm-rust/crates/callbacks-legacy-python/src/python.rs
+++ b/litellm-rust/crates/callbacks-legacy-python/src/python.rs
@@ -42,6 +42,8 @@ pub(crate) enum Logging {
     Update,
     #[strum(serialize = "pre_call")]
     PreCall,
+    #[strum(serialize = "messages_input")]
+    MessagesInput,
     #[strum(serialize = "post_call")]
     PostCall,
     #[strum(serialize = "record_response")]
diff --git a/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs b/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs
index 9e69d0c30e2..c88a6496423 100644
--- a/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs
+++ b/litellm-rust/crates/callbacks-legacy-python/src/test_support.rs
@@ -62,6 +62,7 @@ FAKES = {
         custom_llm_provider=provider,
     ),
     'pre_call': lambda logger, input, api_key, additional_args: logger.pre_call(input, api_key, additional_args),
+    'messages_input': lambda body, kwargs: [{'role': 'user', 'content': json.dumps(body)}],
     'post_call': lambda logger, original_response, api_key, additional_args: logger.post_call(
         original_response, api_key, additional_args
     ),
@@ -134,6 +135,7 @@ class StubCoroutine:
 class StubLogger:
     def __init__(self):
         self.calls = []
+        self.model_call_details = {}
         self.hooks = {}
         self.on_enqueue = lambda coroutine: None
 
diff --git a/litellm-rust/crates/inference-messages/src/handler.rs b/litellm-rust/crates/inference-messages/src/handler.rs
index e0a7aa5adb8..5ab171076ff 100644
--- a/litellm-rust/crates/inference-messages/src/handler.rs
+++ b/litellm-rust/crates/inference-messages/src/handler.rs
@@ -131,6 +131,7 @@ async fn provider_call(
     let transform_context =
         MessagesTransformContext::new(shaping.capabilities, shaping.settings.drop_params);
     let filtered = requested_optional_params(sanitized, provider, &transform_context)?;
+    let requested_params = serde_json::to_value(&filtered.params).map_err(serialize_failure)?;
     let trimmed =
         without_additional_drop_params(filtered, &shaping.settings.additional_drop_params)?;
 
@@ -173,7 +174,7 @@ async fn provider_call(
     let request_context = RequestContext {
         model: body.model.clone(),
         custom_llm_provider: provider.as_str().to_string(),
-        optional_params: serde_json::to_value(&body.params).map_err(serialize_failure)?,
+        optional_params: requested_params,
         secret_fields: Vec::new(),
         api_key: api_key.map(SecretValue::new),
     };
diff --git a/litellm-rust/crates/inference-messages/tests/messages/host.rs b/litellm-rust/crates/inference-messages/tests/messages/host.rs
index c3d06027c32..e8df20e7dcb 100644
--- a/litellm-rust/crates/inference-messages/tests/messages/host.rs
+++ b/litellm-rust/crates/inference-messages/tests/messages/host.rs
@@ -8,7 +8,9 @@ use litellm_host::{
 };
 use litellm_inference_messages::{MessagesCallResponse, route::Messages};
 use litellm_inference_testing::{RecordingSecrets, no_secrets};
-use litellm_llms::base_llm::messages::context::MessagesModelCapabilities as AnthropicModelCapabilities;
+use litellm_llms::base_llm::messages::context::{
+    MessagesModelCapabilities as AnthropicModelCapabilities, SupportedEffortTiers,
+};
 use rstest::rstest;
 
 use super::*;
@@ -419,7 +421,7 @@ async fn no_raw_response_is_emitted_for_a_stream_or_a_failure(
     assert!(host.raw_responses().is_empty());
 }
 
-/// Python logs `optional_params` as what it is about to send, so a dropped param must
+/// Python logs `optional_params` after its requested-param gate, so a dropped param must
 /// not resurface in callbacks.
 #[rstest]
 #[tokio::test]
@@ -450,3 +452,54 @@ async fn the_request_context_carries_the_shaped_params_without_model_or_messages
         .unwrap_or_else(|seen| panic!("before_provider_request runs once, saw {}", seen.len()));
     assert_eq!(optional_params, json!({"max_tokens": 16}));
 }
+
+#[rstest]
+#[case::reasoning_effort_on_an_adaptive_model(
+    AnthropicModelCapabilities {
+        supports_reasoning: true,
+        supports_adaptive_thinking: true,
+        supports_output_config: true,
+        effort_tiers: SupportedEffortTiers { low: true, ..SupportedEffortTiers::default() },
+        ..AnthropicModelCapabilities::default()
+    },
+    json!({"reasoning_effort": "low"})
+)]
+#[case::adaptive_payload_on_a_legacy_model(
+    AnthropicModelCapabilities {
+        supports_reasoning: true,
+        ..AnthropicModelCapabilities::default()
+    },
+    json!({"max_tokens": 3000, "thinking": {"type": "adaptive"}, "output_config": {"effort": "high"}})
+)]
+#[tokio::test]
+async fn the_request_context_carries_the_requested_reasoning_not_its_translation(
+    call: MessagesCall,
+    #[case] capabilities: AnthropicModelCapabilities,
+    #[case] fields: Value,
+) {
+    let upstream = upstream([message_response()]).await;
+    let host = RecordingHost::passthrough(authenticated(
+        MessagesCall {
+            shaping: MessagesShaping {
+                capabilities,
+                ..MessagesShaping::default()
+            },
+            ..with_fields(call, fields.clone())
+        },
+        upstream.uri(),
+    ));
+
+    run_through(&host).await.expect("messages call succeeds");
+
+    let [optional_params] = <[Value; 1]>::try_from(host.optional_params.into_inner().unwrap())
+        .unwrap_or_else(|seen| panic!("before_provider_request runs once, saw {}", seen.len()));
+    assert_eq!(
+        optional_params,
+        Value::Object(
+            object(json!({"max_tokens": 16}))
+                .into_iter()
+                .chain(object(fields))
+                .collect()
+        )
+    );
+}
diff --git a/litellm/rust_bridge/host/callbacks_legacy_python.py b/litellm/rust_bridge/host/callbacks_legacy_python.py
index 92e5436fbb8..8c281c73842 100644
--- a/litellm/rust_bridge/host/callbacks_legacy_python.py
+++ b/litellm/rust_bridge/host/callbacks_legacy_python.py
@@ -9,6 +9,8 @@ from __future__ import annotations
 import asyncio
 import contextvars
 import datetime
+import inspect
+import json
 import traceback
 import uuid
 import weakref
@@ -241,10 +243,33 @@ def update_logging(
     )
 
 
-def pre_call(logger: LoggingSurface, input: str, api_key: str | None, additional_args: dict[str, object]) -> None:
+def pre_call(
+    logger: LoggingSurface,
+    input: str | Sequence[Mapping[str, str]],
+    api_key: str | None,
+    additional_args: dict[str, object],
+) -> None:
     logger.pre_call(input=input, api_key=api_key, additional_args=additional_args)
 
 
+def messages_input(body: Mapping[str, object], kwargs: Mapping[str, object]) -> Sequence[Mapping[str, str]]:
+    from litellm.messages import anthropic_messages_handler
+
+    # Python's Messages handler builds its body as messages, max_tokens, model, then its own parameters in
+    # signature order and the caller's other keywords in their order, and logs that body as serialized
+    order: Final = (
+        "messages",
+        "max_tokens",
+        "model",
+        *inspect.signature(anthropic_messages_handler).parameters,
+        *kwargs,
+        *body,
+    )
+    return [
+        {"role": "user", "content": json.dumps({name: body[name] for name in dict.fromkeys(order) if name in body})}
+    ]
+
+
 def post_call(
     logger: LoggingSurface, original_response: str, api_key: str | None, additional_args: dict[str, object]
 ) -> None:
diff --git a/tests/test_litellm_rust/messages/test_callbacks.py b/tests/test_litellm_rust/messages/test_callbacks.py
index a3c03898a12..3073c1cf7dd 100644
--- a/tests/test_litellm_rust/messages/test_callbacks.py
+++ b/tests/test_litellm_rust/messages/test_callbacks.py
@@ -10,14 +10,15 @@ from pydantic import TypeAdapter
 import litellm
 from litellm.integrations.argilla import ArgillaLogger
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.types.integrations.argilla import ArgillaItem
 from litellm.types.utils import ModelResponse, StandardLoggingPayload
-from tests.test_litellm_rust.messages.data import BAD_REQUEST, MESSAGES_RESPONSE, RATE_LIMITED, call
+from tests.test_litellm_rust.messages.data import BAD_REQUEST, MESSAGES_RESPONSE, RATE_LIMITED, REQUEST, call
 from tests.test_litellm_rust.support.callback_recorder import HookRecorder, RecordingLogger, request_body
 from tests.test_litellm_rust.support.isolation import rebound
 from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec, relaxed
 from tests.test_litellm_rust.support.results import response_id, with_id
-from tests.test_litellm_rust.support.routing import assert_served_natively
+from tests.test_litellm_rust.support.routing import CALL_ID, assert_served_natively
 
 pytestmark = pytest.mark.requires_rust_extension
 
@@ -110,6 +111,33 @@ async def test_pre_api_call_sees_the_wire_body(server: RecordingServer) -> None:
     assert request_body(pre_call) == server.requests[0].body
 
 
+@pytest.mark.asyncio
+async def test_pre_api_call_of_a_retried_call_keeps_the_provider_out_of_the_litellm_params(
+    server: RecordingServer,
+) -> None:
+    recorder: Final = RecordingLogger()
+    logging_obj: Final = Logging(
+        model=REQUEST["model"],
+        messages=[],
+        stream=False,
+        start_time=datetime.now(),
+        call_type="anthropic_messages",
+        litellm_call_id=CALL_ID,
+        function_id=CALL_ID,
+        dynamic_input_callbacks=[recorder],
+    )
+
+    await call(server, litellm_logging_obj=logging_obj)
+    await call(server, litellm_logging_obj=logging_obj)
+
+    pre_calls: Final = recorder.wait_for("log_pre_api_call", count=2)
+
+    assert_served_natively(server, count=2)
+    for pre_call in pre_calls:
+        assert isinstance(pre_call.kwargs, dict)
+        assert "custom_llm_provider" not in _OBJECT.validate_python(pre_call.kwargs["litellm_params"])
+
+
 @pytest.mark.asyncio
 async def test_pre_api_call_edits_reach_the_provider(server: RecordingServer) -> None:
     class Edit(CustomLogger):
@@ -140,6 +168,47 @@ async def test_pre_api_call_sees_the_shaped_optional_params(server: RecordingSer
     assert optional_params["max_tokens"] == sent["max_tokens"]
 
 
+@pytest.mark.asyncio
+async def test_success_event_logs_the_requested_reasoning_effort_not_its_translation(server: RecordingServer) -> None:
+    recorder: Final = RecordingLogger()
+
+    with rebound(litellm, "callbacks", [recorder]):
+        response: Final = await call(server, reasoning_effort="low")
+        success: Final = await recorder.wait_for_async("async_log_success_event")
+
+    assert_served_natively(server, response)
+    sent: Final = server.requests[0].body
+    assert isinstance(sent, dict)
+    assert sent["output_config"] == {"effort": "low"}
+    assert isinstance(success[0].kwargs, dict)
+    logged: Final = _OBJECT.validate_python(success[0].kwargs["standard_logging_object"])
+    assert logged["model_parameters"] == {"max_tokens": 64, "stream": False, "reasoning_effort": "low"}
+
+
+@pytest.mark.asyncio
+async def test_success_event_logs_the_sent_messages_and_the_callers_params(server: RecordingServer) -> None:
+    recorder: Final = RecordingLogger()
+    tool_use: Final = {"type": "tool_use", "id": "functions.Bash:0", "name": "bash", "input": {"command": "ls"}}
+    messages: Final = [
+        {"role": "user", "content": "List /tmp"},
+        {"role": "assistant", "content": [{"type": "text", "text": " \n"}, tool_use]},
+        {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "functions.Bash:0", "content": "a"}]},
+    ]
+
+    with rebound(litellm, "callbacks", [recorder]):
+        response: Final = await call(server, messages=messages, temperature=1)
+        success: Final = await recorder.wait_for_async("async_log_success_event")
+
+    assert_served_natively(server, response)
+    sent: Final = server.requests[0].body
+    assert isinstance(sent, dict)
+    assert sent["messages"][1]["content"] == [{**tool_use, "id": "functions_Bash_0"}]
+    assert isinstance(success[0].kwargs, dict)
+    logged: Final = _OBJECT.validate_python(success[0].kwargs["standard_logging_object"])
+    assert logged["messages"] == sent["messages"]
+    assert repr(_OBJECT.validate_python(logged["model_parameters"])["temperature"]) == "1"
+
+
 @pytest.mark.asyncio
 async def test_logging_hook_and_redaction_edits_reach_the_async_success_event(server: RecordingServer) -> None:
     class Rewrite(HookRecorder):
diff --git a/tests/test_litellm_rust/messages/test_streaming.py b/tests/test_litellm_rust/messages/test_streaming.py
index e57631f48b9..491d257b482 100644
--- a/tests/test_litellm_rust/messages/test_streaming.py
+++ b/tests/test_litellm_rust/messages/test_streaming.py
@@ -1,10 +1,13 @@
+import json
 from collections.abc import AsyncIterator, Iterator
 from typing import Final
 
 import pytest
 
+import litellm
 from tests.test_litellm_rust.messages.data import MESSAGES_EVENTS, MESSAGES_RESPONSE, call
 from tests.test_litellm_rust.support.callback_recorder import RecordingLogger, drain_logging
+from tests.test_litellm_rust.support.isolation import rebound
 from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec, relaxed
 from tests.test_litellm_rust.support.routing import assert_served_natively
 
@@ -68,3 +71,40 @@ async def test_stream_closed_early_logs_success_once(server: RecordingServer) ->
     assert len(await recorder.wait_for_async("async_log_success_event")) == 1
     with pytest.raises(StopAsyncIteration):
         await anext(stream)
+
+
+async def _streamed_pre_call_input(server: RecordingServer) -> object:
+    server.enqueue(STREAM)
+    recorder: Final = RecordingLogger()
+    with rebound(litellm, "callbacks", [recorder]):
+        stream: Final = await call(
+            server, stream=True, system="Be brief", metadata={"user_id": "user-1"}, stop_sequences=["END"]
+        )
+        assert isinstance(stream, AsyncIterator)
+        _ = [chunk async for chunk in stream]
+    pre_call: Final = recorder.wait_for("log_pre_api_call")[0].kwargs
+    assert isinstance(pre_call, dict)
+    return pre_call["input"]
+
+
+@pytest.mark.asyncio
+async def test_stream_logs_the_request_body_as_input_like_python(server: RecordingServer) -> None:
+    logged: Final = await _streamed_pre_call_input(server)
+
+    assert_served_natively(server)
+    assert logged == [
+        {
+            "role": "user",
+            "content": json.dumps(
+                {
+                    "messages": [{"role": "user", "content": "Hello"}],
+                    "max_tokens": 64,
+                    "model": "claude-sonnet-5",
+                    "metadata": {"user_id": "user-1"},
+                    "stop_sequences": ["END"],
+                    "stream": True,
+                    "system": "Be brief",
+                }
+            ),
+        }
+    ]
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 29a6b6f56a8..87b9e70612c 100644
--- a/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py
+++ b/tests/unit/rust_bridge/host/test_callbacks_legacy_python.py
@@ -263,6 +263,30 @@ def test_the_rust_contract_matches_the_shim_signatures() -> None:
     assert contract == {name: list(inspect.signature(getattr(legacy, name)).parameters) for name in contract}
 
 
+def test_messages_input_is_the_body_as_the_python_handler_builds_and_serializes_it() -> None:
+    body: Final = {
+        "model": "claude-sonnet-4-5",
+        "messages": [{"role": "user", "content": "hi"}],
+        "max_tokens": 16,
+        "system": "brief",
+        "stream": True,
+        "context_management": {"edits": []},
+        "output_config": {"effort": "low"},
+    }
+
+    logged: Final = legacy.messages_input(
+        body, {"output_config": {"effort": "low"}, "context_management": {"edits": []}}
+    )
+
+    assert logged == [
+        {
+            "role": "user",
+            "content": '{"messages": [{"role": "user", "content": "hi"}], "max_tokens": 16, "model": "claude-sonnet-4-5", '
+            '"stream": true, "system": "brief", "output_config": {"effort": "low"}, "context_management": {"edits": []}}',
+        }
+    ]
+
+
 def _messages_logger() -> Logging:
     return Logging(
         model="claude-sonnet-4-5",