fix(rust): match Python responses failures and callbacks natively

Provider failures on the native Responses route now carry the provider's
raw body as the message and the model as the caller named it, as the
Python route's exception mapping does. Invalid tool parameters decline to
Python so the caller gets Python's BadRequestError and failure log, and
native responses stamp custom_llm_provider into their hidden params. The
parity harness now also compares failures, callback hooks, hidden params
and cache fields

litellm-rust/crates/python-bridge/src/routes/responses/host.rs

@@ -102,6 +102,15 @@ fn id_scope(
     })
 }
 
+/// Python's `aresponses` stamps the resolved provider into `_hidden_params` once the response id is
+/// encoded, so callbacks can tell which provider served the call (`litellm/responses/main.py`,
+/// mirroring chat completions); a native response carries it the same way
+fn stamp_provider(response: &Bound<'_, PyAny>, provider: Option<&str>) -> PyResult<()> {
+    response
+        .getattr("_hidden_params")?
+        .set_item("custom_llm_provider", provider)
+}
+
 impl PythonBinding for ResponsesPythonHost {
     type Protocol = Responses;
     type Failure = PyErr;
@@ -132,7 +141,9 @@ impl PythonBinding for ResponsesPythonHost {
     ) -> PyResult<Py<PyAny>> {
         let mut response = response;
         ids::encode_response(&mut response, &self.ids);
-        self.host.response(py, &response)
+        let response = self.host.response(py, &response)?;
+        stamp_provider(response.bind(py), self.ids.provider.as_deref())?;
+        Ok(response)
     }
 
     fn encode_stream_head(

litellm/rust_bridge/failures.py

@@ -16,15 +16,15 @@ _UPSTREAM_HEADERS: Final = TypeAdapter(list[tuple[str, str]])
 
 
 class UpstreamFailure(Exception):
-    def __init__(self, response: httpx.Response, cause: Exception) -> None:
-        super().__init__(str(cause))
-        self.message: Final = str(cause)
+    def __init__(self, response: httpx.Response, cause: Exception, message: str | None = None) -> None:
+        super().__init__(message or str(cause))
+        self.message: Final = message or str(cause)
         self.response: Final = response
         self.status_code: Final = response.status_code
         self.__cause__ = cause
 
 
-def _upstream_failure(error: Exception, api_base: str | None) -> Exception:
+def _upstream_failure(error: Exception, api_base: str | None, body_as_message: bool) -> Exception:
     try:
         status, body = _UPSTREAM_ARGS.validate_python(error.args)
         headers: Final = _UPSTREAM_HEADERS.validate_python(getattr(error, "headers", None))
@@ -34,6 +34,7 @@ def _upstream_failure(error: Exception, api_base: str | None) -> Exception:
     return UpstreamFailure(
         httpx.Response(status, content=body.encode(), headers=headers, request=http_request),
         error,
+        body if body_as_message else None,
     )
 
 
@@ -49,13 +50,20 @@ class ExceptionMapper(Protocol):
     ) -> Exception: ...
 
 
-def map_failure(error: Exception, model: str, request_provider: str, kwargs: Mapping[str, object]) -> Exception:
+def map_failure(
+    error: Exception,
+    model: str,
+    request_provider: str,
+    kwargs: Mapping[str, object],
+    *,
+    keep_provider_prefix: bool = False,
+) -> Exception:
     mapper: Final = cast(  # cast-ok: bounded adapter for the legacy public exception mapper
         ExceptionMapper, litellm.exception_type
     )
     try:
         return mapper(
-            model=model.removeprefix(f"{request_provider}/"),
+            model=model if keep_provider_prefix else model.removeprefix(f"{request_provider}/"),
             custom_llm_provider=request_provider,
             original_exception=error,
             completion_kwargs=dict(kwargs),
@@ -67,11 +75,23 @@ def map_failure(error: Exception, model: str, request_provider: str, kwargs: Map
 
 
 def map_native_failure(
-    error: Exception, model: str, request_provider: str, kwargs: Mapping[str, object], api_base: str | None = None
+    error: Exception,
+    model: str,
+    request_provider: str,
+    kwargs: Mapping[str, object],
+    api_base: str | None = None,
+    *,
+    python_responses_shape: bool = False,
 ) -> Exception:
-    """`map_failure`, reading a native `(status, body)` provider failure as the HTTP response it was."""
-    original: Final = _upstream_failure(error, api_base)
-    public_error: Final = map_failure(original, model, request_provider, kwargs)
+    """`map_failure`, reading a native `(status, body)` provider failure as the HTTP response it was.
+
+    `python_responses_shape` matches the Python Responses route, which maps the provider's raw body
+    under the model as the caller named it
+    """
+    original: Final = _upstream_failure(error, api_base, body_as_message=python_responses_shape)
+    public_error: Final = map_failure(
+        original, model, request_provider, kwargs, keep_provider_prefix=python_responses_shape
+    )
     if isinstance(original, UpstreamFailure) and public_error.__context__ is original:
         public_error.__context__ = error
         if isinstance(public_error, openai.APIStatusError):

litellm/rust_bridge/host/routes/responses.py

@@ -36,6 +36,7 @@ def map_failure(error: Exception, request: Mapping[str, object]) -> Exception:
         provider,
         request,
         optional_str(request.get("api_base")) or optional_str(request.get("base_url")),
+        python_responses_shape=True,
     )
 
 
@@ -63,7 +64,7 @@ def _model_rules_decline_reason(request: Mapping[str, object]) -> str | None:
         model_info=request.get("model_info"),
     )
     if _repairs_tool_schemas(model, request.get("tools"), litellm_params):
-        return "native Responses does not implement tool schema repair"
+        return "native Responses does not implement tool schema validation and repair"
     if _carries_cache_control(request.get("input")) and _OPENAI_CONFIG._reads_prompt_cache_breakpoints(  # pyright: ignore[reportPrivateUsage]  # the same rules the Python route applies
         model=model, litellm_params=litellm_params, prompt_cache_options=request.get("prompt_cache_options")
     ):
@@ -77,7 +78,7 @@ def _repairs_tool_schemas(model: str, tools: object, litellm_params: GenericLite
     try:
         decoded: Final = _OPENAI_CONFIG._tools_with_object_parameters(model=model, tools=cast(list[object], tools))  # pyright: ignore[reportPrivateUsage, reportArgumentType]  # cast-ok: caller json
     except litellm.BadRequestError:
-        return False
+        return True
     return (
         _OPENAI_CONFIG._sanitized_tool_schemas_for_openai(model=model, tools=decoded, litellm_params=litellm_params)  # pyright: ignore[reportPrivateUsage]  # the same rules the Python route applies
         is not decoded

tests/rust-python-harness/strategies/e2e_parity/sdk/responses/conftest.py

@@ -1,4 +1,4 @@
-from collections.abc import AsyncIterator, Generator
+from collections.abc import AsyncIterator, Generator, Iterator
 from importlib import import_module
 from pathlib import Path
 from typing import Final
@@ -6,9 +6,15 @@ from typing import Final
 import pytest
 import pytest_asyncio
 
-from tests.test_litellm_rust.support.isolation import isolated_test_state
+from litellm.responses import dispatch
+from litellm.rust_bridge import catalog
+from litellm.rust_bridge.catalog import Route, RouteRule
+from litellm.rust_bridge.configuration import Rollout
+from tests.test_litellm_rust.support.isolation import isolated_test_state, rebound
 from tests.test_litellm_rust.support.recording_server import RecordingServer, recording_service
 
+from .support import FallbackProbe
+
 PARITY_ROOT: Final = Path(__file__).parent
 
 
@@ -24,6 +30,22 @@ def recording_server() -> Generator[RecordingServer]:
         yield server
 
 
+@pytest.fixture
+def native_responses() -> Iterator[FallbackProbe]:
+    probe: Final = FallbackProbe()
+    python_aresponses: Final = dispatch._PYTHON_ARESPONSES  # pyright: ignore[reportPrivateUsage]  # the fallback under observation
+
+    async def python_fallback(*args: object, **kwargs: object) -> object:
+        probe.calls += 1
+        return await python_aresponses(*args, **kwargs)
+
+    with (
+        rebound(catalog, "RULES", (RouteRule(Route.RESPONSES, Rollout.RUST_OPT_OUT), *catalog.RULES)),
+        rebound(dispatch, "_PYTHON_ARESPONSES", python_fallback),
+    ):
+        yield probe
+
+
 def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
     try:
         import_module("litellm.rust_bridge._native")

tests/rust-python-harness/strategies/e2e_parity/sdk/responses/support.py

@@ -0,0 +1,14 @@
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Final, cast
+
+import litellm
+from litellm.responses import main as python_responses
+
+PYTHON_ARESPONSES: Final = cast(Callable[..., Awaitable[object]], python_responses.aresponses)  # noqa: TID251  # the Python baseline under comparison
+PUBLIC_ARESPONSES: Final = cast(Callable[..., Awaitable[object]], litellm.aresponses)
+
+
+@dataclass(slots=True)
+class FallbackProbe:
+    calls: int = 0

tests/rust-python-harness/strategies/e2e_parity/sdk/responses/test_native_error_parity.py

@@ -0,0 +1,117 @@
+"""Native Responses failures must reach the caller and the failure callback exactly as Python's do."""
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Final
+
+import openai
+import pytest
+from pydantic import JsonValue, TypeAdapter
+
+from .support import PUBLIC_ARESPONSES, PYTHON_ARESPONSES, FallbackProbe
+from .test_native_parity import MODEL, WEATHER_TOOL
+from tests.test_litellm_rust.support.callback_recorder import RecordingLogger, drain_logging
+from tests.test_litellm_rust.support.isolation import isolated_callback_registries
+from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec
+
+pytestmark = pytest.mark.requires_rust_extension
+
+_LOGGED: Final = TypeAdapter(dict[str, object])
+_COMPARED_LOG_FIELDS: Final = ("status", "error_str", "model", "custom_llm_provider", "call_type", "response_cost")
+_COMPARED_ERROR_FIELDS: Final = ("error_code", "error_class", "llm_provider")
+
+
+@dataclass(frozen=True, slots=True)
+class Failure:
+    exception: tuple[object, ...]
+    logged: Mapping[str, object]
+    fell_back: bool
+
+
+@dataclass(frozen=True, slots=True)
+class FailureCase:
+    name: str
+    kwargs: Mapping[str, object]
+    upstream: ResponseSpec | None = None
+    falls_back: bool = False
+
+
+def _upstream(status: int, error: Mapping[str, JsonValue]) -> ResponseSpec:
+    return ResponseSpec(body={"error": dict(error)}, status=status)
+
+
+CASES: Final = (
+    FailureCase(
+        "bad_request",
+        {"input": "hi"},
+        _upstream(
+            400,
+            {"message": "Invalid value for 'input'.", "type": "invalid_request_error", "param": "input"},
+        ),
+    ),
+    FailureCase(
+        "invalid_api_key",
+        {"input": "hi"},
+        _upstream(401, {"message": "Incorrect API key provided", "type": "invalid_request_error"}),
+    ),
+    FailureCase(
+        "rate_limited",
+        {"input": "hi"},
+        _upstream(429, {"message": "Rate limit reached", "type": "requests", "code": "rate_limit_exceeded"}),
+    ),
+    FailureCase("server_error", {"input": "hi"}, _upstream(500, {"message": "boom", "type": "server_error"})),
+    FailureCase(
+        "invalid_tool_parameters",
+        {"input": "hi", "tools": [{**WEATHER_TOOL, "parameters": "{"}]},
+        falls_back=True,
+    ),
+)
+
+
+async def _failure(case: FailureCase, server: RecordingServer, probe: FallbackProbe | None) -> Failure:
+    with isolated_callback_registries():
+        recorder: Final = RecordingLogger()
+        if case.upstream is not None:
+            server.enqueue(case.upstream)
+        fallbacks_before: Final = 0 if probe is None else probe.calls
+        call: Final = PYTHON_ARESPONSES if probe is None else PUBLIC_ARESPONSES
+        with pytest.raises(openai.APIError) as raised:
+            await call(model=MODEL, api_key="sk-test", api_base=server.base_url, callbacks=[recorder], **case.kwargs)
+        await drain_logging()
+        failures: Final = await recorder.wait_for_async("async_log_failure_event")
+    standard: Final = _LOGGED.validate_python(_LOGGED.validate_python(failures[0].kwargs)["standard_logging_object"])
+    error_information: Final = _LOGGED.validate_python(standard["error_information"])
+    error: Final = raised.value
+    return Failure(
+        exception=(
+            type(error).__name__,
+            getattr(error, "status_code", None),
+            str(error),
+            str(getattr(error, "llm_provider", None)),
+            getattr(error, "model", None),
+        ),
+        logged=MappingProxyType(
+            {
+                **{name: standard.get(name) for name in _COMPARED_LOG_FIELDS},
+                **{name: str(error_information.get(name)) for name in _COMPARED_ERROR_FIELDS},
+            }
+        ),
+        fell_back=probe is not None and probe.calls > fallbacks_before,
+    )
+
+
+@pytest.fixture
+def failure_server(recording_server: RecordingServer) -> RecordingServer:
+    recording_server.expected_requests = None
+    return recording_server
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("case", tuple(pytest.param(case, id=case.name) for case in CASES))
+async def test_native_failures_match_python(
+    case: FailureCase, failure_server: RecordingServer, native_responses: FallbackProbe
+) -> None:
+    python: Final = await _failure(case, failure_server, probe=None)
+    native: Final = await _failure(case, failure_server, probe=native_responses)
+    assert (native.exception, native.logged, native.fell_back) == (python.exception, python.logged, case.falls_back)

tests/rust-python-harness/strategies/e2e_parity/sdk/responses/test_native_parity.py

@@ -5,24 +5,19 @@ recorded OpenAI response, then compares what reached the provider, what the call
 callback logged. The native leg fails instead of silently falling back to Python
 """
 
-from collections.abc import Awaitable, Callable, Iterator, Mapping
+from collections.abc import Iterator, Mapping
 from dataclasses import dataclass, field
 from types import MappingProxyType
-from typing import Final, cast
+from typing import Final
 
 import pytest
 from pydantic import JsonValue, TypeAdapter
 
-import litellm
-from litellm.responses import dispatch
-from litellm.responses import main as python_responses
 from litellm.responses.utils import ResponsesAPIRequestUtils
-from litellm.rust_bridge import catalog
-from litellm.rust_bridge.catalog import Route, RouteRule
-from litellm.rust_bridge.configuration import Rollout
 from litellm.types.llms.openai import ResponsesAPIResponse
+from .support import PUBLIC_ARESPONSES, PYTHON_ARESPONSES, FallbackProbe
 from tests.test_litellm_rust.support.callback_recorder import RecordingLogger, drain_logging
-from tests.test_litellm_rust.support.isolation import isolated_callback_registries, rebound
+from tests.test_litellm_rust.support.isolation import isolated_callback_registries
 from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec
 
 pytestmark = pytest.mark.requires_rust_extension
@@ -42,7 +37,12 @@ _COMPARED_LOG_FIELDS: Final = (
     "total_tokens",
     "response_cost",
     "status",
+    "api_base",
+    "cache_hit",
+    "saved_cache_cost",
+    "response_cost_failure_debug_info",
 )
+_COMPARED_HIDDEN_PARAMS: Final = ("custom_llm_provider", "model_id", "api_base", "response_cost")
 
 TEXT_MESSAGE: Final[Mapping[str, JsonValue]] = MappingProxyType(
     {
@@ -187,9 +187,6 @@ REASONING_REPLAY_INPUT: Final[list[JsonValue]] = [
     {"role": "user", "content": "And Germany?"},
 ]
 
-_PYTHON_ARESPONSES: Final = cast(Callable[..., Awaitable[object]], python_responses.aresponses)  # noqa: TID251  # the Python baseline under comparison
-_PUBLIC_ARESPONSES: Final = cast(Callable[..., Awaitable[object]], litellm.aresponses)
-
 
 @dataclass(frozen=True, slots=True)
 class Leg:
@@ -197,10 +194,12 @@ class Leg:
     request_headers: Mapping[str, str]
     response: Mapping[str, JsonValue]
     logged: Mapping[str, object]
+    hooks: tuple[str, ...]
+    hidden_params: Mapping[str, object]
     fell_back: bool
 
 
-ASPECTS: Final = ("dispatch", "request", "response", "logged")
+ASPECTS: Final = ("dispatch", "request", "response", "logged", "hooks")
 
 
 @dataclass(frozen=True, slots=True)
@@ -313,27 +312,6 @@ CASES: Final = (
 )
 
 
-@dataclass(slots=True)
-class FallbackProbe:
-    calls: int = 0
-
-
-@pytest.fixture
-def native_responses() -> Iterator[FallbackProbe]:
-    probe: Final = FallbackProbe()
-    python_aresponses: Final = dispatch._PYTHON_ARESPONSES  # pyright: ignore[reportPrivateUsage]  # the fallback under observation
-
-    async def python_fallback(*args: object, **kwargs: object) -> object:
-        probe.calls += 1
-        return await python_aresponses(*args, **kwargs)
-
-    with (
-        rebound(catalog, "RULES", (RouteRule(Route.RESPONSES, Rollout.RUST_OPT_OUT), *catalog.RULES)),
-        rebound(dispatch, "_PYTHON_ARESPONSES", python_fallback),
-    ):
-        yield probe
-
-
 def _kwargs(case: Case, server: RecordingServer, recorder: RecordingLogger) -> dict[str, object]:
     return {"model": MODEL, "api_key": "sk-test", "api_base": server.base_url, "callbacks": [recorder], **case.kwargs}
 
@@ -343,6 +321,12 @@ def _response(result: object) -> Mapping[str, JsonValue]:
     return _OBJECT.validate_python(result.model_dump(mode="json", exclude_none=True))
 
 
+def _hidden_params(result: object) -> Mapping[str, object]:
+    assert isinstance(result, ResponsesAPIResponse)
+    hidden: Final = _LOGGED.validate_python(result._hidden_params)  # pyright: ignore[reportPrivateUsage, reportUnknownMemberType]  # the hidden params callers read
+    return MappingProxyType({name: hidden.get(name) for name in _COMPARED_HIDDEN_PARAMS})
+
+
 async def _leg(case: Case, server: RecordingServer, probe: FallbackProbe | None) -> Leg:
     with isolated_callback_registries():
         return await _isolated_leg(case, server, probe)
@@ -355,9 +339,9 @@ async def _isolated_leg(case: Case, server: RecordingServer, probe: FallbackProb
     kwargs: Final = _kwargs(case, server, recorder)
     fallbacks_before: Final = 0 if probe is None else probe.calls
     if probe is None:
-        result = await _PYTHON_ARESPONSES(**kwargs)
+        result = await PYTHON_ARESPONSES(**kwargs)
     else:
-        result = await _PUBLIC_ARESPONSES(**kwargs)
+        result = await PUBLIC_ARESPONSES(**kwargs)
     await drain_logging()
     success: Final = await recorder.wait_for_async("async_log_success_event")
     assert len(server.requests) == request_count + 1
@@ -370,6 +354,8 @@ async def _isolated_leg(case: Case, server: RecordingServer, probe: FallbackProb
         ),
         response=_response(result),
         logged=MappingProxyType({name: standard.get(name) for name in _COMPARED_LOG_FIELDS}),
+        hooks=recorder.names,
+        hidden_params=_hidden_params(result),
         fell_back=probe is not None and probe.calls > fallbacks_before,
     )
 
@@ -394,5 +380,7 @@ async def test_native_responses_match_python(
             assert (native.request_body, native.request_headers) == (python.request_body, python.request_headers)
         case "response":
             assert native.response == python.response
+        case "logged":
+            assert (native.logged, native.hidden_params) == (python.logged, python.hidden_params)
         case _:
-            assert native.logged == python.logged
+            assert native.hooks == python.hooks

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

@@ -113,7 +113,7 @@ _PLAIN_TOOL: Final = MappingProxyType(
         pytest.param({"input": "hi", "tools": [dict(_PLAIN_TOOL)]}, None, id="valid_tool_schema"),
         pytest.param(
             {"input": "hi", "tools": [dict(_PATTERNED_TOOL)]},
-            "native Responses does not implement tool schema repair",
+            "native Responses does not implement tool schema validation and repair",
             id="python_incompatible_pattern",
         ),
         pytest.param(