scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-12 23:01:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 88

fix: judge model validation, internal-call exclusion from adoption metrics, loop-owned job lifecycle

Browse source
This commit is contained in:
Tin Chi Lo 2026-08-11 00:48:41 -07:00
parent b2feff3f05
commit ece7b16d3c
18 changed files with 724 additions and 283 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm-proxy-extras/litellm_proxy_extras/migrations/20260807000000_add_shadow_eval_job/migration.sql
View file

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

5
litellm-proxy-extras/litellm_proxy_extras/schema.prisma
View file

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

298
litellm/integrations/shadow_eval_logger.py
View file

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

11
litellm/proxy/db/autorouter_session_rollup.py
View file

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

15
litellm/proxy/db/db_spend_update_writer.py
View file

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

64
litellm/proxy/management_endpoints/auto_router_endpoints.py
View file

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

20
litellm/proxy/proxy_server.py
View file

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

5
litellm/proxy/schema.prisma
View file

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

3
litellm/types/management_endpoints/auto_router_endpoints.py
View file

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

5
schema.prisma
View file

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

399
tests/test_litellm/integrations/test_shadow_eval_logger.py
View file

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

9
tests/test_litellm/proxy/db/test_autorouter_session_rollup.py
View file

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

67
tests/test_litellm/proxy/db/test_db_spend_update_writer.py
View file

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

63
tests/test_litellm/proxy/management_endpoints/test_auto_router_endpoints.py
View file

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

16
tests/test_litellm/proxy/proxy_server/test_lifecycle.py
View file

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

14
ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.test.tsx
View file

@ -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(<ShadowEvalSection accessToken="token" />);
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(<ShadowEvalSection accessToken="token" />);
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 });

6
ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/ShadowEvalSection.tsx
View file

@ -298,6 +298,12 @@ const JobResults: React.FC<{
) : null}
</div>
{job.failed_count > 0 && job.last_error ? (
<p className="border-b bg-red-50 px-6 py-2 text-xs text-destructive">
Last failure: <span className="font-mono">{job.last_error}</span>
</p>
) : null}
{results && results.groups.length > 0 ? (
<>
<div className="flex flex-col justify-center gap-1 border-b px-6 py-4">

5
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

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