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 405812a38e4..2242a29fae6 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
@@ -24,6 +24,11 @@ CREATE TABLE IF NOT EXISTS "LiteLLM_ShadowEvalJob" (
CREATE INDEX IF NOT EXISTS "LiteLLM_ShadowEvalJob_team_id_status_idx" ON "LiteLLM_ShadowEvalJob"("team_id", "status");
CREATE INDEX IF NOT EXISTS "LiteLLM_ShadowEvalJob_api_key_id_status_idx" ON "LiteLLM_ShadowEvalJob"("api_key_id", "status");
CREATE INDEX IF NOT EXISTS "LiteLLM_ShadowEvalJob_status_idx" ON "LiteLLM_ShadowEvalJob"("status");
+-- One active job per key, enforced by the database rather than a read-then-create
+-- in the start endpoint, which races against a concurrent start on another pod.
+-- Partial indexes are not expressible in schema.prisma, so this lives here only.
+CREATE UNIQUE INDEX IF NOT EXISTS "LiteLLM_ShadowEvalJob_one_active_per_key"
+ ON "LiteLLM_ShadowEvalJob"("api_key_id") WHERE status IN ('pending', 'running');
CREATE INDEX IF NOT EXISTS "LiteLLM_ShadowEvalJob_created_at_idx" ON "LiteLLM_ShadowEvalJob"("created_at");
CREATE TABLE IF NOT EXISTS "LiteLLM_ShadowEvalVerdict" (
diff --git a/litellm/integrations/shadow_eval_logger.py b/litellm/integrations/shadow_eval_logger.py
index 93b1af00f54..d8292574ac5 100644
--- a/litellm/integrations/shadow_eval_logger.py
+++ b/litellm/integrations/shadow_eval_logger.py
@@ -243,10 +243,12 @@ def _job_is_over_spend_cap(job: ActiveShadowEvalJob) -> bool:
Budgets bound what the *key* may spend; this bounds what a single eval may spend
even under a generous budget, so a bad estimate or a traffic spike cannot quietly
- turn a "$3 eval" into a much larger bill. A job with no estimate (older rows,
- unpriced judge model) has nothing to be a multiple of, so it is uncapped.
+ turn a "$3 eval" into a much larger bill. Only a job with no estimate at all
+ (rows created before estimates existed) is uncapped; a $0 estimate — a key that
+ was quiet during the lookback — still gets the floor, since a later traffic spike
+ on exactly such a job is the scenario the cap exists for.
"""
- if job.cost_estimate is None or job.cost_estimate <= 0.0:
+ if job.cost_estimate is None:
return False
return job.cost_actual >= max(job.cost_estimate * _SPEND_CAP_MULTIPLIER, _SPEND_CAP_FLOOR_USD)
@@ -269,7 +271,7 @@ class ShadowEvalLogger(CustomLogger):
# 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_lock: asyncio.Lock = asyncio.Lock()
+ self._jobs_refresh_task: asyncio.Task[None] | None = None
self._inflight_shadow_tasks: int = 0
self._pending_seen: dict[str, int] = {} # mutable-ok: flush buffer
self._last_seen_flush: float = 0.0
@@ -300,7 +302,7 @@ class ShadowEvalLogger(CustomLogger):
api_key_hash: Final = metadata.get("user_api_key_hash")
if not api_key_hash:
return
- job: Final = await self._get_active_job(api_key_hash)
+ job: Final = self._get_active_job(api_key_hash)
if job is None:
return
if _job_is_past_its_end(job):
@@ -353,46 +355,48 @@ class ShadowEvalLogger(CustomLogger):
#### job lookup ####
- async def _get_active_job(self, api_key_hash: str) -> ActiveShadowEvalJob | None:
+ 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()
- if self._jobs_fetched_at is None or now - self._jobs_fetched_at >= _JOB_CACHE_TTL_SECONDS:
- await self._refresh_active_jobs(now)
+ 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:
- """Reload the active-job set, at most once per TTL across concurrent requests.
-
- 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.
- """
- async with self._jobs_refresh_lock:
- if self._jobs_fetched_at is not None and now - self._jobs_fetched_at < _JOB_CACHE_TTL_SECONDS:
- return
- prisma: Final = self._prisma_provider()
- if prisma is None:
- return
- try:
- records: Final = await prisma.db.litellm_shadowevaljob.find_many(
- where={"status": {"in": ["pending", "running"]}}, # mutable-ok: Prisma filter
- order={"created_at": "desc"}, # mutable-ok: Prisma order
- )
- except Exception as e: # noqa: BLE001 # a DB blip must not break request logging
- verbose_logger.debug("shadow_eval: active-job refresh failed: %s", e)
- return
- jobs_by_key: Final[dict[str, ActiveShadowEvalJob]] = {} # mutable-ok: building the new snapshot
- for record in reversed(records or []):
- jobs_by_key[str(record.api_key_id)] = ActiveShadowEvalJob(
- id=str(record.id),
- router_name=str(record.router_name),
- shadow_percentage=float(record.shadow_percentage),
- judge_model=str(record.judge_model),
- status=str(record.status),
- cost_estimate=float(record.cost_estimate) if record.cost_estimate is not None else None,
- cost_actual=float(record.cost_actual or 0.0),
- 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
+ """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."""
+ prisma: Final = self._prisma_provider()
+ if prisma is None:
+ return
+ try:
+ records: Final = await prisma.db.litellm_shadowevaljob.find_many(
+ where={"status": {"in": ["pending", "running"]}}, # mutable-ok: Prisma filter
+ order={"created_at": "desc"}, # mutable-ok: Prisma order
+ )
+ except Exception as e: # noqa: BLE001 # a DB blip must not break request logging
+ verbose_logger.debug("shadow_eval: active-job refresh failed: %s", e)
+ return
+ jobs_by_key: Final[dict[str, ActiveShadowEvalJob]] = {} # mutable-ok: building the new snapshot
+ for record in reversed(records or []):
+ jobs_by_key[str(record.api_key_id)] = ActiveShadowEvalJob(
+ id=str(record.id),
+ router_name=str(record.router_name),
+ shadow_percentage=float(record.shadow_percentage),
+ judge_model=str(record.judge_model),
+ status=str(record.status),
+ cost_estimate=float(record.cost_estimate) if record.cost_estimate is not None else None,
+ cost_actual=float(record.cost_actual or 0.0),
+ 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.
diff --git a/litellm/proxy/management_endpoints/auto_router_endpoints.py b/litellm/proxy/management_endpoints/auto_router_endpoints.py
index 437bee9fc69..5d456c2aa5b 100644
--- a/litellm/proxy/management_endpoints/auto_router_endpoints.py
+++ b/litellm/proxy/management_endpoints/auto_router_endpoints.py
@@ -482,6 +482,21 @@ async def _recent_request_volume(prisma_client: "PrismaClient", api_key_id: str)
return rows[0].request_count if rows else 0
+def _is_unique_violation(error: Exception) -> bool:
+ """Whether a Prisma create failed on a unique index.
+
+ The one-active-job-per-key guarantee lives in a partial unique index (raw SQL in
+ the migration — schema.prisma cannot express partial indexes), so the read-then-
+ create check above it is advisory: two concurrent starts pass the read, and the
+ loser must surface as the same 409 rather than a 500.
+ """
+ try:
+ from prisma.errors import UniqueViolationError
+ except ImportError:
+ return "unique constraint" in str(error).lower() or "P2002" in str(error)
+ return isinstance(error, UniqueViolationError)
+
+
def _require_admin_viewer(user_api_key_dict: UserAPIKeyAuth, action: str) -> None:
if user_api_key_dict.user_role not in (
LitellmUserRoles.PROXY_ADMIN,
@@ -670,19 +685,27 @@ async def start_shadow_eval(
estimated_cost: Final = round(sampled * per_call, 2)
ends_at: Final = datetime.now(timezone.utc) + timedelta(days=data.duration_days)
- job: Final = await prisma_client.db.litellm_shadowevaljob.create(
- data={ # mutable-ok: Prisma payload
- "api_key_id": data.api_key_id,
- "router_name": data.router_name,
- "shadow_percentage": data.shadow_percentage,
- "judge_model": data.judge_model,
- "team_id": data.team_id,
- "status": "pending",
- "cost_estimate": estimated_cost,
- "created_by": user_api_key_dict.user_id,
- "ends_at": ends_at,
- }
- )
+ try:
+ job: Final = await prisma_client.db.litellm_shadowevaljob.create(
+ data={ # mutable-ok: Prisma payload
+ "api_key_id": data.api_key_id,
+ "router_name": data.router_name,
+ "shadow_percentage": data.shadow_percentage,
+ "judge_model": data.judge_model,
+ "team_id": data.team_id,
+ "status": "pending",
+ "cost_estimate": estimated_cost,
+ "created_by": user_api_key_dict.user_id,
+ "ends_at": ends_at,
+ }
+ )
+ except Exception as e:
+ if not _is_unique_violation(e):
+ raise
+ raise HTTPException(
+ status_code=409,
+ detail="Key already has an active shadow eval job (started concurrently). Stop it first.",
+ ) from e
return StartShadowEvalResponse(
job_id=job.id,
status="pending",
diff --git a/tests/test_litellm/integrations/test_shadow_eval_logger.py b/tests/test_litellm/integrations/test_shadow_eval_logger.py
index 521e2314447..ac55554a7dc 100644
--- a/tests/test_litellm/integrations/test_shadow_eval_logger.py
+++ b/tests/test_litellm/integrations/test_shadow_eval_logger.py
@@ -668,6 +668,28 @@ class TestPerJobSpendCap:
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
+ exactly that job must still hit the floor instead of billing until ends_at."""
+ job = ActiveShadowEvalJob(
+ id="j1",
+ router_name="r",
+ shadow_percentage=100.0,
+ judge_model="m",
+ status="running",
+ cost_estimate=0.0,
+ cost_actual=1.5,
+ )
+ 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)
+
+ logger._run_shadow_eval.assert_not_awaited()
+ 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):
"""1.5x a $0.01 estimate is $0.015; without the floor a single judge call ends the job."""
job = ActiveShadowEvalJob(
@@ -727,14 +749,22 @@ class TestActiveJobSnapshot:
record.ends_at = None
return record
- async def test_one_query_serves_lookups_for_many_distinct_keys(self):
+ @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):
logger, prisma, _ = _logger_with_mocks()
prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[self._record("j1", "key-1")])
- job_hit = await logger._get_active_job("key-1")
- misses = [await logger._get_active_job(f"other-{i}") for i in range(50)]
+ 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)]
- assert job_hit is not None and job_hit.id == "j1"
+ 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)
assert prisma.db.litellm_shadowevaljob.find_many.await_count == 1
@@ -744,21 +774,46 @@ class TestActiveJobSnapshot:
return_value=[self._record("j-newest", "key-1"), self._record("j-oldest", "key-1")]
)
- job = await logger._get_active_job("key-1")
+ logger._get_active_job("key-1")
+ await self._settled(logger)
+ job = logger._get_active_job("key-1")
assert job is not None and job.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")])
- assert (await logger._get_active_job("key-1")) is not None
+ logger._get_active_job("key-1")
+ await self._settled(logger)
logger._jobs_fetched_at = asyncio.get_event_loop().time() - 61.0
prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=RuntimeError("db down"))
- job = await logger._get_active_job("key-1")
+ stale = logger._get_active_job("key-1")
+ await self._settled(logger)
- assert job is not None and job.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):
+ logger, prisma, _ = _logger_with_mocks()
+ started = asyncio.Event()
+ release = asyncio.Event()
+
+ async def slow_find_many(**_kwargs):
+ started.set()
+ await release.wait()
+ return [self._record("j1", "key-1")]
+
+ prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=slow_find_many)
+
+ 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
@@ -839,7 +894,10 @@ class TestJobsStopAtTheirScheduledEnd:
record.ends_at = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(seconds=1)
prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[record])
- job = await logger._get_active_job("key-hash")
+ 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")
assert job is not None
assert job.ends_at is not None and job.ends_at.tzinfo is not None
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 ae19fa56c7c..628e4eec9e7 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
@@ -583,6 +583,24 @@ class TestShadowEvalJobsAreTimeBound:
monkeypatch.setattr(proxy_server, "prisma_client", prisma)
return prisma
+ @pytest.mark.asyncio
+ async def test_losing_a_concurrent_start_race_is_a_409_not_a_500(self, monkeypatch: pytest.MonkeyPatch):
+ """Two admins can pass the advisory read simultaneously; the partial unique
+ index rejects the second create, which must read as the same conflict."""
+ from prisma.errors import UniqueViolationError
+
+ from litellm.proxy.management_endpoints.auto_router_endpoints import start_shadow_eval
+
+ prisma = self._proxy_mocks(monkeypatch, recent_requests=700)
+ prisma.db.litellm_shadowevaljob.create = AsyncMock(
+ side_effect=UniqueViolationError(data={"user_facing_error": {"meta": {"target": "api_key_id"}}})
+ )
+
+ with pytest.raises(HTTPException) as exc:
+ await start_shadow_eval(self._start_request(), ADMIN)
+
+ assert exc.value.status_code == 409
+
@pytest.mark.asyncio
async def test_estimate_reads_the_daily_rollup_not_the_raw_spend_log_table(self, monkeypatch: pytest.MonkeyPatch):
"""LiteLLM_SpendLogs has no api_key index; a per-key count there scans every
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 e3783a55e67..33ff722bf86 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
@@ -285,4 +285,17 @@ describe("ShadowEvalSection", () => {
expect(await screen.findByText("SIMPLE")).toBeInTheDocument();
expect(screen.getByText("REASONING")).toBeInTheDocument();
});
+
+ it("labels an unexpanded previous job as viewable, not verdictless", async () => {
+ const user = userEvent.setup();
+ const current = job({ job_id: "job-new" });
+ const older = job({ job_id: "job-old", status: "completed", results: null });
+ mockHooks({ jobs: [current, older], detailsById: { "job-new": current } });
+ render();
+
+ await user.click(screen.getByRole("button", { name: /Previous evaluations \(1\)/ }));
+
+ expect(screen.getByText("view results")).toBeInTheDocument();
+ expect(screen.queryByText("no verdicts")).not.toBeInTheDocument();
+ });
});
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 7d9d17b1e7e..9210204c0ca 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
@@ -365,6 +365,12 @@ const PreviousJob: React.FC<{
const shown = detail ?? job;
const results = shown.results;
const okOrBetter = results ? results.overall_shadow_win_rate_pct + results.overall_tie_rate_pct : null;
+ let summary = "no verdicts";
+ if (okOrBetter != null) {
+ summary = pct(okOrBetter);
+ } else if (shown.completed_count > 0) {
+ summary = "view results";
+ }
return (
@@ -387,9 +393,7 @@ const PreviousJob: React.FC<{
-
- {okOrBetter != null ? pct(okOrBetter) : "no verdicts"}
-
+ {summary}
{expanded ? (