fix: three review findings — cap bypass, start race, request-path DB read

Zero-estimate cap bypass (cursor): a key quiet during the estimate
lookback gets cost_estimate 0.0, which the spend cap treated the same
as 'no estimate' and left uncapped — a later traffic spike on exactly
that job would bill until ends_at. Only a NULL estimate (rows predating
estimates) is uncapped now; $0 still gets the $1 floor.

Concurrent-start race (cursor): the find_first-then-create check passes
on both sides of a race, giving one key two active jobs and double
judge spend. A partial unique index (api_key_id WHERE status IN
(pending, running)) — raw SQL in the unshipped migration, since
schema.prisma cannot express partial indexes — makes the DB the
arbiter; the losing create surfaces as the same 409 as the advisory
check. The old find_many('desc') + reversed() insertion in the logger
cache already prefers the newest job for any legacy duplicates.

Request-path DB read (greptile): an expired job snapshot awaited
find_many inside the success callback, so every N seconds one request
per pod paid a synchronous Prisma read. The lookup is now sync-only:
it serves the current snapshot and kicks a detached refresh task when
stale. Cost: a cold pod's first ~1 refresh-window of samples are
missed (acceptable for a sampled eval); stale-if-error semantics keep
a DB blip from disabling the feature.

Also from review: a collapsed previous-job row said 'no verdicts' for
jobs with thousands of verdicts, because the list endpoint omits
results by design — it now says 'view results' when completed_count>0.

Co-Authored-By: Claude <noreply@anthropic.com>

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

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.

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

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

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

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(<ShadowEvalSection accessToken="token" />);
+
+    await user.click(screen.getByRole("button", { name: /Previous evaluations \(1\)/ }));
+
+    expect(screen.getByText("view results")).toBeInTheDocument();
+    expect(screen.queryByText("no verdicts")).not.toBeInTheDocument();
+  });
 });

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 (
     <div className="border-b last:border-b-0">
@@ -387,9 +393,7 @@ const PreviousJob: React.FC<{
             </p>
           </div>
         </div>
-        <span className="text-sm font-medium text-foreground">
-          {okOrBetter != null ? pct(okOrBetter) : "no verdicts"}
-        </span>
+        <span className="text-sm font-medium text-foreground">{summary}</span>
       </button>
       {expanded ? (
         <div className="px-6 pb-4">