fix(messages): log native messages request fields like python (#50421)

Python logs the requested optional params, taken before the provider transform, as model_parameters. The native handler logged the transformed body, so a reasoning_effort request was logged as thinking and output_config. The request context now carries the requested params

Python updates model_call_details with the transformed request body before pre_call, so the logged messages are what was sent. The native adapter kept the caller's messages and compared params as JSON values, so an int 0 logged as 0.0

Python only records custom_llm_provider in model_call_details. The native adapter also wrote it into litellm_params, and the proxy reuses one logger across router hops, so a fallback hop's pre_call log showed the provider where Python shows null

Python only stamps cache_hit when a result came from the cache. The native adapter wrote False on every provider result

Python calls pre_call with the request body as a json user message, and a stream's spend-log cache_key is computed over that input. The native adapter logged the string Messages, so a streamed call's cache_key differed
This commit is contained in:
Nate Armstrong 2026-11-09 16:24:19 -08:00 • committed by GitHub
parent 1ddc913002
commit d631f3cf05
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 356 additions and 26 deletions

View file

@ -27,6 +27,10 @@
"api_key",
"additional_args"
],
"messages_input": [
"body",
"kwargs"
],
"post_call": [
"logger",
"original_response",

View file

@ -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]
",
);

View file

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

View file

@ -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")]

View file

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

View file

@ -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),
};

View file

@ -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()
)
);
}

View file

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

View file

@ -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):

View file

@ -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",
}
),
}
]

View file

@ -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",