scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-30 06:43:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 250

fix(logging): mark mirrored silent_model calls with is_silent_experiment in spend logs (#48888)

Browse source 
* 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>
This commit is contained in:
tavren27 2026-10-29 05:12:50 -07:00 • committed by GitHub
parent 67ae3fb527
commit e7d3b24d7c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 631 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
litellm/_internal_context.py
View file

@ -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."""

5
litellm/integrations/datadog/datadog_cost_management.py
View file

@ -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

6
litellm/integrations/langfuse/langfuse.py
View file

@ -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

6
litellm/integrations/prometheus.py
View file

@ -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
),
}

47
litellm/litellm_core_utils/litellm_logging.py
View file

@ -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"))):

1
litellm/proxy/_types.py
View file

@ -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

10
litellm/proxy/spend_tracking/spend_tracking_utils.py
View file

@ -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

25
litellm/router.py
View file

@ -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):
"""

4
litellm/types/router.py
View file

@ -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"

1
litellm/types/utils.py
View file

@ -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):

6
tests/integration/sdk/test_router_silent_model_shadow_sdk.py
View file

@ -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()))

38
tests/unit/integrations/datadog/test_datadog_cost_management.py
View file

@ -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):
"""

25
tests/unit/integrations/test_langfuse.py
View file

@ -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-<hh-mm-ss-us>_<response id>``, so a generation could

35
tests/unit/integrations/test_prometheus_spend_logs_metadata.py
View file

@ -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

22
tests/unit/litellm_core_utils/test_litellm_logging.py
View file

@ -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

36
tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py
View file

@ -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

2
tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py
View file

@ -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)

396
tests/unit/test_router_silent_experiment.py
View file

@ -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}