From e7d3b24d7c0bf23eb1bc44e53f08432efcd8e332 Mon Sep 17 00:00:00 2001 From: tavren27 <348969968+tavren27@users.noreply.github.com> Date: Thu, 29 Oct 2026 05:12:50 -0700 Subject: [PATCH] fix(logging): mark mirrored silent_model calls with is_silent_experiment in spend logs (#48888) * fix(logging): mark mirrored silent_model calls with is_silent_experiment in spend logs and logging payloads * fix(router): keep the mirror flag when litellm_metadata is set and strip a client-sent is_silent_experiment Spend rows read router metadata from litellm_metadata when a call carries it, so the router now sets the flag there too. The proxy drops a client-sent is_silent_experiment with the other router-owned keys, so real traffic cannot be recorded as mirrored. * fix(logging): mark mirrored calls with a router-only key so clients cannot set is_silent_experiment Spend rows and the standard logging payload now derive is_silent_experiment from _litellm_silent_experiment, which only the router writes and the proxy and pass-through routes strip from client metadata. A caller's own is_silent_experiment key stays untouched, so it keeps its current effect on mirroring and on provider metadata, and unmirrored calls log false in both places. * test(langfuse): expect is_silent_experiment false on every logged generation * fix(logging): mark mirrored calls with a router-only enum value so no client route can set is_silent_experiment The router stamps SilentExperimentMarker.MIRRORED, which client JSON can never produce, so batches, Responses WebSocket frames and pass-through need no strip. Langfuse layers the flag after its requester-metadata dedupe. * fix(logging): set is_silent_experiment only on mirrored calls so real calls keep their payloads Real calls no longer gain an is_silent_experiment key in spend rows or the standard logging payload, so OTEL, S3, GCS and every other logger receive what they did before. The Langfuse regression test now runs the real prompt helper. * fix(logging): keep the mirror marker out of callback metadata and read the spend-row flag from the logging payload Logging now strips the router's mirror marker from the request metadata buckets before any callback or vendor sees it, without touching the router's own dicts, so retries keep it. Spend rows read is_silent_experiment from the standard logging payload, which survives the collector's JSON round trip, so collector-mode rows keep the flag too. * fix(logging): mark mirrored calls through a context variable so router usage counts them once * fix(router): leave callback-started calls unmarked and stop client flags from turning mirroring off Logging handlers now run outside the mirrored-call context, so an LLM call a success or failure callback starts while logging a mirrored call is not flagged as mirrored. The guard against mirroring a mirror reads the same context variable instead of request metadata, so a client-sent is_silent_experiment no longer skips the mirrors. The unused module-level get_standard_logging_metadata stops copying resolved keys from raw metadata. * fix(router): keep main's client flag mirroring skip and the other logging metadata keys A client's is_silent_experiment: true skips mirroring again as on main, the module-level get_standard_logging_metadata drops only the flag, and the silent experiment tests wait on a logged-call signal instead of polling a deadline * fix(logging): take the mirror flag in Prometheus labels and Datadog cost tags only from the router * test: type the new parametrized tests and keep the recorder from changing its argument * fix(datadog): keep client request tags from setting the silent experiment cost tag A request tag is_silent_experiment:true set the Datadog cost tag on a direct call. Only the router flag sets it now. Also covers the sync failure handler's mirrored-call reset with a failure function callback test. --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com> --- litellm/_internal_context.py | 26 ++ .../datadog/datadog_cost_management.py | 5 +- litellm/integrations/langfuse/langfuse.py | 6 + litellm/integrations/prometheus.py | 6 + litellm/litellm_core_utils/litellm_logging.py | 47 ++- litellm/proxy/_types.py | 1 + .../spend_tracking/spend_tracking_utils.py | 10 +- litellm/router.py | 25 +- litellm/types/router.py | 4 + litellm/types/utils.py | 1 + .../test_router_silent_model_shadow_sdk.py | 6 +- .../datadog/test_datadog_cost_management.py | 38 ++ tests/unit/integrations/test_langfuse.py | 25 ++ .../test_prometheus_spend_logs_metadata.py | 35 ++ .../test_litellm_logging.py | 22 +- .../test_pass_through_endpoints.py | 36 ++ .../test_spend_tracking_utils.py | 2 +- tests/unit/test_router_silent_experiment.py | 396 +++++++++++++++++- 18 files changed, 631 insertions(+), 60 deletions(-) diff --git a/litellm/_internal_context.py b/litellm/_internal_context.py index 244efc8047b..c2243b90fbf 100644 --- a/litellm/_internal_context.py +++ b/litellm/_internal_context.py @@ -36,6 +36,8 @@ REDIS_FAMILIES_METADATA_KEY: Final = "families" _service_caller: Final[ContextVar[str | None]] = ContextVar("service_caller", default=None) +_mirrored_call: Final[ContextVar[bool]] = ContextVar("mirrored_call", default=False) + @contextmanager def internal_sub_call() -> Generator[None]: @@ -123,6 +125,30 @@ def current_service_caller() -> str | None: return _service_caller.get() +@contextmanager +def mirrored_call() -> Generator[None]: + """A silent_model mirror of a request: every litellm call inside, its retries and fallbacks included.""" + token: Final = _mirrored_call.set(True) + try: + yield + finally: + _restore(_mirrored_call, token) + + +def in_mirrored_call() -> bool: + return _mirrored_call.get() + + +@contextmanager +def outside_mirrored_call() -> Generator[None]: + """Logging a mirrored call: the calls its callbacks start are not mirrors themselves.""" + token: Final = _mirrored_call.set(False) + try: + yield + finally: + _restore(_mirrored_call, token) + + @contextmanager def pinned_billing_time(moment: datetime) -> Generator[None]: """Price every rate lookup inside this block at ``moment`` rather than at each one's own clock read.""" diff --git a/litellm/integrations/datadog/datadog_cost_management.py b/litellm/integrations/datadog/datadog_cost_management.py index 1a81fa019dc..d4b75fc0bce 100644 --- a/litellm/integrations/datadog/datadog_cost_management.py +++ b/litellm/integrations/datadog/datadog_cost_management.py @@ -205,11 +205,12 @@ class DatadogCostManagementLogger(CustomBatchLogger): # see _RESERVED_TAG_KEYS for the rationale. if self.cost_tag_keys: allow: Final = set(self.cost_tag_keys) + client_allow: Final = allow - {"is_silent_experiment"} for rt in log.get("request_tags") or []: if not isinstance(rt, str) or ":" not in rt: continue k, _, v = rt.partition(":") - if k in allow and v: + if k in client_allow and v: self._set_custom_tag(tags, k, v) for k, v in metadata.items(): if k in allow and v is not None and not isinstance(v, (dict, list)): @@ -218,7 +219,7 @@ class DatadogCostManagementLogger(CustomBatchLogger): nested = metadata.get(nested_key) if isinstance(nested, dict): for k, v in nested.items(): - if k in allow and v is not None and not isinstance(v, (dict, list)): + if k in client_allow and v is not None and not isinstance(v, (dict, list)): self._set_custom_tag(tags, k, str(v)) return tags diff --git a/litellm/integrations/langfuse/langfuse.py b/litellm/integrations/langfuse/langfuse.py index b8c6e91d0d7..bd68ffe8cd6 100644 --- a/litellm/integrations/langfuse/langfuse.py +++ b/litellm/integrations/langfuse/langfuse.py @@ -833,6 +833,12 @@ class LangFuseLogger: ("vertex_location", vertex_location, bool(vertex_location)), ("aws_region_name", aws_region_name, bool(aws_region_name)), ("cache_hit", kwargs.get("cache_hit") or False, "cache_hit" in kwargs), + ( + "is_silent_experiment", + "is_silent_experiment" in allowlisted_metadata + and allowlisted_metadata["is_silent_experiment"] is True, + "is_silent_experiment" in allowlisted_metadata, + ), ) enrichments: Final[Mapping[str, object]] = { key: value for key, value, include in candidate_enrichments if include diff --git a/litellm/integrations/prometheus.py b/litellm/integrations/prometheus.py index d15956c0e95..e62b99f8b0f 100644 --- a/litellm/integrations/prometheus.py +++ b/litellm/integrations/prometheus.py @@ -5148,6 +5148,12 @@ def _get_combined_custom_metadata_from_standard_logging_payload( **(requester_metadata if isinstance(requester_metadata, dict) else {}), **(user_api_key_auth_metadata if isinstance(user_api_key_auth_metadata, dict) else {}), **(spend_logs_metadata if isinstance(spend_logs_metadata, dict) else {}), + "is_silent_experiment": ( + "true" + if "is_silent_experiment" in standard_logging_metadata + and standard_logging_metadata["is_silent_experiment"] is True + else None + ), } diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 77c4136efda..545be49ff5e 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -23,7 +23,7 @@ from httpx import Response, Timeout from pydantic import BaseModel, JsonValue, TypeAdapter import litellm -from litellm._internal_context import post_response_phase +from litellm._internal_context import in_mirrored_call, outside_mirrored_call, post_response_phase from litellm._logging import ( is_debugging_on, redact_string, @@ -162,7 +162,7 @@ from litellm.types.llms.openai import ( from litellm.types.mcp import MCPPostCallResponseObject from litellm.types.prompts.init_prompts import PromptSpec from litellm.types.rerank import RerankResponse -from litellm.types.router import RouterRateLimitError +from litellm.types.router import RouterRateLimitError, SilentExperimentFlag from litellm.types.utils import ( DEPLOYMENT_SCOPED_PRICING_FIELDS, AgenticLoopParams, @@ -315,7 +315,7 @@ else: _in_memory_loggers: Final[list[CustomLogger]] = [] _STANDARD_LOGGING_METADATA_RESOLVED_KEYS: Final[frozenset[str]] = frozenset( - ("used_client_oauth_token", "usage_object", KUBERNETES_POD_ROUTING_KEY) + ("used_client_oauth_token", "usage_object", KUBERNETES_POD_ROUTING_KEY, "is_silent_experiment") ) _STANDARD_LOGGING_METADATA_KEYS: Final[frozenset[str]] = ( frozenset(StandardLoggingMetadata.__annotations__.keys()) - _STANDARD_LOGGING_METADATA_RESOLVED_KEYS @@ -809,6 +809,7 @@ class Logging(LiteLLMLoggingBaseClass): litellm_params = scrub_sensitive_keys_in_metadata(litellm_params) self.litellm_params = litellm_params + self.is_silent_experiment: Final = in_mirrored_call() # Initialize cost breakdown field self.cost_breakdown: CostBreakdown | None = None @@ -3253,7 +3254,7 @@ class Logging(LiteLLMLoggingBaseClass): """Restores trace_id/session_id contextvars once this attempt's own success logging (including any nested calls its callbacks trigger) is fully done.""" try: - with post_response_phase(): + with post_response_phase(), outside_mirrored_call(): return self._success_handler_body( result=result, start_time=start_time, @@ -3753,7 +3754,7 @@ class Logging(LiteLLMLoggingBaseClass): """Restores trace_id/session_id contextvars once this attempt's own success logging (including any nested calls its callbacks trigger) is fully done.""" try: - with post_response_phase(): + with post_response_phase(), outside_mirrored_call(): return await self._async_success_handler_body( result=result, start_time=start_time, end_time=end_time, cache_hit=cache_hit, **kwargs ) @@ -4270,13 +4271,14 @@ class Logging(LiteLLMLoggingBaseClass): ``for_async_request`` marks the sync run of a failure whose async run also logs it, so a CustomLogger that gets ``async_log_failure_event`` from that async run skips its sync hook.""" try: - return self._failure_handler_body( - exception=exception, - traceback_exception=traceback_exception, - start_time=start_time, - end_time=end_time, - for_async_request=for_async_request, - ) + with outside_mirrored_call(): + return self._failure_handler_body( + exception=exception, + traceback_exception=traceback_exception, + start_time=start_time, + end_time=end_time, + for_async_request=for_async_request, + ) finally: self._restore_correlation_context() @@ -4469,12 +4471,13 @@ class Logging(LiteLLMLoggingBaseClass): """Restores trace_id/session_id contextvars once this attempt's own failure logging (including any nested calls its callbacks trigger) is fully done.""" try: - return await self._async_failure_handler_body( - exception=exception, - traceback_exception=traceback_exception, - start_time=start_time, - end_time=end_time, - ) + with outside_mirrored_call(): + return await self._async_failure_handler_body( + exception=exception, + traceback_exception=traceback_exception, + start_time=start_time, + end_time=end_time, + ) finally: self._restore_correlation_context() @@ -6563,6 +6566,7 @@ class StandardLoggingPayloadSetup: start_time: dt_object | None = None, response_id: str | None = None, custom_llm_provider: str | None = None, + is_silent_experiment: bool = False, ) -> StandardLoggingMetadata: """ Clean and filter the metadata dictionary to include only the specified keys in StandardLoggingMetadata. @@ -6636,6 +6640,7 @@ class StandardLoggingPayloadSetup: proxy_stamped_used_client_oauth_token(metadata, litellm_params), custom_llm_provider, ), + **(SilentExperimentFlag(is_silent_experiment=True) if is_silent_experiment else SilentExperimentFlag()), ) if isinstance(metadata, dict): for key in metadata.keys() & _STANDARD_LOGGING_METADATA_KEYS: @@ -7436,6 +7441,7 @@ def get_standard_logging_object_payload( start_time=start_time, response_id=id, custom_llm_provider=selected_provider if isinstance(selected_provider, str) else None, + is_silent_experiment=logging_obj.is_silent_experiment is True, ) _request_body: Final = proxy_server_request.get("body", {}) end_user_id: Final = clean_metadata["user_api_key_end_user_id"] or _request_body.get( @@ -7720,9 +7726,8 @@ def get_standard_logging_metadata( ) if isinstance(metadata, dict): # Update the clean_metadata with values from input metadata that match StandardLoggingMetadata fields - for key in StandardLoggingMetadata.__annotations__: - if key in metadata: - clean_metadata[key] = metadata[key] + for key in (metadata.keys() & StandardLoggingMetadata.__annotations__.keys()) - {"is_silent_experiment"}: + clean_metadata[key] = metadata[key] if metadata.get("user_api_key") is not None: if is_valid_sha256_hash(str(metadata.get("user_api_key"))): diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index dd873ae49dd..4047fa3f0ad 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -3748,6 +3748,7 @@ class SpendLogsMetadata(TypedDict): internal_call_origin: InternalCallOrigin | None billed_retry_attempt: ReadOnly[NotRequired[bool | None]] litellm_roi_estimator: ReadOnly[NotRequired[bool | None]] + is_silent_experiment: ReadOnly[NotRequired[bool | None]] guardrail_information: list[StandardLoggingGuardrailInformation] | None eval_information: Any | None status: StandardLoggingPayloadStatus diff --git a/litellm/proxy/spend_tracking/spend_tracking_utils.py b/litellm/proxy/spend_tracking/spend_tracking_utils.py index 94b079163e9..833ee71ae9c 100644 --- a/litellm/proxy/spend_tracking/spend_tracking_utils.py +++ b/litellm/proxy/spend_tracking/spend_tracking_utils.py @@ -80,7 +80,7 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import ( SETTLES_BACKGROUND_RESPONSE_KWARG, BackgroundResponseSettlement, ) -from litellm.types.router import DeploymentTypedDict, LiteLLM_Params +from litellm.types.router import DeploymentTypedDict, LiteLLM_Params, SilentExperimentFlag from litellm.types.utils import ( BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN, PROMPT_CARRYING_GUARDRAIL_FIELDS, @@ -194,6 +194,7 @@ _STAMPED_METADATA_KEYS: Final = frozenset( "used_client_oauth_token", KUBERNETES_POD_ROUTING_KEY, "litellm_roi_estimator", + "is_silent_experiment", BILLED_RETRY_ATTEMPT_METADATA_KEY, "user_api_key_user_email", ) @@ -227,6 +228,7 @@ def _get_spend_logs_metadata( azure_spillover: AzureSpillover | None = None, used_client_oauth_token: bool | None = None, kubernetes_pod_routing: StandardLoggingKubernetesPodRouting | None = None, + is_silent_experiment: bool = False, ) -> SpendLogsMetadata: if metadata is None: return SpendLogsMetadata( @@ -294,6 +296,7 @@ def _get_spend_logs_metadata( used_client_oauth_token=used_client_oauth_token, kubernetes_pod_routing=kubernetes_pod_routing, litellm_roi_estimator=metadata.get("litellm_roi_estimator") is True, + **(SilentExperimentFlag(is_silent_experiment=True) if is_silent_experiment else SilentExperimentFlag()), billed_retry_attempt=billed_retry_attempt, user_api_key_user_email=_stored_user_email(metadata), ) @@ -738,6 +741,11 @@ def get_logging_payload( # clean up litellm metadata clean_metadata = _get_spend_logs_metadata( persisted_metadata, + is_silent_experiment=( + standard_logging_payload is not None + and "is_silent_experiment" in standard_logging_payload["metadata"] + and standard_logging_payload["metadata"]["is_silent_experiment"] + ), applied_guardrails=( standard_logging_payload["metadata"].get("applied_guardrails", None) if standard_logging_payload is not None diff --git a/litellm/router.py b/litellm/router.py index 3809a5d6555..ff31f48c49e 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -75,7 +75,7 @@ from typing_extensions import assert_never, overload import litellm import litellm.litellm_core_utils.exception_mapping_utils from litellm import get_secret_str -from litellm._internal_context import service_target, with_service_target +from litellm._internal_context import in_mirrored_call, mirrored_call, service_target, with_service_target from litellm._logging import verbose_router_logger from litellm._uuid import uuid from litellm.caching.caching import ( @@ -754,7 +754,7 @@ def _every_deployment_refused( def _silent_experiment_targets(silent_model: object) -> tuple[str, ...]: - if silent_model is None: + if silent_model is None or in_mirrored_call(): return () try: targets: Final = _SILENT_MODEL_ADAPTER.validate_python(silent_model) @@ -3372,16 +3372,17 @@ class Router: remaining_kwargs: Final = MappingProxyType( {key: value for key, value in silent_kwargs.items() if key != "stream"} ) - response: Final = await self.acompletion( - model=silent_model, - messages=cast(list[AllMessageValues], messages), - stream=bool(silent_kwargs.get("stream", False)), - **remaining_kwargs, - ) - if not isinstance(response, CustomStreamWrapper): - return - async for _ in response: - pass + with mirrored_call(): + response: Final = await self.acompletion( + model=silent_model, + messages=cast(list[AllMessageValues], messages), + stream=bool(silent_kwargs.get("stream", False)), + **remaining_kwargs, + ) + if not isinstance(response, CustomStreamWrapper): + return + async for _ in response: + pass def _silent_experiment_completion(self, silent_model: str, messages: Sequence[Mapping[str, str]], **kwargs): """ diff --git a/litellm/types/router.py b/litellm/types/router.py index 60c9e596588..db8be9e91a3 100644 --- a/litellm/types/router.py +++ b/litellm/types/router.py @@ -67,6 +67,10 @@ from .utils import ( AllowedModelRegion: TypeAlias = Literal["eu", "us"] +class SilentExperimentFlag(TypedDict, total=False): + is_silent_experiment: ReadOnly[bool] + + class FallbackSkipReason(str, enum.Enum): NOT_AUTHORIZED = "not_authorized" OVER_BUDGET = "over_budget" diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 9c3329d6387..7a349bdc711 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -3374,6 +3374,7 @@ class StandardLoggingMetadata(StandardLoggingUserAPIKeyMetadata): team_alias: str | None team_id: str | None used_client_oauth_token: ReadOnly[bool | None] + is_silent_experiment: ReadOnly[NotRequired[bool]] class AzureSpillover(TypedDict): diff --git a/tests/integration/sdk/test_router_silent_model_shadow_sdk.py b/tests/integration/sdk/test_router_silent_model_shadow_sdk.py index 821e4ddd0b7..be5998e205e 100644 --- a/tests/integration/sdk/test_router_silent_model_shadow_sdk.py +++ b/tests/integration/sdk/test_router_silent_model_shadow_sdk.py @@ -82,9 +82,9 @@ class _ShadowRecorder(CustomLogger): async def async_log_success_event( self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object ) -> None: - metadata: Final = _OBJECT.validate_python(_OBJECT.validate_python(kwargs["litellm_params"])["metadata"] or {}) - if metadata.get("is_silent_experiment") is True: - self.shadow_groups.put(str(metadata.get("model_group"))) + payload: Final = _OBJECT.validate_python(kwargs["standard_logging_object"]) + if _OBJECT.validate_python(payload["metadata"]).get("is_silent_experiment") is True: + self.shadow_groups.put(str(payload["model_group"])) def drained(self) -> tuple[str, ...]: return tuple(self.shadow_groups.get_nowait() for _ in range(self.shadow_groups.qsize())) diff --git a/tests/unit/integrations/datadog/test_datadog_cost_management.py b/tests/unit/integrations/datadog/test_datadog_cost_management.py index 1a50a6991da..3601db12909 100644 --- a/tests/unit/integrations/datadog/test_datadog_cost_management.py +++ b/tests/unit/integrations/datadog/test_datadog_cost_management.py @@ -1,4 +1,5 @@ import time +from typing import Final from unittest.mock import AsyncMock import pytest @@ -314,6 +315,43 @@ async def test_extract_tags_nested_metadata_allowlisted(clean_env): assert "ignored" not in tags +@pytest.mark.parametrize( + ("metadata", "request_tags", "expected_tag"), + [ + ( + {"is_silent_experiment": True, "requester_metadata": {"is_silent_experiment": False}}, + ["is_silent_experiment:false"], + "true", + ), + ( + { + "requester_metadata": {"is_silent_experiment": True}, + "spend_logs_metadata": {"is_silent_experiment": "true"}, + }, + [], + None, + ), + ({}, ["is_silent_experiment:true"], None), + ], + ids=["mirrored call", "direct call with a client flag", "direct call with a client request tag"], +) +@pytest.mark.asyncio +async def test_extract_tags_takes_the_silent_experiment_flag_only_from_the_router( + clean_env: None, metadata: dict[str, object], request_tags: list[str], expected_tag: str | None +): + logger: Final = DatadogCostManagementLogger(cost_tag_keys=["is_silent_experiment"]) + log: Final = StandardLoggingPayload( + custom_llm_provider="openai", + model="gpt-4", + response_cost=0.01, + startTime=time.time(), + metadata=metadata, + request_tags=request_tags, + ) + + assert logger._extract_tags(log).get("is_silent_experiment") == expected_tag + + @pytest.mark.asyncio async def test_extract_tags_allowlist_cannot_override_reserved_dimensions(clean_env): """ diff --git a/tests/unit/integrations/test_langfuse.py b/tests/unit/integrations/test_langfuse.py index 00d85b64566..2d1e49e32df 100644 --- a/tests/unit/integrations/test_langfuse.py +++ b/tests/unit/integrations/test_langfuse.py @@ -830,6 +830,31 @@ class TestLangfuseUsageDetails(unittest.TestCase): assert generation_metadata["litellm_response_cost"] == 0.25 assert generation_metadata["api_base"] == "https://real-api-base" + def test_caller_sent_is_silent_experiment_cannot_hide_the_mirrored_call_flag(self): + payload = self._build_standard_logging_payload(trace_id="canary-trace-id") + payload["metadata"]["is_silent_experiment"] = True + payload["metadata"]["requester_metadata"] = {"is_silent_experiment": False} + kwargs = self._build_langfuse_kwargs(payload) + metadata = self._canary_request_metadata() + self.use_real_langfuse_client() + + self.logger._log_langfuse_v2( + user_id="user-1", + metadata=metadata, + litellm_params={"metadata": metadata}, + output=None, + start_time=datetime.datetime(2024, 1, 1, 12, 0, 0), + end_time=datetime.datetime(2024, 1, 1, 12, 0, 1), + kwargs=kwargs, + optional_params={}, + input=None, + response_obj=None, + level="INFO", + litellm_call_id="canary-call-id", + ) + + assert self.exported_generation_metadata()["is_silent_experiment"] is True + def test_generation_metadata_carries_the_call_id_and_response_id(self): """ v2's generation id was ``time-_``, so a generation could diff --git a/tests/unit/integrations/test_prometheus_spend_logs_metadata.py b/tests/unit/integrations/test_prometheus_spend_logs_metadata.py index e2af6fd2daf..608195a9df8 100644 --- a/tests/unit/integrations/test_prometheus_spend_logs_metadata.py +++ b/tests/unit/integrations/test_prometheus_spend_logs_metadata.py @@ -5,6 +5,10 @@ Verifies that metadata from x-litellm-spend-logs-metadata header is available in Prometheus custom labels via combined_metadata. """ +from typing import Final + +import pytest + from litellm.integrations.prometheus import ( _get_combined_custom_metadata_from_standard_logging_payload, get_custom_labels_from_metadata, @@ -205,3 +209,34 @@ def test_project_alias_accessible_without_prefix(monkeypatch): result = get_custom_labels_from_metadata(combined) assert result == {"user_api_key_project_alias": "hotel-recommendations"} + + +@pytest.mark.parametrize( + ("standard_logging_metadata", "expected_labels"), + [ + ( + {"is_silent_experiment": True, "requester_metadata": {"is_silent_experiment": False}}, + {"metadata_is_silent_experiment": "true"}, + ), + ( + { + "requester_metadata": {"is_silent_experiment": True}, + "spend_logs_metadata": {"is_silent_experiment": "true"}, + }, + {}, + ), + ], + ids=["mirrored call", "direct call with a client flag"], +) +def test_silent_experiment_label_comes_only_from_the_router_flag( + monkeypatch: pytest.MonkeyPatch, + standard_logging_metadata: dict[str, object], + expected_labels: dict[str, str], +): + monkeypatch.setattr("litellm.custom_prometheus_metadata_labels", ["metadata.is_silent_experiment"]) + + combined: Final = _get_combined_custom_metadata_from_standard_logging_payload( + {"metadata": standard_logging_metadata} + ) + + assert get_custom_labels_from_metadata(combined) == expected_labels diff --git a/tests/unit/litellm_core_utils/test_litellm_logging.py b/tests/unit/litellm_core_utils/test_litellm_logging.py index 299088a6c4f..505c110e9a6 100644 --- a/tests/unit/litellm_core_utils/test_litellm_logging.py +++ b/tests/unit/litellm_core_utils/test_litellm_logging.py @@ -11939,7 +11939,7 @@ def test_get_additional_headers(): def all_fields_present(standard_logging_metadata: StandardLoggingMetadata): - for field in StandardLoggingMetadata.__annotations__.keys(): + for field in StandardLoggingMetadata.__required_keys__: assert field in standard_logging_metadata @@ -11975,6 +11975,26 @@ def test_get_standard_logging_metadata(metadata_key, metadata_value): assert standard_logging_metadata[metadata_key] == metadata_value +@pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") +def test_module_level_get_standard_logging_metadata_drops_only_a_caller_sent_silent_experiment_flag(): + from litellm.litellm_core_utils.litellm_logging import get_standard_logging_metadata + + caller_metadata: Final = { + "is_silent_experiment": True, + "user_api_key_alias": "alias", + "usage_object": {"total_tokens": 3}, + "used_client_oauth_token": True, + "kubernetes_pod_routing": {"pod": "a"}, + } + + result = get_standard_logging_metadata(caller_metadata) + + assert "is_silent_experiment" not in result + assert {key: result[key] for key in caller_metadata.keys() - {"is_silent_experiment"}} == { + key: value for key, value in caller_metadata.items() if key != "is_silent_experiment" + } + + @pytest.mark.usefixtures("_vcr_outcome_gate", "drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") def test_get_standard_logging_metadata_user_api_key_hash(): valid_hash = "a" * 64 # 64 character string diff --git a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py index b0504af1b5e..df36d805377 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py +++ b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py @@ -10437,6 +10437,42 @@ def test_init_kwargs_no_authenticated_identity_field_is_client_settable(mock_req survived: Final = {key: metadata.get(key) for key in expected if metadata.get(key) != expected[key]} assert survived == {}, f"client-supplied values survived for: {sorted(survived)}" + +def test_init_kwargs_client_metadata_cannot_mark_a_pass_through_call_as_a_silent_experiment( + mock_user_api_key_dict: UserAPIKeyAuth, +) -> None: + from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload + + started: Final = datetime(2026, 1, 15, 12, 0, 0) + result: Final = HttpPassThroughEndpointHelpers._init_kwargs_for_pass_through_endpoint( + request=Request({"type": "http", "method": "POST", "path": "/test", "headers": [], "query_string": b""}), + user_api_key_dict=mock_user_api_key_dict, + passthrough_logging_payload=PassthroughStandardLoggingPayload(url="https://test.com", request_body={}), + litellm_call_id="test-call-id", + logging_obj=LiteLLMLoggingObj( + model="test-model", + messages=[], + stream=False, + call_type="test-call-type", + start_time=started, + litellm_call_id="test-call-id", + function_id="test-function-id", + ), + _parsed_body={ + "litellm_metadata": {"is_silent_experiment": True, "client_key": "kept"}, + "metadata": {"is_silent_experiment": True}, + }, + ) + + row: Final = get_logging_payload( + kwargs={"model": "test-model", "litellm_params": result["litellm_params"]}, + response_obj=litellm.ModelResponse(id="chatcmpl-test", choices=[]), + start_time=started, + end_time=started, + ) + assert "is_silent_experiment" not in json.loads(row["metadata"]) + assert result["litellm_params"]["metadata"]["client_key"] == "kept" + @pytest.mark.asyncio async def test_direct_passthrough_keeps_mapped_baggage_tags_outside_upstream_body(): from litellm.proxy.common_utils.http_parsing_utils import read_request_body diff --git a/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py b/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py index 86d064050eb..23f9ac0e2d8 100644 --- a/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py +++ b/tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py @@ -6830,7 +6830,7 @@ def test_spend_logs_payload(model_id: str | None): ) assert len(payload["request_id"]) > 0 - expected_metadata_keys: Final = SpendLogsMetadata.__annotations__.keys() + expected_metadata_keys: Final = SpendLogsMetadata.__annotations__.keys() - {"is_silent_experiment"} assert "metadata" in payload assert isinstance(payload["metadata"], str) diff --git a/tests/unit/test_router_silent_experiment.py b/tests/unit/test_router_silent_experiment.py index fd648ceb5d2..b696a61aa3c 100644 --- a/tests/unit/test_router_silent_experiment.py +++ b/tests/unit/test_router_silent_experiment.py @@ -1,9 +1,11 @@ import asyncio +import datetime import json +import threading import time -from collections.abc import Callable, Mapping +from collections.abc import Callable, Generator, Mapping from types import SimpleNamespace -from typing import Final +from typing import Final, Literal from unittest.mock import AsyncMock, MagicMock, patch import httpx @@ -22,9 +24,48 @@ class _RecordingLogger(CustomLogger): def __init__(self) -> None: super().__init__() self.success_kwargs: list[dict[str, object]] = [] + self.failure_kwargs: list[dict[str, object]] = [] + self._logged: Final = threading.Condition() - async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): - self.success_kwargs.append(kwargs) + def _record(self, outcome: Literal["success", "failure"], kwargs: dict[str, object]) -> None: + with self._logged: + match outcome: + case "success": + self.success_kwargs.append(kwargs) + case "failure": + self.failure_kwargs.append(kwargs) + self._logged.notify_all() + + async def async_log_success_event( + self, + kwargs: dict[str, object], + response_obj: object, + start_time: datetime.datetime, + end_time: datetime.datetime, + ) -> None: + self._record("success", kwargs) + + async def async_log_failure_event( + self, + kwargs: dict[str, object], + response_obj: object, + start_time: datetime.datetime, + end_time: datetime.datetime, + ) -> None: + self._record("failure", kwargs) + + def wait_until(self, logged_enough: Callable[[], bool], timeout: float = 30.0) -> None: + with self._logged: + self._logged.wait_for(logged_enough, timeout=timeout) + + def judge_flags(self) -> dict[str, object]: + return { + call["litellm_params"]["metadata"]["judge_of"]: call["standard_logging_object"]["metadata"].get( + "is_silent_experiment" + ) + for call in self.success_kwargs + if "judge_of" in call["litellm_params"]["metadata"] + } def shadow_successes(self) -> list[dict[str, object]]: return [ @@ -34,6 +75,17 @@ class _RecordingLogger(CustomLogger): ] +class _SyncAndAsyncRecordingLogger(_RecordingLogger): + def log_success_event( + self, + kwargs: dict[str, object], + response_obj: object, + start_time: datetime.datetime, + end_time: datetime.datetime, + ) -> None: + self._record("success", kwargs) + + async def _settle_shared_logging_worker() -> None: try: await GLOBAL_LOGGING_WORKER.flush() @@ -41,15 +93,46 @@ async def _settle_shared_logging_worker() -> None: await GLOBAL_LOGGING_WORKER.stop() -@pytest.fixture -def recording_logger(): +class _JudgingLogger(_RecordingLogger): + async def _judge(self, kwargs: Mapping[str, object]) -> None: + metadata: Final = kwargs["litellm_params"]["metadata"] + if "judge_of" in metadata: + return + await litellm.acompletion( + model="openai/gpt-5.4-nano", + api_key="fake-key", + messages=[{"role": "user", "content": "score it"}], + mock_response="ok", + metadata={"judge_of": metadata["model_group"]}, + ) + + async def async_log_success_event( + self, + kwargs: dict[str, object], + response_obj: object, + start_time: datetime.datetime, + end_time: datetime.datetime, + ) -> None: + await super().async_log_success_event(kwargs, response_obj, start_time, end_time) + await self._judge(kwargs) + + async def async_log_failure_event( + self, + kwargs: dict[str, object], + response_obj: object, + start_time: datetime.datetime, + end_time: datetime.datetime, + ) -> None: + await self._judge(kwargs) + + +def _installed(logger: _RecordingLogger) -> Generator[_RecordingLogger]: settle_loop: Final = asyncio.new_event_loop() try: settle_loop.run_until_complete(_settle_shared_logging_worker()) finally: settle_loop.close() original_callbacks: Final = litellm.callbacks - logger: Final = _RecordingLogger() litellm.callbacks = [logger] try: yield logger @@ -57,22 +140,44 @@ def recording_logger(): litellm.callbacks = original_callbacks -async def _wait_for_shadow_successes(logger: _RecordingLogger, expected: int, timeout: float = 5.0) -> None: - deadline: Final = time.monotonic() + timeout - while len(logger.shadow_successes()) < expected and time.monotonic() < deadline: - await asyncio.sleep(0.05) +@pytest.fixture +def recording_logger(): + yield from _installed(_RecordingLogger()) -async def _wait_for_all_successes(logger: _RecordingLogger, expected: int, timeout: float = 5.0) -> None: - deadline: Final = time.monotonic() + timeout - while len(logger.success_kwargs) < expected and time.monotonic() < deadline: - await asyncio.sleep(0.05) +@pytest.fixture +def sync_and_async_recording_logger(): + yield from _installed(_SyncAndAsyncRecordingLogger()) -def _wait_for_shadow_successes_sync(logger: _RecordingLogger, expected: int, timeout: float = 5.0) -> None: - deadline: Final = time.monotonic() + timeout - while len(logger.shadow_successes()) < expected and time.monotonic() < deadline: - time.sleep(0.05) +@pytest.fixture +def judging_logger(): + yield from _installed(_JudgingLogger()) + + +async def _wait_for_shadow_successes(logger: _RecordingLogger, expected: int, timeout: float = 30.0) -> None: + await asyncio.to_thread(logger.wait_until, lambda: len(logger.shadow_successes()) >= expected, timeout) + + +async def _wait_for_all_successes(logger: _RecordingLogger, expected: int) -> None: + await asyncio.to_thread(logger.wait_until, lambda: len(logger.success_kwargs) >= expected) + + +async def _wait_for_judges(logger: _RecordingLogger, judged: set[str]) -> None: + await asyncio.to_thread(logger.wait_until, lambda: judged <= logger.judge_flags().keys()) + + +def _silent_flag_by_group(logger: _RecordingLogger) -> dict[str, object]: + return { + call["standard_logging_object"]["model_group"]: call["standard_logging_object"]["metadata"].get( + "is_silent_experiment" + ) + for call in logger.success_kwargs + } + + +def _wait_for_shadow_successes_sync(logger: _RecordingLogger, expected: int) -> None: + logger.wait_until(lambda: len(logger.shadow_successes()) >= expected) def _streaming_model_list(silent_model: object) -> list[dict[str, object]]: @@ -263,6 +368,197 @@ async def test_multiple_shadow_targets_fan_out_async(recording_logger): assert primary_successes[0]["litellm_params"]["metadata"]["model_group"] == "primary-model" +@pytest.mark.asyncio +async def test_standard_logging_payload_marks_only_the_mirrored_calls_as_silent_experiments( + recording_logger: _RecordingLogger, +) -> None: + router: Final = Router(model_list=_streaming_model_list(["shadow-a", "shadow-b"])) + response: Final = await router.acompletion( + model="primary-model", + messages=[{"role": "user", "content": "hi"}], + mock_response="pong", + ) + assert response.choices[0].message.content == "pong" + await _wait_for_all_successes(recording_logger, expected=3) + + payloads: Final = [call["standard_logging_object"] for call in recording_logger.success_kwargs] + assert sorted(payload["model_group"] for payload in payloads) == ["primary-model", "shadow-a", "shadow-b"] + metadata_by_group: Final = {payload["model_group"]: payload["metadata"] for payload in payloads} + assert metadata_by_group["shadow-a"]["is_silent_experiment"] is True + assert metadata_by_group["shadow-b"]["is_silent_experiment"] is True + assert "is_silent_experiment" not in metadata_by_group["primary-model"] + + +def test_standard_logging_payload_marks_only_the_mirrored_calls_on_the_sync_path( + sync_and_async_recording_logger: _SyncAndAsyncRecordingLogger, +) -> None: + router: Final = Router(model_list=_streaming_model_list(["shadow-a", "shadow-b"])) + response: Final = router.completion( + model="primary-model", + messages=[{"role": "user", "content": "hi"}], + mock_response="pong", + ) + assert response.choices[0].message.content == "pong" + sync_and_async_recording_logger.wait_until( + lambda: ( + _silent_flag_by_group(sync_and_async_recording_logger).keys() >= {"primary-model", "shadow-a", "shadow-b"} + ) + ) + + assert _silent_flag_by_group(sync_and_async_recording_logger) == { + "primary-model": None, + "shadow-a": True, + "shadow-b": True, + } + + +@pytest.mark.asyncio +async def test_a_failed_mirrored_call_reaches_failure_callbacks_marked_as_mirrored( + recording_logger: _RecordingLogger, +) -> None: + router: Final = Router( + model_list=[ + { + "model_name": "primary-model", + "litellm_params": { + "model": "openai/gpt-5.4-mini", + "api_key": "fake-key", + "silent_model": "shadow-a", + "mock_response": "pong", + }, + }, + { + "model_name": "shadow-a", + "litellm_params": { + "model": "openai/gpt-5.4-nano", + "api_key": "fake-key", + "mock_response": "litellm.InternalServerError", + }, + }, + ] + ) + await router.acompletion(model="primary-model", messages=[{"role": "user", "content": "hi"}]) + await asyncio.to_thread(recording_logger.wait_until, lambda: len(recording_logger.failure_kwargs) >= 1) + + assert { + ( + call["standard_logging_object"]["model_group"], + call["standard_logging_object"]["metadata"].get("is_silent_experiment"), + ) + for call in recording_logger.failure_kwargs + } == {("shadow-a", True)} + + +_SPEND_ROW_TIME: Final = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) + + +def _spend_row_is_silent_experiment(call: Mapping[str, object]) -> object: + from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload + + row: Final = get_logging_payload( + kwargs=dict(call), + response_obj=litellm.ModelResponse(id="chatcmpl-test", choices=[]), + start_time=_SPEND_ROW_TIME, + end_time=_SPEND_ROW_TIME, + ) + return json.loads(row["metadata"]).get("is_silent_experiment") + + +def _through_the_spend_collector(call: Mapping[str, object]) -> Mapping[str, object]: + from litellm.proxy.spend_tracking.spend_event import ( + SpendEvent, + build_spend_event, + decode_spend_event, + spend_event_callback_args, + ) + + line: Final = build_spend_event( + call, + litellm.ModelResponse(id="chatcmpl-test", choices=[]), + _SPEND_ROW_TIME, + _SPEND_ROW_TIME, + store_bodies=False, + ) + assert isinstance(line, bytes) + event: Final = decode_spend_event(line) + assert isinstance(event, SpendEvent) + return spend_event_callback_args(event).kwargs + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "spend_kwargs", + [pytest.param(lambda call: call, id="in_process"), pytest.param(_through_the_spend_collector, id="collector")], +) +async def test_spend_rows_mark_only_the_mirrored_calls_when_the_caller_sends_litellm_metadata( + recording_logger: _RecordingLogger, + spend_kwargs: Callable[[Mapping[str, object]], Mapping[str, object]], +) -> None: + router: Final = Router(model_list=_streaming_model_list(["shadow-a", "shadow-b"])) + await router.acompletion( + model="primary-model", + messages=[{"role": "user", "content": "hi"}], + mock_response="pong", + litellm_metadata={"user_api_key": "test-key"}, + ) + await _wait_for_all_successes(recording_logger, expected=3) + + silent_by_group: Final = { + call["standard_logging_object"]["model_group"]: _spend_row_is_silent_experiment(spend_kwargs(call)) + for call in recording_logger.success_kwargs + } + assert silent_by_group == {"primary-model": None, "shadow-a": True, "shadow-b": True} + + +@pytest.mark.asyncio +async def test_router_counts_a_mirrored_call_once_against_its_target_rate_limits( + recording_logger: _RecordingLogger, +) -> None: + router: Final = Router( + model_list=[ + { + "model_name": "primary-model", + "litellm_params": {"model": "openai/gpt-5.4-mini", "api_key": "fake-key", "silent_model": "shadow-b"}, + }, + { + "model_name": "shadow-b", + "litellm_params": {"model": "openai/gpt-5.4-nano", "api_key": "fake-key", "rpm": 100, "tpm": 100000}, + }, + ] + ) + await router.acompletion(model="primary-model", messages=[{"role": "user", "content": "hi"}], mock_response="pong") + await _wait_for_shadow_successes(recording_logger, expected=1) + await GLOBAL_LOGGING_WORKER.flush() + + shadow_tokens: Final = recording_logger.shadow_successes()[0]["standard_logging_object"]["total_tokens"] + assert await router.get_model_group_usage("shadow-b") == (shadow_tokens, 1) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("bucket", "planted"), + [ + ("metadata", {"is_silent_experiment": True}), + ("litellm_metadata", {"is_silent_experiment": True}), + ], +) +async def test_a_caller_sent_silent_experiment_flag_does_not_mark_a_real_call_as_mirrored( + recording_logger: _RecordingLogger, bucket: str, planted: dict[str, object] +) -> None: + router: Final = Router(model_list=_streaming_model_list(["shadow-a", "shadow-b"])) + await router.acompletion( + model="shadow-b", + messages=[{"role": "user", "content": "hi"}], + mock_response="pong", + **{bucket: planted}, + ) + await _wait_for_all_successes(recording_logger, expected=1) + + call: Final = recording_logger.success_kwargs[0] + assert "is_silent_experiment" not in call["standard_logging_object"]["metadata"] + assert _spend_row_is_silent_experiment(call) is None + + def test_multiple_shadow_targets_fan_out_sync(recording_logger): router = Router(model_list=_streaming_model_list(["shadow-a", "shadow-b"])) response = router.completion( @@ -674,6 +970,8 @@ async def test_aembedding_with_silent_model_sends_provider_body_without_it( request_body: Final = json.loads(route.calls.last.request.read()) assert request_body == {"model": "embed-model", "input": ["black dresses"], "input_type": "query"} assert response.data[0]["embedding"] == [0.1, 0.2] + + @pytest.mark.parametrize("run_silent_experiment", SILENT_EXPERIMENT_RUNNERS) def test_silent_experiment_does_not_launch_from_a_shadow_request(run_silent_experiment): router = Router(model_list=_streaming_model_list(["shadow-a"])) @@ -686,3 +984,63 @@ def test_silent_experiment_does_not_launch_from_a_shadow_request(run_silent_expe ) acompletion.assert_not_awaited() + + +def _judged_router(shadow_outcome: str) -> Router: + return Router( + model_list=[ + { + "model_name": "primary-model", + "litellm_params": { + "model": "openai/gpt-5.4-mini", + "api_key": "fake-key", + "silent_model": "shadow-a", + "mock_response": "pong", + }, + }, + { + "model_name": "shadow-a", + "litellm_params": { + "model": "openai/gpt-5.4-nano", + "api_key": "fake-key", + "mock_response": shadow_outcome, + }, + }, + ] + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("shadow_outcome", ["pong", "litellm.InternalServerError"]) +async def test_calls_a_callback_starts_while_logging_a_mirrored_call_are_not_marked_as_mirrored( + judging_logger: _JudgingLogger, shadow_outcome: str +) -> None: + router: Final = _judged_router(shadow_outcome) + await router.acompletion(model="primary-model", messages=[{"role": "user", "content": "hi"}]) + await _wait_for_judges(judging_logger, {"primary-model", "shadow-a"}) + + assert judging_logger.judge_flags() == {"primary-model": None, "shadow-a": None} + + +def _start_judge_call( + kwargs: Mapping[str, object], completion_response: object, start_time: object, end_time: object +) -> None: + litellm.completion( + model="openai/gpt-5.4-nano", + api_key="fake-key", + messages=[{"role": "user", "content": "score it"}], + mock_response="ok", + metadata={"judge_of": kwargs["litellm_params"]["metadata"]["model_group"]}, + ) + + +@pytest.mark.asyncio +async def test_calls_a_failure_function_callback_starts_for_a_mirrored_call_are_not_marked_as_mirrored( + sync_and_async_recording_logger: _SyncAndAsyncRecordingLogger, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "failure_callback", [_start_judge_call]) + router: Final = _judged_router("litellm.InternalServerError") + await router.acompletion(model="primary-model", messages=[{"role": "user", "content": "hi"}]) + await _wait_for_judges(sync_and_async_recording_logger, {"shadow-a"}) + + assert sync_and_async_recording_logger.judge_flags() == {"shadow-a": None}