diff --git a/basedpyright-code-budget.json b/basedpyright-code-budget.json index d14a97f82f4..06010c706e3 100644 --- a/basedpyright-code-budget.json +++ b/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 diff --git a/litellm/integrations/shadow_eval_logger.py b/litellm/integrations/shadow_eval_logger.py index ca9b6982414..99d5ab47f1a 100644 --- a/litellm/integrations/shadow_eval_logger.py +++ b/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({}) diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py b/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py index 69f451973b2..667f9dcaab0 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py +++ b/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. """ diff --git a/tests/test_litellm/integrations/test_shadow_eval_logger.py b/tests/test_litellm/integrations/test_shadow_eval_logger.py index 3a69340109d..8d2f9482fa7 100644 --- a/tests/test_litellm/integrations/test_shadow_eval_logger.py +++ b/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