feat: shadow eval samples /v1/messages and /v1/responses traffic (#36830)

basedpyright-code-budget.json

@@ -57,7 +57,7 @@
     "limit": 5707
   },
   "reportMissingTypeArgument": {
-    "limit": 15642
+    "limit": 15640
   },
   "reportMissingTypeStubs": {
     "limit": 40
@@ -108,7 +108,7 @@
     "limit": 39237
   },
   "reportUnknownParameterType": {
-    "limit": 19969
+    "limit": 19967
   },
   "reportUnknownVariableType": {
     "limit": 30881

litellm/integrations/shadow_eval_logger.py

@@ -1,5 +1,6 @@
-"""Shadow Eval Logger: samples a shadowed key's successful chat requests, duplicates each
-against the job's other arm in a detached task (the auto-router for a forward job, the
+"""Shadow Eval Logger: samples a shadowed key's successful LLM requests (chat completions,
+Anthropic Messages, and Responses API surfaces, each normalized to chat shape), duplicates
+each against the job's other arm in a detached task (the auto-router for a forward job, the
 fixed baseline model for a reverse one), blind-judges real vs shadow, and appends one
 ``LiteLLM_ShadowEvalAttempt`` row (verdict or error) as the feature's only hot-path write.
 Counts, status, and spend derive from those rows at read time, so nothing can disagree
@@ -16,7 +17,7 @@ from operator import itemgetter
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Final
 
-from pydantic import BaseModel, ConfigDict, ValidationError, field_validator, model_validator
+from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError, field_validator, model_validator
 
 from litellm._logging import verbose_logger
 from litellm.caching.in_memory_cache import InMemoryCache
@@ -60,7 +61,240 @@ _MAX_ERROR_CHARS: Final = 500
 
 _EMPTY_METADATA: Final[Mapping[str, object]] = MappingProxyType({})
 
