Merge 42286b9cf7 into ac882742e7

litellm-rust/crates/python-bridge/src/cache/mod.rs

@@ -4,7 +4,7 @@ mod selection;
 
 pub(crate) use native::NativeCacheHandle;
 pub(crate) use python::{CacheCall, PythonCache};
-pub(crate) use selection::{Cached, Selection, admit_native, configure, configured_native};
+pub(crate) use selection::{Cached, Selection, configure, configured_native};
 
 use litellm_cache::Error;
 use pyo3::{

litellm-rust/crates/python-bridge/src/cache/native/v2.rs

@@ -279,7 +279,7 @@ pub(in crate::cache) fn configured(
 )> {
     let handle = native_handle(configured)?.ok_or_else(|| {
         pyo3::exceptions::PyRuntimeError::new_err(
-            "the configured cache changed to a Python cache after native admission",
+            "native inference needs a v2 cache, the catalog keeps Python caches on Python",
         )
     })?;
     let cache = handle.extract::<PyRef<'_, NativeCacheHandle>>()?;

litellm-rust/crates/python-bridge/src/cache/selection.rs

@@ -76,21 +76,6 @@ fn selected_cache<'py>(
     Ok(Some(configured))
 }
 
-pub(crate) fn admit_native(
-    py: Python<'_>,
-    kwargs: &Bound<'_, PyDict>,
-    call_type: &str,
-) -> PyResult<()> {
-    if let Some(configured) = selected_cache(py, kwargs, call_type)?
-        && native::v2::native_handle(&configured)?.is_none()
-    {
-        return Err(crate::errors::RustBridgeDeclined::new_err(
-            "the configured cache requires Python inference",
-        ));
-    }
-    Ok(())
-}
-
 pub(crate) fn configured_native(
     py: Python<'_>,
     kwargs: &Bound<'_, PyDict>,

litellm-rust/crates/python-bridge/src/routes/inference.rs

@@ -177,7 +177,6 @@ where
     } else {
         R::SYNC_CALL_TYPE
     };
-    crate::cache::admit_native(py, &call.kwargs, call_type)?;
     let (arguments, hooks) = super::call_hooks(py, R::OPERATION, &call, asynchronous)?;
     super::run_public_call(
         py,

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

@@ -8,59 +8,19 @@ use litellm_auth::AuthServices;
 use litellm_cache_response::{CachePolicy, ScopedCache};
 use litellm_callbacks_legacy_python::LoggingOperation;
 use litellm_host::{call::HostedMachine, protocol::Protocol};
-use litellm_host_python::present;
 use litellm_inference_responses::{ResponsesRoute, route::Responses};
 use litellm_secrets::source::SecretSource;
 use pyo3::prelude::*;
-use serde_json::Value;
 pub(crate) use websocket::ResponsesWebSocketConnection;
 
 use super::{
     NativeCall,
     inference::{InferenceHost, InferenceRoute, run_inference},
 };
-use crate::errors::RustBridgeDeclined;
 
 const ROUTE_HOST_MODULE: &str = "litellm.rust_bridge.host.routes.responses";
 
 fn run_responses(py: Python<'_>, call: NativeCall<'_>, asynchronous: bool) -> PyResult<Py<PyAny>> {
-    let resolved = call.resolved()?;
-    if let Some(reason) = py
-        .import(ROUTE_HOST_MODULE)?
-        .getattr("decline_reason")?
-        .call1((&resolved,))?
-        .extract::<Option<String>>()?
-    {
-        return Err(RustBridgeDeclined::new_err(reason));
-    }
-    let argument = |name: &str| present(&call.kwargs, &call.base, name);
-    let model = argument("model")?
-        .ok_or_else(|| pyo3::exceptions::PyValueError::new_err("model is required"))?
-        .extract::<String>()?;
-    let provider = argument("custom_llm_provider")?
-        .map(|value| value.extract::<String>())
-        .transpose()?;
-    if provider
-        .as_deref()
-        .is_some_and(|provider| provider != "openai")
-        || model
-            .strip_prefix("openai/")
-            .unwrap_or(&model)
-            .contains('/')
-    {
-        return Err(RustBridgeDeclined::new_err(
-            "native HTTP responses provider",
-        ));
-    }
-    if argument("stream")?
-        .map(|value| litellm_host_python::from_py::<Value>(&value))
-        .transpose()?
-        .is_some_and(|value| value == Value::Bool(true))
-    {
-        return Err(RustBridgeDeclined::new_err(
-            "native Python responses streaming",
-        ));
-    }
     let host = InferenceHost::new(call.view()?, ROUTE_HOST_MODULE);
     run_inference::<ResponsesRoute, _>(py, call, asynchronous, ResponsesPythonHost::new(host))
 }

litellm/chat_completions/dispatch.py

@@ -1,9 +1,11 @@
 import inspect
 from collections.abc import Awaitable, Callable, Coroutine, Mapping
+from types import MappingProxyType
 from typing import Final, TypeAlias, cast  # noqa: TID251  # native binding selects a sync result or an async awaitable
 
 from litellm import main
-from litellm.rust_bridge.catalog import Route, RouteContext
+from litellm.rust_bridge import detectors
+from litellm.rust_bridge.catalog import Feature, Route, RouteContext
 from litellm.rust_bridge.chat_completions.entrypoints import (
     NATIVE_ACOMPLETION,
     NATIVE_COMPLETION,
@@ -63,11 +65,32 @@ def _public_request(
     return native_call(legacy, args, kwargs)
 
 
+_FEATURES: Final[detectors.Detectors] = MappingProxyType(
+    {
+        **detectors.COMMON,
+        Feature.MOCK_RESPONSE: detectors.chat_mock_response,
+        Feature.AGENTIC_LOOP_HOOK: detectors.chat_agentic_loop_hook,
+        Feature.RESPONSE_CACHE: detectors.response_cache,
+        Feature.STREAMING: detectors.streaming,
+        Feature.DROP_PARAMS: detectors.drop_params,
+        Feature.MODIFY_PARAMS: detectors.modify_params,
+        Feature.EXTRA_BODY: detectors.extra_body,
+        Feature.RETRIES_FALLBACKS: detectors.chat_retries_fallbacks,
+        Feature.PROMPT_MANAGEMENT: detectors.prompt_management,
+        Feature.CACHE_CONTROL_INJECTION: detectors.cache_control_injection,
+        Feature.MODEL_PARAM_GATING: detectors.chat_model_param_gating,
+        Feature.MODEL_LIST: detectors.model_list,
+        Feature.MODEL_ALIAS_MAP: detectors.model_alias_map,
+    }
+)
+
+
 def _context(request: NativeCall) -> RouteContext:
     return RouteContext(
         Route.CHAT_COMPLETIONS,
         provider=optional_str(request.resolved.get("custom_llm_provider")),
         model=str(request.resolved["model"]),
+        features=detectors.detected(_FEATURES, request),
     )
 
 

litellm/llms/anthropic/pass_through/messages/interceptors/advisor.py

@@ -15,7 +15,7 @@ How it works:
 """
 
 import uuid
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Final
 
 import litellm
@@ -48,7 +48,7 @@ class AdvisorOrchestrationHandler(MessagesInterceptor):
 
     def can_handle(
         self,
-        tools: list[dict] | None,
+        tools: Sequence[Mapping[str, object]] | None,
         custom_llm_provider: str | None,
     ) -> bool:
         if not tools:

litellm/llms/anthropic/pass_through/messages/interceptors/base.py

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Mapping, Sequence
 
 from litellm.types.llms.anthropic_messages.anthropic_response import (
     AnthropicMessagesResponse,
@@ -21,7 +21,7 @@ class MessagesInterceptor(ABC):
     @abstractmethod
     def can_handle(
         self,
-        tools: list[dict] | None,
+        tools: Sequence[Mapping[str, object]] | None,
         custom_llm_provider: str | None,
     ) -> bool:
         """Return True if this interceptor should handle the request."""

litellm/messages/dispatch.py

@@ -6,7 +6,9 @@ from typing import Final, TypeAlias, cast  # noqa: TID251  # native binding sele
 from litellm.exceptions import BadRequestError
 from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
 from litellm.llms.anthropic.pass_through.messages import handler as main
+from litellm.llms.anthropic.pass_through.messages.interceptors import get_messages_interceptors
 from litellm.llms.anthropic.pass_through.safeguards import native_route_classifies
+from litellm.rust_bridge import detectors
 from litellm.rust_bridge.catalog import Feature, Route, RouteContext, Target
 from litellm.rust_bridge.dispatch import PublicDispatch
 from litellm.rust_bridge.messages.entrypoints import (
@@ -19,6 +21,7 @@ from litellm.rust_bridge.public_call import (
     bind,
     native_call,
     native_call_hook,
+    optional_mapping,
     optional_sequence,
     optional_str,
     signature,
@@ -99,16 +102,41 @@ def _target(request: NativeCall, model: str, provider: str | None) -> Target:
     return Target.CHAT_COMPLETIONS
 
 
-_FEATURES: Final[Mapping[Feature, Callable[[NativeCall], bool]]] = MappingProxyType(
+def _intercepted(request: NativeCall) -> bool:
+    tools: Final = optional_sequence(request.resolved.get("tools"))
+    if not tools:
+        return False
+    mappings: Final = tuple(mapping for mapping in map(optional_mapping, tools) if mapping is not None)
+    provider: Final = _resolved(request)[1]
+    return any(interceptor.can_handle(mappings, provider) for interceptor in get_messages_interceptors())
+
+
+_FEATURES: Final[detectors.Detectors] = MappingProxyType(
     {
+        **detectors.COMMON,
         Feature.SAFEGUARDS_CLASSIFIER: lambda request: native_route_classifies(
             str(request.resolved["model"]), request.resolved.get("safeguards")
         ),
+        Feature.RESPONSE_CACHE: detectors.response_cache,
+        Feature.MODIFY_PARAMS: detectors.modify_params,
+        Feature.WORKLOAD_IDENTITY: detectors.anthropic_workload_identity,
+        Feature.CACHE_CONTROL_INJECTION: detectors.cache_control_injection,
+        Feature.STREAM_COST_INJECTION: detectors.stream_cost_injection,
+        Feature.THINKING_SIGNATURE_RETRY: detectors.thinking_signature_retry,
+    }
+)
+
+_ASYNC_FEATURES: Final[detectors.Detectors] = MappingProxyType(
+    {
+        **_FEATURES,
+        Feature.PRE_REQUEST_HOOK: detectors.pre_request_hook,
+        Feature.WEBSEARCH_INTERCEPTION: detectors.websearch_interception,
+        Feature.MESSAGES_INTERCEPTOR: _intercepted,
     }
 )
 
 
-def _context(request: NativeCall) -> RouteContext:
+def _context(request: NativeCall, features: detectors.Detectors) -> RouteContext:
     requested_model: Final = str(request.resolved["model"])
     model, provider = _resolved(request)
     return RouteContext(
@@ -116,20 +144,20 @@ def _context(request: NativeCall) -> RouteContext:
         provider=provider,
         model=requested_model,
         target=_target(request, model, provider),
-        features=frozenset(feature for feature, detected in _FEATURES.items() if detected(request)),
+        features=detectors.detected(features, request),
     )
 
 
 _DISPATCH: Final = PublicDispatch(
     route=Route.MESSAGES,
     request=lambda args, kwargs: _public_request(_MESSAGES, args, kwargs),
-    context=_context,
+    context=lambda request: _context(request, _FEATURES),
 )
 
 _ADISPATCH: Final = PublicDispatch(
     route=Route.MESSAGES,
     request=lambda args, kwargs: _public_request(_AMESSAGES, args, kwargs),
-    context=_context,
+    context=lambda request: _context(request, _ASYNC_FEATURES),
 )
 
 

litellm/responses/dispatch.py

@@ -1,10 +1,14 @@
 import inspect
 from collections.abc import Awaitable, Callable, Coroutine, Mapping
+from types import MappingProxyType
 from typing import Final, TypeAlias, cast  # noqa: TID251  # native binding selects a sync result or an async awaitable
 
+from litellm.exceptions import BadRequestError
+from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
 from litellm.responses import main
 from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
-from litellm.rust_bridge.catalog import Route, RouteContext
+from litellm.rust_bridge import detectors
+from litellm.rust_bridge.catalog import Feature, Route, RouteContext
 from litellm.rust_bridge.dispatch import PublicDispatch
 from litellm.rust_bridge.public_call import (
     AsyncFlag,
@@ -61,24 +65,64 @@ def _public_request(
     return native_call(legacy, args, kwargs)
 
 
-def _context(request: NativeCall) -> RouteContext:
+def _provider(request: NativeCall) -> str | None:
+    provider: Final = optional_str(request.resolved.get("custom_llm_provider"))
+    try:
+        return get_llm_provider(str(request.resolved["model"]), provider)[1]
+    except BadRequestError:
+        return provider
+
+
+_FEATURES: Final[detectors.Detectors] = MappingProxyType(
+    {
+        **detectors.COMMON,
+        Feature.DROP_PARAMS: detectors.drop_params,
+        Feature.MODIFY_PARAMS: detectors.modify_params,
+        Feature.STREAMING: detectors.streaming,
+        Feature.PROMPT_MANAGEMENT: detectors.prompt_management,
+        Feature.TEXT_FORMAT: detectors.text_format,
+        Feature.CHAT_COMPLETIONS_BRIDGE: detectors.chat_completions_bridge,
+        Feature.RETRIES_FALLBACKS: detectors.retries_fallbacks,
+        Feature.DEPLOYMENT_RESPONSE_ID: detectors.deployment_response_id,
+        Feature.ENCODED_ITEM_IDS: detectors.encoded_item_ids,
+        Feature.INPUT_REWRITE: detectors.input_rewrite,
+        Feature.EXTRA_BODY: detectors.extra_body,
+        Feature.PROVIDER_AFFINITY: detectors.provider_affinity,
+        Feature.RESPONSE_CACHE: detectors.response_cache,
+        Feature.MANAGED_RESOURCE_IDS: detectors.managed_resource_ids,
+        Feature.CACHE_CONTROL_MARKS: detectors.cache_control_marks,
+        Feature.ANTHROPIC_TOOL_SEARCH: detectors.anthropic_tool_search,
+        Feature.TOOL_SCHEMA_REPAIR: detectors.tool_schema_repair,
+        Feature.MODEL_PARAM_GATING: detectors.model_param_gating,
+        Feature.WORKLOAD_IDENTITY: detectors.openai_workload_identity,
+        Feature.CLIENT_HEADERS: detectors.client_headers,
+    }
+)
+
+_SYNC_FEATURES: Final[detectors.Detectors] = MappingProxyType(
+    {**_FEATURES, Feature.DEPLOYMENT_HOOK: detectors.deployment_hook}
+)
+
+
+def _context(request: NativeCall, features: detectors.Detectors) -> RouteContext:
     return RouteContext(
         Route.RESPONSES,
-        provider=optional_str(request.resolved.get("custom_llm_provider")),
+        provider=_provider(request),
         model=str(request.resolved["model"]),
+        features=detectors.detected(features, request),
     )
 
 
 _DISPATCH: Final = PublicDispatch(
     route=Route.RESPONSES,
     request=lambda args, kwargs: _public_request(_RESPONSES, args, kwargs),
-    context=_context,
+    context=lambda request: _context(request, _SYNC_FEATURES),
 )
 
 _ADISPATCH: Final = PublicDispatch(
     route=Route.RESPONSES,
     request=lambda args, kwargs: _public_request(_ARESPONSES, args, kwargs),
-    context=_context,
+    context=lambda request: _context(request, _FEATURES),
 )
 
 

litellm/rust_bridge/AGENTS.md

@@ -1,21 +1,52 @@
-# Python boundary
+# Ownership
 
-This package owns native rollout and fallback selection, Python public API compatibility, settings projection and the Python bindings supplied to the Rust bridge. Rust core owns provider execution. `litellm-host-python` owns CPython runtime mechanics. `callbacks-legacy-python` owns legacy callback sharing and dispatch policy
+- This package owns
+  - choosing Rust or Python for each call, and falling back to Python when Rust is unavailable
+  - Python public API compatibility at the native boundary
+  - projecting Python settings into the values Rust reads
+  - the Python bindings handed to the Rust bridge
+- It does not own
+  - provider execution, which is Rust core
+  - CPython runtime mechanics, which is `litellm-host-python`
+  - legacy callback sharing and dispatch policy, which is `callbacks-legacy-python`
+- `host/` holds the modules Rust imports by name to read Python-owned state. `host/AGENTS.md` states that contract
 
-`host/` holds the Python modules Rust imports by name to read Python-owned state (provider resolution, model capabilities, settings, caches, diagnostics, legacy callbacks, the secret manager, transport and timeout conversions). Its AGENTS.md states that contract
+# Routing
 
-`execution.py` owns generic inline execution and stream iteration, plus the step classes Rust constructs. Native entrypoints return an unstarted `Execution`, and `execution.synchronous` or `execution.asynchronous` hands it to the public caller through `run` or `drive` with the one stream pair, so Rust never names a Python driver or stream class. Keep one driver implementing generator-style `send` (`send(None)` starts), `throw` and idempotent `close`. Do not create a second implementation
+- `catalog.py` decides, `detectors.py` describes the request, `dispatch.py` applies the decision
+- The first matching rule wins. A context no rule matches stays on Python
+- Every Python-only exclusion for a route (a `targets` or `features` rule) sits above any Rust rule for that route
+- A route whose Python side is `NO_PYTHON` resolves to `RUST_REQUIRED` for every context it can produce
+  - `tests/test_litellm_rust/test_public_entries.py` marks such a route `python=None`, so dropping a route's Python changes its entries, its rules and that field together
+- A detector reads the same inputs the legacy Python path reads, so the catalog and the fallback agree on which requests need Python
+  - A route's context, its target as well as its features, reads `NativeCall.resolved`, so an argument counts the same whether it came by position, by keyword or as a signature default
+  - A route's sync and async dispatches read different detector maps when the legacy handlers differ: Messages runs pre-request hooks, web search interception and interceptors only on async calls, and the sync Responses handler runs the deployment hook itself
+- A detector that raises keeps the call on Python. `PublicDispatch` logs it and never fails the call over routing
+- `catalog.RULES` is rollout policy with no tests of its own. Tests pass the rules they need, so check these invariants by review whenever `RULES` changes
 
-Generic execution steps and inline suspension handling must not depend on LiteLLM response metadata. Public stream construction, `_hidden_params` and header compatibility remain product responsibilities. Keep generic stream heads opaque and preserve public header behavior in the product wrappers
+# Public entries
 
-Await each selected suspension in the caller's task so hooks retain thread, loop and context identity. A final awaitable value is returned as a value, not awaited implicitly. Closing, cancellation and `GeneratorExit` release the execution without replaying provider work or starting further callbacks
+- Every route serves `a<entry>(...)`, `<entry>(..., a<entry>=True)` and `a<entry>(..., a<entry>=True)` the same way, handing native one identical `NativeCall` with no async flag on the async binding
+- `tests/test_litellm_rust/test_public_entries.py` holds this for every route in its table, so a new public route gets a row there
 
-Native admission may select legacy before execution starts. A failure after execution starts is terminal and never authorizes fallback. Creating an unstarted native coroutine must not acquire clients or credentials or capture execution context
+# Execution
 
-Verify cross-language behavior using a fresh installed extension, including unstarted-call release, setup ordering, exception identity, sync and async streams, cancellation and GC. Keep `_native.pyi` accurate about Future-returning and coroutine-returning APIs
+- `execution.py` owns generic inline execution, stream iteration and the step classes Rust constructs
+- Native entrypoints return an unstarted `Execution`. `execution.synchronous` or `execution.asynchronous` hands it to the caller through `run` or `drive`, so Rust never names a Python driver or stream class
+- Keep one driver, implementing generator-style `send` (`send(None)` starts), `throw` and idempotent `close`
+- Generic steps and suspension handling never depend on LiteLLM response metadata. Public stream construction, `_hidden_params` and header compatibility stay in the product wrappers, and generic stream heads stay opaque
+- Await each selected suspension in the caller's task, so hooks keep their thread, loop and context
+- Return a final awaitable as a value instead of awaiting it implicitly
+- Closing, cancellation and `GeneratorExit` release the execution without replaying provider work or starting more callbacks
 
-A route with no Python implementation passes `runtime.NO_PYTHON` from both public entries, and every catalog rule for it must be `RUST_REQUIRED`. Dispatch raises `NoPythonImplementationError` for any other rollout instead of falling back. `tests/test_litellm_rust/test_public_entries.py` marks such a route with `python=None`, so dropping a route's Python changes the entries, its rules and that one table field together
+# Fallback
 
-A route's context, its target as well as its features, is built from `NativeCall.resolved`, so an argument counts the same whether it came by position, by keyword or as a signature default
+- Native admission may pick Python before execution starts
+- A failure after execution starts is terminal and never triggers fallback
+- Creating an unstarted native coroutine acquires no clients or credentials and captures no execution context
 
-Every public route serves `a<entry>(...)`, `<entry>(..., a<entry>=True)` and `a<entry>(..., a<entry>=True)` the same way. Each hands native one identical `NativeCall` with no async flag, on the async binding. `test_public_entries.py` holds this for every route in its table, so a new public route gets a row there
+# Verification
+
+- Check cross-language behavior against a freshly installed extension, covering unstarted-call release, setup ordering, exception identity, sync and async streams, cancellation and GC
+- Keep `_native.pyi` accurate about which APIs return a Future and which return a coroutine
+- Unit tests of this package live in `tests/unit/rust_bridge/`, end-to-end Rust path tests in `tests/test_litellm_rust/`

litellm/rust_bridge/catalog.py

@@ -33,6 +33,77 @@ class Target(str, Enum):
 
 class Feature(str, Enum):
     SAFEGUARDS_CLASSIFIER = "safeguards_classifier"
+    """A non-Claude model with dangerous_tool_use safeguards, which the proxy classifier answers."""
+    MOCK_RESPONSE = "mock_response"
+    """mock_response short-circuits before any provider call."""
+    PROXY_MCP_TOOLS = "proxy_mcp_tools"
+    """litellm_proxy MCP tools that Python expands and may auto-execute."""
+    CALLER_CLIENT = "caller_client"
+    """A caller-supplied HTTP client or shared session that Python sends through."""
+    TIMEOUT_VARIANTS = "timeout_variants"
+    """request_timeout, stream_timeout or httpx.Timeout phases beyond a plain timeout."""
+    DROP_PARAMS = "drop_params"
+    """drop_params or additional_drop_params dropping fields per provider."""
+    MODIFY_PARAMS = "modify_params"
+    """litellm.modify_params rewriting the request, like clamping max_tokens."""
+    STREAMING = "streaming"
+    """stream=True."""
+    PROMPT_MANAGEMENT = "prompt_management"
+    """prompt_id resolved through a prompt manager."""
+    TEXT_FORMAT = "text_format"
+    """A text_format model that Python turns into a JSON schema and parses back."""
+    CHAT_COMPLETIONS_BRIDGE = "chat_completions_bridge"
+    """use_chat_completions_api or the openai/chat_completions/ prefix, served through Chat Completions."""
+    RETRIES_FALLBACKS = "retries_fallbacks"
+    """num_retries, max_retries or fallbacks run by the SDK."""
+    DEPLOYMENT_RESPONSE_ID = "deployment_response_id"
+    """A deployment id that Python encodes into the returned response id."""
+    ENCODED_ITEM_IDS = "encoded_item_ids"
+    """Input items carrying litellm-encoded ids that Python restores before sending."""
+    INPUT_REWRITE = "input_rewrite"
+    """Responses input items that Python normalizes or removes before replay."""
+    EXTRA_BODY = "extra_body"
+    """Extra body fields that Python transforms and merges into the provider request."""
+    PROVIDER_AFFINITY = "provider_affinity"
+    """A provider affinity header that Python validates and populates from the session id."""
+    PRE_REQUEST_HOOK = "pre_request_hook"
+    """A logger overriding async_pre_request_hook, which may rewrite the request."""
+    WEBSEARCH_INTERCEPTION = "websearch_interception"
+    """A WebSearchInterceptionLogger that may answer without calling the provider."""
+    MESSAGES_INTERCEPTOR = "messages_interceptor"
+    """A registered MessagesInterceptor that takes over the call."""
+    AGENTIC_LOOP_HOOK = "agentic_loop_hook"
+    """A logger overriding async_should_run_agentic_loop, which may run follow-up calls."""
+    RESPONSE_CACHE = "response_cache"
+    """litellm.cache, whose hit and store order differs from native."""
+    TLS_POLICY = "tls_policy"
+    """An OpenSSL security level or ECDH curve that native TLS cannot apply."""
+    MANAGED_RESOURCE_IDS = "managed_resource_ids"
+    """Managed file or vector store ids that Python maps to provider ids."""
+    CACHE_CONTROL_MARKS = "cache_control_marks"
+    """Anthropic cache_control marks that Python strips or turns into prompt cache breakpoints."""
+    ANTHROPIC_TOOL_SEARCH = "anthropic_tool_search"
+    """Anthropic tool search tools that Python maps to OpenAI tool search."""
+    TOOL_SCHEMA_REPAIR = "tool_schema_repair"
+    """Tool schemas that Python decodes, flattens or strips of non-Python regex patterns."""
+    MODEL_PARAM_GATING = "model_param_gating"
+    """Params that Python clamps, drops or rejects from the model map, like gpt-5 temperature."""
+    WORKLOAD_IDENTITY = "workload_identity"
+    """No static key, so Python mints a workload identity federation token."""
+    CACHE_CONTROL_INJECTION = "cache_control_injection"
+    """Prompt caching injection that Python applies to messages and system."""
+    STREAM_COST_INJECTION = "stream_cost_injection"
+    """litellm.include_cost_in_streaming_usage, where Python adds cost to streamed usage."""
+    THINKING_SIGNATURE_RETRY = "thinking_signature_retry"
+    """Replayed thinking blocks, which Python strips and retries once if Anthropic rejects their signature."""
+    CLIENT_HEADERS = "client_headers"
+    """Proxy-forwarded headers that Python merges into extra_headers."""
+    DEPLOYMENT_HOOK = "deployment_hook"
+    """A logger overriding async_pre_call_deployment_hook, which the sync Python handler runs itself."""
+    MODEL_LIST = "model_list"
+    """A model_list that Python fans out across its deployments."""
+    MODEL_ALIAS_MAP = "model_alias_map"
+    """A model that litellm.model_alias_map rewrites before the call."""
 
 
 @dataclass(frozen=True, slots=True)
@@ -68,14 +139,78 @@ class RouteRule:
 
 Rules: TypeAlias = tuple[RouteRule, ...]
 
+COMMON_PYTHON_FEATURES: Final = frozenset(
+    {
+        Feature.MOCK_RESPONSE,
+        Feature.PROXY_MCP_TOOLS,
+        Feature.CALLER_CLIENT,
+        Feature.TIMEOUT_VARIANTS,
+        Feature.AGENTIC_LOOP_HOOK,
+        Feature.TLS_POLICY,
+    }
+)
+
+CHAT_PYTHON_FEATURES: Final = COMMON_PYTHON_FEATURES | {
+    Feature.RESPONSE_CACHE,
+    Feature.STREAMING,
+    Feature.DROP_PARAMS,
+    Feature.MODIFY_PARAMS,
+    Feature.EXTRA_BODY,
+    Feature.RETRIES_FALLBACKS,
+    Feature.PROMPT_MANAGEMENT,
+    Feature.CACHE_CONTROL_INJECTION,
+    Feature.MODEL_PARAM_GATING,
+    Feature.MODEL_LIST,
+    Feature.MODEL_ALIAS_MAP,
+}
+
+MESSAGES_PYTHON_FEATURES: Final = COMMON_PYTHON_FEATURES | {
+    Feature.SAFEGUARDS_CLASSIFIER,
+    Feature.PRE_REQUEST_HOOK,
+    Feature.WEBSEARCH_INTERCEPTION,
+    Feature.MESSAGES_INTERCEPTOR,
+    Feature.RESPONSE_CACHE,
+    Feature.MODIFY_PARAMS,
+    Feature.WORKLOAD_IDENTITY,
+    Feature.CACHE_CONTROL_INJECTION,
+    Feature.STREAM_COST_INJECTION,
+    Feature.THINKING_SIGNATURE_RETRY,
+}
+
+RESPONSES_PYTHON_FEATURES: Final = COMMON_PYTHON_FEATURES | {
+    Feature.DROP_PARAMS,
+    Feature.MODIFY_PARAMS,
+    Feature.STREAMING,
+    Feature.PROMPT_MANAGEMENT,
+    Feature.TEXT_FORMAT,
+    Feature.CHAT_COMPLETIONS_BRIDGE,
+    Feature.RETRIES_FALLBACKS,
+    Feature.DEPLOYMENT_RESPONSE_ID,
+    Feature.ENCODED_ITEM_IDS,
+    Feature.INPUT_REWRITE,
+    Feature.EXTRA_BODY,
+    Feature.PROVIDER_AFFINITY,
+    Feature.RESPONSE_CACHE,
+    Feature.MANAGED_RESOURCE_IDS,
+    Feature.CACHE_CONTROL_MARKS,
+    Feature.ANTHROPIC_TOOL_SEARCH,
+    Feature.TOOL_SCHEMA_REPAIR,
+    Feature.MODEL_PARAM_GATING,
+    Feature.WORKLOAD_IDENTITY,
+    Feature.CLIENT_HEADERS,
+    Feature.DEPLOYMENT_HOOK,
+}
+
 RULES: Final[Rules] = (
+    RouteRule(Route.CHAT_COMPLETIONS, Rollout.PYTHON_ONLY, features=CHAT_PYTHON_FEATURES),
     RouteRule(Route.CHAT_COMPLETIONS, Rollout.PYTHON_ONLY),
     RouteRule(Route.EMBEDDINGS, Rollout.PYTHON_ONLY),
     RouteRule(Route.OCR, Rollout.RUST_REQUIRED),
     RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY, targets=frozenset({Target.CHAT_COMPLETIONS, Target.RESPONSES})),
-    RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY, features=frozenset({Feature.SAFEGUARDS_CLASSIFIER})),
+    RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY, features=MESSAGES_PYTHON_FEATURES),
     RouteRule(Route.MESSAGES, Rollout.RUST_OPT_IN, providers=frozenset({"anthropic"})),
     RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY),
+    RouteRule(Route.RESPONSES, Rollout.PYTHON_ONLY, features=RESPONSES_PYTHON_FEATURES),
     RouteRule(Route.RESPONSES, Rollout.PYTHON_ONLY),
     RouteRule(Route.TOKEN_COUNTER, Rollout.PYTHON_ONLY),
     RouteRule(Route.TOKENIZER, Rollout.PYTHON_ONLY),

litellm/rust_bridge/detectors.py

@@ -0,0 +1,498 @@
+from __future__ import annotations
+
+import os
+from collections.abc import Callable, Iterable, Mapping, Sequence
+from itertools import accumulate, chain, repeat, takewhile
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final, TypeAlias, cast  # noqa: TID251  # tool lists stay caller-owned
+
+import httpx
+from pydantic import TypeAdapter
+
+import litellm
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.core_helpers import normalize_drop_params
+from litellm.rust_bridge.catalog import Feature
+from litellm.rust_bridge.public_call import NativeCall, optional_mapping, optional_sequence, optional_str
+
+if TYPE_CHECKING:
+    from openai.types.responses import ResponseInputParam
+
+    from litellm.responses.mcp.litellm_proxy_mcp_handler import ToolParam
+    from litellm.types.llms.openai import ALL_RESPONSES_API_TOOL_PARAMS, ResponsesAPIOptionalRequestParams
+
+Detector: TypeAlias = Callable[[NativeCall], bool]
+Detectors: TypeAlias = Mapping[Feature, Detector]
+
+
+def detected(detectors: Detectors, request: NativeCall) -> frozenset[Feature]:
+    return frozenset(feature for feature, detector in detectors.items() if detector(request))
+
+
+def _present(request: NativeCall, name: str) -> bool:
+    return request.resolved.get(name) is not None
+
+
+def mock_response(request: NativeCall) -> bool:
+    return _present(request, "mock_response")
+
+
+def chat_mock_response(request: NativeCall) -> bool:
+    return (
+        mock_response(request)
+        or bool(request.resolved.get("mock_tool_calls"))
+        or bool(request.resolved.get("mock_timeout"))
+    )
+
+
+def client_headers(request: NativeCall) -> bool:
+    return bool(optional_mapping(request.resolved.get("headers")))
+
+
+def model_list(request: NativeCall) -> bool:
+    return _present(request, "model_list")
+
+
+def model_alias_map(request: NativeCall) -> bool:
+    return bool(litellm.model_alias_map) and str(request.resolved["model"]) in litellm.model_alias_map
+
+
+def caller_client(request: NativeCall) -> bool:
+    return _present(request, "client") or _present(request, "shared_session")
+
+
+def extra_body(request: NativeCall) -> bool:
+    return bool(request.resolved.get("extra_body"))
+
+
+def provider_affinity(request: NativeCall) -> bool:
+    return _present(request, "provider_affinity_header")
+
+
+def streaming(request: NativeCall) -> bool:
+    return request.resolved.get("stream") is True
+
+
+def timeout_variants(request: NativeCall) -> bool:
+    return (
+        _present(request, "request_timeout")
+        or (streaming(request) and _present(request, "stream_timeout"))
+        or isinstance(request.resolved.get("timeout"), httpx.Timeout)
+    )
+
+
+def proxy_mcp_tools(request: NativeCall) -> bool:
+    from litellm.responses.mcp.litellm_proxy_mcp_handler import LiteLLM_Proxy_MCP_Handler
+
+    tools: Final = optional_sequence(request.resolved.get("tools"))
+    return tools is not None and LiteLLM_Proxy_MCP_Handler.should_use_litellm_mcp_gateway(
+        cast("Iterable[ToolParam]", tools)  # cast-ok: the predicate skips entries that are not MCP tool dicts
+    )
+
+
+def prompt_management(request: NativeCall) -> bool:
+    from litellm.integrations.litellm_agent.litellm_agent_model_resolver import LITELLM_AGENT_PREFIX
+    from litellm.types.utils import DynamicPromptManagementParamLiteral
+
+    return (
+        _present(request, "prompt_id")
+        or str(request.resolved["model"]).startswith(LITELLM_AGENT_PREFIX)
+        or any(request.resolved.get(param) for param in DynamicPromptManagementParamLiteral.list_all_params())
+    )
+
+
+def text_format(request: NativeCall) -> bool:
+    return _present(request, "text_format")
+
+
+def chat_completions_bridge(request: NativeCall) -> bool:
+    from litellm.responses.main import (
+        _normalize_openai_chat_completions_responses_model,  # pyright: ignore[reportPrivateUsage]  # the handler's own bridge prefix check
+    )
+
+    return (
+        bool(request.resolved.get("use_chat_completions_api"))
+        or _normalize_openai_chat_completions_responses_model(str(request.resolved["model"]))[1]
+    )
+
+
+def retries_fallbacks(request: NativeCall) -> bool:
+    return bool(litellm.num_retries) or any(
+        _present(request, name) for name in ("num_retries", "max_retries", "fallbacks", "retry_policy")
+    )
+
+
+def chat_retries_fallbacks(request: NativeCall) -> bool:
+    return (
+        retries_fallbacks(request)
+        or bool(optional_sequence(litellm.model_fallbacks))  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]  # litellm annotates it with a bare list
+        or _present(request, "context_window_fallback_dict")
+    )
+
+
+def _openai_model(request: NativeCall) -> str:
+    from litellm.exceptions import BadRequestError
+    from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
+
+    model: Final = str(request.resolved["model"])
+    try:
+        return get_llm_provider(model, optional_str(request.resolved.get("custom_llm_provider")))[0]
+    except BadRequestError:
+        return model
+
+
+def _api_base(request: NativeCall) -> str | None:
+    return optional_str(request.resolved.get("api_base")) or optional_str(request.resolved.get("base_url"))
+
+
+def _model_info(request: NativeCall) -> Mapping[str, object] | None:
+    return optional_mapping(request.resolved.get("model_info"))
+
+
+def _tools(request: NativeCall) -> Sequence[object]:
+    return optional_sequence(request.resolved.get("tools")) or ()
+
+
+def managed_resource_ids(request: NativeCall) -> bool:
+    from litellm.litellm_core_utils.prompt_templates.common_utils import (
+        update_responses_input_with_model_file_ids,
+        update_responses_tools_with_model_file_ids,
+    )
+
+    model_id: Final = optional_str((_model_info(request) or {}).get("id"))
+    mapping: Final = request.resolved.get("model_file_id_mapping")
+    file_ids: Final = cast("dict[str, dict[str, str]] | None", mapping)  # cast-ok: handler passes it unvalidated
+    input_value: Final = request.resolved.get("input")
+    tools: Final = [dict(tool) for tool in map(optional_mapping, _tools(request)) if tool is not None]
+    rewritten_tools: Final = update_responses_tools_with_model_file_ids(tools, model_id, file_ids) if tools else tools
+    return update_responses_input_with_model_file_ids(input_value, model_id, file_ids) != input_value or (
+        rewritten_tools != tools
+    )
+
+
+def input_rewrite(request: NativeCall) -> bool:
+    from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
+
+    input_value: Final = request.resolved.get("input")
+    if not isinstance(input_value, list):
+        return False
+    config: Final = OpenAIResponsesAPIConfig()
+    replayable: Final = config._drop_bridge_minted_reasoning_items(  # pyright: ignore[reportPrivateUsage]  # use the handler's replay policy
+        cast("ResponseInputParam", input_value),  # cast-ok: the handler accepts unvalidated input items
+    )
+    normalized: Final = config._validate_input_param(replayable)  # pyright: ignore[reportPrivateUsage]  # use the handler's normalization
+    cleaned: Final = config._drop_foreign_tool_call_item_ids(normalized)  # pyright: ignore[reportPrivateUsage]  # use the handler's id cleanup
+    return config.strip_custom_tool_call_namespace_from_responses_input(cleaned) != input_value
+
+
+def _children(value: object) -> tuple[object, ...]:
+    if isinstance(value, Mapping):
+        return tuple(cast("Mapping[object, object]", value).values())  # cast-ok: only values are walked
+    if isinstance(value, list | tuple):
+        return tuple(cast("Sequence[object]", value))  # cast-ok: only items are walked
+    return ()
+
+
+def _next_level(level: tuple[object, ...], _: None) -> tuple[object, ...]:
+    return tuple(chain.from_iterable(map(_children, level)))
+
+
+def _mentions(value: object, key: str) -> bool:
+    levels: Final = takewhile(bool, accumulate(repeat(None), _next_level, initial=(value,)))
+    return any(
+        key in cast("Mapping[object, object]", node)  # cast-ok: node passed the Mapping guard
+        for node in chain.from_iterable(levels)
+        if isinstance(node, Mapping)
+    )
+
+
+def cache_control_marks(request: NativeCall) -> bool:
+    return _mentions(request.resolved.get("input"), "cache_control") or _mentions(_tools(request), "cache_control")
+
+
+def anthropic_tool_search(request: NativeCall) -> bool:
+    from litellm.responses.litellm_completion_transformation.tool_search import is_anthropic_tool_search_chat_tool
+
+    return any(is_anthropic_tool_search_chat_tool(tool) for tool in _tools(request))
+
+
+def _undecoded_schema(tool: object) -> bool:
+    mapping: Final = optional_mapping(tool)
+    parameters: Final = None if mapping is None else mapping.get("parameters")
+    return parameters is not None and not isinstance(parameters, Mapping)
+
+
+def tool_schema_repair(request: NativeCall) -> bool:
+    from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
+    from litellm.types.router import GenericLiteLLMParams
+
+    tools: Final = _tools(request)
+    if any(_undecoded_schema(tool) for tool in tools):
+        return True
+    sanitized: Final = OpenAIResponsesAPIConfig()._sanitized_tool_schemas_for_openai(  # pyright: ignore[reportPrivateUsage]  # the handler's own schema rewrite
+        model=_openai_model(request),
+        tools=cast("Sequence[ALL_RESPONSES_API_TOOL_PARAMS]", tools),  # cast-ok: sanitizer skips non-dict tools
+        litellm_params=GenericLiteLLMParams(model_info=_model_info(request)),
+    )
+    return bool(tools) and sanitized is not tools
+
+
+_PARAMS: Final = TypeAdapter(dict[str, object])
+
+
+def model_param_gating(request: NativeCall) -> bool:
+    from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
+    from litellm.rust_bridge.host.routes.responses import PARAMETERS
+
+    params: Final = {name: request.resolved[name] for name in PARAMETERS if request.resolved.get(name) is not None}
+    mapped: Final = _PARAMS.validate_python(
+        OpenAIResponsesAPIConfig().map_openai_params(  # pyright: ignore[reportUnknownMemberType]  # returns a bare dict
+            cast("ResponsesAPIOptionalRequestParams", params),  # cast-ok: the handler passes these kwargs
+            _openai_model(request),
+            drop_params=True,
+        )
+    )
+    return mapped != params
+
+
+_CHAT_UNGATED: Final = frozenset(
+    {
+        "model",
+        "messages",
+        "custom_llm_provider",
+        "api_version",
+        "drop_params",
+        "allowed_openai_params",
+        "additional_drop_params",
+    }
+)
+_NAMES: Final = TypeAdapter(list[str])
+
+
+def _chat_non_default_params(request: NativeCall) -> Mapping[str, object]:
+    from litellm.constants import DEFAULT_CHAT_COMPLETION_PARAM_VALUES
+
+    defaults: Final = _PARAMS.validate_python(DEFAULT_CHAT_COMPLETION_PARAM_VALUES)
+    return {
+        name: value
+        for name, value in request.resolved.items()
+        if name in defaults and name not in _CHAT_UNGATED and value != defaults[name]
+    }
+
+
+def _chat_provider(request: NativeCall) -> tuple[str, str | None]:
+    from litellm.exceptions import BadRequestError
+    from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
+
+    model: Final = str(request.resolved["model"])
+    provider: Final = optional_str(request.resolved.get("custom_llm_provider"))
+    try:
+        resolved_model, resolved_provider, _, _ = get_llm_provider(model, provider, api_base=_api_base(request))
+    except BadRequestError:
+        return model, provider
+    return resolved_model, resolved_provider
+
+
+def _unvalidated(params: Mapping[str, object]) -> frozenset[str]:
+    from litellm.utils import PROVIDER_UNVALIDATED_PARAMS
+
+    return PROVIDER_UNVALIDATED_PARAMS | (frozenset({"n"}) if params.get("n") == 1 else frozenset[str]())
+
+
+def chat_model_param_gating(request: NativeCall) -> bool:
+    from litellm.exceptions import UnsupportedParamsError
+    from litellm.litellm_core_utils.get_supported_openai_params import (
+        get_supported_openai_params,  # pyright: ignore[reportUnknownVariableType]  # returns a bare list
+    )
+    from litellm.types.utils import LlmProviders
+    from litellm.utils import ProviderConfigManager
+
+    params: Final = _chat_non_default_params(request)
+    model, provider = _chat_provider(request)
+    if not params or provider is None:
+        return False
+    supported: Final = frozenset(_NAMES.validate_python(get_supported_openai_params(model, provider) or []))
+    if any(name not in supported for name in params.keys() - _unvalidated(params)):
+        return True
+    try:
+        llm_provider: Final = LlmProviders(provider)
+    except ValueError:
+        return False
+    config: Final = ProviderConfigManager.get_provider_chat_config(model, llm_provider)
+    if config is None:
+        return False
+    try:
+        strict: Final = _PARAMS.validate_python(
+            config.map_openai_params(dict(params), {}, model, drop_params=False)  # pyright: ignore[reportUnknownMemberType]  # returns a bare dict
+        )
+    except UnsupportedParamsError:
+        return True
+    return strict != _PARAMS.validate_python(
+        config.map_openai_params(dict(params), {}, model, drop_params=True)  # pyright: ignore[reportUnknownMemberType]  # returns a bare dict
+    )
+
+
+def openai_workload_identity(request: NativeCall) -> bool:
+    from litellm.llms.openai.workload_identity import resolve_openai_workload_identity_config
+
+    api_key: Final = optional_str(request.resolved.get("api_key")) or litellm.api_key or litellm.openai_key
+    return resolve_openai_workload_identity_config(api_key, _api_base(request), request.resolved) is not None
+
+
+def anthropic_workload_identity(request: NativeCall) -> bool:
+    from litellm.llms.anthropic.common_utils import AnthropicModelInfo
+    from litellm.secret_managers.main import get_secret_str
+
+    def configured(param: str, env: str) -> bool:
+        return _present(request, param) or get_secret_str(env) is not None
+
+    return (
+        AnthropicModelInfo.get_api_key(optional_str(request.resolved.get("api_key"))) is None
+        and AnthropicModelInfo.get_auth_token() is None
+        and configured("anthropic_federation_rule_id", "ANTHROPIC_FEDERATION_RULE_ID")
+        and configured("anthropic_organization_id", "ANTHROPIC_ORGANIZATION_ID")
+    )
+
+
+def cache_control_injection(request: NativeCall) -> bool:
+    from litellm.integrations.anthropic_cache_control_hook import configured_injection_points
+
+    return (
+        litellm.enable_anthropic_prompt_caching is True
+        or request.resolved.get("enable_prompt_caching") is True
+        or bool(configured_injection_points(request.resolved.get("cache_control_injection_points")))
+    )
+
+
+_THINKING_BLOCKS: Final = frozenset({"thinking", "redacted_thinking"})
+
+
+def _replays_thinking(message: object) -> bool:
+    content: Final = optional_sequence((optional_mapping(message) or {}).get("content")) or ()
+    return any((optional_mapping(block) or {}).get("type") in _THINKING_BLOCKS for block in content)
+
+
+def thinking_signature_retry(request: NativeCall) -> bool:
+    return any(_replays_thinking(message) for message in optional_sequence(request.resolved.get("messages")) or ())
+
+
+def stream_cost_injection(request: NativeCall) -> bool:
+    return streaming(request) and bool(litellm.include_cost_in_streaming_usage)
+
+
+def drop_params(request: NativeCall) -> bool:
+    return (
+        bool(litellm.drop_params)
+        or normalize_drop_params(request.resolved.get("drop_params")) is True
+        or bool(optional_sequence(request.resolved.get("additional_drop_params")))
+    )
+
+
+def modify_params(_request: NativeCall) -> bool:
+    return bool(litellm.modify_params)
+
+
+def deployment_response_id(request: NativeCall) -> bool:
+    metadata: Final = optional_mapping(request.resolved.get("litellm_metadata")) or {}
+    model_info: Final = optional_mapping(metadata.get("model_info")) or {}
+    return model_info.get("id") is not None
+
+
+def _encoded_item(item: Mapping[str, object]) -> bool:
+    from litellm.responses.utils import ResponsesAPIRequestUtils
+
+    item_id: Final = optional_str(item.get("id"))
+    encrypted_content: Final = optional_str(item.get("encrypted_content"))
+    return (item_id is not None and ResponsesAPIRequestUtils.decode_encrypted_item_id(item_id) is not None) or (
+        encrypted_content is not None
+        and ResponsesAPIRequestUtils.unwrap_encrypted_content_with_model_id(encrypted_content)[1] != encrypted_content
+    )
+
+
+def encoded_item_ids(request: NativeCall) -> bool:
+    items: Final = optional_sequence(request.resolved.get("input")) or ()
+    mappings: Final = (optional_mapping(item) for item in items)
+    return any(_encoded_item(item) for item in mappings if item is not None)
+
+
+def _custom_loggers(callbacks: Iterable[object]) -> tuple[CustomLogger, ...]:
+    from litellm.litellm_core_utils.litellm_logging import get_custom_logger_compatible_class
+
+    resolved: Final = (
+        get_custom_logger_compatible_class(callback) if isinstance(callback, str) else callback
+        for callback in callbacks
+    )
+    return tuple(callback for callback in resolved if isinstance(callback, CustomLogger))
+
+
+def _overridden(method: str, callbacks: Iterable[object]) -> bool:
+    base: Final = getattr(CustomLogger, method)
+    return any(getattr(type(callback), method, base) is not base for callback in _custom_loggers(callbacks))
+
+
+def _request_success_callbacks(request: NativeCall) -> tuple[object, ...]:
+    from litellm.litellm_core_utils.litellm_logging import Logging
+
+    supplied: Final = optional_sequence(request.resolved.get("success_callback")) or ()
+    logging_obj: Final = request.resolved.get("litellm_logging_obj")
+    dynamic: Final = (
+        optional_sequence(logging_obj.dynamic_success_callbacks)  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]  # Logging annotates the list with a bare Callable
+        if isinstance(logging_obj, Logging)
+        else None
+    )
+    return (*supplied, *(dynamic or ()))
+
+
+def pre_request_hook(_request: NativeCall) -> bool:
+    return _overridden("async_pre_request_hook", litellm.callbacks)
+
+
+def _call_callbacks(request: NativeCall) -> tuple[object, ...]:
+    return (*litellm.callbacks, *_request_success_callbacks(request))
+
+
+def agentic_loop_hook(request: NativeCall) -> bool:
+    return _overridden("async_should_run_agentic_loop", _call_callbacks(request))
+
+
+def chat_agentic_loop_hook(request: NativeCall) -> bool:
+    return agentic_loop_hook(request) or _overridden(
+        "async_should_run_chat_completion_agentic_loop", _call_callbacks(request)
+    )
+
+
+def deployment_hook(request: NativeCall) -> bool:
+    return _overridden("async_pre_call_deployment_hook", _call_callbacks(request))
+
+
+def websearch_interception(_request: NativeCall) -> bool:
+    from litellm.integrations.websearch_interception.handler import WebSearchInterceptionLogger
+
+    return any(isinstance(callback, WebSearchInterceptionLogger) for callback in litellm.callbacks)
+
+
+def response_cache(request: NativeCall) -> bool:
+    cache: Final = optional_mapping(request.resolved.get("cache")) or {}
+    return litellm.cache is not None and cache.get("no-cache") is not True
+
+
+_NATIVE_ECDH_CURVES: Final = frozenset({"x25519", "prime256v1", "secp256r1", "p-256", "secp384r1", "p-384"})
+
+
+def tls_policy(_request: NativeCall) -> bool:
+    curve: Final = os.getenv("SSL_ECDH_CURVE", litellm.ssl_ecdh_curve)
+    return bool(os.getenv("SSL_SECURITY_LEVEL", litellm.ssl_security_level)) or (
+        bool(curve) and str(curve).strip().lower() not in _NATIVE_ECDH_CURVES
+    )
+
+
+COMMON: Final[Detectors] = MappingProxyType(
+    {
+        Feature.MOCK_RESPONSE: mock_response,
+        Feature.PROXY_MCP_TOOLS: proxy_mcp_tools,
+        Feature.CALLER_CLIENT: caller_client,
+        Feature.TIMEOUT_VARIANTS: timeout_variants,
+        Feature.AGENTIC_LOOP_HOOK: agentic_loop_hook,
+        Feature.TLS_POLICY: tls_policy,
+    }
+)

litellm/rust_bridge/dispatch.py

@@ -4,6 +4,7 @@ from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass
 from typing import Final, Generic, TypeAlias, TypeVar
 
+from litellm._logging import verbose_logger
 from litellm.rust_bridge import catalog, runtime
 from litellm.rust_bridge.bindings import NativeBinding
 from litellm.rust_bridge.catalog import Route, RouteContext, Rules
@@ -38,6 +39,15 @@ class PublicDispatch(Generic[RequestT]):
             )
         return request
 
+    def _guarded_context(self, request: RequestT) -> RouteContext | None:
+        try:
+            return self.context(request)
+        except Exception:  # noqa: BLE001  # detectors call legacy handlers, so any error must fall back to Python
+            verbose_logger.warning(
+                "%s route detection failed, serving the call on Python", self.route.value, exc_info=True
+            )
+            return None
+
     def run(
         self,
         args: tuple[object, ...],
@@ -63,8 +73,11 @@ class PublicDispatch(Generic[RequestT]):
         request: Final = self.request(args, kwargs)
         if request is None:
             return python(*args, **kwargs)
+        context: Final = self._guarded_context(request)
+        if context is None:
+            return python(*args, **kwargs)
         return runtime.run(
-            self.context(request),
+            context,
             binding=binding,
             native=lambda hook: native(hook, request, args, kwargs),
             python=lambda: python(*args, **kwargs),
@@ -96,8 +109,11 @@ class PublicDispatch(Generic[RequestT]):
         request: Final = self.request(args, kwargs)
         if request is None:
             return await python(*args, **kwargs)
+        context: Final = self._guarded_context(request)
+        if context is None:
+            return await python(*args, **kwargs)
         return await runtime.arun(
-            self.context(request),
+            context,
             binding=binding,
             native=lambda hook: native(hook, request, args, kwargs),
             python=lambda: python(*args, **kwargs),

litellm/rust_bridge/host/routes/responses.py

@@ -4,9 +4,8 @@ from collections.abc import Mapping
 from typing import Final
 
 import litellm
-from litellm import get_llm_provider
 from litellm.rust_bridge import failures
-from litellm.rust_bridge.public_call import inference_decline_reason, known_params, optional_str
+from litellm.rust_bridge.public_call import known_params, optional_str
 from litellm.types.llms.openai import ResponsesAPIOptionalRequestParams, ResponsesAPIResponse
 
 PARAMETERS: Final = tuple(ResponsesAPIOptionalRequestParams.__annotations__)
@@ -34,14 +33,3 @@ def map_failure(error: Exception, request: Mapping[str, object]) -> Exception:
         request,
         optional_str(request.get("api_base")) or optional_str(request.get("base_url")),
     )
-
-
-def decline_reason(request: Mapping[str, object]) -> str | None:
-    if optional_str(request.get("custom_llm_provider")) is None and "/" not in str(request["model"]):
-        try:
-            _, provider, _, _ = get_llm_provider(model=str(request["model"]))
-        except litellm.exceptions.BadRequestError:
-            return "native Responses could not resolve the provider"
-        if provider != "openai":
-            return "native HTTP responses provider"
-    return inference_decline_reason(PARAMETERS, request)

litellm/rust_bridge/public_call.py

@@ -8,33 +8,6 @@ from dataclasses import dataclass
 from types import MappingProxyType
 from typing import Final, TypeVar, cast  # noqa: TID251  # narrows caller-owned containers without copying them
 
-import litellm
-
-_INFERENCE_CONTEXT: Final = frozenset(
-    {
-        "model",
-        "messages",
-        "input",
-        "api_key",
-        "api_base",
-        "base_url",
-        "custom_llm_provider",
-        "extra_headers",
-        "timeout",
-        "request_timeout",
-        "callbacks",
-        "success_callback",
-        "failure_callback",
-        "metadata",
-        "litellm_metadata",
-        "litellm_call_id",
-        "litellm_trace_id",
-        "litellm_logging_obj",
-        "litellm_credential_name",
-        "proxy_server_request",
-    }
-)
-
 
 def signature(legacy: Callable[..., object]) -> inspect.Signature:
     return inspect.signature(legacy)
@@ -75,19 +48,6 @@ def known_params(names: Collection[str], request: Mapping[str, object]) -> Mappi
     return {name: value for name, value in request.items() if name in names and value is not None}
 
 
-def inference_decline_reason(parameters: tuple[str, ...], kwargs: Mapping[str, object]) -> str | None:
-    if litellm.drop_params or litellm.modify_params:
-        return "native inference does not implement the configured parameter rewrites"
-    for name, value in kwargs.items():
-        if value is None:
-            continue
-        if name in {"cache", "caching"}:
-            continue
-        if name not in parameters and name not in _INFERENCE_CONTEXT:
-            return f"native inference does not implement {name}"
-    return None
-
-
 _BAGS: Final = frozenset({inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD})
 
 

tests/test_litellm_rust/AGENTS.md

@@ -1,33 +1,62 @@
-# Invariants
+# Scope
 
-- Outside the dispatch tests below, this directory tests the Rust path and nothing else. Treat the Python implementation of every route as already removed. Never call it on purpose, never run one request through both engines, and never compare Python and Rust outputs, headers, cache keys or logs
-- A test that would still pass with Python serving the call does not belong here. Every test except the dispatch tests asserts the call was served natively with `support.routing.assert_served_natively`
-- Dispatch tests check which path a public call takes under given rules. They are each route folder's `test_dispatch.py`, plus `test_dispatch.py` and `test_public_entries.py` at the top of this directory
-  - They pass `rules=` to a dispatch directly or wrap the whole test in one `support.routing.routed(...)` block, and they stub the native bindings
-  - They may call Python, since the choice between the two paths is what they test
-  - They live here until they can run in `tests/unit/` without the compiled extension
-- Select the Rust path with `support.routing.native_route(...)`, which prepends `RUST_REQUIRED` rules to `catalog.RULES` and restores them on exit. Each route folder's `conftest.py` applies it to every test through an autouse fixture. A test outside a route folder wraps its whole body in one `native_route(...)` block, and a child interpreter script assigns `catalog.RULES` once at its top. Never select a backend by reading or setting `LITELLM_RUST`, calling `litellm.rust(...)` or rebinding `catalog.RULES` any other way. The rollout in `litellm/rust_bridge/catalog.py` is a product decision, and a test here must pass no matter what it says, including for a route marked `PYTHON_ONLY`
-- `catalog.RULES` is static at runtime and `native_route(...)` swaps a module global, so open it once per test and never per call. Two overlapping blocks, such as one per task inside an `asyncio.gather`, restore out of order and leave the wrong rules behind for later tests
-- No test here depends on the rollout switch. Only no rule, `PYTHON_ONLY` and `RUST_REQUIRED` pick a path without it, so those are the only rollouts a test here uses, and Rust with Python fallback is not reachable from this directory. What `LITELLM_RUST` and `litellm.rust(...)` mean is tested in `tests/unit/rust_bridge/test_configuration.py`, mostly through `decide`, which takes both as plain values
-- A test lives here only when it cannot be written in Rust. That means it needs the Python public API, Python-owned state such as `litellm.cache` and callbacks, or the process boundary (fork, threads, event loops). Anything checkable from the Rust side goes in the crate's own tests
+- This directory tests the Rust path end to end through the Python public API, plus the dispatch tests below, and nothing else
+- A test belongs here only when it cannot be written in Rust, meaning it needs
+  - the Python public API
+  - Python-owned state such as `litellm.cache` and callbacks
+  - the process boundary (fork, threads, event loops)
+- Anything checkable from the Rust side goes in the crate's own tests
 - The wire is the local recording server. No test here reaches a real provider
-- Never parametrize one test across routes. Each route folder owns its tests, its `server` fixture, its request data and its `call`. The one exception is `test_public_entries.py`, whose subject is that every route serves its public call forms the same way
-- Call `_native` directly only for a case the public API cannot reach
+
+# Rust only
+
+- Outside the dispatch tests, treat every route's Python implementation as already removed
+  - never call it on purpose
+  - never run one request through both engines
+  - never compare Python and Rust outputs, headers, cache keys or logs
+- A test that would still pass with Python serving the call does not belong here
+- Every test except the dispatch tests asserts native serving with `support.routing.assert_served_natively`
+
+# Dispatch tests
+
+- They check which path a public call takes under given rules
+  - each route folder's `test_dispatch.py`
+  - `test_dispatch.py` at the top, which tests `PublicDispatch` with stub routes
+  - `test_public_entries.py` at the top, which runs every route's public call forms through one table
+- They pass `rules=` to a dispatch directly or wrap the whole test in one `support.routing.routed(...)` block, and they stub the native bindings
+- They may call Python, since the choice between the two paths is what they test
+- They live here until they can run in `tests/unit/` without the compiled extension
+
+# Selecting a path
+
+- Use `support.routing.native_route(...)`, which prepends `RUST_REQUIRED` rules to `catalog.RULES` and restores them on exit
+  - route folders apply it to every test through the autouse fixture in their `conftest.py`
+  - a test outside a route folder wraps its whole body in one `native_route(...)` block
+  - a child interpreter script assigns `catalog.RULES` once at its top
+- Dispatch tests use `support.routing.routed(...)` instead, which sets `catalog.RULES` to exactly the rules given
+- Open either once per test, never per call. They swap a module global, so overlapping blocks (one per task in an `asyncio.gather`) restore out of order and leak rules into later tests
+- Never select a backend through `LITELLM_RUST`, `litellm.rust(...)` or any other rebinding of `catalog.RULES`
+- No test here depends on the rollout switch
+  - only no rule, `PYTHON_ONLY` and `RUST_REQUIRED` pick a path without it, so those are the only rollouts used here
+  - Rust with Python fallback needs the switch, so it is not reachable from this directory
+  - what `LITELLM_RUST` and `litellm.rust(...)` mean is tested in `tests/unit/rust_bridge/test_configuration.py`, mostly through `decide`, which takes both as plain values
+- Tests pass whatever the shipped rollout in `litellm/rust_bridge/catalog.py` says, including for a `PYTHON_ONLY` route
 
 # Layout
 
-Each route folder (`ocr/`, `messages/`, `responses/`, `chat_completions/`) uses the same file names, one concern per file
-
-- `test_dispatch.py` the call reaches the native route, and which path the rules pick for each public call
-- `test_requests.py` Python inputs that reach the wire
-- `test_failures.py` errors raised to the caller and rejection before provider IO
-- `test_settings.py` globals, env, named credentials and their precedence
-- `test_streaming.py` stream relay, early close and stream-time logging
-- `test_callbacks.py` the logger hook contract
-- `test_body_hooks.py` handler body hooks such as interceptors and agentic loops
-- `test_lifecycle.py` cancellation, concurrency, context and nested calls
-- `test_cache.py` the configured cache seen from this route
-
-`embeddings/` holds only `test_dispatch.py` so far. At the top, `test_dispatch.py` tests `PublicDispatch` with stub routes and `test_public_entries.py` runs every route's public call forms through one table
-
-Route data and the route's `call` live in `<route>/data.py`, and `<route>/conftest.py` owns the `server` fixture and the autouse `native` fixture. `support/` holds only route-neutral helpers
+- One folder per route (`ocr/`, `messages/`, `responses/`, `chat_completions/`), all using the same file names
+  - `test_dispatch.py` which path the rules pick for each public call, and that the call reaches the native route
+  - `test_requests.py` Python inputs that reach the wire
+  - `test_failures.py` errors raised to the caller and rejection before provider IO
+  - `test_settings.py` globals, env, named credentials and their precedence
+  - `test_streaming.py` stream relay, early close and stream-time logging
+  - `test_callbacks.py` the logger hook contract
+  - `test_body_hooks.py` handler body hooks such as interceptors and agentic loops
+  - `test_lifecycle.py` cancellation, concurrency, context and nested calls
+  - `test_cache.py` the configured cache seen from this route
+- `embeddings/` holds only `test_dispatch.py` so far
+- `<route>/data.py` holds the route's request data and its `call`
+- `<route>/conftest.py` owns the `server` fixture and the autouse `native` fixture
+- `support/` holds only route-neutral helpers
+- Never parametrize one test across routes, except in `test_public_entries.py`, whose subject is that every route serves its public call forms the same way
+- Call `_native` directly only for a case the public API cannot reach

tests/test_litellm_rust/chat_completions/test_dispatch.py

@@ -4,13 +4,15 @@ from typing import Final, cast  # noqa: TID251  # narrows legacy callable signat
 import pytest
 
 import litellm
+from litellm.caching.caching import Cache
 from litellm.chat_completions import dispatch
 from litellm.chat_completions.dispatch import (
     _ADISPATCH,  # pyright: ignore[reportPrivateUsage]  # tests configured dispatch
     _DISPATCH,  # pyright: ignore[reportPrivateUsage]  # tests configured dispatch
 )
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.rust_bridge.bindings import NativeBinding
-from litellm.rust_bridge.catalog import Route, RouteRule, Rules
+from litellm.rust_bridge.catalog import Feature, Route, RouteRule, Rules
 from litellm.rust_bridge.chat_completions.entrypoints import (
     NATIVE_ACOMPLETION,
     NATIVE_COMPLETION,
@@ -228,6 +230,93 @@ async def test_acompletion_marker_binds_positionals_by_the_sync_signature() -> N
     assert [(request.resolved["timeout"], request.resolved["functions"]) for request in captured] == [(30.0, None)]
 
 
+class _ChatAgentic(CustomLogger):
+    async def async_should_run_chat_completion_agentic_loop(
+        self, *args: object, **kwargs: object
+    ) -> tuple[bool, dict[str, object]]:  # pyright: ignore[reportIncompatibleMethodOverride]  # test double only needs the override
+        return False, {}
+
+
+@pytest.mark.parametrize(
+    ("feature", "kwargs", "settings"),
+    (
+        (Feature.MOCK_RESPONSE, {"mock_response": "hi"}, {}),
+        pytest.param(Feature.MOCK_RESPONSE, {"mock_tool_calls": [{"id": "call_1"}]}, {}, id="mock-tool-calls"),
+        pytest.param(Feature.MOCK_RESPONSE, {"mock_timeout": True}, {}, id="mock-timeout"),
+        (Feature.CALLER_CLIENT, {"client": object()}, {}),
+        (Feature.TIMEOUT_VARIANTS, {"request_timeout": 5}, {}),
+        (Feature.AGENTIC_LOOP_HOOK, {}, {"callbacks": [_ChatAgentic()]}),
+        (Feature.TLS_POLICY, {}, {"ssl_security_level": "DEFAULT@SECLEVEL=1"}),
+        (Feature.RESPONSE_CACHE, {}, {"cache": Cache()}),
+        (Feature.STREAMING, {"stream": True}, {}),
+        (Feature.DROP_PARAMS, {}, {"drop_params": True}),
+        (Feature.MODIFY_PARAMS, {}, {"modify_params": True}),
+        (Feature.EXTRA_BODY, {"extra_body": {"top_k": 1}}, {}),
+        (Feature.RETRIES_FALLBACKS, {"num_retries": 2}, {}),
+        pytest.param(
+            Feature.RETRIES_FALLBACKS,
+            {"context_window_fallback_dict": {"anthropic/claude-sonnet-4-5": "anthropic/claude-opus-4-5"}},
+            {},
+            id="context-window-fallbacks",
+        ),
+        pytest.param(
+            Feature.RETRIES_FALLBACKS,
+            {},
+            {"model_fallbacks": [{"anthropic/claude-sonnet-4-5": ["anthropic/claude-opus-4-5"]}]},
+            id="model-fallbacks",
+        ),
+        (Feature.PROMPT_MANAGEMENT, {"prompt_id": "welcome"}, {}),
+        pytest.param(Feature.PROMPT_MANAGEMENT, {"model": "litellm_agent/gpt-5"}, {}, id="litellm-agent-model"),
+        (Feature.CACHE_CONTROL_INJECTION, {"enable_prompt_caching": True}, {}),
+        (Feature.MODEL_PARAM_GATING, {"logit_bias": {"1": 1}}, {}),
+        (
+            Feature.MODEL_LIST,
+            {"model_list": [{"model_name": "anthropic/claude-sonnet-4-5", "litellm_params": {"model": "gpt-5"}}]},
+            {},
+        ),
+        (
+            Feature.MODEL_ALIAS_MAP,
+            {},
+            {"model_alias_map": {"anthropic/claude-sonnet-4-5": "anthropic/claude-opus-4-5"}},
+        ),
+    ),
+)
+def test_each_detected_feature_reaches_its_rule(
+    monkeypatch: pytest.MonkeyPatch, feature: Feature, kwargs: Mapping[str, object], settings: Mapping[str, object]
+) -> None:
+    rules: Final = (
+        RouteRule(Route.CHAT_COMPLETIONS, Rollout.PYTHON_ONLY, features=frozenset({feature})),
+        RouteRule(Route.CHAT_COMPLETIONS, Rollout.RUST_REQUIRED),
+    )
+    python_result: Final = ModelResponse()
+    native_result: Final = ModelResponse()
+
+    def served(call_kwargs: Mapping[str, object]) -> object:
+        return _DISPATCH.run(
+            (),
+            {"model": "anthropic/claude-sonnet-4-5", "messages": MESSAGES, **call_kwargs},
+            python=lambda *call_args, **python_kwargs: python_result,
+            binding=completion_binding(lambda request: native_result),
+            native=native_call_hook,
+            rules=rules,
+        )
+
+    monkeypatch.setattr(litellm, "callbacks", [])
+    monkeypatch.setattr(litellm, "ssl_security_level", None)
+    monkeypatch.setattr(litellm, "cache", None)
+    monkeypatch.setattr(litellm, "model_alias_map", {})
+    for name in ("num_retries", "model_fallbacks"):
+        monkeypatch.setattr(litellm, name, None)
+    for name in ("drop_params", "modify_params", "enable_anthropic_prompt_caching"):
+        monkeypatch.setattr(litellm, name, False)
+    for name in ("SSL_SECURITY_LEVEL", "SSL_ECDH_CURVE"):
+        monkeypatch.delenv(name, raising=False)
+    assert served({}) is native_result
+    for name, value in settings.items():
+        monkeypatch.setattr(litellm, name, value)
+    assert served(kwargs) is python_result
+
+
 def test_acompletion_flag_hands_off_only_when_true() -> None:
     expected: Final = ModelResponse()
 

tests/test_litellm_rust/messages/test_dispatch.py

@@ -5,6 +5,9 @@ from typing import Final, cast  # noqa: TID251  # narrows legacy callable signat
 import pytest
 
 import litellm
+from litellm.caching.caching import Cache
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.integrations.websearch_interception.handler import WebSearchInterceptionLogger
 from litellm.llms.anthropic.pass_through.messages import handler as python_messages
 from litellm.messages import dispatch
 from litellm.messages.dispatch import (
@@ -17,6 +20,7 @@ from litellm.rust_bridge.catalog import Feature, Route, RouteRule, Rules, Target
 from litellm.rust_bridge.configuration import Rollout
 from litellm.rust_bridge.messages.entrypoints import NATIVE_AMESSAGES, NATIVE_MESSAGES, NativeAmessages, NativeMessages
 from litellm.rust_bridge.public_call import NativeCall
+from litellm.types.llms.anthropic import ANTHROPIC_ADVISOR_TOOL_TYPE
 from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse
 
 MESSAGES: Final = [{"role": "user", "content": "hi"}]
@@ -425,3 +429,175 @@ def test_messages_binding_error_delegates_unchanged_to_python() -> None:
     )
 
     assert response is expected
+
+
+class _Rewrite(CustomLogger):
+    async def async_pre_request_hook(self, model: str, messages: list, kwargs: dict) -> dict | None:  # pyright: ignore[reportMissingTypeArgument]  # mirrors the CustomLogger signature
+        return kwargs
+
+
+class _Agentic(CustomLogger):
+    async def async_should_run_agentic_loop(self, *args: object, **kwargs: object) -> tuple[bool, dict]:  # pyright: ignore[reportIncompatibleMethodOverride, reportMissingTypeArgument]  # test double only needs the override
+        return False, {}
+
+
+ADVISOR_TOOL: Final = {"type": ANTHROPIC_ADVISOR_TOOL_TYPE, "name": "advisor", "model": "claude-opus-5-5"}
+
+
+@pytest.mark.parametrize(
+    ("feature", "model", "kwargs", "settings"),
+    (
+        (Feature.MOCK_RESPONSE, "anthropic/claude-sonnet-4-5", {"mock_response": "hi"}, {}),
+        (
+            Feature.PROXY_MCP_TOOLS,
+            "anthropic/claude-sonnet-4-5",
+            {"tools": [{"type": "mcp", "server_label": "gateway", "server_url": "litellm_proxy"}]},
+            {},
+        ),
+        (Feature.CALLER_CLIENT, "anthropic/claude-sonnet-4-5", {"client": object()}, {}),
+        (Feature.TIMEOUT_VARIANTS, "anthropic/claude-sonnet-4-5", {"request_timeout": 5}, {}),
+        (Feature.AGENTIC_LOOP_HOOK, "anthropic/claude-sonnet-4-5", {}, {"callbacks": [_Agentic()]}),
+        (Feature.RESPONSE_CACHE, "anthropic/claude-sonnet-4-5", {}, {"cache": Cache()}),
+        (Feature.TLS_POLICY, "anthropic/claude-sonnet-4-5", {}, {"ssl_security_level": "DEFAULT@SECLEVEL=1"}),
+        (Feature.MODIFY_PARAMS, "anthropic/claude-sonnet-4-5", {}, {"modify_params": True}),
+        (
+            Feature.WORKLOAD_IDENTITY,
+            "anthropic/claude-sonnet-4-5",
+            {"anthropic_federation_rule_id": "rule-1", "anthropic_organization_id": "org-1"},
+            {},
+        ),
+        (Feature.CACHE_CONTROL_INJECTION, "anthropic/claude-sonnet-4-5", {"enable_prompt_caching": True}, {}),
+        (
+            Feature.THINKING_SIGNATURE_RETRY,
+            "anthropic/claude-sonnet-4-5",
+            {"messages": [{"role": "assistant", "content": [{"type": "redacted_thinking", "data": "x"}]}]},
+            {},
+        ),
+        (
+            Feature.STREAM_COST_INJECTION,
+            "anthropic/claude-sonnet-4-5",
+            {"stream": True},
+            {"include_cost_in_streaming_usage": True},
+        ),
+    ),
+)
+def test_each_detected_feature_reaches_its_rule(
+    monkeypatch: pytest.MonkeyPatch,
+    feature: Feature,
+    model: str,
+    kwargs: Mapping[str, object],
+    settings: Mapping[str, object],
+) -> None:
+    rules: Final[Rules] = (
+        RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY, features=frozenset({feature})),
+        RouteRule(Route.MESSAGES, Rollout.RUST_REQUIRED),
+    )
+    python_result: Final = response(model)
+    native_result: Final = response(model)
+
+    def served(call_kwargs: Mapping[str, object]) -> AnthropicMessagesResponse:
+        messages: Final = call_kwargs.get("messages", MESSAGES)
+        return _DISPATCH.run(
+            (16, messages, model),
+            {name: value for name, value in call_kwargs.items() if name != "messages"},
+            python=lambda *call_args, **python_kwargs: python_result,
+            binding=messages_binding(lambda request: native_result),
+            native=lambda hook, request, call_args, native_kwargs: hook(request),
+            rules=rules,
+        )
+
+    monkeypatch.setattr(litellm, "callbacks", [])
+    monkeypatch.setattr(litellm, "cache", None)
+    monkeypatch.setattr(litellm, "ssl_security_level", None)
+    for setting in ("modify_params", "enable_anthropic_prompt_caching", "include_cost_in_streaming_usage"):
+        monkeypatch.setattr(litellm, setting, False)
+    for name in ("SSL_SECURITY_LEVEL", "SSL_ECDH_CURVE", "ANTHROPIC_FEDERATION_RULE_ID", "ANTHROPIC_ORGANIZATION_ID"):
+        monkeypatch.delenv(name, raising=False)
+    monkeypatch.setenv("ANTHROPIC_API_KEY", "")
+    monkeypatch.delenv("ANTHROPIC_AUTH_TOKEN", raising=False)
+    assert served({}) is native_result
+    for name, value in settings.items():
+        monkeypatch.setattr(litellm, name, value)
+    assert served(kwargs) is python_result
+
+
+ASYNC_ONLY_FEATURES: Final = (
+    pytest.param(Feature.PRE_REQUEST_HOOK, {}, {"callbacks": [_Rewrite()]}, id="pre-request-hook"),
+    pytest.param(
+        Feature.WEBSEARCH_INTERCEPTION, {}, {"callbacks": [WebSearchInterceptionLogger()]}, id="websearch-interception"
+    ),
+    pytest.param(
+        Feature.MESSAGES_INTERCEPTOR,
+        {"model": "deepseek/deepseek-v4-flash", "tools": [ADVISOR_TOOL]},
+        {},
+        id="messages-interceptor",
+    ),
+)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("feature", "kwargs", "settings"), ASYNC_ONLY_FEATURES)
+async def test_handler_hooks_the_sync_entry_never_runs_keep_only_async_calls_on_python(
+    monkeypatch: pytest.MonkeyPatch, feature: Feature, kwargs: Mapping[str, object], settings: Mapping[str, object]
+) -> None:
+    rules: Final[Rules] = (
+        RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY, features=frozenset({feature})),
+        RouteRule(Route.MESSAGES, Rollout.RUST_REQUIRED),
+    )
+    python_result: Final = response()
+    native_result: Final = response()
+    model: Final = str(kwargs.get("model", "anthropic/claude-sonnet-4-5"))
+    call_kwargs: Final = {name: value for name, value in kwargs.items() if name != "model"}
+
+    async def python(*call_args: object, **python_kwargs: object) -> AnthropicMessagesResponse:  # kwargs-ok: stub
+        return python_result
+
+    async def native(request: NativeCall) -> AnthropicMessagesResponse:
+        return native_result
+
+    monkeypatch.setattr(litellm, "callbacks", [])
+    for name, value in settings.items():
+        monkeypatch.setattr(litellm, name, value)
+    served_async: Final = await _ADISPATCH.arun(
+        (16, MESSAGES, model),
+        call_kwargs,
+        python=python,
+        binding=amessages_binding(native),
+        native=lambda hook, request, args, native_kwargs: hook(request),
+        rules=rules,
+    )
+    served_sync: Final = _DISPATCH.run(
+        (16, MESSAGES, model),
+        call_kwargs,
+        python=lambda *call_args, **python_kwargs: python_result,
+        binding=messages_binding(lambda request: native_result),
+        native=lambda hook, request, args, native_kwargs: hook(request),
+        rules=rules,
+    )
+    assert served_async is python_result
+    assert served_sync is native_result
+
+
+@pytest.mark.asyncio
+async def test_interceptor_feature_follows_the_resolved_provider() -> None:
+    rules: Final[Rules] = (
+        RouteRule(Route.MESSAGES, Rollout.PYTHON_ONLY, features=frozenset({Feature.MESSAGES_INTERCEPTOR})),
+        RouteRule(Route.MESSAGES, Rollout.RUST_REQUIRED),
+    )
+    native_result: Final = response()
+
+    async def python(*call_args: object, **call_kwargs: object) -> AnthropicMessagesResponse:  # kwargs-ok: stub
+        pytest.fail("anthropic serves the advisor tool itself")
+
+    async def native(request: NativeCall) -> AnthropicMessagesResponse:
+        return native_result
+
+    result: Final = await _ADISPATCH.arun(
+        (16, MESSAGES, "claude-sonnet-4-5"),
+        {"tools": [ADVISOR_TOOL], "custom_llm_provider": "anthropic"},
+        python=python,
+        binding=amessages_binding(native),
+        native=lambda hook, request, call_args, call_kwargs: hook(request),
+        rules=rules,
+    )
+    assert result is native_result

tests/test_litellm_rust/responses/data.py

@@ -69,5 +69,7 @@ async def call(server: RecordingServer, *, asynchronous: bool = True, **kwargs:
 
 
 SYNC_DEPLOYMENT_HOOK: Final = gap("native sync calls skip the async pre-call deployment hook")
-STREAMING: Final = gap("native Responses streaming declines to Python", raises=RuntimeError)
+STREAMING: Final = gap(
+    "native Responses streaming is unsupported, the catalog keeps it on Python", raises=litellm.APIConnectionError
+)
 BODY_HOOKS: Final = gap("native Responses does not run the handler body hooks")

tests/test_litellm_rust/responses/test_dispatch.py

@@ -5,20 +5,24 @@ from typing import Final, cast  # noqa: TID251  # narrows legacy callable signat
 import pytest
 
 import litellm
+from litellm.caching.caching import Cache
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.responses import dispatch as responses_dispatch
 from litellm.responses import main as python_responses
 from litellm.responses.dispatch import (
     _ADISPATCH,  # pyright: ignore[reportPrivateUsage]  # tests configured dispatch
     _DISPATCH,  # pyright: ignore[reportPrivateUsage]  # tests configured dispatch
 )
+from litellm.responses.utils import ResponsesAPIRequestUtils
 from litellm.rust_bridge.bindings import NativeBinding
-from litellm.rust_bridge.catalog import Route, RouteRule
+from litellm.rust_bridge.catalog import Feature, Route, RouteRule
 from litellm.rust_bridge.configuration import Rollout
 from litellm.rust_bridge.public_call import NativeCall
 from litellm.rust_bridge.responses.entrypoints import (
     NativeAresponses,
     NativeResponses,
 )
+from litellm.types.llms.anthropic import ANTHROPIC_TOOL_SEARCH_TOOL_TYPES
 from litellm.types.llms.openai import ResponsesAPIResponse
 
 INPUT: Final = [{"role": "user", "content": "hi"}]
@@ -235,3 +239,167 @@ def test_positional_parameters_remain_available_to_native_projection() -> None:
     assert request.resolved["include"] is include
     assert request.resolved["instructions"] == "Be brief"
     assert request.resolved["max_output_tokens"] == 16
+
+
+class _Agentic(CustomLogger):
+    async def async_should_run_agentic_loop(self, *args: object, **kwargs: object) -> tuple[bool, dict]:  # pyright: ignore[reportIncompatibleMethodOverride, reportMissingTypeArgument]  # test double only needs the override
+        return False, {}
+
+
+class _Deployment(CustomLogger):
+    async def async_pre_call_deployment_hook(self, *args: object, **kwargs: object) -> dict[str, object] | None:  # pyright: ignore[reportIncompatibleMethodOverride]  # test double only needs the override
+        return None
+
+
+ENCODED_ID: Final = ResponsesAPIRequestUtils._build_encrypted_item_id("deployment-1", "rs_1")  # pyright: ignore[reportPrivateUsage]  # the encoder litellm stamps on earlier turns
+
+
+@pytest.mark.parametrize(
+    ("feature", "kwargs", "settings"),
+    (
+        (Feature.MOCK_RESPONSE, {"mock_response": "hi"}, {}),
+        (Feature.PROXY_MCP_TOOLS, {"tools": [{"type": "mcp", "server_label": "g", "server_url": "litellm_proxy"}]}, {}),
+        (Feature.CALLER_CLIENT, {"client": object()}, {}),
+        pytest.param(Feature.CALLER_CLIENT, {"shared_session": object()}, {}, id="shared-session"),
+        pytest.param(Feature.EXTRA_BODY, {"extra_body": {"stream": True}}, {}, id="extra-body-stream"),
+        pytest.param(
+            Feature.PROVIDER_AFFINITY,
+            {"provider_affinity_header": "X-Session-ID", "session_id": "session-1"},
+            {},
+            id="provider-affinity",
+        ),
+        pytest.param(
+            Feature.INPUT_REWRITE,
+            {"input": [{"type": "reasoning", "id": "rs_1", "summary": list[object](), "status": None}]},
+            {},
+            id="input-rewrite",
+        ),
+        (Feature.TIMEOUT_VARIANTS, {"request_timeout": 5}, {}),
+        (Feature.AGENTIC_LOOP_HOOK, {}, {"callbacks": [_Agentic()]}),
+        (Feature.TLS_POLICY, {}, {"ssl_security_level": "DEFAULT@SECLEVEL=1"}),
+        (Feature.DROP_PARAMS, {}, {"drop_params": True}),
+        (Feature.STREAMING, {"stream": True}, {}),
+        (Feature.PROMPT_MANAGEMENT, {"prompt_id": "welcome"}, {}),
+        (Feature.TEXT_FORMAT, {"text_format": dict}, {}),
+        (Feature.CHAT_COMPLETIONS_BRIDGE, {"use_chat_completions_api": True}, {}),
+        (Feature.CHAT_COMPLETIONS_BRIDGE, {"model": "openai/chat_completions/gpt-5"}, {}),
+        (Feature.RETRIES_FALLBACKS, {"num_retries": 2}, {}),
+        (Feature.DEPLOYMENT_RESPONSE_ID, {"litellm_metadata": {"model_info": {"id": "deployment-1"}}}, {}),
+        (Feature.ENCODED_ITEM_IDS, {"input": [{"type": "reasoning", "id": ENCODED_ID, "summary": []}]}, {}),
+        (Feature.MODIFY_PARAMS, {}, {"modify_params": True}),
+        (Feature.RESPONSE_CACHE, {}, {"cache": Cache()}),
+        (
+            Feature.MANAGED_RESOURCE_IDS,
+            {
+                "input": [{"role": "user", "content": [{"type": "input_file", "file_id": "litellm-file"}]}],
+                "model_file_id_mapping": {"litellm-file": {"deployment-1": "file-abc"}},
+                "model_info": {"id": "deployment-1"},
+            },
+            {},
+        ),
+        (Feature.CACHE_CONTROL_MARKS, {"tools": [{"type": "function", "name": "f", "cache_control": {}}]}, {}),
+        (
+            Feature.ANTHROPIC_TOOL_SEARCH,
+            {"tools": [{"type": sorted(ANTHROPIC_TOOL_SEARCH_TOOL_TYPES)[0], "name": "s"}]},
+            {},
+        ),
+        (Feature.TOOL_SCHEMA_REPAIR, {"tools": [{"type": "function", "name": "f", "parameters": "{}"}]}, {}),
+        (Feature.MODEL_PARAM_GATING, {"max_output_tokens": 1}, {}),
+        (Feature.CLIENT_HEADERS, {"headers": {"x-tenant": "a"}}, {}),
+        (Feature.DEPLOYMENT_HOOK, {}, {"callbacks": [_Deployment()]}),
+        (
+            Feature.WORKLOAD_IDENTITY,
+            {
+                "openai_identity_provider_id": "idp-1",
+                "openai_service_account_id": "sa-1",
+                "openai_identity_token_file": "/var/run/token",
+            },
+            {},
+        ),
+    ),
+)
+def test_each_detected_feature_reaches_its_rule(
+    monkeypatch: pytest.MonkeyPatch, feature: Feature, kwargs: Mapping[str, object], settings: Mapping[str, object]
+) -> None:
+    rules: Final = (
+        RouteRule(Route.RESPONSES, Rollout.PYTHON_ONLY, features=frozenset({feature})),
+        RouteRule(Route.RESPONSES, Rollout.RUST_REQUIRED),
+    )
+    python_result: Final = _response()
+    native_result: Final = _response()
+
+    def served(call_kwargs: Mapping[str, object]) -> object:
+        return _DISPATCH.run(
+            (),
+            {"model": "gpt-5", "input": INPUT, **call_kwargs},
+            python=lambda *call_args, **python_kwargs: python_result,
+            binding=responses_binding(lambda request: native_result),
+            native=lambda hook, request, call_args, native_kwargs: hook(request),
+            rules=rules,
+        )
+
+    monkeypatch.setattr(litellm, "callbacks", [])
+    monkeypatch.setattr(litellm, "drop_params", False)
+    monkeypatch.setattr(litellm, "modify_params", False)
+    monkeypatch.setattr(litellm, "ssl_security_level", None)
+    monkeypatch.setattr(litellm, "cache", None)
+    for setting in ("api_key", "openai_key", "api_base", "num_retries"):
+        monkeypatch.setattr(litellm, setting, None)
+    for name in ("SSL_SECURITY_LEVEL", "SSL_ECDH_CURVE", "OPENAI_API_KEY"):
+        monkeypatch.delenv(name, raising=False)
+    assert served({}) is native_result
+    for name, value in settings.items():
+        monkeypatch.setattr(litellm, name, value)
+    assert served(kwargs) is python_result
+
+
+@pytest.mark.asyncio
+async def test_deployment_hook_keeps_only_sync_calls_on_python(monkeypatch: pytest.MonkeyPatch) -> None:
+    rules: Final = (
+        RouteRule(Route.RESPONSES, Rollout.PYTHON_ONLY, features=frozenset({Feature.DEPLOYMENT_HOOK})),
+        RouteRule(Route.RESPONSES, Rollout.RUST_REQUIRED),
+    )
+    native_result: Final = _response()
+
+    async def python(*call_args: object, **call_kwargs: object) -> ResponsesAPIResponse:  # kwargs-ok: stub
+        pytest.fail("the async handler leaves the deployment hook to the shared call wrapper")
+
+    async def native(request: NativeCall) -> ResponsesAPIResponse:
+        return native_result
+
+    monkeypatch.setattr(litellm, "callbacks", [_Deployment()])
+    result: Final = await _ADISPATCH.arun(
+        (),
+        {"model": "gpt-5", "input": INPUT},
+        python=python,
+        binding=aresponses_binding(native),
+        native=lambda hook, request, call_args, call_kwargs: hook(request),
+        rules=rules,
+    )
+    assert result is native_result
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "native"),
+    (
+        ({"model": "gpt-5"}, True),
+        ({"model": "openai/gpt-5"}, True),
+        ({"model": "gpt-5", "custom_llm_provider": "openai"}, True),
+        ({"model": "anthropic/claude-sonnet-4-5"}, False),
+        ({"model": "claude-sonnet-4-5"}, False),
+        ({"model": "gpt-5", "custom_llm_provider": "azure"}, False),
+    ),
+)
+def test_provider_rule_sees_the_resolved_provider(kwargs: Mapping[str, object], native: bool) -> None:
+    python_result: Final = _response()
+    native_result: Final = _response()
+
+    result: Final = _DISPATCH.run(
+        (),
+        {"input": INPUT, **kwargs},
+        python=lambda *call_args, **call_kwargs: python_result,
+        binding=responses_binding(lambda request: native_result),
+        native=lambda hook, request, call_args, call_kwargs: hook(request),
+        rules=(RouteRule(Route.RESPONSES, Rollout.RUST_REQUIRED, providers=frozenset({"openai"})),),
+    )
+    assert result is (native_result if native else python_result)

tests/test_litellm_rust/responses/test_failures.py

@@ -56,28 +56,3 @@ async def test_provider_failure_is_terminal_and_shared_with_callbacks(
     assert len(failure) == 1
     assert _OBJECT.validate_python(failure[0].kwargs)["exception"] is caught.value
     assert not any("success" in name for name in recorder.names)
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize(
-    "options",
-    (
-        {"stream": True},
-        {"extra_body": {"provider_option": True}},
-        {"mock_response": "mock"},
-        {"num_retries": 1},
-        {"use_chat_completions_api": True},
-        {"model_list": []},
-    ),
-)
-async def test_unsupported_requests_fail_before_callbacks_and_provider_io(
-    server: RecordingServer, options: dict[str, object]
-) -> None:
-    recorder: Final = RecordingLogger()
-    server.expected_requests = 0
-
-    with pytest.raises(RuntimeError, match="declined the request"):
-        await call(server, callbacks=[recorder], **options)
-
-    assert not server.requests
-    assert not recorder.events

tests/test_litellm_rust/test_dispatch.py

@@ -1,6 +1,6 @@
 from collections.abc import AsyncGenerator, Awaitable, Callable, Iterator, Mapping
 from dataclasses import dataclass
-from typing import Final, TypeAlias
+from typing import Final, NoReturn, TypeAlias
 
 import pytest
 
@@ -245,6 +245,50 @@ async def dispatch_without_python(
     return await dispatch.arun(("model",), {"page": 1}, python=NO_PYTHON, binding=bound, native=native, rules=rules)
 
 
+def failing_context(request: Request) -> RouteContext:
+    raise RuntimeError("detector exploded")
+
+
+def failing_dispatch() -> PublicDispatch[Request]:
+    return PublicDispatch(
+        route=Route.CHAT_COMPLETIONS, request=lambda args, kwargs: Request(model="model"), context=failing_context
+    )
+
+
+def reject_native(hook: object, request: Request, args: tuple[object, ...], kwargs: Mapping[str, object]) -> NoReturn:
+    pytest.fail("a failed context build must not reach native")
+
+
+def test_a_detector_error_keeps_the_call_on_python() -> None:
+    expected: Final = object()
+    result: Final = failing_dispatch().run(
+        ("model",),
+        {},
+        python=lambda *args, **kwargs: expected,
+        binding=binding(),
+        native=reject_native,
+        rules=(RouteRule(Route.CHAT_COMPLETIONS, Rollout.RUST_REQUIRED),),
+    )
+    assert result is expected
+
+
+async def test_a_detector_error_keeps_the_async_call_on_python() -> None:
+    expected: Final = object()
+
+    async def python(*args: object, **kwargs: object) -> object:  # kwargs-ok: stub
+        return expected
+
+    result: Final = await failing_dispatch().arun(
+        ("model",),
+        {},
+        python=python,
+        binding=binding(),
+        native=reject_native,
+        rules=(RouteRule(Route.CHAT_COMPLETIONS, Rollout.RUST_REQUIRED),),
+    )
+    assert result is expected
+
+
 def ocr_dispatch(request: Request | None) -> PublicDispatch[Request]:
     return PublicDispatch(
         route=Route.OCR,

tests/unit/rust_bridge/test_detectors.py

@@ -0,0 +1,456 @@
+from collections.abc import Callable, Mapping
+from copy import deepcopy
+from datetime import datetime
+from functools import reduce
+from typing import Final
+
+import httpx
+import pytest
+
+import litellm
+from litellm.caching.caching import Cache
+from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.litellm_logging import Logging
+from litellm.responses.utils import ResponsesAPIRequestUtils
+from litellm.rust_bridge import detectors
+from litellm.rust_bridge.public_call import NativeCall
+from litellm.types.llms.anthropic import ANTHROPIC_TOOL_SEARCH_TOOL_TYPES
+
+GATEWAY_TOOL: Final = {"type": "mcp", "server_label": "gateway", "server_url": "litellm_proxy"}
+REMOTE_TOOL: Final = {"type": "mcp", "server_label": "docs", "server_url": "https://mcp.example.com/sse"}
+ENCODED_ID: Final = ResponsesAPIRequestUtils._build_encrypted_item_id("deployment-1", "rs_1")  # pyright: ignore[reportPrivateUsage]  # the encoder litellm stamps on earlier turns
+WRAPPED_CONTENT: Final = ResponsesAPIRequestUtils.wrap_encrypted_content_with_model_id("opaque", "deployment-1")
+
+
+MAPPED_FILE: Final = {
+    "model_file_id_mapping": {"litellm-file": {"deployment-1": "file-abc"}},
+    "model_info": {"id": "deployment-1"},
+}
+MAPPED_FILE_INPUT: Final = [{"role": "user", "content": [{"type": "input_file", "file_id": "litellm-file"}]}]
+CACHED_TEXT: Final = {"type": "input_text", "text": "hi", "cache_control": {"type": "ephemeral"}}
+THINKING_TURN: Final = {"role": "assistant", "content": [{"type": "thinking", "thinking": "x", "signature": "s"}]}
+REDACTED: Final = {"type": "redacted_thinking", "data": "x"}
+TOOL_SEARCH_TYPE: Final = sorted(ANTHROPIC_TOOL_SEARCH_TOOL_TYPES)[0]
+
+
+def function_tool(parameters: object) -> Mapping[str, object]:
+    return {"type": "function", "name": "f", "parameters": parameters}
+
+
+def call(**kwargs: object) -> NativeCall:
+    return NativeCall(args=(), kwargs=kwargs, base={"model": "gpt-5"})
+
+
+def _wrap_nested_level(value: object, depth: int) -> object:
+    return {"nested": value} if depth % 2 else [value]
+
+
+@pytest.mark.parametrize(
+    ("detector", "kwargs", "expected"),
+    (
+        (detectors.mock_response, {"mock_response": "hi"}, True),
+        (detectors.mock_response, {"mock_response": None}, False),
+        (detectors.mock_response, {"mock_tool_calls": [{"id": "call_1"}]}, False),
+        (detectors.chat_mock_response, {"mock_response": "hi"}, True),
+        (detectors.chat_mock_response, {"mock_tool_calls": [{"id": "call_1"}]}, True),
+        (detectors.chat_mock_response, {"mock_timeout": True}, True),
+        (detectors.chat_mock_response, {"mock_tool_calls": [], "mock_timeout": False}, False),
+        (detectors.client_headers, {"headers": {"x-tenant": "a"}}, True),
+        (detectors.client_headers, {"headers": {}}, False),
+        (detectors.client_headers, {"headers": None}, False),
+        (detectors.model_list, {"model_list": [{"model_name": "gpt-5", "litellm_params": {"model": "gpt-5"}}]}, True),
+        (detectors.model_list, {"model_list": None}, False),
+        (detectors.chat_retries_fallbacks, {"context_window_fallback_dict": {"gpt-5": "gpt-5-mini"}}, True),
+        (detectors.chat_retries_fallbacks, {"fallbacks": ["gpt-5-mini"]}, True),
+        (detectors.retries_fallbacks, {"context_window_fallback_dict": {"gpt-5": "gpt-5-mini"}}, False),
+        (detectors.prompt_management, {"model": "litellm_agent/gpt-5"}, True),
+        (detectors.prompt_management, {"model": "gpt-5"}, False),
+        (detectors.caller_client, {"client": object()}, True),
+        (detectors.caller_client, {"shared_session": object()}, True),
+        (detectors.caller_client, {"client": None, "shared_session": None}, False),
+        (detectors.caller_client, {}, False),
+        (detectors.extra_body, {"extra_body": {"stream": True}}, True),
+        (detectors.extra_body, {"extra_body": {"stream": False}}, True),
+        (detectors.extra_body, {"extra_body": {}}, False),
+        (detectors.extra_body, {"extra_body": None}, False),
+        (detectors.extra_body, {}, False),
+        (detectors.provider_affinity, {"provider_affinity_header": "X-Session-ID", "session_id": "session-1"}, True),
+        (detectors.provider_affinity, {"provider_affinity_header": "X-Session-ID"}, True),
+        (
+            detectors.provider_affinity,
+            {"provider_affinity_header": "X-Session-ID", "extra_headers": {"x-session-id": "session-1"}},
+            True,
+        ),
+        (detectors.provider_affinity, {"provider_affinity_header": "Authorization"}, True),
+        (detectors.provider_affinity, {"provider_affinity_header": "invalid header"}, True),
+        (detectors.provider_affinity, {"provider_affinity_header": ""}, True),
+        (detectors.provider_affinity, {"provider_affinity_header": None, "session_id": "session-1"}, False),
+        (detectors.provider_affinity, {}, False),
+        (detectors.streaming, {"stream": True}, True),
+        (detectors.streaming, {"stream": False}, False),
+        (detectors.timeout_variants, {"request_timeout": 5}, True),
+        (detectors.timeout_variants, {"stream": True, "stream_timeout": 5}, True),
+        (detectors.timeout_variants, {"stream_timeout": 5}, False),
+        (detectors.timeout_variants, {"timeout": httpx.Timeout(60, connect=2)}, True),
+        (detectors.timeout_variants, {"timeout": 60}, False),
+        (detectors.proxy_mcp_tools, {"tools": [GATEWAY_TOOL]}, True),
+        (detectors.proxy_mcp_tools, {"tools": [REMOTE_TOOL]}, False),
+        (detectors.proxy_mcp_tools, {"tools": None}, False),
+        (detectors.prompt_management, {"prompt_id": "welcome"}, True),
+        (detectors.prompt_management, {"prompt_variables": {"name": "x"}}, False),
+        (detectors.text_format, {"text_format": dict}, True),
+        (detectors.text_format, {}, False),
+        (detectors.chat_completions_bridge, {"use_chat_completions_api": True}, True),
+        (detectors.chat_completions_bridge, {"use_chat_completions_api": False}, False),
+        (detectors.chat_completions_bridge, {"model": "openai/chat_completions/gpt-5"}, True),
+        (detectors.chat_completions_bridge, {"model": "openai/gpt-5"}, False),
+        (detectors.retries_fallbacks, {"num_retries": 0}, True),
+        (detectors.retries_fallbacks, {"max_retries": 2}, True),
+        (detectors.retries_fallbacks, {"fallbacks": ["gpt-5-mini"]}, True),
+        (detectors.retries_fallbacks, {"retry_policy": {"TimeoutErrorRetries": 2}}, True),
+        (detectors.retries_fallbacks, {}, False),
+        (detectors.prompt_management, {"knowledge_bases": ["kb-1"]}, True),
+        (detectors.prompt_management, {"vector_store_ids": []}, False),
+        (detectors.managed_resource_ids, {"input": MAPPED_FILE_INPUT, **MAPPED_FILE}, True),
+        (detectors.managed_resource_ids, {"input": MAPPED_FILE_INPUT}, False),
+        (detectors.managed_resource_ids, {"input": "hi"}, False),
+        (detectors.cache_control_marks, {"input": [{"role": "user", "content": [CACHED_TEXT]}]}, True),
+        (detectors.cache_control_marks, {"tools": [{"type": "function", "name": "f", "cache_control": {}}]}, True),
+        (detectors.cache_control_marks, {"input": [{"role": "user", "content": [{"type": "input_text"}]}]}, False),
+        (detectors.anthropic_tool_search, {"tools": [{"type": TOOL_SEARCH_TYPE, "name": "tool_search"}]}, True),
+        (detectors.anthropic_tool_search, {"tools": [{"type": "function", "name": "f"}]}, False),
+        (detectors.tool_schema_repair, {"tools": [function_tool('{"type": "object"}')]}, True),
+        (detectors.tool_schema_repair, {"tools": [function_tool({"type": "string", "pattern": r"\p{L}"})]}, True),
+        (detectors.tool_schema_repair, {"tools": [function_tool({"type": "object"})]}, False),
+        (detectors.tool_schema_repair, {}, False),
+        (detectors.model_param_gating, {"max_output_tokens": 1}, True),
+        (detectors.model_param_gating, {"max_output_tokens": 1024}, False),
+        (detectors.cache_control_injection, {"enable_prompt_caching": True}, True),
+        (detectors.cache_control_injection, {"cache_control_injection_points": [{"location": "message"}]}, True),
+        (detectors.cache_control_injection, {"cache_control_injection_points": []}, False),
+        (detectors.thinking_signature_retry, {"messages": [THINKING_TURN]}, True),
+        (detectors.thinking_signature_retry, {"messages": [{"role": "assistant", "content": [REDACTED]}]}, True),
+        (detectors.thinking_signature_retry, {"messages": [{"role": "assistant", "content": "done"}]}, False),
+        (detectors.thinking_signature_retry, {"messages": [{"role": "user", "content": [{"type": "text"}]}]}, False),
+        (detectors.drop_params, {"drop_params": True}, True),
+        (detectors.drop_params, {"drop_params": False}, False),
+        (detectors.drop_params, {"drop_params": " True "}, True),
+        (detectors.drop_params, {"drop_params": "false"}, False),
+        (detectors.drop_params, {"drop_params": "maybe"}, False),
+        (detectors.drop_params, {"additional_drop_params": ["metadata.user_id"]}, True),
+        (detectors.drop_params, {"additional_drop_params": []}, False),
+        (detectors.deployment_response_id, {"litellm_metadata": {"model_info": {"id": "deployment-1"}}}, True),
+        (detectors.deployment_response_id, {"litellm_metadata": {"model_info": {}}}, False),
+        (detectors.deployment_response_id, {"litellm_metadata": None}, False),
+        (detectors.encoded_item_ids, {"input": [{"type": "reasoning", "id": ENCODED_ID}]}, True),
+        (detectors.encoded_item_ids, {"input": [{"type": "reasoning", "encrypted_content": WRAPPED_CONTENT}]}, True),
+        (detectors.encoded_item_ids, {"input": [{"type": "reasoning", "id": "rs_1", "encrypted_content": "x"}]}, False),
+        (detectors.encoded_item_ids, {"input": "hi"}, False),
+    ),
+)
+def test_request_detectors(
+    detector: Callable[[NativeCall], bool], kwargs: Mapping[str, object], expected: bool
+) -> None:
+    assert detector(call(**kwargs)) is expected
+
+
+def test_cache_control_marks_handles_deeply_nested_input() -> None:
+    nested_input: Final = reduce(_wrap_nested_level, range(5_000), {"cache_control": {}})
+    assert detectors.cache_control_marks(call(input=nested_input))
+
+
+@pytest.mark.parametrize(
+    ("input_value", "expected"),
+    (
+        pytest.param(
+            [{"type": "reasoning", "id": "rs_00000000-0000-4000-8000-000000000000", "summary": list[object]()}],
+            True,
+            id="bridge-minted-reasoning",
+        ),
+        pytest.param(
+            [{"type": "reasoning", "encrypted_content": '[{"type":"thinking","thinking":"x","signature":"s"}]'}],
+            True,
+            id="bridge-thinking-blocks",
+        ),
+        pytest.param(
+            [{"type": "function_call", "id": "foreign-id", "call_id": "call_1", "name": "f", "arguments": "{}"}],
+            True,
+            id="foreign-tool-call-id",
+        ),
+        pytest.param(
+            [{"type": "reasoning", "id": "rs_1", "summary": list[object](), "status": None}],
+            True,
+            id="null-reasoning-status",
+        ),
+        pytest.param(
+            [{"type": "function_call", "id": "fc_1", "call_id": "call_1", "name": "f", "arguments": "{}"}],
+            False,
+            id="native-tool-call-id",
+        ),
+        pytest.param(
+            [{"type": "custom_tool_call", "call_id": "call_1", "name": "f", "input": "x", "namespace": "ns"}],
+            True,
+            id="custom-tool-call-namespace",
+        ),
+        pytest.param(
+            [{"type": "custom_tool_call", "call_id": "call_1", "name": "f", "input": "x"}],
+            False,
+            id="custom-tool-call",
+        ),
+        pytest.param([{"type": "reasoning", "id": "rs_1", "summary": list[object]()}], False, id="native-reasoning"),
+        pytest.param([{"role": "user", "content": "hi"}], False, id="message"),
+        pytest.param("hi", False, id="string"),
+        pytest.param([], False, id="empty"),
+        pytest.param(None, False, id="absent"),
+    ),
+)
+def test_input_rewrite_preserves_caller_input(input_value: object, expected: bool) -> None:
+    original: Final = deepcopy(input_value)
+    assert detectors.input_rewrite(call(input=input_value)) is expected
+    assert input_value == original
+
+
+@pytest.mark.parametrize(
+    ("detector", "setting", "value", "kwargs"),
+    (
+        (detectors.drop_params, "drop_params", True, {}),
+        (detectors.modify_params, "modify_params", True, {}),
+        (detectors.retries_fallbacks, "num_retries", 3, {}),
+        (detectors.cache_control_injection, "enable_anthropic_prompt_caching", True, {}),
+        (detectors.stream_cost_injection, "include_cost_in_streaming_usage", True, {"stream": True}),
+        (detectors.stream_cost_injection, "include_cost_in_streaming_usage", "true", {"stream": True}),
+        (detectors.chat_retries_fallbacks, "model_fallbacks", [{"gpt-5": ["gpt-5-mini"]}], {}),
+        (detectors.chat_retries_fallbacks, "num_retries", 3, {}),
+        (detectors.model_alias_map, "model_alias_map", {"gpt-5": "openai/gpt-5"}, {}),
+    ),
+)
+def test_global_settings_are_detected(
+    monkeypatch: pytest.MonkeyPatch,
+    detector: Callable[[NativeCall], bool],
+    setting: str,
+    value: object,
+    kwargs: Mapping[str, object],
+) -> None:
+    for name in ("drop_params", "modify_params", "enable_anthropic_prompt_caching", "include_cost_in_streaming_usage"):
+        monkeypatch.setattr(litellm, name, False)
+    for name in ("num_retries", "model_fallbacks"):
+        monkeypatch.setattr(litellm, name, None)
+    monkeypatch.setattr(litellm, "model_alias_map", {})
+    assert not detector(call(**kwargs))
+    monkeypatch.setattr(litellm, setting, value)
+    assert detector(call(**kwargs))
+
+
+def test_stream_cost_injection_needs_a_stream(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", True)
+    assert not detectors.stream_cost_injection(call())
+
+
+def test_model_alias_map_applies_only_to_the_requested_model(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(litellm, "model_alias_map", {"fast": "openai/gpt-5-mini"})
+    assert detectors.model_alias_map(call(model="fast"))
+    assert not detectors.model_alias_map(call())
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "expected"),
+    (
+        pytest.param({"temperature": 0.2}, True, id="gpt-5-temperature"),
+        pytest.param({"max_tokens": 16}, False, id="gpt-5-max-tokens"),
+        pytest.param({"model": "gpt-4o", "temperature": 0.2}, False, id="gpt-4o-temperature"),
+        pytest.param({"model": "anthropic/claude-sonnet-4-5", "logit_bias": {"1": 1}}, True, id="unsupported-param"),
+        pytest.param({"model": "anthropic/claude-sonnet-4-5", "n": 1}, False, id="langchain-default-n"),
+        pytest.param({"model": "anthropic/claude-sonnet-4-5", "n": 2}, True, id="unsupported-n"),
+        pytest.param({"model": "anthropic/claude-sonnet-4-5", "max_tokens": 16, "stream": True}, False, id="supported"),
+        pytest.param({"model": "unknown-model", "temperature": 0.2}, False, id="unresolved-provider"),
+        pytest.param({}, False, id="no-params"),
+    ),
+)
+def test_chat_model_param_gating_follows_the_provider_param_rules(
+    monkeypatch: pytest.MonkeyPatch, kwargs: Mapping[str, object], expected: bool
+) -> None:
+    monkeypatch.setattr(litellm, "drop_params", False)
+    assert detectors.chat_model_param_gating(call(**kwargs)) is expected
+
+
+def test_openai_workload_identity_applies_only_without_a_static_key(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("OPENAI_API_KEY", raising=False)
+    monkeypatch.setattr(litellm, "api_key", None)
+    monkeypatch.setattr(litellm, "openai_key", None)
+    monkeypatch.setattr(litellm, "api_base", None)
+    identity: Final = {
+        "openai_identity_provider_id": "idp-1",
+        "openai_service_account_id": "sa-1",
+        "openai_identity_token_file": "/var/run/token",
+    }
+    assert detectors.openai_workload_identity(call(**identity))
+    assert not detectors.openai_workload_identity(call(**identity, api_key="sk-static"))
+    assert not detectors.openai_workload_identity(call())
+
+
+def test_anthropic_workload_identity_applies_only_without_a_static_key(monkeypatch: pytest.MonkeyPatch) -> None:
+    for name in (
+        "ANTHROPIC_API_KEY",
+        "ANTHROPIC_AUTH_TOKEN",
+        "ANTHROPIC_FEDERATION_RULE_ID",
+        "ANTHROPIC_ORGANIZATION_ID",
+    ):
+        monkeypatch.delenv(name, raising=False)
+    federation: Final = {"anthropic_federation_rule_id": "rule-1", "anthropic_organization_id": "org-1"}
+    assert detectors.anthropic_workload_identity(call(**federation))
+    assert not detectors.anthropic_workload_identity(call(**federation, api_key="sk-ant-static"))
+    assert not detectors.anthropic_workload_identity(call(anthropic_federation_rule_id="rule-1"))
+
+
+class _Logger(CustomLogger):
+    pass
+
+
+class _Rewrite(CustomLogger):
+    async def async_pre_request_hook(self, model: str, messages: list, kwargs: dict) -> dict | None:  # pyright: ignore[reportMissingTypeArgument]  # mirrors the CustomLogger signature
+        return kwargs
+
+
+class _InheritedRewrite(_Rewrite):
+    pass
+
+
+class _Agentic(CustomLogger):
+    async def async_should_run_agentic_loop(self, *args: object, **kwargs: object) -> tuple[bool, dict]:  # pyright: ignore[reportIncompatibleMethodOverride, reportMissingTypeArgument]  # test double only needs the override
+        return False, {}
+
+
+class _ChatAgentic(CustomLogger):
+    async def async_should_run_chat_completion_agentic_loop(
+        self, *args: object, **kwargs: object
+    ) -> tuple[bool, dict[str, object]]:  # pyright: ignore[reportIncompatibleMethodOverride]  # test double only needs the override
+        return False, {}
+
+
+class _Deployment(CustomLogger):
+    async def async_pre_call_deployment_hook(self, *args: object, **kwargs: object) -> dict[str, object] | None:  # pyright: ignore[reportIncompatibleMethodOverride]  # test double only needs the override
+        return None
+
+
+@pytest.mark.parametrize(
+    ("detector", "callbacks", "expected"),
+    (
+        (detectors.pre_request_hook, [_Rewrite()], True),
+        (detectors.pre_request_hook, [_InheritedRewrite()], True),
+        (detectors.pre_request_hook, [_Logger(), _Agentic()], False),
+        (detectors.pre_request_hook, ["not-a-logger"], False),
+        (detectors.agentic_loop_hook, [_Agentic()], True),
+        (detectors.agentic_loop_hook, [_Logger(), _Rewrite()], False),
+        (detectors.agentic_loop_hook, [_ChatAgentic()], False),
+        (detectors.chat_agentic_loop_hook, [_ChatAgentic()], True),
+        (detectors.chat_agentic_loop_hook, [_Agentic()], True),
+        (detectors.chat_agentic_loop_hook, [_Logger(), _Deployment()], False),
+        (detectors.deployment_hook, [_Deployment()], True),
+        (detectors.deployment_hook, [_Logger(), _Agentic(), _Rewrite()], False),
+    ),
+)
+def test_callback_detectors_need_an_override(
+    monkeypatch: pytest.MonkeyPatch, detector: Callable[[NativeCall], bool], callbacks: list[object], expected: bool
+) -> None:
+    monkeypatch.setattr(litellm, "callbacks", callbacks)
+    assert detector(call()) is expected
+
+
+def _logging_with(callback: CustomLogger) -> Logging:
+    logging_obj: Final = Logging(
+        model="claude-sonnet-4-5",
+        messages=[],
+        stream=False,
+        call_type="anthropic_messages",
+        start_time=datetime.now(),
+        litellm_call_id="call-1",
+        function_id="function-1",
+    )
+    logging_obj.add_dynamic_callback(callback)
+    return logging_obj
+
+
+@pytest.mark.parametrize(
+    ("detector", "kwargs", "expected"),
+    (
+        (detectors.agentic_loop_hook, {"success_callback": [_Agentic()]}, True),
+        (detectors.agentic_loop_hook, {"litellm_logging_obj": _logging_with(_Agentic())}, True),
+        (detectors.agentic_loop_hook, {"success_callback": [_Logger()]}, False),
+        (detectors.pre_request_hook, {"success_callback": [_Rewrite()]}, False),
+        (detectors.chat_agentic_loop_hook, {"litellm_logging_obj": _logging_with(_ChatAgentic())}, True),
+        (detectors.deployment_hook, {"success_callback": [_Deployment()]}, True),
+    ),
+    ids=(
+        "success-callback",
+        "logging-object",
+        "plain-logger",
+        "pre-request-reads-globals-only",
+        "chat-agentic-logging-object",
+        "deployment-success-callback",
+    ),
+)
+def test_callback_detectors_read_the_request_callbacks_the_handler_reads(
+    monkeypatch: pytest.MonkeyPatch,
+    detector: Callable[[NativeCall], bool],
+    kwargs: Mapping[str, object],
+    expected: bool,
+) -> None:
+    monkeypatch.setattr(litellm, "callbacks", [])
+    assert detector(call(**kwargs)) is expected
+
+
+def test_websearch_interception_needs_the_interception_logger(monkeypatch: pytest.MonkeyPatch) -> None:
+    from litellm.integrations.websearch_interception.handler import WebSearchInterceptionLogger
+
+    monkeypatch.setattr(litellm, "callbacks", [_Rewrite()])
+    assert not detectors.websearch_interception(call())
+    monkeypatch.setattr(litellm, "callbacks", [WebSearchInterceptionLogger()])
+    assert detectors.websearch_interception(call())
+
+
+@pytest.mark.parametrize(
+    ("cache", "kwargs", "expected"),
+    (
+        (None, {}, False),
+        (Cache(), {}, True),
+        (Cache(), {"cache": {"ttl": 60}}, True),
+        (Cache(), {"cache": {"no-cache": True}}, False),
+    ),
+)
+def test_response_cache_is_skipped_only_by_no_cache(
+    monkeypatch: pytest.MonkeyPatch, cache: Cache | None, kwargs: Mapping[str, object], expected: bool
+) -> None:
+    monkeypatch.setattr(litellm, "cache", cache)
+    assert detectors.response_cache(call(**kwargs)) is expected
+
+
+@pytest.mark.parametrize(
+    ("environment", "setting", "expected"),
+    ((None, None, False), ("DEFAULT@SECLEVEL=1", None, True), (None, "DEFAULT@SECLEVEL=1", True)),
+)
+def test_tls_policy_reads_the_security_level_environment_then_the_setting(
+    monkeypatch: pytest.MonkeyPatch, environment: str | None, setting: str | None, expected: bool
+) -> None:
+    monkeypatch.delenv("SSL_SECURITY_LEVEL", raising=False)
+    monkeypatch.delenv("SSL_ECDH_CURVE", raising=False)
+    monkeypatch.setattr(litellm, "ssl_ecdh_curve", None)
+    if environment is not None:
+        monkeypatch.setenv("SSL_SECURITY_LEVEL", environment)
+    monkeypatch.setattr(litellm, "ssl_security_level", setting)
+    assert detectors.tls_policy(call()) is expected
+
+
+@pytest.mark.parametrize(
+    ("environment", "setting", "expected"),
+    ((None, "X25519", False), (None, " P-256 ", False), ("secp521r1", None, True), (None, "secp521r1", True)),
+)
+def test_tls_policy_flags_only_curves_native_tls_lacks(
+    monkeypatch: pytest.MonkeyPatch, environment: str | None, setting: str | None, expected: bool
+) -> None:
+    monkeypatch.delenv("SSL_SECURITY_LEVEL", raising=False)
+    monkeypatch.delenv("SSL_ECDH_CURVE", raising=False)
+    monkeypatch.setattr(litellm, "ssl_security_level", None)
+    if environment is not None:
+        monkeypatch.setenv("SSL_ECDH_CURVE", environment)
+    monkeypatch.setattr(litellm, "ssl_ecdh_curve", setting)
+    assert detectors.tls_policy(call()) is expected