diff --git a/litellm/integrations/shadow_eval_logger.py b/litellm/integrations/shadow_eval_logger.py index df0e7366689..10ee32af282 100644 --- a/litellm/integrations/shadow_eval_logger.py +++ b/litellm/integrations/shadow_eval_logger.py @@ -179,12 +179,13 @@ class _CallFailure: @dataclass(frozen=True, slots=True) class _ShadowResponse: - """A successful shadow call: (text, routed model, tier, completion tokens).""" + """A successful shadow call, with the ids and token counts the verdict row records.""" text: str model: str tier: str | None completion_tokens: int | None + request_id: str | None @dataclass(frozen=True, slots=True) @@ -320,10 +321,20 @@ class ShadowEvalLogger(CustomLogger): self._lifecycle_task: asyncio.Task[None] | None = None def start_lifecycle_loop(self) -> None: - """Idempotently start the periodic loop that owns job lifecycle off the request path.""" + """Idempotently start the periodic loop that owns job lifecycle off the request path. + + Without the loop, jobs never finalize and counters never flush, so a caller + registering this logger outside a running event loop gets a warning rather + than a silent no-op. + """ if self._lifecycle_task is not None and not self._lifecycle_task.done(): return - self._lifecycle_task = asyncio.create_task(self._lifecycle_loop()) + try: + self._lifecycle_task = asyncio.create_task(self._lifecycle_loop()) + except RuntimeError: + verbose_logger.warning( + "shadow_eval: no running event loop; lifecycle loop not started, jobs will not finalize on this process" + ) async def _lifecycle_loop(self) -> None: while True: @@ -406,6 +417,7 @@ class ShadowEvalLogger(CustomLogger): else (), response_obj=response_obj, real_model=payload.get("model") or "", + real_response_tokens=payload.get("completion_tokens"), model_parameters=MappingProxyType( dict(payload.get("model_parameters") or {}) # mutable-ok: frozen snapshot ), @@ -507,6 +519,7 @@ class ShadowEvalLogger(CustomLogger): messages: Sequence[Mapping[str, object]], response_obj: object, real_model: str, + real_response_tokens: int | None, model_parameters: Mapping[str, object], parent_metadata: Mapping[str, object], budget_metadata: Mapping[str, object] = _EMPTY_METADATA, @@ -547,8 +560,10 @@ class ShadowEvalLogger(CustomLogger): data={ # mutable-ok: Prisma payload "job_id": job.id, "request_id": request_id, + "shadow_request_id": shadow.request_id, "tier_classification": shadow.tier, "real_model": real_model, + "real_response_tokens": real_response_tokens, "shadow_model": shadow.model, "shadow_response_tokens": shadow.completion_tokens, "judge_preference": verdict.preference, @@ -631,11 +646,13 @@ class ShadowEvalLogger(CustomLogger): raw_tier: Final = routing_decision.get("tier_label") or routing_decision.get("tier") usage: Final = getattr(response, "usage", None) raw_tokens: Final = getattr(usage, "completion_tokens", None) if usage is not None else None + raw_id: Final = getattr(response, "id", None) return _ShadowResponse( text=text, model=str(getattr(response, "model", None) or routing_decision.get("routed_model") or ""), tier=str(raw_tier) if raw_tier is not None else None, completion_tokens=int(raw_tokens) if isinstance(raw_tokens, int) else None, + request_id=str(raw_id) if raw_id else None, ) async def _call_judge( diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 0803d9b8b69..b21dd701ae8 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -1184,8 +1184,6 @@ async def proxy_startup_event(app: FastAPI): _tagged.strategy._state_loaded = True asyncio.create_task(_adaptive_router_flusher_loop()) - _start_shadow_eval_lifecycle_loop() - ## [Optional] Initialize dd tracer ProxyStartupEvent._init_dd_tracer() @@ -2217,27 +2215,18 @@ def cost_tracking(): def _register_shadow_eval_logger() -> None: - """Register the shadow-eval success hook. + """Register the shadow-eval success hook and start its lifecycle loop. Cheap when idle: with no active LiteLLM_ShadowEvalJob rows the hook is one - cached dict lookup per request. Registered alongside cost tracking because - it has the same hard dependency on prisma_client. Its lifecycle loop is - started separately at proxy startup, once an event loop exists. + dict lookup per request. Registered alongside cost tracking because it has + the same hard dependency on prisma_client. Registration owns the loop start + so the two cannot be reordered apart; the loop is started on whichever + instance is actually registered, since the callback manager dedups by class + and discards a second instance. """ from litellm.integrations.shadow_eval_logger import ShadowEvalLogger litellm.logging_callback_manager.add_litellm_callback(ShadowEvalLogger()) - - -def _start_shadow_eval_lifecycle_loop() -> None: - """Start the registered shadow-eval logger's periodic lifecycle loop. - - The loop owns counter flushes, snapshot refreshes, and finishing jobs whose - window or spend cap has passed, so a job on a key that goes quiet still ends - on schedule with its final counters written. - """ - from litellm.integrations.shadow_eval_logger import ShadowEvalLogger - for callback in litellm.callbacks: if isinstance(callback, ShadowEvalLogger): callback.start_lifecycle_loop() diff --git a/tests/test_litellm/integrations/test_shadow_eval_logger.py b/tests/test_litellm/integrations/test_shadow_eval_logger.py index 4cc1224d4bb..bf8b5b86cf1 100644 --- a/tests/test_litellm/integrations/test_shadow_eval_logger.py +++ b/tests/test_litellm/integrations/test_shadow_eval_logger.py @@ -362,7 +362,7 @@ class TestStoppedJobCannotBeReactivated: logger = ShadowEvalLogger(router_provider=lambda: MagicMock(), prisma_provider=lambda: prisma) logger._call_router_shadow = AsyncMock( - return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10) + return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10, request_id="shadow-req-1") ) logger._call_judge = AsyncMock( return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.01) @@ -375,6 +375,7 @@ class TestStoppedJobCannotBeReactivated: messages=[{"role": "user", "content": "hi"}], response_obj={"choices": [{"message": {"content": "real text"}}]}, real_model="gpt-4o", + real_response_tokens=42, model_parameters={}, parent_metadata={}, ) @@ -387,6 +388,44 @@ class TestStoppedJobCannotBeReactivated: assert call_kwargs["data"]["status"] == "running" +@pytest.mark.asyncio +class TestVerdictRowRecordsBothSides: + async def test_verdict_row_carries_ids_and_token_counts_for_both_arms(self): + """real_response_tokens next to shadow_response_tokens is what lets a reader + check the judge's verdicts for verbosity bias, and shadow_request_id joins the + verdict to the shadow call's own spend log for drill-down. Declared columns + that are never written are worse than absent ones.""" + prisma = MagicMock() + prisma.db.litellm_shadowevalverdict.create = AsyncMock() + prisma.db.litellm_shadowevaljob.update_many = AsyncMock() + logger = ShadowEvalLogger(router_provider=_router_mock, prisma_provider=lambda: prisma) + logger._call_router_shadow = AsyncMock( + return_value=_ShadowResponse( + text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10, request_id="shadow-req-1" + ) + ) + logger._call_judge = AsyncMock( + return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.01) + ) + + job = ActiveShadowEvalJob(id="j1", router_name="r", judge_model="m", shadow_percentage=100.0, status="running") + await logger._run_shadow_eval( + job=job, + request_id="req-1", + messages=[{"role": "user", "content": "hi"}], + response_obj={"choices": [{"message": {"content": "real text"}}]}, + real_model="gpt-4o", + real_response_tokens=42, + model_parameters={}, + parent_metadata={}, + ) + + data = prisma.db.litellm_shadowevalverdict.create.call_args.kwargs["data"] + assert data["shadow_request_id"] == "shadow-req-1" + assert data["real_response_tokens"] == 42 + assert data["shadow_response_tokens"] == 10 + + @pytest.mark.asyncio class TestVerdictWriteAccumulatesCost: async def test_cost_actual_uses_increment_not_a_raw_set(self): @@ -403,7 +442,7 @@ class TestVerdictWriteAccumulatesCost: logger = ShadowEvalLogger(router_provider=lambda: MagicMock(), prisma_provider=lambda: prisma) logger._call_router_shadow = AsyncMock( - return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10) + return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10, request_id="shadow-req-1") ) logger._call_judge = AsyncMock( return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.05) @@ -416,6 +455,7 @@ class TestVerdictWriteAccumulatesCost: messages=[{"role": "user", "content": "hi"}], response_obj={"choices": [{"message": {"content": "real text"}}]}, real_model="gpt-4o", + real_response_tokens=42, model_parameters={}, parent_metadata={}, ) @@ -534,7 +574,7 @@ class TestSubCallsAreAttributedToTheShadowedKey: logger, _, _ = _logger_with_mocks(job) logger._prisma_provider = lambda: prisma logger._call_router_shadow = AsyncMock( - return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10) + return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10, request_id="shadow-req-1") ) logger._call_judge = AsyncMock( return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.01) @@ -1109,6 +1149,7 @@ class TestJudgeFailureModes: messages=({"role": "user", "content": "hi"},), response_obj={"choices": [{"message": {"content": "real says"}}]}, real_model="gpt-4o", + real_response_tokens=42, model_parameters={}, parent_metadata={}, ) diff --git a/tests/test_litellm/proxy/proxy_server/test_lifecycle.py b/tests/test_litellm/proxy/proxy_server/test_lifecycle.py index 5edd68cbbb3..2b75e24d053 100644 --- a/tests/test_litellm/proxy/proxy_server/test_lifecycle.py +++ b/tests/test_litellm/proxy/proxy_server/test_lifecycle.py @@ -512,19 +512,45 @@ def test_cost_tracking_adds_two_callbacks_when_prisma_set(monkeypatch): @pytest.mark.asyncio -async def test_startup_starts_the_shadow_eval_lifecycle_loop(monkeypatch): - """The loop owns counter flushes and finishing expired jobs, so a job on a key - that goes quiet still ends on schedule; it must be running after startup.""" +async def test_registration_starts_the_lifecycle_loop_on_the_registered_instance(monkeypatch): + """Registration owns the loop start, so there is no startup ordering to get + wrong, and the start lands on the instance the callback manager kept, which + is the ORIGINAL one when its class dedup discards a re-registration.""" import litellm from litellm.integrations.shadow_eval_logger import ShadowEvalLogger - logger = ShadowEvalLogger(router_provider=lambda: None, prisma_provider=lambda: None) - monkeypatch.setattr(litellm, "callbacks", [logger], raising=False) + fake_prisma = MagicMock() + monkeypatch.setattr(ps, "prisma_client", fake_prisma, raising=False) + monkeypatch.setattr(litellm, "callbacks", [], raising=False) + monkeypatch.setattr(litellm, "_async_success_callback", [], raising=False) - ps._start_shadow_eval_lifecycle_loop() + cost_tracking() + cost_tracking() - assert logger._lifecycle_task is not None and not logger._lifecycle_task.done() - logger._lifecycle_task.cancel() + loggers = [cb for cb in litellm.callbacks if isinstance(cb, ShadowEvalLogger)] + assert len(loggers) == 1 + task = loggers[0]._lifecycle_task + assert task is not None and not task.done() + task.cancel() + + +def test_cost_tracking_twice_registers_one_shadow_eval_logger(monkeypatch): + """cost_tracking can run again (config reload); the callback manager dedups + CustomLogger instances by class, so a second call must not produce a second + ShadowEvalLogger with its own snapshot, buffer, and in-flight counter.""" + import litellm + from litellm.integrations.shadow_eval_logger import ShadowEvalLogger + + fake_prisma = MagicMock() + monkeypatch.setattr(ps, "prisma_client", fake_prisma, raising=False) + monkeypatch.setattr(litellm, "callbacks", [], raising=False) + monkeypatch.setattr(litellm, "_async_success_callback", [], raising=False) + + cost_tracking() + cost_tracking() + + shadow_loggers = [cb for cb in litellm.callbacks if isinstance(cb, ShadowEvalLogger)] + assert len(shadow_loggers) == 1 def test_cost_tracking_no_op_when_prisma_missing(monkeypatch):