-_SAMPLED_CALL_TYPES: Final = frozenset({"completion", "acompletion"})
+# Typed boundaries around the owner transformations, which declare untyped returns:
+# a request or message that fails this lenient shape check is skipped, never sampled.
+_CHAT_REQUEST_ADAPTER: Final = TypeAdapter(Mapping[str, object])
+_CHAT_MESSAGES_ADAPTER: Final = TypeAdapter(tuple[Mapping[str, object], ...])
+_MESSAGE_ITEMS_ADAPTER: Final = TypeAdapter(tuple[object, ...])
+
+
+def _chat_messages(kwargs: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:
+    raw: Final = kwargs.get("messages")
+    return tuple(m for m in raw if isinstance(m, Mapping)) if isinstance(raw, Sequence) else ()
+
+
+def _proxy_wire_body(kwargs: Mapping[str, object]) -> Mapping[str, object]:
+    litellm_params: Final = kwargs.get("litellm_params")
+    request: Final = litellm_params.get("proxy_server_request") if isinstance(litellm_params, Mapping) else None
+    body: Final = request.get("body") if isinstance(request, Mapping) else None
+    return body if isinstance(body, Mapping) else _EMPTY_METADATA
+
+
+def _chat_request_from_chat(
+    kwargs: Mapping[str, object], model_parameters: Mapping[str, object]
+) -> Mapping[str, object]:
+    """Chat requests are already chat-shaped: the logged model_parameters forward as-is."""
+    return MappingProxyType({**model_parameters, "messages": _chat_messages(kwargs)})
+
+
+# Anthropic params the adapter copies through untranslated; the translatable set comes
+# from the adapter itself at call time.
+_ANTHROPIC_SAMPLING_PARAM_KEYS: Final = frozenset(("max_tokens", "temperature", "top_p", "top_k", "reasoning_effort"))
+
+
+def _chat_request_from_anthropic_messages(
+    kwargs: Mapping[str, object], _model_parameters: Mapping[str, object]
+) -> Mapping[str, object]:
+    """/v1/messages logs surface-native block messages with ``system`` top-level: the
+    native provider path carries it in kwargs, the openai-compatible bridge path only in
+    the proxy's snapshot of the client's wire body. Params come from the wire body alone,
+    because the logged optional_params switch dialect per provider path (the bridge's
+    inner completion rewrites them to chat shape mid-flight); the adapter translates
+    them alongside the messages, and sampling params copy through untranslated."""
+    from litellm.llms.anthropic.experimental_pass_through.adapters.transformation import (
+        LiteLLMAnthropicMessagesAdapter,
+    )
+
+    adapter: Final = LiteLLMAnthropicMessagesAdapter()
+    wire_body: Final = _proxy_wire_body(kwargs)
+    system: Final = kwargs.get("system") or wire_body.get("system")
+    param_keys: Final = (
+        frozenset(adapter.translatable_anthropic_params()) | _ANTHROPIC_SAMPLING_PARAM_KEYS
+    ) - frozenset(("messages", "system"))
+    request: Final = MappingProxyType(
+        dict(
+            (
+                *((k, v) for k, v in wire_body.items() if k in param_keys),
+                ("model", str(kwargs.get("model") or "")),
+                ("messages", _CHAT_MESSAGES_ADAPTER.validate_python(kwargs.get("messages") or ())),
+                *((("system", system),) if system is not None else ()),
+            )
+        )
+    )
+    translated, _ = adapter.translate_anthropic_to_openai(request)  # pyright: ignore[reportArgumentType]  # wire-body mapping is the surface's native request shape; the adapter is duck-typed and read-only here
+    return translated
+
+
+def _chat_request_from_responses(
+    kwargs: Mapping[str, object], _model_parameters: Mapping[str, object]
+) -> Mapping[str, object]:
+    """/v1/responses logs the raw ``input`` under ``kwargs["messages"]``, an alias
+    function_setup creates for responses call types: a bare string, chat-shaped dicts,
+    or item dicts; ``instructions`` is the system prompt. Params come from the wire body
+    for the same reason as the messages surface; the transformer translates them with
+    the input (max_output_tokens to max_tokens, Responses tools to chat tools, reasoning
+    to reasoning_effort) and never reads surface-only keys like previous_response_id."""
+    from litellm.responses.litellm_completion_transformation.transformation import (
+        LiteLLMCompletionResponsesConfig,
+    )
+    from litellm.types.llms.openai import ResponsesAPIOptionalRequestParams
+
+    wire_body: Final = _proxy_wire_body(kwargs)
+    instructions: Final = kwargs.get("instructions") or wire_body.get("instructions")
+    responses_request: Final = MappingProxyType(
+        dict(
+            (
+                *((k, v) for k, v in wire_body.items() if k in ResponsesAPIOptionalRequestParams.__annotations__),
+                *((("instructions", instructions),) if instructions is not None else ()),
+            )
+        )
+    )
+    return _CHAT_REQUEST_ADAPTER.validate_python(
+        LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request(  # pyright: ignore[reportUnknownMemberType]  # transformer declares a bare dict return
+            model=str(kwargs.get("model") or ""),
+            input=kwargs.get("messages"),  # pyright: ignore[reportArgumentType]  # untyped callback kwargs; transformer validates shapes
+            responses_api_request=responses_request,  # pyright: ignore[reportArgumentType]  # wire-body dict filtered to the surface's own request keys; the transformer is duck-typed
+        )
+    )
+
+
+def _chat_final_text(response_obj: object) -> str:
+    """The assistant's text, or empty when the turn carries tool calls: only text-final
+    turns produce a judgeable A/B comparison."""
+    try:
+        message: Final = (
+            response_obj["choices"][0]["message"]
+            if isinstance(response_obj, Mapping)
+            else response_obj.choices[0].message  # pyright: ignore[reportAttributeAccessIssue]  # duck-typed ModelResponse
+        )
+    except (AttributeError, KeyError, IndexError, TypeError):
+        return ""
+    read: Final = message.get if isinstance(message, Mapping) else lambda key: getattr(message, key, None)
+    if read("tool_calls") or read("function_call"):
+        return ""
+    return extract_text_from_content(read("content"))
+
+
+def _responses_final_text(response_obj: object) -> str:
+    """The turn's aggregated output text, or empty when the turn carries tool calls. A
+    dict-shaped payload is validated into the owner type first, because ``output_text``
+    is a derived property rather than a serialized field, so it never exists on a dict;
+    a dict the owner type rejects is unjudgeable and skipped."""
+    from litellm.types.llms.openai import ResponsesAPIResponse
+
+    try:
+        response: Final = (
+            ResponsesAPIResponse.model_validate(response_obj) if isinstance(response_obj, Mapping) else response_obj
+        )
+    except ValidationError:
+        return ""
+    output: Final = getattr(response, "output", None)
+    if not isinstance(output, Sequence):
+        return ""
+    items: Final = tuple(item.model_dump() if isinstance(item, BaseModel) else item for item in output)
+    if any(
+        not isinstance(item, Mapping) or item.get("type") in ("function_call", "custom_tool_call") for item in items
+    ):
+        return ""
+    return str(getattr(response, "output_text", "") or "")
+
+
+class _SurfaceOps:
+    """One row per sampled call_type: how its logged request becomes a chat-shaped
+    request (messages plus translated generation params) and how its response yields
+    the judgeable final text. Membership in this table IS the sampling allowlist;
+    unknown call types fail closed. ``wire_params`` marks the surfaces whose params
+    come from the proxy's wire-body snapshot, which is taken before the guardrail
+    pre-call hook: those rows must not sample a request a pre-call guardrail rewrote,
+    or the shadow call would replay content (tools, unmasked entities) the guardrail
+    removed."""
+
+    __slots__ = ("chat_request", "final_text", "wire_params")
+
+    def __init__(
+        self,
+        chat_request: Callable[[Mapping[str, object], Mapping[str, object]], Mapping[str, object]],
+        final_text: Callable[[object], str],
+        wire_params: bool,
+    ) -> None:
+        self.chat_request = chat_request
+        self.final_text = final_text
+        self.wire_params = wire_params
+
+
+_CHAT_OPS: Final = _SurfaceOps(_chat_request_from_chat, _chat_final_text, wire_params=False)
+_ANTHROPIC_OPS: Final = _SurfaceOps(_chat_request_from_anthropic_messages, _chat_final_text, wire_params=True)
+_RESPONSES_OPS: Final = _SurfaceOps(_chat_request_from_responses, _responses_final_text, wire_params=True)
+
+# Guardrail hooks that never rewrite the outbound request: they run in parallel with
+# the call, on the response, or on logged copies. Anything else (pre_call, pre_mcp_call,
+# a future mode) counts as request-mutating, failing closed.
+_NON_MUTATING_GUARDRAIL_MODES: Final = frozenset(
+    ("during_call", "post_call", "logging_only", "during_mcp_call", "post_mcp_call", "realtime_input_transcription")
+)
+
+
+def _request_mutating_guardrail_ran(request_metadata: Mapping[str, object]) -> bool:
+    """Whether a guardrail that can rewrite the outbound request ran on this one, read
+    from the same guardrail-information entries spend logging uses. str-enum modes
+    compare equal to their plain-string values, and an entry whose mode is missing or
+    unrecognized counts as mutating."""
+    raw: Final = request_metadata.get("standard_logging_guardrail_information")
+    entries: Final = raw if isinstance(raw, Sequence) else ()
+    modes_per_entry: Final = tuple(entry.get("guardrail_mode") for entry in entries if isinstance(entry, Mapping))
+    return any(
+        not all(
+            mode in _NON_MUTATING_GUARDRAIL_MODES for mode in (modes if isinstance(modes, list | tuple) else (modes,))
+        )
+        for modes in modes_per_entry
+    )
+
+
+# Translated-request keys that never forward to the shadow call: identity and transport,
+# not generation. Empty-list values (e.g. tools) carry nothing and are dropped with them.
+_UNFORWARDED_REQUEST_KEYS: Final = frozenset(("model", "messages", "stream", "stream_options", "metadata"))
+
+
+def _forwards_nothing(value: object) -> bool:
+    return value is None or (isinstance(value, list) and len(value) == 0)
+
+
+def _judgeable_sample(
+    ops: _SurfaceOps,
+    kwargs: Mapping[str, object],
+    model_parameters: Mapping[str, object],
+    response_obj: object,
+) -> tuple[tuple[Mapping[str, object], ...], Mapping[str, object], str] | None:
+    """The normalized chat conversation, the forwardable generation params, and the
+    judgeable final text; None when this request's shapes cannot be sampled (tool-final
+    turn, empty text, or a shape the owner transformations reject)."""
+    try:
+        request: Final = ops.chat_request(kwargs, model_parameters)
+        items: Final = _MESSAGE_ITEMS_ADAPTER.validate_python(request.get("messages"))
+        messages: Final = _CHAT_MESSAGES_ADAPTER.validate_python(
+            tuple(m.model_dump(exclude_none=True) if isinstance(m, BaseModel) else m for m in items)
+        )
+    except Exception as e:  # noqa: BLE001  # a rejected shape is skipped, never sampled
+        verbose_logger.debug("shadow_eval: request normalization failed, skipping: %s", e)
+        return None
+    real_text: Final = ops.final_text(response_obj)
+    if not messages or not real_text:
+        return None
+    params: Final = MappingProxyType(
+        {k: v for k, v in request.items() if k not in _UNFORWARDED_REQUEST_KEYS and not _forwards_nothing(v)}
+    )
+    return messages, params, real_text
+
+
+_SURFACE_OPS: Final[Mapping[str, _SurfaceOps]] = MappingProxyType(
+    {
+        "completion": _CHAT_OPS,
+        "acompletion": _CHAT_OPS,
+        "anthropic_messages": _ANTHROPIC_OPS,
+        "aresponses": _RESPONSES_OPS,
+        "responses": _RESPONSES_OPS,
+    }
+)
 
 PAIRWISE_JUDGE_SYSTEM_PROMPT: Final = """You are an impartial quality judge comparing two responses to the same conversation.
 
@@ -361,25 +595,36 @@ class ShadowEvalLogger(CustomLogger):
             request_id: Final = payload.get("id") or ""
             if not request_id:
                 return
-            if payload.get("call_type") not in _SAMPLED_CALL_TYPES:
-                return  # only known chat-shaped traffic is comparable; unknown or missing types fail closed
-            raw_messages: Final = kwargs.get("messages")
-            messages: Final = (
-                tuple(m for m in raw_messages if isinstance(m, Mapping)) if isinstance(raw_messages, Sequence) else ()
-            )
-            control_tier: Final = _routed_tier(request_metadata)
+            ops: Final = _SURFACE_OPS.get(str(payload.get("call_type") or ""))
+            if ops is None:
+                return  # only surfaces this table can normalize are comparable; unknown types fail closed
+            if ops.wire_params and _request_mutating_guardrail_ran(request_metadata):
+                return  # the wire-body snapshot predates the rewrite; replaying it would resurrect stripped content
             # A key can hold one job per direction, and a request routed by one job's
             # router while bypassing the other's qualifies for both. Each is separately
-            # budgeted, so both fire.
-            for job in (await self._active_jobs()).get(str(api_key_hash), ()):
-                if datetime.now(timezone.utc) >= job.ends_at:
-                    continue
-                if job.attempts + self._job_starts.get(job.id, 0) >= job.max_turns:
-                    continue
-                if not _sample_hits(request_id, job.id, job.shadow_percentage):
-                    continue
-                if _request_was_routed_by(request_metadata, job.router_name) != (job.direction == "reverse"):
-                    continue
+            # budgeted, so both fire; the request is normalized once, and only when at
+            # least one job sampled it.
+            eligible: Final = tuple(
+                job
+                for job in (await self._active_jobs()).get(str(api_key_hash), ())
+                if datetime.now(timezone.utc) < job.ends_at
+                and job.attempts + self._job_starts.get(job.id, 0) < job.max_turns
+                and _sample_hits(request_id, job.id, job.shadow_percentage)
+                and _request_was_routed_by(request_metadata, job.router_name) == (job.direction == "reverse")
+            )
+            if not eligible:
+                return
+            sample: Final = _judgeable_sample(
+                ops,
+                kwargs,
+                MappingProxyType(dict(payload.get("model_parameters") or {})),  # mutable-ok: frozen snapshot
+                response_obj,
+            )
+            if sample is None:
+                return
+            messages, shadow_params, real_text = sample
+            control_tier: Final = _routed_tier(request_metadata)
+            for job in eligible:
                 if self._inflight_shadow_tasks >= _MAX_CONCURRENT_SHADOW_TASKS:
                     return
                 self._job_starts[job.id] = self._job_starts.get(job.id, 0) + 1
@@ -389,12 +634,10 @@ class ShadowEvalLogger(CustomLogger):
                         job=job,
                         request_id=request_id,
                         messages=messages,
-                        response_obj=response_obj,
+                        real_text=real_text,
                         real_model=payload.get("model") or "",
                         control_tier=control_tier,
-                        model_parameters=MappingProxyType(
-                            dict(payload.get("model_parameters") or {})  # mutable-ok: frozen snapshot
-                        ),
+                        shadow_params=shadow_params,
                         parent_metadata=MappingProxyType(dict(request_metadata)),  # mutable-ok: frozen snapshot
                     )
                 ).add_done_callback(self._release_shadow_slot)
@@ -411,10 +654,10 @@ class ShadowEvalLogger(CustomLogger):
         job: ActiveShadowEvalJob,
         request_id: str,
         messages: Sequence[Mapping[str, object]],
-        response_obj: object,
+        real_text: str,
         real_model: str,
         control_tier: str | None,
-        model_parameters: Mapping[str, object],
+        shadow_params: Mapping[str, object],
         parent_metadata: Mapping[str, object],
     ) -> None:
         """Budget gate -> shadow call -> blind judge -> one attempt row. The prisma gate
@@ -424,15 +667,10 @@ class ShadowEvalLogger(CustomLogger):
         try:
             if prisma is None:
                 return
-            real_text: Final = self._extract_response_text(response_obj)
-            if not real_text or not messages:
-                return
             if await _key_or_team_is_over_budget(parent_metadata):
                 return
 
-            shadow: Final = await self._call_router_shadow(
-                job.shadow_target, messages, model_parameters, parent_metadata
-            )
+            shadow: Final = await self._call_router_shadow(job.shadow_target, messages, shadow_params, parent_metadata)
             if isinstance(shadow, _CallFailure):
                 await self._record_attempt(prisma, job, request_id, control_tier, outcome="error", error=shadow.error)
                 return
@@ -510,7 +748,7 @@ class ShadowEvalLogger(CustomLogger):
         self,
         target_model: str,
         messages: Sequence[Mapping[str, object]],
-        model_parameters: Mapping[str, object],
+        shadow_params: Mapping[str, object],
         parent_metadata: Mapping[str, object],
     ) -> "_ShadowResponse | _CallFailure":
         """Send the prompt through the arm nobody was served: the auto-router under
@@ -523,9 +761,6 @@ class ShadowEvalLogger(CustomLogger):
         shadow_metadata: Final[dict[str, object]] = (  # mutable-ok: router writes its routing decision back
             sanitized_forwardable_call_metadata(parent_metadata, SHADOW_EVAL_ROUTER_CALL_ORIGIN)
         )
-        shadow_params: Final = {  # mutable-ok: splatted as kwargs
-            k: v for k, v in model_parameters.items() if k not in ("stream", "metadata")
-        }
         try:
             response: Final = await router.acompletion(
                 model=target_model,
@@ -538,7 +773,7 @@ class ShadowEvalLogger(CustomLogger):
         except Exception as e:  # noqa: BLE001  # provider errors become error rows, not crashes
             verbose_logger.debug("shadow_eval: router call failed: %s", e)
             return _CallFailure(f"shadow router call failed: {e}")
-        text: Final = self._extract_response_text(response)
+        text: Final = _chat_final_text(response)
         if not text:
             return _CallFailure("shadow router returned an empty response")
         return _ShadowResponse(
@@ -597,19 +832,6 @@ class ShadowEvalLogger(CustomLogger):
             cost=_judge_call_cost(response),
         )
 
-    @staticmethod
-    def _extract_response_text(response_obj: object) -> str:
-        """Extract the assistant's text from a ModelResponse-shaped object or dict."""
-        try:
-            content: Final = (
-                response_obj["choices"][0]["message"]["content"]
-                if isinstance(response_obj, Mapping)
-                else response_obj.choices[0].message.content  # pyright: ignore[reportAttributeAccessIssue]  # duck-typed ModelResponse
-            )
-        except (AttributeError, KeyError, IndexError, TypeError):
-            return ""
-        return extract_text_from_content(content)
-
 
 _EMPTY_JOBS: Final[Mapping[str, tuple[ActiveShadowEvalJob, ...]]] = MappingProxyType({})
 

litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py

@@ -307,7 +307,7 @@ class LiteLLMAnthropicMessagesAdapter:
                 # Fallback for non-dict objects (shouldn't happen in practice)
                 cast(dict[str, Any], target)["cache_control"] = cache_control
 
-    def translatable_anthropic_params(self) -> list:
+    def translatable_anthropic_params(self) -> list[str]:
         """
         Which anthropic params, we need to translate to the openai format.
         """

tests/test_litellm/integrations/test_shadow_eval_logger.py

@@ -20,6 +20,7 @@ from litellm.integrations.shadow_eval_logger import (
     _sample_hits,
     _unmask_preference,
 )
+from litellm.types.guardrails import GuardrailEventHooks
 from litellm.types.utils import SHADOW_EVAL_JUDGE_CALL_ORIGIN, SHADOW_EVAL_ROUTER_CALL_ORIGIN, ModelResponse
 
 
@@ -123,6 +124,32 @@ def _success_kwargs(
 
 RESPONSE = {"choices": [{"message": {"content": "real answer"}}]}
 
+RESPONSES_API_RESPONSE = {
+    "id": "resp_1",
+    "created_at": 1,
+    "model": "gpt-5",
+    "object": "response",
+    "output": [
+        {
+            "type": "message",
+            "id": "msg_1",
+            "status": "completed",
+            "role": "assistant",
+            "content": [{"type": "output_text", "text": "real answer", "annotations": []}],
+        }
+    ],
+    "parallel_tool_calls": True,
+    "error": None,
+    "incomplete_details": None,
+    "instructions": None,
+    "metadata": None,
+    "temperature": None,
+    "tool_choice": "auto",
+    "tools": [],
+    "top_p": None,
+    "status": "completed",
+}
+
 
 async def _drain(logger: ShadowEvalLogger, target: int = 0):
     for _ in range(100):
@@ -132,6 +159,251 @@ async def _drain(logger: ShadowEvalLogger, target: int = 0):
     raise AssertionError("shadow tasks never drained")
 
 
+@pytest.mark.asyncio
+class TestSurfaceNormalization:
+    """/v1/messages and /v1/responses arms: the hook normalizes each surface's logged
+    request through litellm's own transformations and judges only text-final turns."""
+
+    async def _drive(self, hook_kwargs, response_obj):
+        prisma = _prisma()
+        router = _router()
+        logger = _logger(router=router, prisma=prisma, jobs=(_job(),))
+        await logger.async_log_success_event(hook_kwargs, response_obj, None, None)
+        await _drain(logger)
+        return prisma, router
+
+    async def test_anthropic_messages_arm_normalizes_blocks_and_system(self):
+        hook_kwargs = _success_kwargs(call_type="anthropic_messages")
+        hook_kwargs["messages"] = [{"role": "user", "content": [{"type": "text", "text": "what is 2+2"}]}]
+        hook_kwargs["system"] = "you are terse"
+
+        prisma, router = await self._drive(hook_kwargs, RESPONSE)
+
+        shadow_messages = router.acompletion.call_args_list[0].kwargs["messages"]
+        assert shadow_messages[0]["role"] == "system"
+        assert shadow_messages[0]["content"] == "you are terse"
+        assert shadow_messages[1]["role"] == "user"
+        prisma.db.litellm_shadowevalattempt.create.assert_called_once()
+
+    async def test_anthropic_bridge_path_recovers_system_from_proxy_wire_body(self):
+        """On the openai-compatible bridge path kwargs carry no system (live-probed:
+        kwargs["system"] is None and complete_input_dict is empty); the proxy's snapshot
+        of the client's wire body is the only remaining source."""
+        hook_kwargs = _success_kwargs(call_type="anthropic_messages")
+        hook_kwargs["messages"] = [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
+        hook_kwargs["litellm_params"]["proxy_server_request"] = {
+            "body": {"model": "gpt-5", "max_tokens": 100, "system": "from the wire body", "messages": []}
+        }
+
+        _, router = await self._drive(hook_kwargs, RESPONSE)
+
+        shadow_messages = router.acompletion.call_args_list[0].kwargs["messages"]
+        assert shadow_messages[0] == {"role": "system", "content": "from the wire body"}
+
+    async def test_anthropic_arm_translates_wire_body_params_not_logged_optional_params(self):
+        """The wire body is the only surface-native param source on both provider paths
+        (the bridge's inner completion rewrites the logged optional_params to chat
+        shape); anthropic tools and stop_sequences reach the shadow call translated,
+        transport and litellm keys never do."""
+        hook_kwargs = _success_kwargs(call_type="anthropic_messages")
+        hook_kwargs["messages"] = [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
+        hook_kwargs["standard_logging_object"]["model_parameters"] = {"temperature": 0.9}
+        hook_kwargs["litellm_params"]["proxy_server_request"] = {
+            "body": {
+                "model": "claude-x",
+                "messages": [],
+                "system": "you are terse",
+                "max_tokens": 100,
+                "temperature": 0.1,
+                "top_k": 5,
+                "stop_sequences": ["END"],
+                "stream": True,
+                "tools": [
+                    {"name": "get_weather", "description": "d", "input_schema": {"type": "object", "properties": {}}}
+                ],
+                "litellm_metadata": {"user_api_key_hash": "key-hash"},
+            }
+        }
+
+        _, router = await self._drive(hook_kwargs, RESPONSE)
+
+        shadow_call = router.acompletion.call_args_list[0].kwargs
+        assert shadow_call["max_tokens"] == 100
+        assert shadow_call["temperature"] == 0.1
+        assert shadow_call["top_k"] == 5
+        assert shadow_call["stop"] == ["END"]
+        assert shadow_call["tools"][0]["type"] == "function"
+        assert shadow_call["tools"][0]["function"]["name"] == "get_weather"
+        assert "stop_sequences" not in shadow_call
+        assert "stream" not in shadow_call
+        assert shadow_call["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_ROUTER_CALL_ORIGIN
+
+    async def test_responses_arm_translates_wire_body_params_and_drops_surface_only_keys(self):
+        from litellm.types.llms.openai import ResponsesAPIResponse
+
+        hook_kwargs = _success_kwargs(call_type="aresponses")
+        hook_kwargs["messages"] = "what is 8+8"
+        hook_kwargs["litellm_params"]["proxy_server_request"] = {
+            "body": {
+                "model": "gpt-5",
+                "input": "what is 8+8",
+                "instructions": "you are terse",
+                "max_output_tokens": 128,
+                "temperature": 0.3,
+                "previous_response_id": "resp_0",
+                "tools": [
+                    {
+                        "type": "function",
+                        "name": "get_weather",
+                        "description": "d",
+                        "parameters": {"type": "object", "properties": {}},
+                    }
+                ],
+            }
+        }
+        response = ResponsesAPIResponse.model_validate(RESPONSES_API_RESPONSE)
+
+        _, router = await self._drive(hook_kwargs, response)
+
+        shadow_call = router.acompletion.call_args_list[0].kwargs
+        assert shadow_call["messages"][0] == {"role": "system", "content": "you are terse"}
+        assert shadow_call["max_tokens"] == 128
+        assert shadow_call["temperature"] == 0.3
+        assert shadow_call["tools"][0]["function"]["name"] == "get_weather"
+        assert "max_output_tokens" not in shadow_call
+        assert "previous_response_id" not in shadow_call
+        assert "instructions" not in shadow_call
+
+    @pytest.mark.parametrize("payload_shape", ["typed", "dict"])
+    @pytest.mark.parametrize("call_type", ["aresponses", "responses"])
+    async def test_responses_arms_normalize_bare_string_input_and_instructions(self, call_type, payload_shape):
+        from litellm.types.llms.openai import ResponsesAPIResponse
+
+        hook_kwargs = _success_kwargs(call_type=call_type)
+        hook_kwargs["messages"] = "what is 8+8"
+        hook_kwargs["instructions"] = "you are terse"
+        response = (
+            ResponsesAPIResponse.model_validate(RESPONSES_API_RESPONSE)
+            if payload_shape == "typed"
+            else RESPONSES_API_RESPONSE
+        )
+
+        prisma, router = await self._drive(hook_kwargs, response)
+
+        shadow_call = router.acompletion.call_args_list[0].kwargs
+        shadow_messages = shadow_call["messages"]
+        assert shadow_messages[0]["role"] == "system"
+        assert shadow_messages[1]["role"] == "user"
+        assert shadow_messages[1]["content"] == "what is 8+8"
+        assert "tools" not in shadow_call
+        prisma.db.litellm_shadowevalattempt.create.assert_called_once()
+
+    @pytest.mark.parametrize(
+        "response_mutation,kwargs_mutation",
+        [
+            ("chat-tool-calls", {}),
+            ("responses-function-call", {"call_type": "aresponses"}),
+        ],
+        ids=["tool-final-chat-turn", "tool-final-responses-turn"],
+    )
+    async def test_unjudgeable_turns_are_skipped_without_consuming_budget(self, response_mutation, kwargs_mutation):
+        from litellm.types.llms.openai import ResponsesAPIResponse
+
+        hook_kwargs = _success_kwargs(**({"call_type": "acompletion"} | kwargs_mutation))
+        response = RESPONSE
+        if response_mutation == "chat-tool-calls":
+            response = {
+                "choices": [
+                    {
+                        "message": {
+                            "content": "let me check",
+                            "tool_calls": [
+                                {"id": "t1", "type": "function", "function": {"name": "f", "arguments": "{}"}}
+                            ],
+                        }
+                    }
+                ]
+            }
+        elif response_mutation == "responses-function-call":
+            hook_kwargs["messages"] = "do the thing"
+            response = ResponsesAPIResponse.model_validate(
+                RESPONSES_API_RESPONSE
+                | {
+                    "output": [
+                        {
+                            "type": "function_call",
+                            "name": "f",
+                            "arguments": "{}",
+                            "call_id": "c1",
+                            "id": "fc1",
+                            "status": "completed",
+                        }
+                    ]
+                }
+            )
+
+        prisma, router = await self._drive(hook_kwargs, response)
+
+        router.acompletion.assert_not_called()
+        prisma.db.litellm_shadowevalattempt.create.assert_not_called()
+
+    @pytest.mark.parametrize(
+        "call_type,guardrail_mode,sampled",
+        [
+            ("anthropic_messages", ["logging_only", "pre_call"], False),
+            ("aresponses", GuardrailEventHooks.pre_call, False),
+            ("anthropic_messages", "post_call", True),
+            ("acompletion", "pre_call", True),
+        ],
+        ids=["anthropic-pre-call-list", "responses-pre-call-enum", "anthropic-post-call-only", "chat-pre-call"],
+    )
+    async def test_guardrail_rewritten_requests_never_replay_the_wire_body(self, call_type, guardrail_mode, sampled):
+        """The proxy snapshots the wire body before the guardrail pre-call hook, so the
+        wire-sourced surfaces skip requests a request-mutating guardrail ran on rather
+        than replay stripped tools or unmasked content; chat sources the dispatched
+        call and keeps sampling, as do requests only response-mode guardrails touched."""
+        hook_kwargs = _success_kwargs(
+            call_type=call_type,
+            request_metadata={
+                "standard_logging_guardrail_information": [{"guardrail_name": "g", "guardrail_mode": guardrail_mode}]
+            },
+        )
+        response = RESPONSE
+        if call_type == "anthropic_messages":
+            hook_kwargs["messages"] = [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
+        elif call_type == "aresponses":
+            hook_kwargs["messages"] = "hi"
+            response = RESPONSES_API_RESPONSE
+
+        prisma, router = await self._drive(hook_kwargs, response)
+
+        if sampled:
+            prisma.db.litellm_shadowevalattempt.create.assert_called_once()
+        else:
+            router.acompletion.assert_not_called()
+            prisma.db.litellm_shadowevalattempt.create.assert_not_called()
+
+    @pytest.mark.parametrize(
+        "call_type,messages,response_obj",
+        [
+            ("anthropic_messages", "not-a-message-list", RESPONSE),
+            ("acompletion", [{"role": "user", "content": "hi"}], {"unexpected": "shape"}),
+            ("aresponses", "hi", RESPONSE),
+        ],
+        ids=["rejected-request-shape", "malformed-chat-response", "responses-response-without-output"],
+    )
+    async def test_unsampleable_shapes_fail_closed(self, call_type, messages, response_obj):
+        """A request or response shape the normalizers reject is skipped without a
+        provider call or an attempt row, never raised."""
+        hook_kwargs = _success_kwargs(call_type=call_type)
+        hook_kwargs["messages"] = messages
+
+        prisma, router = await self._drive(hook_kwargs, response_obj)
+
+        router.acompletion.assert_not_called()
+        prisma.db.litellm_shadowevalattempt.create.assert_not_called()
+
+
 class TestSampling:
     def test_boundaries_and_determinism(self):
         assert not any(_sample_hits(f"req-{i}", "job", 0.0) for i in range(100))
@@ -187,6 +459,9 @@ class TestSuccessHookSkipChain:
         await logger.async_log_success_event(_success_kwargs(), RESPONSE, None, None)
         await _drain(logger)
 
+        shadow_call = router.acompletion.call_args_list[0].kwargs
+        assert shadow_call["temperature"] == 0.5
+        assert "stream" not in shadow_call
         create = prisma.db.litellm_shadowevalattempt.create
         create.assert_awaited_once()
         row = create.call_args.kwargs["data"]
@@ -369,10 +644,10 @@ class TestShadowPipeline:
             job=_job(),
             request_id="req-1",
             messages=({"role": "user", "content": "hi"},),
-            response_obj=RESPONSE,
+            real_text="real answer",
             real_model="claude-opus",
             control_tier=None,
-            model_parameters={},
+            shadow_params={},
             parent_metadata={},
         )
 
@@ -398,10 +673,10 @@ class TestShadowPipeline:
             job=_job(),
             request_id="req-1",
             messages=({"role": "user", "content": "hi"},),
-            response_obj=RESPONSE,
+            real_text="real answer",
             real_model="claude-opus",
             control_tier=None,
-            model_parameters={},
+            shadow_params={},
             parent_metadata={"user_api_key_auth": UserAPIKeyAuth(api_key="sk-abc", max_budget=10.0)},
         )
 
@@ -429,10 +704,10 @@ class TestShadowPipeline:
             job=_job(),
             request_id="req-1",
             messages=({"role": "user", "content": "hi"},),
-            response_obj=RESPONSE,
+            real_text="real answer",
             real_model="claude-opus",
             control_tier=None,
-            model_parameters={},
+            shadow_params={},
             parent_metadata={},
         )
 
@@ -457,10 +732,10 @@ class TestShadowPipeline:
             job=_job(),
             request_id="req-1",
             messages=({"role": "user", "content": "hi"},),
-            response_obj=RESPONSE,
+            real_text="real answer",
             real_model="claude-opus",
             control_tier=None,
-            model_parameters={"stream": True, "temperature": 0.2, "metadata": {"x": 1}},
+            shadow_params={"temperature": 0.2},
             parent_metadata=parent_metadata,
         )
 
@@ -475,7 +750,6 @@ class TestShadowPipeline:
         assert shadow_call["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_ROUTER_CALL_ORIGIN
         assert judge_call["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_JUDGE_CALL_ORIGIN
         assert "routing_decision" not in judge_call["metadata"]
-        assert "stream" not in shadow_call
         assert shadow_call["temperature"] == 0.2
         assert judge_call["max_tokens"] == JUDGE_MAX_OUTPUT_TOKENS