diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20260807000000_add_shadow_eval_job/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260807000000_add_shadow_eval_job/migration.sql index 2242a29fae6..9f643adfa81 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/migrations/20260807000000_add_shadow_eval_job/migration.sql +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260807000000_add_shadow_eval_job/migration.sql @@ -10,7 +10,7 @@ CREATE TABLE IF NOT EXISTS "LiteLLM_ShadowEvalJob" ( "request_count" INTEGER NOT NULL DEFAULT 0, "completed_count" INTEGER NOT NULL DEFAULT 0, "failed_count" INTEGER NOT NULL DEFAULT 0, - "result_json" JSONB, + "last_error" TEXT, "cost_estimate" DOUBLE PRECISION, "cost_actual" DOUBLE PRECISION NOT NULL DEFAULT 0, "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma index c955d5265b2..a430b17be2b 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma +++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma @@ -1466,10 +1466,7 @@ model LiteLLM_ShadowEvalJob { request_count Int @default(0) // total requests on the shadowed key during the job completed_count Int @default(0) // verdicts written failed_count Int @default(0) // shadow calls or judge calls that failed - - // Results (written incrementally or at end) - // { groups: [ { tier: "SIMPLE", turn_count: N, real_win_rate_pct: X, shadow_win_rate_pct: Y, tie_rate_pct: Z, avg_confidence: C } ] } - result_json Json? + last_error String? // most recent shadow/judge failure, for diagnosing a growing failed_count cost_estimate Float? // estimated cost of running the judge cost_actual Float @default(0) // actual cost (judge calls * price) diff --git a/litellm/integrations/shadow_eval_logger.py b/litellm/integrations/shadow_eval_logger.py index 65147abb609..059141b3c57 100644 --- a/litellm/integrations/shadow_eval_logger.py +++ b/litellm/integrations/shadow_eval_logger.py @@ -15,15 +15,21 @@ Flow per sampled request (all in a detached background task, zero added latency) the judge cannot learn a position bias. 3. Write a ``LiteLLM_ShadowEvalVerdict`` row and bump the job's counters. -Shadow and judge calls carry ``shadow_eval_internal`` metadata so this logger -ignores its own traffic and cannot recurse. They also carry the shadowed key's -identity metadata, so the provider spend they incur is attributed to that key -(and its team/org/user) and counts against every budget that key is subject to, -including the global proxy budget. A shadow/judge pair is skipped outright if -the shadowed key or its team is already at or over budget, read from the same -cross-pod spend counters the normal auth path reserves against. A job -additionally stops itself once its sampling window (``ends_at``) closes or its -judge spend reaches a multiple of the estimate quoted when it started. +Shadow and judge calls carry an ``internal_call_origin`` stamp, the same field +every internal sub-call (including the auto-router's own classifier) is marked +with, and the logger skips any request carrying it: its own traffic cannot +recurse, and internal sub-calls billed to the shadowed key are never judged as +if a user sent them. The calls also carry the shadowed key's identity metadata, +so the provider spend they incur is attributed to that key (and its team/org/ +user) and counts against every budget that key is subject to, including the +global proxy budget. A shadow/judge pair is skipped outright if the shadowed +key or its team is already at or over budget, read from the same cross-pod +spend counters the normal auth path reserves against. + +Job lifecycle (counter flushes, snapshot refresh, stopping a job at its +``ends_at`` or judge-spend cap) runs on a periodic loop started at proxy +startup, never on the request path: an idle key's job still ends on schedule, +and the final counter batch lands without needing another request to arrive. """ import asyncio @@ -41,6 +47,7 @@ from pydantic import BaseModel, TypeAdapter import litellm from litellm._logging import verbose_logger +from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.internal_call_metadata import sanitized_forwardable_call_metadata from litellm.types.utils import SHADOW_EVAL_JUDGE_CALL_ORIGIN, SHADOW_EVAL_ROUTER_CALL_ORIGIN @@ -52,13 +59,10 @@ if TYPE_CHECKING: _JSON_FENCE_RE: Final = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL | re.IGNORECASE) -# Metadata marker that tags shadow/judge calls made by this logger, so the -# success hook skips them instead of shadowing the shadow. -SHADOW_EVAL_INTERNAL_MARKER: Final = "shadow_eval_internal" - -# How long the active-job snapshot is cached. A shadow-eval job starting -# or stopping takes up to this long to be noticed by running pods. -_JOB_CACHE_TTL_SECONDS: Final = 30.0 +# Cadence of the lifecycle loop: snapshot refresh, counter flush, and job +# finalization all run on this tick. A job starting or stopping takes up to +# one tick to be noticed by running pods. +_LIFECYCLE_TICK_SECONDS: Final = 10.0 # Upper bound on concurrent shadow+judge pipelines per pod, so a traffic spike # turns into skipped samples rather than an unbounded task pileup. @@ -73,13 +77,14 @@ _MAX_JUDGE_CHARS: Final = 16_000 # to failed_count, while a larger one buys nothing but cost exposure. JUDGE_MAX_OUTPUT_TOKENS: Final = 500 -_SEEN_FLUSH_INTERVAL_SECONDS: Final = 10.0 +# Persisted diagnosis bound: the most recent shadow/judge failure, truncated. +_MAX_LAST_ERROR_CHARS: Final = 500 # A job stops sampling once its judge spend reaches this multiple of the estimate shown # at start. The headroom absorbs an estimate that undershot the real traffic mix without # letting the job run away; the floor keeps a cent-sized estimate from stopping a job on -# its first verdict. cost_actual is read from the job cache, so overshoot is bounded by -# _JOB_CACHE_TTL_SECONDS worth of judge calls. +# its first verdict. cost_actual is read from the job snapshot, so overshoot is bounded +# by _LIFECYCLE_TICK_SECONDS worth of judge calls. _SPEND_CAP_MULTIPLIER: Final = 1.5 _SPEND_CAP_FLOOR_USD: Final = 1.0 @@ -165,6 +170,33 @@ def _unmask_preference(raw_preference: str, real_is_a: bool) -> str: return "tie" +@dataclass(frozen=True, slots=True) +class _CallFailure: + """A shadow or judge call that produced no usable response, with why.""" + + error: str + + +@dataclass(frozen=True, slots=True) +class _ShadowResponse: + """A successful shadow call: (text, routed model, tier, completion tokens).""" + + text: str + model: str + tier: str | None + completion_tokens: int | None + + +@dataclass(frozen=True, slots=True) +class _JudgeVerdict: + """A parsed judge verdict, unmasked back to real/shadow/tie.""" + + preference: str + confidence: float + reasoning: str + cost: float + + @dataclass(frozen=True, slots=True) class ActiveShadowEvalJob: """The subset of a shadow-eval job row the request path actually needs.""" @@ -279,15 +311,44 @@ class ShadowEvalLogger(CustomLogger): self._router_provider = router_provider or _default_router_provider self._prisma_provider = prisma_provider or _default_prisma_provider # Snapshot of every active job, keyed by shadowed api_key_id. Refreshed as a - # whole: active jobs are admin-started and capped at one per key, so the set is - # tiny, and one find_many per pod per TTL keeps DB load flat no matter how many - # distinct keys the proxy serves. - self._jobs_by_key: dict[str, ActiveShadowEvalJob] = {} # mutable-ok: TTL cache - self._jobs_fetched_at: float | None = None - self._jobs_refresh_task: asyncio.Task[None] | None = None + # whole by the lifecycle loop: active jobs are admin-started and capped at one + # per key, so the set is tiny, and one find_many per pod per tick keeps DB load + # flat no matter how many distinct keys the proxy serves. + self._jobs_by_key: dict[str, ActiveShadowEvalJob] = {} # mutable-ok: loop-refreshed snapshot self._inflight_shadow_tasks: int = 0 self._pending_seen: dict[str, int] = {} # mutable-ok: flush buffer - self._last_seen_flush: float = 0.0 + 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.""" + if self._lifecycle_task is not None and not self._lifecycle_task.done(): + return + self._lifecycle_task = asyncio.create_task(self._lifecycle_loop()) + + async def _lifecycle_loop(self) -> None: + while True: + try: + await self._lifecycle_tick() + except Exception as e: # noqa: BLE001 # the loop must survive any single tick failing + verbose_logger.debug("shadow_eval: lifecycle tick failed: %s", e) + await asyncio.sleep(_LIFECYCLE_TICK_SECONDS) + + async def _lifecycle_tick(self) -> None: + """One tick: flush counters while jobs are still active, refresh, then finalize. + + Flushing before finalizing lets an expiring job's last counter batch land while + its row still passes the active-status guard. + """ + await self._flush_seen_counts() + await self._refresh_active_jobs() + for job in tuple(self._jobs_by_key.values()): + if _job_is_past_its_end(job): + await self._finalize_job(job, "reached its scheduled end") + elif _job_is_over_spend_cap(job): + await self._finalize_job( + job, + f"spend ${job.cost_actual:.4f} reached the cap for its ${job.cost_estimate or 0.0:.4f} estimate", + ) #### hook #### @@ -310,33 +371,22 @@ class ShadowEvalLogger(CustomLogger): request_metadata: Final = ( raw_request_metadata if isinstance(raw_request_metadata, Mapping) else _EMPTY_METADATA ) - if request_metadata.get(SHADOW_EVAL_INTERNAL_MARKER): - return # our own shadow/judge traffic + if request_metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY): + return # internal sub-call (our own shadow/judge, a classifier, ...), not user traffic api_key_hash: Final = metadata.get("user_api_key_hash") if not api_key_hash: return - job: Final = self._get_active_job(api_key_hash) + job: Final = self._jobs_by_key.get(str(api_key_hash)) if job is None: return - if _job_is_past_its_end(job): - await self._finalize_job(job, "reached its scheduled end") - return - if _job_is_over_spend_cap(job): - await self._finalize_job( - job, - f"spend ${job.cost_actual:.4f} reached the cap for its ${job.cost_estimate or 0.0:.4f} estimate", - ) - return + if _job_is_past_its_end(job) or _job_is_over_spend_cap(job): + return # stop sampling now; the lifecycle loop finalizes the row request_id: Final = payload.get("id") or "" if not request_id: return # The job tracks every request it saw, sampled or not, so the UI can - # show "N of M requests shadowed". + # show "N of M requests shadowed". Flushed by the lifecycle loop. self._pending_seen[job.id] = self._pending_seen.get(job.id, 0) + 1 - now: Final = asyncio.get_event_loop().time() - if now - self._last_seen_flush >= _SEEN_FLUSH_INTERVAL_SECONDS: - self._last_seen_flush = now - asyncio.create_task(self._flush_seen_counts()) if not _sample_hits(request_id, job.id, job.shadow_percentage): return if payload.get("call_type") not in (None, "completion", "acompletion", "chat_completion"): @@ -369,23 +419,9 @@ class ShadowEvalLogger(CustomLogger): #### job lookup #### - def _get_active_job(self, api_key_hash: str) -> ActiveShadowEvalJob | None: - """Serve from the snapshot; never await the DB on the request path. - - An expired (or missing) snapshot kicks a detached refresh and this request is - answered from whatever is already in memory — possibly stale, on a cold pod - possibly empty. A sampled eval tolerates a missed slice far better than every - request on the proxy tolerating a synchronous Prisma read in its success hook. - """ - now: Final = asyncio.get_event_loop().time() - expired: Final = self._jobs_fetched_at is None or now - self._jobs_fetched_at >= _JOB_CACHE_TTL_SECONDS - if expired and (self._jobs_refresh_task is None or self._jobs_refresh_task.done()): - self._jobs_refresh_task = asyncio.create_task(self._refresh_active_jobs(now)) - return self._jobs_by_key.get(api_key_hash) - - async def _refresh_active_jobs(self, now: float) -> None: + async def _refresh_active_jobs(self) -> None: """Reload the active-job set. On a DB blip the stale snapshot is kept and the - next TTL retries, so a blip degrades freshness rather than turning the feature off.""" + next tick retries, so a blip degrades freshness rather than turning the feature off.""" prisma: Final = self._prisma_provider() if prisma is None: return @@ -410,7 +446,6 @@ class ShadowEvalLogger(CustomLogger): ends_at=_as_utc(getattr(record, "ends_at", None)), ) self._jobs_by_key = jobs_by_key # mutable-ok: atomic snapshot swap - self._jobs_fetched_at = now async def _finalize_job(self, job: ActiveShadowEvalJob, reason: str) -> None: """Flip a finished job to completed, keeping the verdicts it already produced. @@ -440,15 +475,24 @@ class ShadowEvalLogger(CustomLogger): verbose_logger.debug("shadow_eval: failed to stop job %s: %s", job.id, e) async def _flush_seen_counts(self) -> None: + """Write the buffered request counts, guarded on the job still being active. + + The guard makes stopping a job freeze its counter: a pod serving a stale + snapshot keeps buffering for up to one tick, and this write drops those + increments instead of growing a stopped job's request_count. + """ prisma: Final = self._prisma_provider() - if prisma is None: + if prisma is None or not self._pending_seen: return pending: Final = self._pending_seen self._pending_seen = {} # mutable-ok: fresh flush buffer for job_id, count in pending.items(): try: - await prisma.db.litellm_shadowevaljob.update( - where={"id": job_id}, # mutable-ok: Prisma filter + await prisma.db.litellm_shadowevaljob.update_many( + where={ # mutable-ok: Prisma filter + "id": job_id, + "status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter + }, data={"request_count": {"increment": count}}, # mutable-ok: Prisma payload ) except Exception as e: # noqa: BLE001 # counter drift is acceptable; failing the loop is not @@ -482,22 +526,20 @@ class ShadowEvalLogger(CustomLogger): return shadow: Final = await self._call_router_shadow(job.router_name, messages, model_parameters, parent_metadata) - if shadow is None: - await self._bump_failed(job.id) + if isinstance(shadow, _CallFailure): + await self._bump_failed(job.id, shadow.error) return - shadow_text, shadow_model, tier, shadow_tokens = shadow verdict: Final = await self._call_judge( judge_model=job.judge_model, messages=messages, real_text=real_text, - shadow_text=shadow_text, + shadow_text=shadow.text, parent_metadata=parent_metadata, ) - if verdict is None: - await self._bump_failed(job.id) + if isinstance(verdict, _CallFailure): + await self._bump_failed(job.id, verdict.error) return - preference, confidence, reasoning, judge_cost = verdict if prisma is None: return @@ -505,13 +547,13 @@ class ShadowEvalLogger(CustomLogger): data={ # mutable-ok: Prisma payload "job_id": job.id, "request_id": request_id, - "tier_classification": tier, + "tier_classification": shadow.tier, "real_model": real_model, - "shadow_model": shadow_model, - "shadow_response_tokens": shadow_tokens, - "judge_preference": preference, - "judge_confidence": confidence, - "judge_reasoning": reasoning[:1000] if reasoning else None, + "shadow_model": shadow.model, + "shadow_response_tokens": shadow.completion_tokens, + "judge_preference": verdict.preference, + "judge_confidence": verdict.confidence, + "judge_reasoning": verdict.reasoning[:1000] if verdict.reasoning else None, "judge_model": job.judge_model, } ) @@ -522,22 +564,28 @@ class ShadowEvalLogger(CustomLogger): }, data={ # mutable-ok: Prisma payload "completed_count": {"increment": 1}, # mutable-ok: Prisma operator - "cost_actual": {"increment": judge_cost}, # mutable-ok: Prisma operator + "cost_actual": {"increment": verdict.cost}, # mutable-ok: Prisma operator "status": "running", }, ) except Exception as e: # noqa: BLE001 # detached task: log, count, never raise verbose_logger.debug("shadow_eval: pipeline failed for %s: %s", request_id, e) - await self._bump_failed(job.id) + await self._bump_failed(job.id, f"pipeline error: {e}") - async def _bump_failed(self, job_id: str) -> None: + async def _bump_failed(self, job_id: str, error: str) -> None: prisma: Final = self._prisma_provider() if prisma is None: return try: - await prisma.db.litellm_shadowevaljob.update( - where={"id": job_id}, # mutable-ok: Prisma filter - data={"failed_count": {"increment": 1}}, # mutable-ok: Prisma payload + await prisma.db.litellm_shadowevaljob.update_many( + where={ # mutable-ok: Prisma filter + "id": job_id, + "status": {"in": ["pending", "running"]}, # mutable-ok: Prisma filter + }, + data={ # mutable-ok: Prisma payload + "failed_count": {"increment": 1}, # mutable-ok: Prisma operator + "last_error": error[:_MAX_LAST_ERROR_CHARS], + }, ) except Exception as e: # noqa: BLE001 # counter drift is acceptable verbose_logger.debug("shadow_eval: failed_count increment failed: %s", e) @@ -548,20 +596,18 @@ class ShadowEvalLogger(CustomLogger): messages: Sequence[Mapping[str, object]], model_parameters: Mapping[str, object], parent_metadata: Mapping[str, object], - ) -> tuple[str, str, str | None, int | None] | None: - """Send the prompt through the auto-router; return (text, model, tier, completion_tokens).""" + ) -> "_ShadowResponse | _CallFailure": + """Send the prompt through the auto-router being evaluated.""" router: Final = self._router_provider() if router is None: - verbose_logger.debug("shadow_eval: no router available") - return None + return _CallFailure("no router configured on this pod") # Carries the shadowed key's identity so this call's real provider spend is # attributed to, and budget-checked against, the key whose traffic it copies. # The router's pre-routing hook also writes its routing decision into this dict; # read it back after the call for tier attribution. - attribution: Final = sanitized_forwardable_call_metadata(parent_metadata, SHADOW_EVAL_ROUTER_CALL_ORIGIN) - shadow_metadata: Final[dict[str, object]] = attribution | { # mutable-ok: router writes back - SHADOW_EVAL_INTERNAL_MARKER: True - } + shadow_metadata: Final[dict[str, object]] = sanitized_forwardable_call_metadata( # mutable-ok: router writes back + parent_metadata, SHADOW_EVAL_ROUTER_CALL_ORIGIN + ) shadow_params: Final = { # mutable-ok: splatted as kwargs k: v for k, v in model_parameters.items() if k not in ("stream", "metadata") } @@ -574,19 +620,21 @@ class ShadowEvalLogger(CustomLogger): ) except Exception as e: # noqa: BLE001 # provider errors are a counted failure, not a crash verbose_logger.debug("shadow_eval: router call failed: %s", e) - return None + return _CallFailure(f"shadow router call failed: {e}") text: Final = self._extract_response_text(response) if not text: - return None + return _CallFailure("shadow router returned an empty response") raw_decision: Final = shadow_metadata.get("routing_decision") routing_decision: Final = raw_decision if isinstance(raw_decision, Mapping) else _EMPTY_METADATA raw_tier: Final = routing_decision.get("tier_label") or routing_decision.get("tier") - tier: Final = str(raw_tier) if raw_tier is not None else None - model: Final = str(getattr(response, "model", None) or routing_decision.get("routed_model") or "") usage: Final = getattr(response, "usage", None) raw_tokens: Final = getattr(usage, "completion_tokens", None) if usage is not None else None - completion_tokens: Final = int(raw_tokens) if isinstance(raw_tokens, int) else None - return text, model, tier, completion_tokens + 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, + ) async def _call_judge( self, @@ -595,8 +643,10 @@ class ShadowEvalLogger(CustomLogger): real_text: str, shadow_text: str, parent_metadata: Mapping[str, object], - ) -> tuple[str, float, str, float] | None: - """Blind pairwise judge. Returns (preference, confidence, reasoning, cost).""" + ) -> "_JudgeVerdict | _CallFailure": + """Blind pairwise judge, dispatched through the proxy's router when the judge + model is a configured deployment (so DB-stored credentials work), the SDK + otherwise (provider-qualified public names with credentials in the env).""" real_is_a: Final = random.random() < 0.5 response_a: Final = real_text if real_is_a else shadow_text response_b: Final = shadow_text if real_is_a else real_text @@ -612,32 +662,50 @@ class ShadowEvalLogger(CustomLogger): f"Response B:\n{response_b[:_MAX_JUDGE_CHARS]}\n\n" "Which response is better?" ) - judge_metadata: Final = sanitized_forwardable_call_metadata(parent_metadata, SHADOW_EVAL_JUDGE_CALL_ORIGIN) | { - SHADOW_EVAL_INTERNAL_MARKER: True - } + judge_metadata: Final = sanitized_forwardable_call_metadata(parent_metadata, SHADOW_EVAL_JUDGE_CALL_ORIGIN) + judge_messages: Final = [ # mutable-ok: SDK takes a list + {"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT}, # mutable-ok: SDK message + {"role": "user", "content": user_prompt}, # mutable-ok: SDK message + ] + router: Final = self._router_provider() + use_router: Final = router is not None and ( + judge_model in router.model_group_alias or router.get_model_list(model_name=judge_model) + ) try: - response: Final = await litellm.acompletion( - model=judge_model, - messages=[ # mutable-ok: SDK takes a list - {"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT}, # mutable-ok: SDK message - {"role": "user", "content": user_prompt}, # mutable-ok: SDK message - ], - temperature=0, - max_tokens=JUDGE_MAX_OUTPUT_TOKENS, - metadata=judge_metadata, + response: Final = ( + await router.acompletion( # pyright: ignore[reportOptionalMemberAccess] # use_router implies router is not None + model=judge_model, + messages=judge_messages, + temperature=0, + max_tokens=JUDGE_MAX_OUTPUT_TOKENS, + metadata=judge_metadata, + num_retries=0, + fallbacks=[], + ) + if use_router + else await litellm.acompletion( + model=judge_model, + messages=judge_messages, + temperature=0, + max_tokens=JUDGE_MAX_OUTPUT_TOKENS, + metadata=judge_metadata, + ) ) except Exception as e: # noqa: BLE001 # judge outages are a counted failure, not a crash verbose_logger.debug("shadow_eval: judge call failed: %s", e) - return None + return _CallFailure(f"judge call failed: {e}") try: raw: Final = response["choices"][0]["message"]["content"] or "" verdict: Final = _parse_pairwise_verdict(raw) except (KeyError, IndexError, TypeError, ValueError, json.JSONDecodeError) as e: verbose_logger.debug("shadow_eval: unparseable judge verdict: %s", e) - return None - preference: Final = _unmask_preference(verdict.preference, real_is_a) - confidence: Final = max(0.0, min(1.0, verdict.confidence)) - return preference, confidence, verdict.reasoning, _judge_call_cost(response) + return _CallFailure(f"unparseable judge verdict: {e}") + return _JudgeVerdict( + preference=_unmask_preference(verdict.preference, real_is_a), + confidence=max(0.0, min(1.0, verdict.confidence)), + reasoning=verdict.reasoning, + cost=_judge_call_cost(response), + ) @staticmethod def _extract_response_text(response_obj: object) -> str: diff --git a/litellm/proxy/db/autorouter_session_rollup.py b/litellm/proxy/db/autorouter_session_rollup.py index 9732b1d7402..1c8b72e3c61 100644 --- a/litellm/proxy/db/autorouter_session_rollup.py +++ b/litellm/proxy/db/autorouter_session_rollup.py @@ -180,12 +180,17 @@ def build_autorouter_turn_transaction( The routing_decision record is what says a request was auto-routed at all, so a request without one (including the auto-router's own classifier sub-calls) never - reaches the rollup. Failed requests served nothing and are excluded. Cache facts - are derived from the payload's own usage record through the savings owner, never - handed in beside it. + reaches the rollup. Internal sub-calls that DO carry one (a shadow eval's duplicate + of a request through the router) are excluded by their internal_call_origin stamp: + they are not traffic a user sent, so counting them would manufacture sessions and + savings in the adoption metrics. Failed requests served nothing and are excluded. + Cache facts are derived from the payload's own usage record through the savings + owner, never handed in beside it. """ if payload.get("status") != "success": return None + if metadata.get("internal_call_origin"): + return None routing_decision: Final = metadata.get("routing_decision") if not isinstance(routing_decision, Mapping) or not routing_decision: return None diff --git a/litellm/proxy/db/db_spend_update_writer.py b/litellm/proxy/db/db_spend_update_writer.py index b0130db232a..a0c0ec1ec7d 100644 --- a/litellm/proxy/db/db_spend_update_writer.py +++ b/litellm/proxy/db/db_spend_update_writer.py @@ -1794,6 +1794,7 @@ class DBSpendUpdateWriter: if call_type: endpoint = ROUTE_ENDPOINT_MAPPING.get(call_type, None) + is_internal_call: Final = bool(_metadata.get("internal_call_origin")) cache_read_input_tokens: Final = extract_cache_read_tokens(usage_obj) compression_saved_tokens: Final = extract_compression_saved_tokens(_metadata) savings_spend: Final = compute_savings_spend( @@ -1818,15 +1819,21 @@ class DBSpendUpdateWriter: prompt_tokens=payload["prompt_tokens"], completion_tokens=payload["completion_tokens"], spend=payload["spend"], - api_requests=1, - successful_requests=1 if request_status == "success" else 0, - failed_requests=1 if request_status != "success" else 0, + # Internal sub-calls (auto-router classifier, shadow eval's shadow and + # judge) bill real spend and tokens to the key, but they are not + # requests the caller made: counting them inflates request-volume + # readers (usage dashboards, the shadow-eval cost estimate), and an + # auto-router savings figure computed on a shadow duplicate credits + # savings for traffic no user sent. + api_requests=0 if is_internal_call else 1, + successful_requests=1 if not is_internal_call and request_status == "success" else 0, + failed_requests=1 if not is_internal_call and request_status != "success" else 0, cache_read_input_tokens=cache_read_input_tokens, cache_creation_input_tokens=extract_cache_creation_tokens(usage_obj), compression_saved_tokens=compression_saved_tokens, compression_savings_spend=savings_spend.compression, prompt_caching_savings_spend=savings_spend.prompt_caching, - autorouter_savings_spend=savings_spend.autorouter, + autorouter_savings_spend=0.0 if is_internal_call else savings_spend.autorouter, ) return daily_transaction except Exception as e: diff --git a/litellm/proxy/management_endpoints/auto_router_endpoints.py b/litellm/proxy/management_endpoints/auto_router_endpoints.py index 2ad7da7877b..62217e230a9 100644 --- a/litellm/proxy/management_endpoints/auto_router_endpoints.py +++ b/litellm/proxy/management_endpoints/auto_router_endpoints.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Annotated, Final from pydantic import BaseModel, ConfigDict, TypeAdapter +import litellm from litellm._logging import verbose_proxy_logger from litellm.exceptions import BudgetExceededError from litellm.integrations.shadow_eval_logger import JUDGE_MAX_OUTPUT_TOKENS @@ -524,19 +525,61 @@ def _is_configured_pre_routing_strategy(llm_router: "Router", router_name: str) ) -def _estimate_judge_cost_per_call(judge_model: str) -> float: - """Price one judge call: ~4k prompt tokens (two responses + conversation) + the judge's output budget.""" - try: - import litellm as _litellm +def _judge_model_is_router_resolvable(llm_router: "Router | None", judge_model: str) -> bool: + """Whether the judge name resolves through the proxy's router, the same check the + judge dispatch itself makes (and llm_as_a_judge before it), so validation cannot + accept a name the call path then fails on.""" + return llm_router is not None and bool( + judge_model in llm_router.model_group_alias or llm_router.get_model_list(model_name=judge_model) + ) - prompt_cost, completion_cost = _litellm.cost_per_token( - model=judge_model, prompt_tokens=_JUDGE_PROMPT_TOKENS_ESTIMATE, completion_tokens=JUDGE_MAX_OUTPUT_TOKENS + +def _validate_judge_model(llm_router: "Router | None", judge_model: str) -> None: + """Reject a judge model the dispatch path cannot resolve, at start rather than as a + silently growing failed_count once the job is already sampling and billing.""" + if llm_router is not None and _is_configured_pre_routing_strategy(llm_router, judge_model): + raise HTTPException( + status_code=400, + detail=f"judge_model '{judge_model}' is an auto-router; the judge must be a plain model", + ) + if _judge_model_is_router_resolvable(llm_router, judge_model): + return + try: + litellm.get_llm_provider(model=judge_model) + except Exception as e: + raise HTTPException( + status_code=400, + detail=( + f"judge_model '{judge_model}' is neither a model configured on this proxy nor a " + "provider-qualified public model name (e.g. 'anthropic/claude-sonnet-5')" + ), + ) from e + + +def _judge_pricing_model(llm_router: "Router | None", judge_model: str) -> str: + """The name to price the judge under: a configured deployment's underlying + provider model when the judge is a deployment (the deployment name itself is + admin-arbitrary and not a pricing key), the given name otherwise.""" + deployments: Final = llm_router.get_model_list(model_name=judge_model) if llm_router is not None else None + if deployments: + underlying: Final = deployments[0].get("litellm_params", {}).get("model") + if isinstance(underlying, str) and underlying: + return underlying + return judge_model + + +def _estimate_judge_cost_per_call(llm_router: "Router | None", judge_model: str) -> float: + """Price one judge call: ~4k prompt tokens (two responses + conversation) + the judge's output budget.""" + pricing_model: Final = _judge_pricing_model(llm_router, judge_model) + try: + prompt_cost, completion_cost = litellm.cost_per_token( + model=pricing_model, prompt_tokens=_JUDGE_PROMPT_TOKENS_ESTIMATE, completion_tokens=JUDGE_MAX_OUTPUT_TOKENS ) estimated: Final = prompt_cost + completion_cost if estimated > 0: return estimated except Exception as e: # noqa: BLE001 # unknown judge model: fall back to a flat per-call figure - verbose_proxy_logger.debug("shadow_eval: judge cost lookup failed for %s: %s", judge_model, e) + verbose_proxy_logger.debug("shadow_eval: judge cost lookup failed for %s: %s", pricing_model, e) return _FALLBACK_JUDGE_COST_PER_CALL @@ -652,6 +695,7 @@ class _ShadowEvalJobRow(BaseModel): request_count: int completed_count: int failed_count: int + last_error: str | None = None cost_estimate: float | None = None cost_actual: float = 0.0 created_at: datetime @@ -674,6 +718,7 @@ def _job_to_response(record: object, results: ShadowEvalResult | None) -> GetSha request_count=row.request_count, completed_count=row.completed_count, failed_count=row.failed_count, + last_error=row.last_error, results=results, cost_estimate=row.cost_estimate, cost_actual=row.cost_actual, @@ -715,6 +760,7 @@ async def start_shadow_eval( status_code=400, detail=f"'{data.router_name}' is not a configured auto-router", ) + _validate_judge_model(llm_router, data.judge_model) existing: Final = await prisma_client.db.litellm_shadowevaljob.find_first( where={ # mutable-ok: Prisma filter @@ -732,7 +778,7 @@ async def start_shadow_eval( sampled: Final = int( recent_requests * (data.duration_days / _ESTIMATE_LOOKBACK_DAYS) * data.shadow_percentage / 100.0 ) - per_call: Final = _estimate_judge_cost_per_call(data.judge_model) + per_call: Final = _estimate_judge_cost_per_call(llm_router, data.judge_model) estimated_cost: Final = round(sampled * per_call, 2) ends_at: Final = datetime.now(timezone.utc) + timedelta(days=data.duration_days) @@ -829,7 +875,7 @@ async def stop_shadow_eval_job( job_id: str, user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], ) -> GetShadowEvalJobResponse: - """Stop an active shadow eval job. Existing verdicts are kept; sampling halts within ~30s.""" + """Stop an active shadow eval job. Existing verdicts are kept; sampling halts within ~10s.""" from litellm.proxy.proxy_server import prisma_client _require_admin_writer(user_api_key_dict, "stop a shadow eval") diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 67652b8c59c..0803d9b8b69 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -1184,6 +1184,8 @@ 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() @@ -2219,13 +2221,29 @@ def _register_shadow_eval_logger() -> None: 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. + it has the same hard dependency on prisma_client. Its lifecycle loop is + started separately at proxy startup, once an event loop exists. """ 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() + return + + # Bounds authoritative DB re-reads when enforcing a budget against a # stale-low spend counter: at most one DB read per counter per window. SPEND_DB_FLOOR_CACHE_TTL_SECONDS: Final = 5 diff --git a/litellm/proxy/schema.prisma b/litellm/proxy/schema.prisma index c955d5265b2..a430b17be2b 100644 --- a/litellm/proxy/schema.prisma +++ b/litellm/proxy/schema.prisma @@ -1466,10 +1466,7 @@ model LiteLLM_ShadowEvalJob { request_count Int @default(0) // total requests on the shadowed key during the job completed_count Int @default(0) // verdicts written failed_count Int @default(0) // shadow calls or judge calls that failed - - // Results (written incrementally or at end) - // { groups: [ { tier: "SIMPLE", turn_count: N, real_win_rate_pct: X, shadow_win_rate_pct: Y, tie_rate_pct: Z, avg_confidence: C } ] } - result_json Json? + last_error String? // most recent shadow/judge failure, for diagnosing a growing failed_count cost_estimate Float? // estimated cost of running the judge cost_actual Float @default(0) // actual cost (judge calls * price) diff --git a/litellm/types/management_endpoints/auto_router_endpoints.py b/litellm/types/management_endpoints/auto_router_endpoints.py index 3aa9acd84d9..a7e48c1a469 100644 --- a/litellm/types/management_endpoints/auto_router_endpoints.py +++ b/litellm/types/management_endpoints/auto_router_endpoints.py @@ -250,6 +250,9 @@ class GetShadowEvalJobResponse(BaseModel): request_count: int = Field(description="Total requests observed on the shadowed key since the job started") completed_count: int = Field(description="Verdicts written so far") failed_count: int = Field(description="Shadow or judge calls that errored and were skipped") + last_error: str | None = Field( + default=None, description="The most recent shadow or judge failure, so a growing failed_count is diagnosable" + ) results: ShadowEvalResult | None = Field( default=None, description="Present once at least one verdict has been recorded" ) diff --git a/schema.prisma b/schema.prisma index c955d5265b2..a430b17be2b 100644 --- a/schema.prisma +++ b/schema.prisma @@ -1466,10 +1466,7 @@ model LiteLLM_ShadowEvalJob { request_count Int @default(0) // total requests on the shadowed key during the job completed_count Int @default(0) // verdicts written failed_count Int @default(0) // shadow calls or judge calls that failed - - // Results (written incrementally or at end) - // { groups: [ { tier: "SIMPLE", turn_count: N, real_win_rate_pct: X, shadow_win_rate_pct: Y, tie_rate_pct: Z, avg_confidence: C } ] } - result_json Json? + last_error String? // most recent shadow/judge failure, for diagnosing a growing failed_count cost_estimate Float? // estimated cost of running the judge cost_actual Float @default(0) // actual cost (judge calls * price) diff --git a/tests/test_litellm/integrations/test_shadow_eval_logger.py b/tests/test_litellm/integrations/test_shadow_eval_logger.py index 0d4ff456dc4..4cc1224d4bb 100644 --- a/tests/test_litellm/integrations/test_shadow_eval_logger.py +++ b/tests/test_litellm/integrations/test_shadow_eval_logger.py @@ -10,12 +10,14 @@ import pytest from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY from litellm.integrations.shadow_eval_logger import ( JUDGE_MAX_OUTPUT_TOKENS, - SHADOW_EVAL_INTERNAL_MARKER, ActiveShadowEvalJob, ShadowEvalLogger, + _CallFailure, + _JudgeVerdict, _key_or_team_is_over_budget, _parse_pairwise_verdict, _sample_hits, + _ShadowResponse, _unmask_preference, ) from litellm.types.utils import SHADOW_EVAL_JUDGE_CALL_ORIGIN, SHADOW_EVAL_ROUTER_CALL_ORIGIN @@ -99,12 +101,12 @@ class TestJudgeOutputBudget: ) monkeypatch.setattr(litellm_module, "acompletion", acompletion) - logger = ShadowEvalLogger(router_provider=lambda: MagicMock(), prisma_provider=lambda: MagicMock()) + logger = ShadowEvalLogger(router_provider=_router_mock, prisma_provider=lambda: MagicMock()) verdict = await logger._call_judge( "gpt-4o-mini", [{"role": "user", "content": "hi"}], "real text", "shadow text", {} ) - assert verdict is not None + assert not isinstance(verdict, _CallFailure) _, kwargs = acompletion.call_args assert kwargs["max_tokens"] == JUDGE_MAX_OUTPUT_TOKENS @@ -115,13 +117,36 @@ class TestJudgeOutputBudgetStaticChecks: assert JUDGE_MAX_OUTPUT_TOKENS >= 500 +def _job_record(job: ActiveShadowEvalJob, api_key_id: str = "key-hash") -> MagicMock: + """A Prisma row double mirroring an ActiveShadowEvalJob, for snapshot refreshes.""" + record = MagicMock() + record.id = job.id + record.api_key_id = api_key_id + record.router_name = job.router_name + record.shadow_percentage = job.shadow_percentage + record.judge_model = job.judge_model + record.status = job.status + record.cost_estimate = job.cost_estimate + record.cost_actual = job.cost_actual + record.ends_at = job.ends_at + return record + + +def _router_mock(resolves_judge: bool = False): + """A router double that, by default, does not resolve judge names, so judge + dispatch falls through to the SDK exactly as it did for public model names.""" + router = MagicMock() + router.model_group_alias = {} + router.get_model_list = MagicMock(return_value=[{"litellm_params": {"model": "openai/gpt-4o"}}] if resolves_judge else None) + return router + + def _logger_with_mocks(job=None): prisma = MagicMock() - router = MagicMock() + router = _router_mock() logger = ShadowEvalLogger(router_provider=lambda: router, prisma_provider=lambda: prisma) if job is not None: logger._jobs_by_key["key-hash"] = job - logger._jobs_fetched_at = asyncio.get_event_loop().time() return logger, prisma, router @@ -132,7 +157,16 @@ class TestSuccessHookSkipPaths: await logger.async_log_success_event({"messages": []}, MagicMock(), None, None) prisma.db.litellm_shadowevaljob.find_many.assert_not_called() - async def test_skips_own_internal_traffic(self): + @pytest.mark.parametrize( + "origin", + [SHADOW_EVAL_ROUTER_CALL_ORIGIN, SHADOW_EVAL_JUDGE_CALL_ORIGIN, "autorouter_classifier"], + ) + async def test_skips_any_internal_sub_call(self, origin): + """Regression: the auto-router's own classifier calls are logged as ordinary + successes on the shadowed key; sampling them judges a tier label against a + shadow answer and burns judge spend on traffic no user sent. One stamp, + internal_call_origin, classifies every internal sub-call, including this + logger's own shadow/judge traffic (the recursion guard).""" job = ActiveShadowEvalJob(id="j1", router_name="r", shadow_percentage=100.0, judge_model="m", status="running") logger, prisma, _ = _logger_with_mocks(job) kwargs = { @@ -141,7 +175,7 @@ class TestSuccessHookSkipPaths: "model": "gpt-4o", "metadata": {"user_api_key_hash": "key-hash"}, }, - "litellm_params": {"metadata": {SHADOW_EVAL_INTERNAL_MARKER: True}}, + "litellm_params": {"metadata": {INTERNAL_CALL_ORIGIN_METADATA_KEY: origin}}, "messages": [{"role": "user", "content": "hi"}], } await logger.async_log_success_event(kwargs, MagicMock(), None, None) @@ -224,7 +258,7 @@ class TestSuccessHookSkipPaths: @pytest.mark.asyncio class TestCallRouterShadowForwardsParameters: async def test_forwards_non_default_params(self): - router = MagicMock() + router = _router_mock() router.acompletion = AsyncMock(return_value={"choices": [{"message": {"content": "shadow reply"}}]}) logger = ShadowEvalLogger(router_provider=lambda: router, prisma_provider=lambda: MagicMock()) @@ -241,7 +275,7 @@ class TestCallRouterShadowForwardsParameters: assert kwargs["max_tokens"] == 500 async def test_drops_stream_and_metadata_from_forwarded_params(self): - router = MagicMock() + router = _router_mock() router.acompletion = AsyncMock(return_value={"choices": [{"message": {"content": "shadow reply"}}]}) logger = ShadowEvalLogger(router_provider=lambda: router, prisma_provider=lambda: MagicMock()) @@ -258,7 +292,7 @@ class TestCallRouterShadowForwardsParameters: # model_parameters carries a stale copy of the parent's metadata; the shadow call # builds its own from the parent metadata argument instead. assert "leaked" not in str(kwargs["metadata"]) - assert kwargs["metadata"][SHADOW_EVAL_INTERNAL_MARKER] is True + assert kwargs["metadata"][INTERNAL_CALL_ORIGIN_METADATA_KEY] == SHADOW_EVAL_ROUTER_CALL_ORIGIN @pytest.mark.asyncio @@ -327,8 +361,12 @@ class TestStoppedJobCannotBeReactivated: del prisma.db.litellm_shadowevaljob.update logger = ShadowEvalLogger(router_provider=lambda: MagicMock(), prisma_provider=lambda: prisma) - logger._call_router_shadow = AsyncMock(return_value=("shadow text", "shadow-model", "SIMPLE", 10)) - logger._call_judge = AsyncMock(return_value=("real", 0.9, "clearer", 0.01)) + logger._call_router_shadow = AsyncMock( + return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10) + ) + 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( @@ -364,8 +402,12 @@ class TestVerdictWriteAccumulatesCost: prisma.db.litellm_shadowevaljob.update_many = AsyncMock() logger = ShadowEvalLogger(router_provider=lambda: MagicMock(), prisma_provider=lambda: prisma) - logger._call_router_shadow = AsyncMock(return_value=("shadow text", "shadow-model", "SIMPLE", 10)) - logger._call_judge = AsyncMock(return_value=("real", 0.9, "clearer", 0.05)) + logger._call_router_shadow = AsyncMock( + return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10) + ) + logger._call_judge = AsyncMock( + return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.05) + ) job = ActiveShadowEvalJob(id="j1", router_name="r", judge_model="m", shadow_percentage=100.0, status="running") await logger._run_shadow_eval( @@ -439,19 +481,18 @@ class TestSubCallsAreAttributedToTheShadowedKey: } ) monkeypatch.setattr(litellm_module, "acompletion", acompletion) - logger = ShadowEvalLogger(router_provider=lambda: MagicMock(), prisma_provider=lambda: MagicMock()) + logger = ShadowEvalLogger(router_provider=_router_mock, prisma_provider=lambda: MagicMock()) verdict = await logger._call_judge( "gpt-4o-mini", [{"role": "user", "content": "hi"}], "real text", "shadow text", _PARENT_METADATA ) - assert verdict is not None + assert not isinstance(verdict, _CallFailure) metadata = acompletion.call_args.kwargs["metadata"] _assert_attributed(metadata, SHADOW_EVAL_JUDGE_CALL_ORIGIN) - assert metadata[SHADOW_EVAL_INTERNAL_MARKER] is True async def test_shadow_router_call_carries_the_key_identity_and_the_recursion_guard(self): - router = MagicMock() + router = _router_mock() router.acompletion = AsyncMock(return_value={"choices": [{"message": {"content": "shadow reply"}}]}) logger = ShadowEvalLogger(router_provider=lambda: router, prisma_provider=lambda: MagicMock()) @@ -459,15 +500,15 @@ class TestSubCallsAreAttributedToTheShadowedKey: "claude-auto", [{"role": "user", "content": "hi"}], {}, _PARENT_METADATA ) - assert result is not None + assert not isinstance(result, _CallFailure) metadata = router.acompletion.call_args.kwargs["metadata"] + # The origin stamp doubles as the recursion guard: the hook skips any + # request carrying it, so the shadow call cannot shadow itself. _assert_attributed(metadata, SHADOW_EVAL_ROUTER_CALL_ORIGIN) - # Without the marker the shadow call's own success event would shadow itself. - assert metadata[SHADOW_EVAL_INTERNAL_MARKER] is True async def test_shadow_metadata_stays_writable_for_the_routing_decision_read_back(self): """Tier attribution reads routing_decision back out of the dict the router was given.""" - router = MagicMock() + router = _router_mock() async def acompletion(**kwargs): kwargs["metadata"]["routing_decision"] = {"tier_label": "COMPLEX", "routed_model": "o1"} @@ -480,8 +521,8 @@ class TestSubCallsAreAttributedToTheShadowedKey: "claude-auto", [{"role": "user", "content": "hi"}], {}, _PARENT_METADATA ) - assert result is not None - assert result[2] == "COMPLEX" + assert not isinstance(result, _CallFailure) + assert result.tier == "COMPLEX" async def test_parent_metadata_reaches_both_legs_from_the_success_hook(self): """The hook is the only place the parent's metadata exists; a break here is invisible @@ -492,8 +533,12 @@ class TestSubCallsAreAttributedToTheShadowedKey: job = ActiveShadowEvalJob(id="j1", router_name="r", shadow_percentage=100.0, judge_model="m", status="running") logger, _, _ = _logger_with_mocks(job) logger._prisma_provider = lambda: prisma - logger._call_router_shadow = AsyncMock(return_value=("shadow text", "shadow-model", "SIMPLE", 10)) - logger._call_judge = AsyncMock(return_value=("real", 0.9, "clearer", 0.01)) + logger._call_router_shadow = AsyncMock( + return_value=_ShadowResponse(text="shadow text", model="shadow-model", tier="SIMPLE", completion_tokens=10) + ) + logger._call_judge = AsyncMock( + return_value=_JudgeVerdict(preference="real", confidence=0.9, reasoning="clearer", cost=0.01) + ) await logger.async_log_success_event( { @@ -648,7 +693,7 @@ class TestPerJobSpendCap: "messages": [{"role": "user", "content": "hi"}], } - async def test_job_over_the_cap_stops_sampling_and_completes_the_job(self): + async def test_job_over_the_cap_stops_sampling(self): job = ActiveShadowEvalJob( id="j1", router_name="r", @@ -659,7 +704,6 @@ class TestPerJobSpendCap: cost_actual=15.0, ) logger, prisma, router = _logger_with_mocks(job) - prisma.db.litellm_shadowevaljob.update_many = AsyncMock() logger._run_shadow_eval = AsyncMock() await logger.async_log_success_event(self._success_kwargs(), MagicMock(), None, None) @@ -667,7 +711,24 @@ class TestPerJobSpendCap: logger._run_shadow_eval.assert_not_awaited() router.acompletion.assert_not_called() - prisma.db.litellm_shadowevaljob.update_many.assert_awaited_once() + + async def test_lifecycle_tick_completes_a_job_over_the_cap(self): + """The cap is enforced by the loop, so it lands even on a key gone quiet.""" + job = ActiveShadowEvalJob( + id="j1", + router_name="r", + shadow_percentage=100.0, + judge_model="m", + status="running", + cost_estimate=10.0, + cost_actual=15.0, + ) + logger, prisma, _ = _logger_with_mocks(job) + prisma.db.litellm_shadowevaljob.update_many = AsyncMock() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[_job_record(job)]) + + await logger._lifecycle_tick() + call_kwargs = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs assert call_kwargs["where"]["id"] == "j1" # Guarded so a pod breaching the cap cannot resurrect a job an admin already stopped. @@ -686,14 +747,12 @@ class TestPerJobSpendCap: cost_actual=14.99, ) logger, prisma, _ = _logger_with_mocks(job) - prisma.db.litellm_shadowevaljob.update_many = AsyncMock() logger._run_shadow_eval = AsyncMock() await logger.async_log_success_event(self._success_kwargs(), MagicMock(), None, None) await asyncio.sleep(0.01) logger._run_shadow_eval.assert_awaited_once() - prisma.db.litellm_shadowevaljob.update_many.assert_not_awaited() async def test_job_without_an_estimate_is_uncapped(self): """A missing estimate is no multiple to compare against; treating it as 0 would @@ -708,14 +767,12 @@ class TestPerJobSpendCap: cost_actual=500.0, ) logger, prisma, _ = _logger_with_mocks(job) - prisma.db.litellm_shadowevaljob.update_many = AsyncMock() logger._run_shadow_eval = AsyncMock() await logger.async_log_success_event(self._success_kwargs(), MagicMock(), None, None) await asyncio.sleep(0.01) logger._run_shadow_eval.assert_awaited_once() - prisma.db.litellm_shadowevaljob.update_many.assert_not_awaited() async def test_a_zero_estimate_job_is_still_capped_at_the_floor(self): """A key quiet during the lookback gets estimate $0.00; a later traffic spike on @@ -731,12 +788,14 @@ class TestPerJobSpendCap: ) logger, prisma, _ = _logger_with_mocks(job) prisma.db.litellm_shadowevaljob.update_many = AsyncMock() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[_job_record(job)]) logger._run_shadow_eval = AsyncMock() await logger.async_log_success_event(self._success_kwargs(), MagicMock(), None, None) await asyncio.sleep(0) - logger._run_shadow_eval.assert_not_awaited() + + await logger._lifecycle_tick() assert prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs["data"]["status"] == "completed" async def test_a_cent_sized_estimate_is_not_stopped_by_its_first_verdict(self): @@ -759,7 +818,7 @@ class TestPerJobSpendCap: logger._run_shadow_eval.assert_awaited_once() - async def test_stopping_evicts_the_cached_job_so_later_requests_do_not_rewrite_it(self): + async def test_finalizing_evicts_the_job_so_a_second_tick_does_not_rewrite_it(self): job = ActiveShadowEvalJob( id="j1", router_name="r", @@ -771,104 +830,154 @@ class TestPerJobSpendCap: ) logger, prisma, _ = _logger_with_mocks(job) prisma.db.litellm_shadowevaljob.update_many = AsyncMock() - prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=[[_job_record(job)], []]) - await logger.async_log_success_event(self._success_kwargs(), MagicMock(), None, None) - await logger.async_log_success_event(self._success_kwargs(), MagicMock(), None, None) + await logger._lifecycle_tick() + await logger._lifecycle_tick() assert prisma.db.litellm_shadowevaljob.update_many.await_count == 1 @pytest.mark.asyncio class TestActiveJobSnapshot: - """One find_many per pod per TTL serves every key, so DB load stays flat no + """One find_many per pod per tick serves every key, so DB load stays flat no matter how many distinct keys send traffic through the proxy.""" - @staticmethod - def _record(job_id: str, api_key_id: str) -> MagicMock: - record = MagicMock() - record.id = job_id - record.api_key_id = api_key_id - record.router_name = "r" - record.shadow_percentage = 100.0 - record.judge_model = "m" - record.status = "running" - record.cost_estimate = 10.0 - record.cost_actual = 0.0 - record.ends_at = None - return record - - @staticmethod - async def _settled(logger): - if logger._jobs_refresh_task is not None: - await logger._jobs_refresh_task - - async def test_lookup_never_awaits_the_db_and_one_query_serves_many_keys(self): + async def test_hook_lookup_never_awaits_the_db(self): + """The success hook reads the loop-maintained snapshot; a cold pod answers + from the empty snapshot instead of blocking on Prisma.""" logger, prisma, _ = _logger_with_mocks() - prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[self._record("j1", "key-1")]) + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[]) + kwargs = { + "standard_logging_object": { + "id": "req-1", + "model": "gpt-4o", + "call_type": "acompletion", + "metadata": {"user_api_key_hash": "key-1"}, + }, + "litellm_params": {"metadata": {}}, + "messages": [{"role": "user", "content": "hi"}], + } + await logger.async_log_success_event(kwargs, MagicMock(), None, None) + prisma.db.litellm_shadowevaljob.find_many.assert_not_awaited() - cold = logger._get_active_job("key-1") - await self._settled(logger) - warm = logger._get_active_job("key-1") - misses = [logger._get_active_job(f"other-{i}") for i in range(50)] + async def test_one_refresh_serves_many_keys(self): + logger, prisma, _ = _logger_with_mocks() + job = ActiveShadowEvalJob(id="j1", router_name="r", shadow_percentage=100.0, judge_model="m", status="running") + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[_job_record(job, api_key_id="key-1")]) - assert cold is None, "cold pod answers from the empty snapshot instead of blocking on Prisma" - assert warm is not None and warm.id == "j1" - assert all(m is None for m in misses) + await logger._refresh_active_jobs() + + assert logger._jobs_by_key["key-1"].id == "j1" + assert all(logger._jobs_by_key.get(f"other-{i}") is None for i in range(50)) assert prisma.db.litellm_shadowevaljob.find_many.await_count == 1 async def test_newest_job_wins_when_a_key_somehow_has_two(self): logger, prisma, _ = _logger_with_mocks() + newest = ActiveShadowEvalJob( + id="j-newest", router_name="r", shadow_percentage=100.0, judge_model="m", status="running" + ) + oldest = ActiveShadowEvalJob( + id="j-oldest", router_name="r", shadow_percentage=100.0, judge_model="m", status="running" + ) prisma.db.litellm_shadowevaljob.find_many = AsyncMock( - return_value=[self._record("j-newest", "key-1"), self._record("j-oldest", "key-1")] + return_value=[_job_record(newest, "key-1"), _job_record(oldest, "key-1")] ) - logger._get_active_job("key-1") - await self._settled(logger) - job = logger._get_active_job("key-1") + await logger._refresh_active_jobs() - assert job is not None and job.id == "j-newest" + assert logger._jobs_by_key["key-1"].id == "j-newest" async def test_db_blip_keeps_the_stale_snapshot_instead_of_disabling_the_feature(self): logger, prisma, _ = _logger_with_mocks() - prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[self._record("j1", "key-1")]) - logger._get_active_job("key-1") - await self._settled(logger) + job = ActiveShadowEvalJob(id="j1", router_name="r", shadow_percentage=100.0, judge_model="m", status="running") + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[_job_record(job, "key-1")]) + await logger._refresh_active_jobs() - logger._jobs_fetched_at = asyncio.get_event_loop().time() - 61.0 prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=RuntimeError("db down")) + await logger._refresh_active_jobs() - stale = logger._get_active_job("key-1") - await self._settled(logger) + assert logger._jobs_by_key["key-1"].id == "j1" - assert stale is not None and stale.id == "j1" - assert logger._get_active_job("key-1") is not None - - async def test_a_slow_refresh_is_not_stacked_by_concurrent_lookups(self): + async def test_a_failing_tick_does_not_kill_the_loop(self): logger, prisma, _ = _logger_with_mocks() - started = asyncio.Event() - release = asyncio.Event() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=RuntimeError("db down")) + logger._pending_seen = {"j1": 3} + prisma.db.litellm_shadowevaljob.update_many = AsyncMock(side_effect=RuntimeError("db down")) - async def slow_find_many(**_kwargs): - started.set() - await release.wait() - return [self._record("j1", "key-1")] + await logger._lifecycle_tick() - prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=slow_find_many) + async def test_start_lifecycle_loop_is_idempotent(self): + logger, _, _ = _logger_with_mocks() + logger.start_lifecycle_loop() + first = logger._lifecycle_task + logger.start_lifecycle_loop() + assert logger._lifecycle_task is first + first.cancel() - for _ in range(10): - logger._get_active_job("key-1") - await started.wait() - release.set() - await self._settled(logger) - assert prisma.db.litellm_shadowevaljob.find_many.await_count == 1 +@pytest.mark.asyncio +class TestRequestCountFlush: + """request_count is flushed by the loop, not by the next request arriving, so + the final batch lands and a stopped job's counter freezes.""" + + async def test_tick_flushes_buffered_counts_without_new_requests(self): + """Regression: the old flush was piggybacked on a later request 10s+ after + the last flush, so the final batch of a job was never written and idle jobs + read 'N judged of 0 seen'.""" + logger, prisma, _ = _logger_with_mocks() + prisma.db.litellm_shadowevaljob.update_many = AsyncMock() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[]) + logger._pending_seen = {"j1": 7} + + await logger._lifecycle_tick() + + flush_call = prisma.db.litellm_shadowevaljob.update_many.call_args_list[0] + assert flush_call.kwargs["where"]["id"] == "j1" + assert flush_call.kwargs["data"] == {"request_count": {"increment": 7}} + assert logger._pending_seen == {} + + async def test_flush_is_guarded_on_the_job_still_being_active(self): + """Stopping a job freezes its request_count: a pod serving a stale snapshot + keeps buffering for up to one tick, and the status-guarded write drops those + increments instead of growing a stopped job's counter for ~30s after stop.""" + logger, prisma, _ = _logger_with_mocks() + prisma.db.litellm_shadowevaljob.update_many = AsyncMock() + logger._pending_seen = {"j1": 2} + + await logger._flush_seen_counts() + + where = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs["where"] + assert set(where["status"]["in"]) == {"pending", "running"} + + async def test_flush_before_finalize_lands_the_tail_batch_of_an_expiring_job(self): + """The tick flushes first, so an expiring job's last counts pass the active + guard before the same tick flips it to completed.""" + expired = ActiveShadowEvalJob( + id="j1", + router_name="r", + shadow_percentage=100.0, + judge_model="m", + status="running", + ends_at=datetime.now(timezone.utc) - timedelta(seconds=1), + ) + logger, prisma, _ = _logger_with_mocks(expired) + prisma.db.litellm_shadowevaljob.update_many = AsyncMock() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[_job_record(expired)]) + logger._pending_seen = {"j1": 4} + + await logger._lifecycle_tick() + + calls = prisma.db.litellm_shadowevaljob.update_many.call_args_list + assert calls[0].kwargs["data"] == {"request_count": {"increment": 4}} + assert calls[-1].kwargs["data"]["status"] == "completed" @pytest.mark.asyncio class TestJobsStopAtTheirScheduledEnd: """A shadow eval samples ongoing traffic, so a job whose window has closed must - stop billing judge calls even if nobody remembers to stop it by hand.""" + stop billing judge calls even if nobody remembers to stop it by hand, and even + if its key never sends another request.""" @staticmethod def _job(ends_at): @@ -883,9 +992,8 @@ class TestJobsStopAtTheirScheduledEnd: ends_at=ends_at, ) - async def test_job_past_its_end_stops_sampling_and_completes(self): + async def test_job_past_its_end_stops_sampling(self): logger, prisma, router = _logger_with_mocks(self._job(datetime.now(timezone.utc) - timedelta(seconds=1))) - prisma.db.litellm_shadowevaljob.update_many = AsyncMock() logger._run_shadow_eval = AsyncMock() await logger.async_log_success_event(TestPerJobSpendCap._success_kwargs(), MagicMock(), None, None) @@ -893,6 +1001,17 @@ class TestJobsStopAtTheirScheduledEnd: logger._run_shadow_eval.assert_not_awaited() router.acompletion.assert_not_called() + + async def test_lifecycle_tick_completes_an_expired_job_with_no_traffic_at_all(self): + """Regression: expiry used to be checked only inside the success hook, so a + job on a key gone quiet stayed pending/running indefinitely.""" + expired = self._job(datetime.now(timezone.utc) - timedelta(seconds=1)) + logger, prisma, _ = _logger_with_mocks() + prisma.db.litellm_shadowevaljob.update_many = AsyncMock() + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[_job_record(expired)]) + + await logger._lifecycle_tick() + call_kwargs = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs assert call_kwargs["where"]["id"] == "j1" assert set(call_kwargs["where"]["status"]["in"]) == {"pending", "running"} @@ -919,36 +1038,14 @@ class TestJobsStopAtTheirScheduledEnd: logger._run_shadow_eval.assert_awaited_once() - async def test_expiry_evicts_the_cached_job_so_later_requests_do_not_rewrite_it(self): - logger, prisma, _ = _logger_with_mocks(self._job(datetime.now(timezone.utc) - timedelta(seconds=1))) - prisma.db.litellm_shadowevaljob.update_many = AsyncMock() - prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[]) - - await logger.async_log_success_event(TestPerJobSpendCap._success_kwargs(), MagicMock(), None, None) - await logger.async_log_success_event(TestPerJobSpendCap._success_kwargs(), MagicMock(), None, None) - - assert prisma.db.litellm_shadowevaljob.update_many.await_count == 1 - async def test_naive_db_datetime_is_treated_as_utc(self): logger, prisma, _ = _logger_with_mocks() - record = MagicMock() - record.id = "j1" - record.api_key_id = "key-hash" - record.router_name = "r" - record.shadow_percentage = 100.0 - record.judge_model = "m" - record.status = "running" - record.cost_estimate = 10.0 - record.cost_actual = 0.0 - record.ends_at = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(seconds=1) - prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[record]) + naive_expired = self._job(datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(seconds=1)) + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[_job_record(naive_expired)]) - logger._get_active_job("key-hash") - if logger._jobs_refresh_task is not None: - await logger._jobs_refresh_task - job = logger._get_active_job("key-hash") + await logger._refresh_active_jobs() + job = logger._jobs_by_key["key-hash"] - assert job is not None assert job.ends_at is not None and job.ends_at.tzinfo is not None from litellm.integrations.shadow_eval_logger import _job_is_past_its_end @@ -963,22 +1060,24 @@ class TestJudgeFailureModes: @staticmethod def _logger(): prisma = MagicMock() - return ShadowEvalLogger(router_provider=lambda: MagicMock(), prisma_provider=lambda: prisma), prisma + router = _router_mock() + return ShadowEvalLogger(router_provider=lambda: router, prisma_provider=lambda: prisma), prisma, router - async def test_judge_provider_error_returns_none(self, monkeypatch: pytest.MonkeyPatch): + async def test_judge_provider_error_is_a_described_failure(self, monkeypatch: pytest.MonkeyPatch): import litellm as litellm_module - logger, _ = self._logger() + logger, _, _ = self._logger() monkeypatch.setattr(litellm_module, "acompletion", AsyncMock(side_effect=RuntimeError("provider down"))) verdict = await logger._call_judge("m", [{"role": "user", "content": "hi"}], "real", "shadow", {}) - assert verdict is None + assert isinstance(verdict, _CallFailure) + assert "provider down" in verdict.error - async def test_unparseable_judge_output_returns_none(self, monkeypatch: pytest.MonkeyPatch): + async def test_unparseable_judge_output_is_a_described_failure(self, monkeypatch: pytest.MonkeyPatch): import litellm as litellm_module - logger, _ = self._logger() + logger, _, _ = self._logger() monkeypatch.setattr( litellm_module, "acompletion", @@ -987,16 +1086,21 @@ class TestJudgeFailureModes: verdict = await logger._call_judge("m", [{"role": "user", "content": "hi"}], "real", "shadow", {}) - assert verdict is None + assert isinstance(verdict, _CallFailure) + assert "unparseable" in verdict.error - async def test_failed_judge_bumps_failed_count_and_writes_no_verdict(self, monkeypatch: pytest.MonkeyPatch): + async def test_failed_judge_bumps_failed_count_with_last_error_and_writes_no_verdict( + self, monkeypatch: pytest.MonkeyPatch + ): + """Regression: a misconfigured judge model failed every turn with nothing but + a debug log, so the admin saw only a growing failed_count and null results. + The most recent failure is persisted on the job for the UI to show.""" import litellm as litellm_module - logger, prisma = self._logger() + logger, prisma, router = self._logger() monkeypatch.setattr(litellm_module, "acompletion", AsyncMock(side_effect=RuntimeError("down"))) - router = logger._router_provider() router.acompletion = AsyncMock(return_value={"choices": [{"message": {"content": "shadow says"}}]}) - prisma.db.litellm_shadowevaljob.update = AsyncMock() + prisma.db.litellm_shadowevaljob.update_many = AsyncMock() job = ActiveShadowEvalJob(id="j1", router_name="r", shadow_percentage=100.0, judge_model="m", status="running") await logger._run_shadow_eval( @@ -1010,7 +1114,30 @@ class TestJudgeFailureModes: ) prisma.db.litellm_shadowevalverdict.create.assert_not_called() - assert prisma.db.litellm_shadowevaljob.update.call_args.kwargs["data"] == {"failed_count": {"increment": 1}} + call_kwargs = prisma.db.litellm_shadowevaljob.update_many.call_args.kwargs + assert call_kwargs["data"]["failed_count"] == {"increment": 1} + assert "down" in call_kwargs["data"]["last_error"] + assert set(call_kwargs["where"]["status"]["in"]) == {"pending", "running"} + + async def test_judge_configured_as_a_proxy_deployment_dispatches_through_the_router(self): + """Regression: the judge always went through the SDK, so a judge named after a + proxy deployment failed every turn with 'LLM Provider NOT provided' while + still paying for the shadow call.""" + logger, _, router = self._logger() + router.get_model_list = MagicMock(return_value=[{"litellm_params": {"model": "openai/gpt-4o"}}]) + router.acompletion = AsyncMock( + return_value={ + "choices": [{"message": {"content": '{"preference": "A", "confidence": 0.8, "reasoning": "x"}'}}] + } + ) + + verdict = await logger._call_judge( + "my-deployment", [{"role": "user", "content": "hi"}], "real", "shadow", {} + ) + + assert not isinstance(verdict, _CallFailure) + router.acompletion.assert_awaited_once() + assert router.acompletion.call_args.kwargs["model"] == "my-deployment" class TestExtractResponseText: diff --git a/tests/test_litellm/proxy/db/test_autorouter_session_rollup.py b/tests/test_litellm/proxy/db/test_autorouter_session_rollup.py index 0df11f224a2..99a699e0225 100644 --- a/tests/test_litellm/proxy/db/test_autorouter_session_rollup.py +++ b/tests/test_litellm/proxy/db/test_autorouter_session_rollup.py @@ -93,6 +93,15 @@ class TestBuildTransaction: def test_requests_without_a_routing_decision_are_skipped(self, metadata: dict): assert _build(metadata=metadata) is None + @pytest.mark.parametrize("origin", ["shadow_eval_router", "shadow_eval_judge", "autorouter_classifier"]) + def test_internal_sub_calls_are_excluded_even_with_a_routing_decision(self, origin: str): + """Regression: a shadow eval duplicates requests through the router, so its + sub-calls carry a genuine routing_decision and rolled up as real routed + sessions, inflating session counts and saved_spend in the benchmarks a + pre-adoption eval renders next to. Traffic no user sent must never + become adoption metrics.""" + assert _build(metadata=_metadata(internal_call_origin=origin)) is None + def test_the_tier_the_decision_recorded_is_carried_onto_the_transaction(self): transaction = _build(metadata=_metadata(routing_decision={**ROUTING_DECISION, "tier": "reasoning"})) assert transaction is not None and transaction.tier == "reasoning" diff --git a/tests/test_litellm/proxy/db/test_db_spend_update_writer.py b/tests/test_litellm/proxy/db/test_db_spend_update_writer.py index 51810a28cdd..6199e79f318 100644 --- a/tests/test_litellm/proxy/db/test_db_spend_update_writer.py +++ b/tests/test_litellm/proxy/db/test_db_spend_update_writer.py @@ -2157,3 +2157,70 @@ async def test_daily_transaction_compression_saved_tokens_zero_when_absent(): assert transaction["compression_saved_tokens"] == 0 assert transaction["compression_savings_spend"] == 0 assert transaction["prompt_caching_savings_spend"] == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("origin", ["shadow_eval_router", "shadow_eval_judge", "autorouter_classifier"]) +async def test_internal_sub_calls_keep_spend_but_are_not_counted_as_requests(origin: str): + """Regression: shadow-eval and classifier sub-calls counted toward api_requests, so + the shadow-eval cost estimate (SUM(api_requests) over LiteLLM_DailyUserSpend) fed on + its own output and usage dashboards inflated. Spend and tokens must still land so + budgets and billing stay whole; the derived auto-router savings of a shadow + duplicate is not a saving anyone realized.""" + db_writer = DBSpendUpdateWriter() + payload = { + "user": "user-1", + "api_key": "hashed-key", + "model": "gpt-4o", + "model_group": "gpt-4o", + "custom_llm_provider": "openai", + "startTime": datetime(2026, 8, 10, 12, 0, tzinfo=timezone.utc), + "spend": 0.05, + "prompt_tokens": 90, + "completion_tokens": 10, + "metadata": json.dumps( + { + "internal_call_origin": origin, + "routing_decision": {"router_model_name": "claude-auto", "router_type": "complexity"}, + } + ), + } + prisma = MagicMock() + prisma.get_request_status = MagicMock(return_value="success") + + transaction = await db_writer._common_add_spend_log_transaction_to_daily_transaction(payload, prisma) + + assert transaction is not None + assert transaction["api_requests"] == 0 + assert transaction["successful_requests"] == 0 + assert transaction["failed_requests"] == 0 + assert transaction["spend"] == 0.05 + assert transaction["prompt_tokens"] == 90 + assert transaction["autorouter_savings_spend"] == 0.0 + + +@pytest.mark.asyncio +async def test_user_requests_still_count_as_requests(): + """The negative class for the internal-origin exclusion: an ordinary user request + keeps counting, so the exclusion cannot silently zero the whole rollup.""" + db_writer = DBSpendUpdateWriter() + payload = { + "user": "user-1", + "api_key": "hashed-key", + "model": "gpt-4o", + "model_group": "gpt-4o", + "custom_llm_provider": "openai", + "startTime": datetime(2026, 8, 10, 12, 0, tzinfo=timezone.utc), + "spend": 0.05, + "prompt_tokens": 90, + "completion_tokens": 10, + "metadata": json.dumps({}), + } + prisma = MagicMock() + prisma.get_request_status = MagicMock(return_value="success") + + transaction = await db_writer._common_add_spend_log_transaction_to_daily_transaction(payload, prisma) + + assert transaction is not None + assert transaction["api_requests"] == 1 + assert transaction["successful_requests"] == 1 diff --git a/tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py index 015bd46ab8a..e48f5eedc9a 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py @@ -531,7 +531,7 @@ class TestJudgeCostEstimate: completion_tokens=JUDGE_MAX_OUTPUT_TOKENS, ) assert completion_cost > 0 - assert _estimate_judge_cost_per_call(self.JUDGE_MODEL) == prompt_cost + completion_cost + assert _estimate_judge_cost_per_call(None, self.JUDGE_MODEL) == prompt_cost + completion_cost def test_estimate_is_not_still_pinned_to_the_old_200_token_budget(self): import litellm as litellm_module @@ -545,7 +545,65 @@ class TestJudgeCostEstimate: model=self.JUDGE_MODEL, prompt_tokens=_JUDGE_PROMPT_TOKENS_ESTIMATE, completion_tokens=200 ) ) - assert _estimate_judge_cost_per_call(self.JUDGE_MODEL) > stale + assert _estimate_judge_cost_per_call(None, self.JUDGE_MODEL) > stale + + +class TestJudgeModelValidation: + """A judge the dispatch path cannot resolve fails every sampled turn silently, so + start must reject it up front instead of accepting a job that only ever bills + shadow calls.""" + + @staticmethod + def _router() -> MagicMock: + router = MagicMock() + router.auto_routers = {} + router.complexity_routers = {"claude-auto": [MagicMock()]} + router.adaptive_routers = {} + router.quality_routers = {} + router.model_group_alias = {} + router.get_model_list = MagicMock(return_value=None) + return router + + def test_unresolvable_judge_model_is_a_400(self): + from litellm.proxy.management_endpoints.auto_router_endpoints import _validate_judge_model + + with pytest.raises(HTTPException) as exc: + _validate_judge_model(self._router(), "opus") + + assert exc.value.status_code == 400 + assert "opus" in str(exc.value.detail) + + def test_configured_deployment_name_is_accepted(self): + from litellm.proxy.management_endpoints.auto_router_endpoints import _validate_judge_model + + router = self._router() + router.get_model_list = MagicMock(return_value=[{"litellm_params": {"model": "openai/gpt-4o"}}]) + _validate_judge_model(router, "opus") + + def test_provider_qualified_public_name_is_accepted_without_a_router(self): + from litellm.proxy.management_endpoints.auto_router_endpoints import _validate_judge_model + + _validate_judge_model(None, "openai/gpt-4o") + + def test_an_auto_router_is_rejected_as_judge(self): + from litellm.proxy.management_endpoints.auto_router_endpoints import _validate_judge_model + + with pytest.raises(HTTPException) as exc: + _validate_judge_model(self._router(), "claude-auto") + + assert exc.value.status_code == 400 + assert "auto-router" in str(exc.value.detail) + + def test_estimate_prices_a_deployment_by_its_underlying_model(self): + """The deployment name an admin typed is not a pricing key; the estimate must + price the provider model behind it, not fall back to the flat $0.01.""" + from litellm.proxy.management_endpoints.auto_router_endpoints import _judge_pricing_model + + router = self._router() + router.get_model_list = MagicMock(return_value=[{"litellm_params": {"model": "anthropic/claude-sonnet-5"}}]) + + assert _judge_pricing_model(router, "opus") == "anthropic/claude-sonnet-5" + assert _judge_pricing_model(None, "openai/gpt-4o") == "openai/gpt-4o" class TestShadowEvalJobsAreTimeBound: @@ -765,6 +823,7 @@ class TestShadowEvalJobLifecycleEndpoints: record.request_count = 100 record.completed_count = 9 record.failed_count = 1 + record.last_error = None record.cost_estimate = 3.0 record.cost_actual = 0.42 record.created_at = datetime(2026, 8, 1, tzinfo=timezone.utc) diff --git a/tests/test_litellm/proxy/proxy_server/test_lifecycle.py b/tests/test_litellm/proxy/proxy_server/test_lifecycle.py index b88ecd4029d..5edd68cbbb3 100644 --- a/tests/test_litellm/proxy/proxy_server/test_lifecycle.py +++ b/tests/test_litellm/proxy/proxy_server/test_lifecycle.py @@ -511,6 +511,22 @@ 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.""" + 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) + + ps._start_shadow_eval_lifecycle_loop() + + assert logger._lifecycle_task is not None and not logger._lifecycle_task.done() + logger._lifecycle_task.cancel() + + def test_cost_tracking_no_op_when_prisma_missing(monkeypatch): """Without a prisma_client cost_tracking is a no-op — not an error.""" import litellm diff --git a/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.test.tsx b/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.test.tsx index 98decd7f20a..5176936670c 100644 --- a/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.test.tsx +++ b/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.test.tsx @@ -225,6 +225,20 @@ describe("ShadowEvalSection", () => { expect(screen.queryByText(/By prompt difficulty/)).not.toBeInTheDocument(); }); + it("surfaces the last shadow/judge failure so a growing failed_count is diagnosable", () => { + const j = job({ failed_count: 7, last_error: "judge call failed: LLM Provider NOT provided" }); + mockHooks({ jobs: [j], detail: j }); + render(); + expect(screen.getByText(/LLM Provider NOT provided/)).toBeInTheDocument(); + }); + + it("shows no failure banner when nothing has failed", () => { + const j = job({ failed_count: 0, last_error: null }); + mockHooks({ jobs: [j], detail: j }); + render(); + expect(screen.queryByText(/Last failure/)).not.toBeInTheDocument(); + }); + it("shows a stop button for running jobs but not completed ones", () => { const running = job({ status: "running" }); mockHooks({ jobs: [running], detail: running }); diff --git a/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.tsx b/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.tsx index 46a0759fd9c..5677173f01f 100644 --- a/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.tsx +++ b/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.tsx @@ -298,6 +298,12 @@ const JobResults: React.FC<{ ) : null} + {job.failed_count > 0 && job.last_error ? ( +

+ Last failure: {job.last_error} +

+ ) : null} + {results && results.groups.length > 0 ? ( <>
diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index f96fed69c3a..4f9ac085774 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -25391,6 +25391,11 @@ export interface components { failed_count: number; /** Job Id */ job_id: string; + /** + * Last Error + * @description The most recent shadow or judge failure, so a growing failed_count is diagnosable + */ + last_error?: string | null; /** * Request Count * @description Total requests observed on the shadowed key since the job started