diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20260817000000_shadow_eval_multi_key/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260817000000_shadow_eval_multi_key/migration.sql new file mode 100644 index 00000000000..18ef5c40662 --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260817000000_shadow_eval_multi_key/migration.sql @@ -0,0 +1,7 @@ +ALTER TABLE "LiteLLM_ShadowEvalJob" ADD COLUMN IF NOT EXISTS "group_id" TEXT; + +UPDATE "LiteLLM_ShadowEvalJob" SET "group_id" = "id" WHERE "group_id" IS NULL; + +ALTER TABLE "LiteLLM_ShadowEvalJob" ALTER COLUMN "group_id" SET NOT NULL; + +CREATE INDEX IF NOT EXISTS "LiteLLM_ShadowEvalJob_group_id_idx" ON "LiteLLM_ShadowEvalJob"("group_id"); diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20260818224500_add_shadow_eval_stopped_by/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260818224500_add_shadow_eval_stopped_by/migration.sql new file mode 100644 index 00000000000..9efa3fdd052 --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260818224500_add_shadow_eval_stopped_by/migration.sql @@ -0,0 +1,5 @@ +-- AlterTable +ALTER TABLE "LiteLLM_ShadowEvalJob" ADD COLUMN "stopped_by" TEXT; + +UPDATE "LiteLLM_ShadowEvalJob" SET stopped_by = 'unknown' +WHERE stopped_at IS NOT NULL AND ends_at > (NOW() AT TIME ZONE 'utc'); diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma index 52fb447157b..f79e2bb0c18 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma +++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma @@ -1467,28 +1467,38 @@ model LiteLLM_AutoRouterSession { @@index([last_turn_at], map: "idx_autorouter_session_last_turn") } -// Shadow eval: evaluation of an auto-router against a key's live traffic, in either -// direction. forward duplicates the requests the key did not route through the router -// through it, answering whether the key should adopt it; reverse duplicates the requests -// the router did serve against a fixed baseline model, answering whether a key already on -// it still benefits. Either way a sampled slice runs in a detached task and an LLM judge -// compares real vs shadow responses blind. The job row is immutable config plus -// stopped_at; every count, status, and spend figure is derived from the append-only -// attempt rows, so nothing can disagree across pods or stop races. +// Shadow eval: evaluation of an auto-router against one or more keys' live traffic, in +// either direction. forward duplicates the requests the keys did not route through the +// router through it, answering whether they should adopt it; reverse duplicates the +// requests the router did serve against a fixed baseline model, answering whether a key +// already on it still benefits. Either way a sampled slice runs in a detached task and an +// LLM judge compares real vs shadow responses blind. Each row is ONE key's leg of a job: +// immutable config plus that key's own turn budget and stop state, so one key exhausting +// its budget never ends a sibling's sampling. A job is the set of legs sharing group_id +// (the id the API reports), written together by one atomic create_many with identical +// config; single-key jobs predating group_id were backfilled group_id = id. "One active +// job per (key, direction)" is a partial unique index on (api_key_id, direction) WHERE +// stopped_at IS NULL, expressed only in the migration because schema.prisma cannot state +// partial indexes; it is what makes a concurrent start on another pod race-safe rather +// than read-then-create. Every count, status, and spend figure is derived from the +// append-only attempt rows, so nothing can disagree across pods or stop races. model LiteLLM_ShadowEvalJob { id String @id @default(cuid()) - api_key_id String // hashed virtual key whose traffic is shadowed + group_id String // legs of one job share this; the API's job id + api_key_id String // hashed virtual key whose traffic this leg shadows router_name String // the auto-router under evaluation, in either direction direction String @default("forward") // forward | reverse baseline_model String? // reverse only: the fixed model the router is judged against judge_model String shadow_percentage Float - max_turns Int // sample budget: judge at most this many turns + max_turns Int // this key's sample budget: judge at most this many turns created_at DateTime @default(now()) created_by String? ends_at DateTime stopped_at DateTime? + stopped_by String? // operator who stopped it early; null when it ended on its own + @@index([group_id]) @@index([api_key_id]) @@index([created_at]) } diff --git a/litellm/proxy/management_endpoints/auto_router_endpoints.py b/litellm/proxy/management_endpoints/auto_router_endpoints.py index d8b7414f32c..0112ad1f6ed 100644 --- a/litellm/proxy/management_endpoints/auto_router_endpoints.py +++ b/litellm/proxy/management_endpoints/auto_router_endpoints.py @@ -6,10 +6,13 @@ POST /auto_router/test_routing - Route one prompt through an unsaved complexity- from collections.abc import Mapping, Sequence from datetime import datetime, timedelta, timezone +from itertools import groupby +from operator import attrgetter from types import MappingProxyType from typing import TYPE_CHECKING, Annotated, Final, Protocol +from uuid import uuid4 -from pydantic import BaseModel, TypeAdapter +from pydantic import BaseModel, ConfigDict, TypeAdapter, field_validator from litellm._logging import verbose_proxy_logger from litellm.exceptions import BudgetExceededError @@ -41,6 +44,8 @@ from litellm.types.management_endpoints.auto_router_endpoints import ( AutoRouterRoutingTestRequest, AutoRouterRoutingTestResponse, RequestComplexityRouterConfig, + ShadowEvalDirection, + ShadowEvalJobKeyResponse, ShadowEvalJobResponse, ShadowEvalResult, ShadowEvalSlice, @@ -89,17 +94,9 @@ class _ShadowEvalJobRow(Protocol): class _ShadowEvalJobTable(Protocol): - async def find_unique(self, *, where: Mapping[str, object]) -> _ShadowEvalJobRow | None: ... + async def find_many(self, *, where: Mapping[str, object]) -> Sequence[_ShadowEvalJobRow]: ... - async def find_first(self, *, where: Mapping[str, object]) -> _ShadowEvalJobRow | None: ... - - async def find_many( - self, *, where: Mapping[str, object], order: Mapping[str, str], take: int - ) -> Sequence[_ShadowEvalJobRow]: ... - - async def create(self, data: Mapping[str, object]) -> _ShadowEvalJobRow: ... - - async def update(self, *, where: Mapping[str, object], data: Mapping[str, object]) -> _ShadowEvalJobRow | None: ... + async def create_many(self, data: Sequence[Mapping[str, object]]) -> int: ... class _ShadowEvalAttemptRow(Protocol): @@ -606,18 +603,19 @@ _ATTEMPT_AGG_SELECT: Final = """ COUNT(*) FILTER (WHERE outcome = 'tie')::int AS ties, AVG(confidence)::float AS avg_confidence FROM "LiteLLM_ShadowEvalAttempt" -WHERE job_id = $1 AND outcome != 'error' +WHERE job_id = ANY($1::text[]) AND outcome != 'error' GROUP BY 1 """ _ATTEMPT_AGG_BY_TIER_SQL: Final = "SELECT COALESCE(tier, 'UNCLASSIFIED') AS grp," + _ATTEMPT_AGG_SELECT _ATTEMPT_AGG_BY_MODEL_SQL: Final = "SELECT COALESCE(real_model, 'unknown') AS grp," + _ATTEMPT_AGG_SELECT +_ATTEMPT_AGG_BY_LEG_SQL: Final = "SELECT job_id AS grp," + _ATTEMPT_AGG_SELECT _SWEEP_FINISHED_JOBS_SQL: Final = """ -UPDATE "LiteLLM_ShadowEvalJob" j SET stopped_at = NOW() -WHERE j.api_key_id = $1 AND j.stopped_at IS NULL +UPDATE "LiteLLM_ShadowEvalJob" j SET stopped_at = (NOW() AT TIME ZONE 'utc') +WHERE j.api_key_id = ANY($1::text[]) AND j.stopped_at IS NULL AND ( - j.ends_at <= NOW() + j.ends_at <= (NOW() AT TIME ZONE 'utc') OR (SELECT COUNT(*) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = j.id) >= j.max_turns ) """ @@ -628,7 +626,52 @@ SELECT COUNT(*) FILTER (WHERE outcome = 'error')::int AS error_count, COALESCE(SUM(judge_cost), 0)::float AS judge_spend FROM "LiteLLM_ShadowEvalAttempt" -WHERE job_id = $1 +WHERE job_id = ANY($1::text[]) +""" + +_ATTEMPT_COUNTS_SQL: Final = """ +SELECT a.job_id, COUNT(*)::int AS attempt_count +FROM "LiteLLM_ShadowEvalAttempt" a +JOIN "LiteLLM_ShadowEvalJob" j ON j.id = a.job_id +WHERE a.job_id = ANY($1::text[]) AND (j.stopped_at IS NULL OR a.created_at <= j.stopped_at) +GROUP BY a.job_id +""" + +_STOP_JOB_SQL: Final = """ +UPDATE "LiteLLM_ShadowEvalJob" +SET stopped_by = $2, stopped_at = COALESCE(stopped_at, $3::timestamp) +WHERE group_id = $1 AND stopped_by IS NULL + AND ends_at > (NOW() AT TIME ZONE 'utc') + AND EXISTS ( + SELECT 1 FROM "LiteLLM_ShadowEvalJob" k + WHERE k.group_id = $1 AND k.stopped_at IS NULL + AND (SELECT COUNT(*) FROM "LiteLLM_ShadowEvalAttempt" a WHERE a.job_id = k.id) < k.max_turns + ) +""" + + +class _AttemptCountRow(BaseModel): + job_id: str + attempt_count: int + + +_ATTEMPT_COUNT_ROWS: Final = TypeAdapter(list[_AttemptCountRow]) + + +_LIST_LEGS_SQL: Final = """ +SELECT * FROM "LiteLLM_ShadowEvalJob" +WHERE group_id IN ( + SELECT group_id FROM "LiteLLM_ShadowEvalJob" + GROUP BY group_id ORDER BY MAX(created_at) DESC LIMIT $1::int +) +""" + +_LIST_LEGS_BY_KEY_SQL: Final = """ +SELECT * FROM "LiteLLM_ShadowEvalJob" +WHERE group_id IN ( + SELECT group_id FROM "LiteLLM_ShadowEvalJob" WHERE api_key_id = $2 + GROUP BY group_id ORDER BY MAX(created_at) DESC LIMIT $1::int +) """ @@ -659,18 +702,98 @@ def _slices(rows: Sequence[_AttemptAggRow]) -> tuple[ShadowEvalSlice, ...]: ) +class _LegRow(BaseModel): + """One LiteLLM_ShadowEvalJob row, validated off the untyped prisma record. A row is + one key's leg of a job; the legs of a job share group_id and identical config, written + together by one create_many. The API's job id is the group id, so leg ids never leave + the server (attempts reference them internally).""" + + model_config = ConfigDict(from_attributes=True) + + id: str + group_id: str + api_key_id: str + router_name: str + direction: ShadowEvalDirection + baseline_model: str | None = None + judge_model: str + shadow_percentage: float + max_turns: int + created_at: datetime + ends_at: datetime + stopped_at: datetime | None = None + stopped_by: str | None = None + + @field_validator("created_at", "ends_at", "stopped_at") + @classmethod + def _as_aware_utc(cls, value: datetime | None) -> datetime | None: + """The columns store naive UTC wall time (prisma's convention); prisma reads hand + back aware datetimes while raw SQL reads hand back naive ones, so this boundary + makes every read aware UTC before anything compares or serializes them.""" + if value is None or value.tzinfo is not None: + return value + return value.replace(tzinfo=timezone.utc) + + +_LEG_ROWS: Final = TypeAdapter(list[_LegRow]) + + +async def _leg_attempt_counts(prisma_client: "PrismaClient", legs: Sequence[_LegRow]) -> Mapping[str, int]: + """Each leg's attempt count by leg id, judged and errored alike, in one grouped read. + It is the same count the sampler budgets against max_turns, so the derived status + flips to completed exactly when sampling actually ends. A stamped leg's count freezes + at its stopped_at: in-flight attempts that land after the stamp are excluded, so they + can never reclassify a leg that was stopped under budget as budget-spent.""" + if not legs: + return MappingProxyType({}) + rows: Final = _ATTEMPT_COUNT_ROWS.validate_python( + await _query_raw(prisma_client, _ATTEMPT_COUNTS_SQL, [leg.id for leg in legs]) # mutable-ok: query param + or () + ) + return MappingProxyType({row.job_id: row.attempt_count for row in rows}) + + +def _group_response(group_id: str, legs: Sequence[_LegRow], attempt_counts: Mapping[str, int]) -> ShadowEvalJobResponse: + """The one constructor of a job response: the caller names the group and passes that + group's legs. Config is read off the first leg because every leg carries the same copy, + written by one create_many. No caller may serialize a raw row (that would leak a leg id + as the job id).""" + first: Final = legs[0] + return ShadowEvalJobResponse( + job_id=group_id, + keys=tuple( + ShadowEvalJobKeyResponse( + api_key_id=leg.api_key_id, + max_turns=leg.max_turns, + stopped_at=leg.stopped_at, + attempt_count=attempt_counts.get(leg.id, 0), + ) + for leg in sorted(legs, key=lambda leg: leg.api_key_id) + ), + router_name=first.router_name, + direction=first.direction, + baseline_model=first.baseline_model, + judge_model=first.judge_model, + shadow_percentage=first.shadow_percentage, + created_at=first.created_at, + ends_at=first.ends_at, + stopped_by=next((leg.stopped_by for leg in legs if leg.stopped_by is not None), None), + ) + + _NO_KEY_LABELS: Final[tuple[str | None, str | None]] = (None, None) async def _with_key_labels( prisma_client: "PrismaClient", responses: Sequence[ShadowEvalJobResponse] ) -> tuple[ShadowEvalJobResponse, ...]: - """Resolve each job's key hash to the key's alias and masked name in one batched read, + """Resolve every scoped key's hash to its alias and masked name in one batched read, so the UI can say whose traffic a job shadows. Deleted keys resolve to None.""" if not responses: return () + tokens: Final = sorted(frozenset(key.api_key_id for response in responses for key in response.keys)) key_rows: Final = await _verification_tokens(prisma_client).find_many( - where={"token": {"in": sorted({response.api_key_id for response in responses})}} # mutable-ok: Prisma filter + where={"token": {"in": tokens}} # mutable-ok: Prisma filter ) labels: Final[Mapping[str, tuple[str | None, str | None]]] = { row.token: (row.key_alias, row.key_name) for row in key_rows or () @@ -678,32 +801,50 @@ async def _with_key_labels( return tuple( response.model_copy( update={ # mutable-ok: pydantic update payload - "key_alias": labels.get(response.api_key_id, _NO_KEY_LABELS)[0], - "key_name": labels.get(response.api_key_id, _NO_KEY_LABELS)[1], + "keys": tuple( + key.model_copy( + update={ # mutable-ok: pydantic update payload + "key_alias": labels.get(key.api_key_id, _NO_KEY_LABELS)[0], + "key_name": labels.get(key.api_key_id, _NO_KEY_LABELS)[1], + } + ) + for key in response.keys + ) } ) for response in responses ) -async def _shadow_eval_results(prisma_client: "PrismaClient", job_id: str) -> ShadowEvalResult | None: - """Both stratifications of one job's verdicts. Tier answers "where does the router do - well"; the model stratification groups by whichever model served the real arm, so it - answers "which of the models this key uses today would the router beat" forward, and - "for the turns the router sent to X, did X beat the baseline" in reverse. Reads are - bounded by the job's own attempts (<= max_turns) via the job_id index.""" +async def _shadow_eval_results(prisma_client: "PrismaClient", legs: Sequence[_LegRow]) -> ShadowEvalResult | None: + """All three stratifications of one job's verdicts. Tier answers "where does the router + do well"; the model stratification groups by whichever model served the real arm, so it + answers "which of the models these keys use today would the router beat" forward, and + "for the turns the router sent to X, did X beat the baseline" in reverse; key answers + "which key's traffic does the router suit". Reads are bounded by the job's own attempts + (<= the sum of its keys' max_turns) via the job_id index.""" + leg_ids: Final = [leg.id for leg in legs] # mutable-ok: query param by_tier: Final = _ATTEMPT_AGG_ROWS.validate_python( - await _query_raw(prisma_client, _ATTEMPT_AGG_BY_TIER_SQL, job_id) or () + await _query_raw(prisma_client, _ATTEMPT_AGG_BY_TIER_SQL, leg_ids) or () ) if not by_tier: return None by_model: Final = _ATTEMPT_AGG_ROWS.validate_python( - await _query_raw(prisma_client, _ATTEMPT_AGG_BY_MODEL_SQL, job_id) or () + await _query_raw(prisma_client, _ATTEMPT_AGG_BY_MODEL_SQL, leg_ids) or () + ) + key_by_leg: Final = MappingProxyType({leg.id: leg.api_key_id for leg in legs}) + by_leg: Final = _ATTEMPT_AGG_ROWS.validate_python( + await _query_raw(prisma_client, _ATTEMPT_AGG_BY_LEG_SQL, leg_ids) or () + ) + by_key: Final = tuple( + row.model_copy(update={"grp": key_by_leg[row.grp]}) # mutable-ok: pydantic update payload + for row in by_leg ) total_turns: Final = sum(r.turn_count for r in by_tier) return ShadowEvalResult( by_tier=_slices(by_tier), by_current_model=_slices(by_model), + by_key=_slices(by_key), overall_shadow_win_rate_pct=_pct_of(sum(r.shadow_wins for r in by_tier), total_turns), overall_tie_rate_pct=_pct_of(sum(r.ties for r in by_tier), total_turns), ) @@ -721,20 +862,21 @@ async def start_shadow_eval( user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], ) -> ShadowEvalJobResponse: """ - Start a shadow eval: duplicate a sampled slice of a key's live traffic against a second - arm, judge the two responses blind, and stratify win rates by tier and by the model that - served the real arm. + Start a shadow eval: duplicate a sampled slice of one or more keys' live traffic against + a second arm, judge the two responses blind, and stratify win rates by tier, by the model + that served the real arm, and by key. - A forward job answers whether the key should adopt router_name: it samples the requests + A forward job answers whether the keys should adopt router_name: it samples the requests the router did not serve and duplicates them through it. A reverse job answers whether a key already on the router still gains from it: it samples the requests the router did serve and duplicates them against baseline_model. A key can hold one active job per direction, so both questions can run at once. - Shadow responses are never served to users. The job samples until it has judged - max_turns turns, reaches the end of its window, or is stopped; sampling changes - propagate to pods within about 10 seconds. Shadow and judge calls bill to the - shadowed key but are excluded from request counts and auto-router adoption metrics. + Shadow responses are never served to users. Each key samples until it has judged + max_turns turns of its own traffic, the job's window ends, or the job is stopped, so one + key running out of budget does not end sampling for the others; sampling changes + propagate to pods within about 10 seconds. Shadow and judge calls bill to the shadowed + key but are excluded from request counts and auto-router adoption metrics. """ from litellm.proxy.proxy_server import llm_router, prisma_client @@ -746,48 +888,58 @@ async def start_shadow_eval( _validate_plain_model(llm_router, data.judge_model, "judge_model") if data.baseline_model is not None: _validate_plain_model(llm_router, data.baseline_model, "baseline_model") - key_row: Final = await _verification_tokens(prisma_client).find_unique( - where={"token": data.api_key_id} # mutable-ok: Prisma filter + token_rows: Final = await _verification_tokens(prisma_client).find_many( + where={"token": {"in": list(data.api_key_ids)}} # mutable-ok: Prisma filter ) - if key_row is None: + unknown: Final = tuple(sorted(frozenset(data.api_key_ids) - frozenset(row.token for row in token_rows or ()))) + if unknown: raise HTTPException( status_code=400, detail=( - f"api_key_id '{data.api_key_id}' is not a key on this proxy; pass the key's token hash, " + f"api_key_ids not on this proxy: {', '.join(unknown)}; pass each key's token hash, " "the value the key list and key info endpoints report" ), ) - # A job that expired or exhausted its turn budget stopped sampling on its own, but - # still holds its slot in the per-key, per-direction partial unique index until - # stamped; free it so a new eval can start. Sweeping both directions is deliberate. - await prisma_client.db.execute_raw(_SWEEP_FINISHED_JOBS_SQL, data.api_key_id) - active: Final = await _shadow_eval_jobs(prisma_client).find_first( + # A job whose window passed or whose turn budget ran out stopped sampling on its own, + # but its legs still hold their slots in the per-key, per-direction partial unique index + # until stamped; free them so a new eval can start. Sweeping both directions is deliberate. + requested: Final = list(data.api_key_ids) # mutable-ok: query param + await prisma_client.db.execute_raw(_SWEEP_FINISHED_JOBS_SQL, requested) + claimed: Final = await _shadow_eval_jobs(prisma_client).find_many( where={ # mutable-ok: Prisma filter - "api_key_id": data.api_key_id, + "api_key_id": {"in": requested}, # mutable-ok: Prisma filter "direction": data.direction, "stopped_at": None, }, ) - if active is not None: + if claimed: raise HTTPException( status_code=409, - detail=f"Key already has an active {data.direction} shadow eval job ({active.id}). Stop it first.", + detail=( + f"Already in an active {data.direction} shadow eval job: " + + ", ".join(sorted(f"{row.api_key_id} (job {row.group_id})" for row in claimed)) + + ". Stop it first." + ), ) now: Final = datetime.now(timezone.utc) + group_id: Final = str(uuid4()) + ends_at: Final = now + timedelta(days=data.duration_days) + shared_config: Final = { # mutable-ok: Prisma payload + "group_id": group_id, + "router_name": data.router_name, + "direction": data.direction, + "baseline_model": data.baseline_model, + "judge_model": data.judge_model, + "shadow_percentage": data.shadow_percentage, + "max_turns": data.max_turns, + "created_by": user_api_key_dict.user_id, + "created_at": now, + "ends_at": ends_at, + } try: - job: Final = await _shadow_eval_jobs(prisma_client).create( - data={ # mutable-ok: Prisma payload - "api_key_id": data.api_key_id, - "router_name": data.router_name, - "direction": data.direction, - "baseline_model": data.baseline_model, - "judge_model": data.judge_model, - "shadow_percentage": data.shadow_percentage, - "max_turns": data.max_turns, - "created_by": user_api_key_dict.user_id, - "ends_at": now + timedelta(days=data.duration_days), - } + await _shadow_eval_jobs(prisma_client).create_many( + data=[{**shared_config, "api_key_id": key} for key in data.api_key_ids] # mutable-ok: Prisma payload ) except Exception as e: if not _is_unique_violation(e): @@ -795,11 +947,28 @@ async def start_shadow_eval( raise HTTPException( status_code=409, detail=( - f"Key already has an active {data.direction} shadow eval job (started concurrently). Stop it first." + f"A requested key was claimed by another {data.direction} shadow eval job concurrently. Stop it first." ), ) from e - return ShadowEvalJobResponse.model_validate(job, from_attributes=True).model_copy( - update={"key_alias": key_row.key_alias, "key_name": key_row.key_name} # mutable-ok: pydantic update payload + labels: Final = MappingProxyType({row.token: row for row in token_rows}) + return ShadowEvalJobResponse( + job_id=group_id, + keys=tuple( + ShadowEvalJobKeyResponse( + api_key_id=api_key_id, + max_turns=data.max_turns, + key_alias=labels[api_key_id].key_alias, + key_name=labels[api_key_id].key_name, + ) + for api_key_id in sorted(data.api_key_ids) + ), + router_name=data.router_name, + direction=data.direction, + baseline_model=data.baseline_model, + judge_model=data.judge_model, + shadow_percentage=data.shadow_percentage, + created_at=now, + ends_at=ends_at, ) @@ -811,23 +980,38 @@ async def start_shadow_eval( ) async def list_shadow_eval_jobs( user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], - api_key_id: Annotated[str | None, Query(description="Filter to jobs shadowing this key")] = None, + api_key_id: Annotated[ + str | None, Query(description="Filter to jobs that shadow this key, alone or alongside others") + ] = None, limit: Annotated[int, Query(ge=1, le=200, description="Newest jobs to return")] = 50, ) -> tuple[ShadowEvalJobResponse, ...]: - """List shadow eval jobs, newest first. Counts and results ride the detail endpoint only.""" + """List shadow eval jobs, newest first, each key with its attempt count so status is + accurate. Judged counts, spend, and results ride the detail endpoint only.""" from litellm.proxy.proxy_server import prisma_client _require_admin_viewer(user_api_key_dict, "view shadow evals") if prisma_client is None: raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value) - records: Final = await _shadow_eval_jobs(prisma_client).find_many( - where={"api_key_id": api_key_id} if api_key_id else {}, # mutable-ok: Prisma filter - order={"created_at": "desc"}, # mutable-ok: Prisma order - take=limit, + legs: Final = _LEG_ROWS.validate_python( + ( + await _query_raw(prisma_client, _LIST_LEGS_BY_KEY_SQL, limit, api_key_id) + if api_key_id + else await _query_raw(prisma_client, _LIST_LEGS_SQL, limit) + ) + or () ) + by_group: Final[Mapping[str, tuple[_LegRow, ...]]] = MappingProxyType( + { + group_id: tuple(group) + for group_id, group in groupby(sorted(legs, key=attrgetter("group_id")), key=attrgetter("group_id")) + } + ) + newest_first: Final = sorted( + by_group, key=lambda group_id: max(leg.created_at for leg in by_group[group_id]), reverse=True + ) + counts: Final = await _leg_attempt_counts(prisma_client, legs) return await _with_key_labels( - prisma_client, - tuple(ShadowEvalJobResponse.model_validate(record, from_attributes=True) for record in records or ()), + prisma_client, tuple(_group_response(group_id, by_group[group_id], counts) for group_id in newest_first) ) @@ -847,20 +1031,24 @@ async def get_shadow_eval_job( _require_admin_viewer(user_api_key_dict, "view shadow evals") if prisma_client is None: raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value) - record: Final = await _shadow_eval_jobs(prisma_client).find_unique( - where={"id": job_id} # mutable-ok: Prisma filter + legs: Final = _LEG_ROWS.validate_python( + await _shadow_eval_jobs(prisma_client).find_many( + where={"group_id": job_id} # mutable-ok: Prisma filter + ) + or () ) - if record is None: + if not legs: raise HTTPException(status_code=404, detail=f"No shadow eval job {job_id}") + leg_ids: Final = [leg.id for leg in legs] # mutable-ok: query param totals: Final = _ATTEMPT_TOTALS_ROWS.validate_python( - await _query_raw(prisma_client, _ATTEMPT_TOTALS_SQL, job_id) or () + await _query_raw(prisma_client, _ATTEMPT_TOTALS_SQL, leg_ids) or () ) latest_error: Final = await _shadow_eval_attempts(prisma_client).find_first( - where={"job_id": job_id, "outcome": "error"}, # mutable-ok: Prisma filter + where={"job_id": {"in": leg_ids}, "outcome": "error"}, # mutable-ok: Prisma filter order={"created_at": "desc"}, # mutable-ok: Prisma order ) labeled: Final = await _with_key_labels( - prisma_client, (ShadowEvalJobResponse.model_validate(record, from_attributes=True),) + prisma_client, (_group_response(job_id, legs, await _leg_attempt_counts(prisma_client, legs)),) ) return labeled[0].model_copy( update={ # mutable-ok: pydantic update payload @@ -868,7 +1056,7 @@ async def get_shadow_eval_job( "error_count": totals[0].error_count if totals else 0, "judge_spend": round(totals[0].judge_spend, 6) if totals else 0.0, "last_error": latest_error.error if latest_error else None, - "results": await _shadow_eval_results(prisma_client, job_id), + "results": await _shadow_eval_results(prisma_client, legs), } ) @@ -883,25 +1071,33 @@ async def stop_shadow_eval_job( job_id: str, user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], ) -> ShadowEvalJobResponse: - """Stop an active shadow eval job. Attempts are kept; sampling halts within ~10s.""" + """Stop an active shadow eval job, every key it scopes at once. Attempts are kept; + sampling halts within ~10s. Keys that already stopped on their own budget keep the + stopped_at they earned. The statement is the whole state machine: it claims the job + only while a leg still samples inside the window with no stop recorded, so a racing + operator, a same-instant budget spend, and a repeat stop all read the same 400 with + the status the job actually holds.""" from litellm.proxy.proxy_server import prisma_client _require_admin_writer(user_api_key_dict, "stop a shadow eval") if prisma_client is None: raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value) - record: Final = await _shadow_eval_jobs(prisma_client).find_unique( - where={"id": job_id} # mutable-ok: Prisma filter + stamp: Final = datetime.now(timezone.utc) + operator: Final = user_api_key_dict.user_id or "operator" + claimed: Final = await prisma_client.db.execute_raw( + _STOP_JOB_SQL, job_id, operator, stamp.replace(tzinfo=None).isoformat() ) - if record is None: + legs: Final = _LEG_ROWS.validate_python( + await _shadow_eval_jobs(prisma_client).find_many( + where={"group_id": job_id} # mutable-ok: Prisma filter + ) + or () + ) + if not legs: raise HTTPException(status_code=404, detail=f"No shadow eval job {job_id}") - current: Final = ShadowEvalJobResponse.model_validate(record, from_attributes=True) - if current.status != "running": + counts: Final = await _leg_attempt_counts(prisma_client, legs) + current: Final = _group_response(job_id, legs, counts) + if claimed == 0: raise HTTPException(status_code=400, detail=f"Job {job_id} is already {current.status}") - updated: Final = await _shadow_eval_jobs(prisma_client).update( - where={"id": job_id}, # mutable-ok: Prisma filter - data={"stopped_at": datetime.now(timezone.utc)}, # mutable-ok: Prisma payload - ) - labeled: Final = await _with_key_labels( - prisma_client, (ShadowEvalJobResponse.model_validate(updated, from_attributes=True),) - ) + labeled: Final = await _with_key_labels(prisma_client, (current,)) return labeled[0] diff --git a/litellm/proxy/schema.prisma b/litellm/proxy/schema.prisma index 52fb447157b..f79e2bb0c18 100644 --- a/litellm/proxy/schema.prisma +++ b/litellm/proxy/schema.prisma @@ -1467,28 +1467,38 @@ model LiteLLM_AutoRouterSession { @@index([last_turn_at], map: "idx_autorouter_session_last_turn") } -// Shadow eval: evaluation of an auto-router against a key's live traffic, in either -// direction. forward duplicates the requests the key did not route through the router -// through it, answering whether the key should adopt it; reverse duplicates the requests -// the router did serve against a fixed baseline model, answering whether a key already on -// it still benefits. Either way a sampled slice runs in a detached task and an LLM judge -// compares real vs shadow responses blind. The job row is immutable config plus -// stopped_at; every count, status, and spend figure is derived from the append-only -// attempt rows, so nothing can disagree across pods or stop races. +// Shadow eval: evaluation of an auto-router against one or more keys' live traffic, in +// either direction. forward duplicates the requests the keys did not route through the +// router through it, answering whether they should adopt it; reverse duplicates the +// requests the router did serve against a fixed baseline model, answering whether a key +// already on it still benefits. Either way a sampled slice runs in a detached task and an +// LLM judge compares real vs shadow responses blind. Each row is ONE key's leg of a job: +// immutable config plus that key's own turn budget and stop state, so one key exhausting +// its budget never ends a sibling's sampling. A job is the set of legs sharing group_id +// (the id the API reports), written together by one atomic create_many with identical +// config; single-key jobs predating group_id were backfilled group_id = id. "One active +// job per (key, direction)" is a partial unique index on (api_key_id, direction) WHERE +// stopped_at IS NULL, expressed only in the migration because schema.prisma cannot state +// partial indexes; it is what makes a concurrent start on another pod race-safe rather +// than read-then-create. Every count, status, and spend figure is derived from the +// append-only attempt rows, so nothing can disagree across pods or stop races. model LiteLLM_ShadowEvalJob { id String @id @default(cuid()) - api_key_id String // hashed virtual key whose traffic is shadowed + group_id String // legs of one job share this; the API's job id + api_key_id String // hashed virtual key whose traffic this leg shadows router_name String // the auto-router under evaluation, in either direction direction String @default("forward") // forward | reverse baseline_model String? // reverse only: the fixed model the router is judged against judge_model String shadow_percentage Float - max_turns Int // sample budget: judge at most this many turns + max_turns Int // this key's sample budget: judge at most this many turns created_at DateTime @default(now()) created_by String? ends_at DateTime stopped_at DateTime? + stopped_by String? // operator who stopped it early; null when it ended on its own + @@index([group_id]) @@index([api_key_id]) @@index([created_at]) } diff --git a/litellm/types/management_endpoints/auto_router_endpoints.py b/litellm/types/management_endpoints/auto_router_endpoints.py index 9461297feca..63c93e0f268 100644 --- a/litellm/types/management_endpoints/auto_router_endpoints.py +++ b/litellm/types/management_endpoints/auto_router_endpoints.py @@ -6,7 +6,7 @@ from collections.abc import Mapping from datetime import datetime, timezone from typing import Final, Literal, TypeAlias -from pydantic import AliasChoices, BaseModel, ConfigDict, Field, computed_field, field_validator, model_validator +from pydantic import BaseModel, Field, computed_field, field_validator, model_validator from litellm.router_strategy.complexity_router.config import ComplexityRouterConfig from litellm.types.utils import StandardLoggingRoutingDecision @@ -155,13 +155,17 @@ DEFAULT_SHADOW_EVAL_JUDGE_MODEL: Final[str] = "anthropic/claude-sonnet-5" class StartShadowEvalRequest(BaseModel): - """Start duplicating a key's traffic for blind comparison against an auto-router.""" + """Start duplicating one or more keys' traffic for blind comparison against an auto-router.""" - api_key_id: str = Field( + api_key_ids: tuple[str, ...] = Field( + min_length=1, + max_length=100, description=( - "The hashed virtual key whose traffic will be shadowed. Shadow evaluation runs ONLY on this " - "key's traffic; requests made with any other key are not sampled." - ) + "The hashed virtual keys whose traffic will be shadowed. Shadow evaluation runs ONLY on these " + "keys' traffic; requests made with any other key are not sampled. Each key carries its own " + "max_turns budget, so one key exhausting its budget leaves the others sampling. At most 100 " + "keys per job, which also bounds every read the job's endpoints make." + ), ) router_name: str = Field(description="The auto-router under evaluation, in either direction") direction: ShadowEvalDirection = Field( @@ -204,8 +208,9 @@ class StartShadowEvalRequest(BaseModel): ge=1, le=2000, description=( - "Sample budget: the job judges at most this many turns, then completes. This is also the spend " - "bound; expected judge cost is roughly max_turns times one judge call" + "Per-key sample budget: the job judges at most this many turns of EACH scoped key's traffic, " + "so a job over N keys judges at most N times max_turns turns. This is also the spend bound; " + "expected judge cost is roughly that turn ceiling times one judge call" ), ) @@ -214,6 +219,12 @@ class StartShadowEvalRequest(BaseModel): def _round_percentage(cls, value: float) -> float: return round(value, 2) + @field_validator("api_key_ids") + @classmethod + def _dedupe_keys(cls, value: tuple[str, ...]) -> tuple[str, ...]: + """A key named twice would collide with itself on the one-active-per-(key, direction) index.""" + return tuple(dict.fromkeys(value)) + @model_validator(mode="after") def _baseline_model_matches_direction(self) -> "StartShadowEvalRequest": if self.direction == "reverse" and self.baseline_model is None: @@ -251,24 +262,46 @@ class ShadowEvalResult(BaseModel): by_tier: tuple[ShadowEvalSlice, ...] by_current_model: tuple[ShadowEvalSlice, ...] = Field( description=( - "Sliced by the model that served the real arm: the key's incumbent models in forward mode, " + "Sliced by the model that served the real arm: the keys' incumbent models in forward mode, " "and in reverse the models the router itself picked" ) ) + by_key: tuple[ShadowEvalSlice, ...] = Field( + description=( + "One slice per scoped key that has judged verdicts, grouped on the raw key hash. Keys the job " + "scopes but has not judged a turn for yet are absent rather than reported as zero" + ), + ) overall_shadow_win_rate_pct: float overall_tie_rate_pct: float -class ShadowEvalJobResponse(BaseModel): - """A shadow-eval job. Validates directly from the prisma record (job_id reads the - row's id); status is derived from stopped_at and ends_at, never stored, so no writer - anywhere can produce an inconsistent one. Aggregate fields are populated by the - detail endpoint only and stay None on list responses.""" +class ShadowEvalJobKeyResponse(BaseModel): + """One key a job shadows, with its own budget and stop state.""" - model_config = ConfigDict(from_attributes=True, populate_by_name=True) + api_key_id: str = Field(description="The hashed virtual key whose traffic this entry scopes") + max_turns: int = Field(description="This key's own sample budget, independent of its siblings'") + stopped_at: datetime | None = Field( + default=None, + description=( + "When this key's slot was stamped free, whether its own budget ran out, the window closed, " + "or an operator stopped the job; status is derived, so a spent budget reads completed even " + "while this is still unset" + ), + ) + attempt_count: int | None = Field( + default=None, + description=( + "This key's sampled attempts so far, judged and errored alike, the same count the sampler " + "budgets against max_turns; populated on list and detail responses. Frozen at stopped_at " + "once the key is stamped, so in-flight attempts landing after a stop never reclassify it" + ), + ) + + @property + def budget_spent(self) -> bool: + return self.attempt_count is not None and self.attempt_count >= self.max_turns - job_id: str = Field(validation_alias=AliasChoices("id", "job_id")) - api_key_id: str = Field(description="The hashed virtual key whose traffic this job evaluates, and only that key's") key_alias: str | None = Field( default=None, description="Alias of the shadowed key, resolved from the key row at read time; None when unset or deleted", @@ -277,15 +310,34 @@ class ShadowEvalJobResponse(BaseModel): default=None, description="Masked display name (sk-...) of the shadowed key, resolved at read time like key_alias", ) + + +class ShadowEvalJobResponse(BaseModel): + """A shadow-eval job over one or more keys, each with its own budget and stop state; + status is derived from stopped_by, the keys' stop and budget state, and ends_at, + never stored, so no writer anywhere can produce an inconsistent one. Aggregate + fields are populated by the detail endpoint only and stay None on list responses.""" + + job_id: str + keys: tuple[ShadowEvalJobKeyResponse, ...] = Field( + min_length=1, + description="The keys whose traffic this job evaluates, and only those keys', each with its own budget", + ) router_name: str direction: ShadowEvalDirection = "forward" baseline_model: str | None = None judge_model: str shadow_percentage: float - max_turns: int created_at: datetime ends_at: datetime - stopped_at: datetime | None = None + stopped_by: str | None = Field( + default=None, + description=( + "The operator who stopped the job early, recorded by the stop endpoint; 'unknown' backfilled " + "by migration for jobs that displayed stopped when the column arrived; None when the job " + "ended on its own. Its presence is what makes a job read stopped rather than completed" + ), + ) judged_count: int | None = Field(default=None, description="Verdicts recorded; detail endpoint only") error_count: int | None = Field(default=None, description="Sampled attempts that errored; detail endpoint only") @@ -296,12 +348,19 @@ class ShadowEvalJobResponse(BaseModel): @computed_field @property def status(self) -> ShadowEvalStatus: - """A job whose window has passed reads completed even if a later sweep stamped - stopped_at; stopped means sampling ended before the window did.""" + """Three recorded facts, no history-guessing: a stop is stopped_by (the migration + backfills it for every job that displayed stopped when the column arrived, so the + pre-column population is closed), completion is the window passing or every key + spending its budget, and anything else is running. The all-keys-stamped fallback + covers only stops written by pre-column pods during a rolling deploy.""" + if self.stopped_by is not None: + return "stopped" if datetime.now(timezone.utc) >= ( self.ends_at if self.ends_at.tzinfo else self.ends_at.replace(tzinfo=timezone.utc) ): return "completed" - if self.stopped_at is not None: + if all(key.budget_spent for key in self.keys): + return "completed" + if all(key.stopped_at is not None for key in self.keys): return "stopped" return "running" diff --git a/schema.prisma b/schema.prisma index 52fb447157b..f79e2bb0c18 100644 --- a/schema.prisma +++ b/schema.prisma @@ -1467,28 +1467,38 @@ model LiteLLM_AutoRouterSession { @@index([last_turn_at], map: "idx_autorouter_session_last_turn") } -// Shadow eval: evaluation of an auto-router against a key's live traffic, in either -// direction. forward duplicates the requests the key did not route through the router -// through it, answering whether the key should adopt it; reverse duplicates the requests -// the router did serve against a fixed baseline model, answering whether a key already on -// it still benefits. Either way a sampled slice runs in a detached task and an LLM judge -// compares real vs shadow responses blind. The job row is immutable config plus -// stopped_at; every count, status, and spend figure is derived from the append-only -// attempt rows, so nothing can disagree across pods or stop races. +// Shadow eval: evaluation of an auto-router against one or more keys' live traffic, in +// either direction. forward duplicates the requests the keys did not route through the +// router through it, answering whether they should adopt it; reverse duplicates the +// requests the router did serve against a fixed baseline model, answering whether a key +// already on it still benefits. Either way a sampled slice runs in a detached task and an +// LLM judge compares real vs shadow responses blind. Each row is ONE key's leg of a job: +// immutable config plus that key's own turn budget and stop state, so one key exhausting +// its budget never ends a sibling's sampling. A job is the set of legs sharing group_id +// (the id the API reports), written together by one atomic create_many with identical +// config; single-key jobs predating group_id were backfilled group_id = id. "One active +// job per (key, direction)" is a partial unique index on (api_key_id, direction) WHERE +// stopped_at IS NULL, expressed only in the migration because schema.prisma cannot state +// partial indexes; it is what makes a concurrent start on another pod race-safe rather +// than read-then-create. Every count, status, and spend figure is derived from the +// append-only attempt rows, so nothing can disagree across pods or stop races. model LiteLLM_ShadowEvalJob { id String @id @default(cuid()) - api_key_id String // hashed virtual key whose traffic is shadowed + group_id String // legs of one job share this; the API's job id + api_key_id String // hashed virtual key whose traffic this leg shadows router_name String // the auto-router under evaluation, in either direction direction String @default("forward") // forward | reverse baseline_model String? // reverse only: the fixed model the router is judged against judge_model String shadow_percentage Float - max_turns Int // sample budget: judge at most this many turns + max_turns Int // this key's sample budget: judge at most this many turns created_at DateTime @default(now()) created_by String? ends_at DateTime stopped_at DateTime? + stopped_by String? // operator who stopped it early; null when it ended on its own + @@index([group_id]) @@index([api_key_id]) @@index([created_at]) } 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 77149457e82..3fd023552f5 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 @@ -4,6 +4,7 @@ Unit tests for auto router management endpoints import os import sys +from pathlib import Path import pytest from fastapi import HTTPException @@ -325,9 +326,7 @@ class TestAutoRouterBenchmarks: from litellm.proxy.management_endpoints.auto_router_endpoints import _benchmark_totals totals = _benchmark_totals(self.ROW) - bucket_hits = ( - totals.cache.same_model.hits + totals.cache.first_visit.hits + totals.cache.return_to_tier.hits - ) + bucket_hits = totals.cache.same_model.hits + totals.cache.first_visit.hits + totals.cache.return_to_tier.hits assert bucket_hits == 27 assert totals.cache.hit_rate_pct == pytest.approx(100.0 * 28 / 38, abs=0.1) @@ -490,7 +489,7 @@ from litellm.proxy.management_endpoints.auto_router_endpoints import ( start_shadow_eval, stop_shadow_eval_job, ) -from litellm.types.management_endpoints.auto_router_endpoints import ShadowEvalJobResponse, StartShadowEvalRequest +from litellm.types.management_endpoints.auto_router_endpoints import StartShadowEvalRequest VIEWER = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, api_key="sk-view", user_id="viewer") NON_ADMIN = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, api_key="sk-user", user_id="user") @@ -507,19 +506,23 @@ def _shadow_router() -> MagicMock: return router -def _job_record(**overrides: object) -> MagicMock: - """Spec'd like a real prisma row: only the table's columns exist as attributes, so - from_attributes validation falls back to model defaults for everything else.""" +def _leg_record(**overrides: object) -> MagicMock: + """Spec'd like a real prisma row: only the table's columns exist as attributes. One + row is one key's leg of a job; legs sharing group_id are one job.""" defaults = { - "id": "job-1", + "id": "leg-1", + "group_id": "job-1", "api_key_id": "key-hash", "router_name": "my-router", + "direction": "forward", + "baseline_model": None, "judge_model": "anthropic/claude-sonnet-5", "shadow_percentage": 10.0, "max_turns": 200, "created_at": datetime(2026, 8, 11, tzinfo=timezone.utc), "ends_at": datetime.now(timezone.utc) + timedelta(days=7), "stopped_at": None, + "stopped_by": None, } fields = {**defaults, **overrides} record = MagicMock(spec=list(fields)) @@ -538,23 +541,90 @@ def _key_record( return record -def _shadow_prisma(active_job=None, agg_rows=None) -> MagicMock: +def _shadow_prisma(legs=(), agg_rows=None, by_leg_rows=None, known_keys=("key-hash", "key-hash-2")) -> MagicMock: + """The job-table fake honours the filters it is handed, so a read that forgets + stopped_at sees rows the partial index would have released, one that forgets + direction sees the opposite-direction legs a key may hold at the same time, and a + group read that matched on a leg id would come back empty.""" prisma = MagicMock() - prisma.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=_key_record()) - prisma.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[_key_record()]) - prisma.db.execute_raw = AsyncMock(return_value=0) - prisma.db.litellm_shadowevaljob.find_first = AsyncMock(return_value=active_job) - prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=None) - prisma.db.litellm_shadowevaljob.find_many = AsyncMock(return_value=[]) - prisma.db.litellm_shadowevaljob.create = AsyncMock(return_value=_job_record()) - prisma.db.litellm_shadowevaljob.update = AsyncMock( - return_value=_job_record(stopped_at=datetime.now(timezone.utc)) - ) + prisma.db.litellm_verificationtoken.find_many = AsyncMock(return_value=[_key_record(token) for token in known_keys]) + async def execute_raw(sql: str, *params: object): + if "SET stopped_by" in sql: + group = [row for row in stored if row.group_id == params[0]] + counts = {row["job_id"]: row["attempt_count"] for row in prisma.attempt_rows} + sampling = any(row.stopped_at is None and counts.get(row.id, 0) < row.max_turns for row in group) + window_open = bool(group) and group[0].ends_at > datetime.now(timezone.utc) + claimable = [row for row in group if row.stopped_by is None] + if not (claimable and sampling and window_open): + return 0 + for row in claimable: + row.stopped_by = params[1] + if row.stopped_at is None: + row.stopped_at = datetime.fromisoformat(str(params[2])).replace(tzinfo=timezone.utc) + return len(claimable) + return 0 + + prisma.db.execute_raw = AsyncMock(side_effect=execute_raw) + stored = legs if isinstance(legs, list) else list(legs) + + async def find_many_legs(where=None, **_: object): + current = list(stored) + w = dict(where or {}) + if "api_key_id" in w: + wanted = w["api_key_id"]["in"] if isinstance(w["api_key_id"], dict) else [w["api_key_id"]] + current = [row for row in current if row.api_key_id in wanted] + if "direction" in w: + current = [row for row in current if row.direction == w["direction"]] + if "stopped_at" in w: + current = [row for row in current if row.stopped_at is w["stopped_at"]] + if "group_id" in w: + wanted = w["group_id"]["in"] if isinstance(w["group_id"], dict) else [w["group_id"]] + current = [row for row in current if row.group_id in wanted] + return current + + def newest_groups(rows, limit): + latest: dict = {} + for row in rows: + if row.group_id not in latest or row.created_at > latest[row.group_id]: + latest[row.group_id] = row.created_at + ordered = sorted(latest, key=lambda group_id: latest[group_id], reverse=True) + return ordered[: int(limit)] + + def leg_dict(row): + fields = ( + "id", + "group_id", + "api_key_id", + "router_name", + "direction", + "baseline_model", + "judge_model", + "shadow_percentage", + "max_turns", + "created_at", + "ends_at", + "stopped_at", + "stopped_by", + ) + return {field: getattr(row, field) for field in fields} + + prisma.db.litellm_shadowevaljob.find_many = AsyncMock(side_effect=find_many_legs) + prisma.db.litellm_shadowevaljob.create_many = AsyncMock(return_value=1) + prisma.db.litellm_shadowevaljob.update_many = AsyncMock(return_value=1) prisma.db.litellm_shadowevalattempt.find_first = AsyncMock(return_value=None) + prisma.attempt_rows = [] async def query_raw(sql: str, *params: object): + if "AS attempt_count" in sql: + return prisma.attempt_rows + if "GROUP BY group_id" in sql: + scoped = [row for row in stored if "api_key_id = $2" not in sql or row.api_key_id == params[1]] + keep = set(newest_groups(scoped, params[0])) + return [leg_dict(row) for row in stored if row.group_id in keep] if "FILTER (WHERE outcome != 'error')::int AS judged_count" in sql: return [{"judged_count": 10, "error_count": 2, "judge_spend": 0.031}] + if "SELECT job_id AS grp" in sql: + return by_leg_rows if by_leg_rows is not None else [] return agg_rows if agg_rows is not None else [] prisma.db.query_raw = AsyncMock(side_effect=query_raw) @@ -563,7 +633,7 @@ def _shadow_prisma(active_job=None, agg_rows=None) -> MagicMock: def _start_request(**overrides: object) -> StartShadowEvalRequest: payload = { - "api_key_id": "key-hash", + "api_key_ids": ("key-hash",), "router_name": "my-router", "shadow_percentage": 10.0, "judge_model": "anthropic/claude-sonnet-5", @@ -575,44 +645,55 @@ def _start_request(**overrides: object) -> StartShadowEvalRequest: @pytest.mark.asyncio -async def test_start_shadow_eval_creates_job_and_frees_expired_or_exhausted_ones(monkeypatch: pytest.MonkeyPatch): - """Expiry and turn-budget exhaustion both end sampling on their own; either must - release the key's slot in the active-job index so a new eval can start.""" +async def test_start_shadow_eval_writes_one_leg_per_key_in_one_statement(monkeypatch: pytest.MonkeyPatch): + """N keys become N sibling rows sharing group_id and identical config, written by a + single create_many so a unique-index loser rolls back the whole claim, and expiry or + budget exhaustion frees every requested key's slot first.""" import litellm.proxy.proxy_server as proxy_server prisma = _shadow_prisma() monkeypatch.setattr(proxy_server, "prisma_client", prisma) monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) - response = await start_shadow_eval(_start_request(), ADMIN) + response = await start_shadow_eval(_start_request(api_key_ids=("key-hash", "key-hash-2")), ADMIN) - assert response.status == "running" - assert response.max_turns == 200 - assert response.judged_count is None - sweep_sql, sweep_key = prisma.db.execute_raw.call_args.args + sweep_sql, sweep_keys = prisma.db.execute_raw.call_args.args assert "stopped_at IS NULL" in sweep_sql - assert "ends_at <= NOW()" in sweep_sql + assert "j.ends_at <= (NOW() AT TIME ZONE 'utc')" in sweep_sql + assert "SET stopped_at = (NOW() AT TIME ZONE 'utc')" in sweep_sql assert ">= j.max_turns" in sweep_sql - assert sweep_key == "key-hash" - create_data = prisma.db.litellm_shadowevaljob.create.call_args.kwargs["data"] - assert create_data["api_key_id"] == "key-hash" - assert create_data["created_by"] == "admin" - assert "status" not in create_data + assert "j.api_key_id = ANY($1::text[])" in sweep_sql + assert sweep_keys == ["key-hash", "key-hash-2"] + prisma.db.litellm_shadowevaljob.create_many.assert_awaited_once() + rows = prisma.db.litellm_shadowevaljob.create_many.call_args.kwargs["data"] + assert [row["api_key_id"] for row in rows] == ["key-hash", "key-hash-2"] + assert len({frozenset((k, v) for k, v in row.items() if k != "api_key_id") for row in rows}) == 1 + assert len({row["group_id"] for row in rows}) == 1 + assert all(row["max_turns"] == 200 and row["created_by"] == "admin" for row in rows) + assert all("status" not in row and "id" not in row for row in rows) + assert response.job_id == rows[0]["group_id"] + assert response.status == "running" + assert response.judged_count is None + assert [(key.api_key_id, key.max_turns, key.key_alias) for key in response.keys] == [ + ("key-hash", 200, "prod-alpha"), + ("key-hash-2", 200, "prod-alpha"), + ] @pytest.mark.asyncio @pytest.mark.parametrize( - "caller,request_overrides,active,expected_status", + "caller,request_overrides,claimed,expected_status", [ - (NON_ADMIN, {}, None, 403), - (VIEWER, {}, None, 403), - (ADMIN, {"router_name": "not-a-router"}, None, 400), - (ADMIN, {"judge_model": "not/a real model!"}, None, 400), - (ADMIN, {"judge_model": "my-router"}, None, 400), - (ADMIN, {}, "active", 409), - (ADMIN, {"direction": "reverse", "baseline_model": "my-router"}, None, 400), - (ADMIN, {"direction": "reverse", "baseline_model": "not/a real model!"}, None, 400), - (ADMIN, {"direction": "reverse", "baseline_model": "openai/gpt-4o", "router_name": "not-a-router"}, None, 400), + (NON_ADMIN, {}, (), 403), + (VIEWER, {}, (), 403), + (ADMIN, {"router_name": "not-a-router"}, (), 400), + (ADMIN, {"judge_model": "not/a real model!"}, (), 400), + (ADMIN, {"judge_model": "my-router"}, (), 400), + (ADMIN, {}, ("key-hash",), 409), + (ADMIN, {"api_key_ids": ("key-hash", "key-hash-2")}, ("key-hash-2",), 409), + (ADMIN, {"direction": "reverse", "baseline_model": "my-router"}, (), 400), + (ADMIN, {"direction": "reverse", "baseline_model": "not/a real model!"}, (), 400), + (ADMIN, {"direction": "reverse", "baseline_model": "openai/gpt-4o", "router_name": "not-a-router"}, (), 400), ], ids=[ "non-admin", @@ -621,23 +702,143 @@ async def test_start_shadow_eval_creates_job_and_frees_expired_or_exhausted_ones "unresolvable-judge", "router-as-judge", "already-active", + "one-of-several-keys-already-active", "router-as-baseline", "unresolvable-baseline", "reverse-still-needs-an-auto-router", ], ) async def test_start_shadow_eval_rejections( - monkeypatch: pytest.MonkeyPatch, caller, request_overrides, active, expected_status + monkeypatch: pytest.MonkeyPatch, caller, request_overrides, claimed, expected_status ): import litellm.proxy.proxy_server as proxy_server - prisma = _shadow_prisma(active_job=_job_record() if active else None) + prisma = _shadow_prisma(legs=[_leg_record(id=f"leg-{key}", group_id="job-7", api_key_id=key) for key in claimed]) monkeypatch.setattr(proxy_server, "prisma_client", prisma) monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) with pytest.raises(HTTPException) as exc: await start_shadow_eval(_start_request(**request_overrides), caller) assert exc.value.status_code == expected_status + prisma.db.litellm_shadowevaljob.create_many.assert_not_called() + + +@pytest.mark.asyncio +async def test_start_shadow_eval_names_the_busy_key_and_its_job(monkeypatch: pytest.MonkeyPatch): + """A key busy elsewhere blocks the whole start rather than being silently dropped from + it, and the 409 names which key and which job so the caller can stop or drop it.""" + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma(legs=[_leg_record(id="leg-b", group_id="job-7", api_key_id="key-hash-2")]) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + with pytest.raises(HTTPException) as exc: + await start_shadow_eval(_start_request(api_key_ids=("key-hash", "key-hash-2")), ADMIN) + assert exc.value.status_code == 409 + assert "key-hash-2 (job job-7)" in exc.value.detail + + +@pytest.mark.asyncio +async def test_start_shadow_eval_reuses_a_key_whose_previous_job_already_stopped(monkeypatch: pytest.MonkeyPatch): + """The claim is held by unstopped legs only, matching the partial unique index. A read + that forgets that would strand every key that has ever finished a job.""" + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma(legs=[_leg_record(group_id="job-7", stopped_at=datetime.now(timezone.utc))]) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + job = await start_shadow_eval(_start_request(), ADMIN) + + assert job.status == "running" + prisma.db.litellm_shadowevaljob.create_many.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_start_shadow_eval_reverse_records_its_arms_and_holds_its_own_slot(monkeypatch: pytest.MonkeyPatch): + """The two directions ask opposite questions of the same key, so a forward job holding + the slot must not block a reverse one. The second reverse start still 409s.""" + import litellm.proxy.proxy_server as proxy_server + + legs = [_leg_record(group_id="job-fwd")] + prisma = _shadow_prisma(legs=legs) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + reverse = _start_request(direction="reverse", baseline_model="openai/gpt-4o") + response = await start_shadow_eval(reverse, ADMIN) + + assert (response.direction, response.baseline_model) == ("reverse", "openai/gpt-4o") + rows = prisma.db.litellm_shadowevaljob.create_many.call_args.kwargs["data"] + assert rows[0]["direction"] == "reverse" + assert rows[0]["baseline_model"] == "openai/gpt-4o" + + legs.append(_leg_record(id="leg-2", group_id="job-rev", direction="reverse")) + with pytest.raises(HTTPException) as exc: + await start_shadow_eval(reverse, ADMIN) + assert exc.value.status_code == 409 + + +@pytest.mark.asyncio +async def test_start_shadow_eval_forward_leaves_the_baseline_column_empty(monkeypatch: pytest.MonkeyPatch): + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma() + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + await start_shadow_eval(_start_request(), ADMIN) + + rows = prisma.db.litellm_shadowevaljob.create_many.call_args.kwargs["data"] + assert rows[0]["direction"] == "forward" + assert rows[0]["baseline_model"] is None + + +@pytest.mark.asyncio +async def test_start_shadow_eval_rejects_keys_this_proxy_does_not_know(monkeypatch: pytest.MonkeyPatch): + """A typo'd api_key_id would otherwise create a leg no traffic can ever match. Every + unknown key is named at once, so a caller passing several fixes them in one round.""" + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma(known_keys=("key-hash",)) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + with pytest.raises(HTTPException) as exc: + await start_shadow_eval(_start_request(api_key_ids=("key-hash", "typo-a", "typo-b")), ADMIN) + assert exc.value.status_code == 400 + assert "typo-a, typo-b" in exc.value.detail + assert "key-hash," not in exc.value.detail + prisma.db.litellm_shadowevaljob.create_many.assert_not_called() + + +def test_start_shadow_eval_request_dedupes_and_bounds_the_key_set(): + """A key named twice would collide with itself on the one-active-per-key index, a job + scoping no key samples nothing, and the key-count cap bounds every downstream read.""" + assert _start_request(api_key_ids=("a", "b", "a")).api_key_ids == ("a", "b") + assert len(_start_request(api_key_ids=tuple(f"k{i}" for i in range(100))).api_key_ids) == 100 + with pytest.raises(ValidationError): + _start_request(api_key_ids=()) + with pytest.raises(ValidationError): + _start_request(api_key_ids=tuple(f"k{i}" for i in range(101))) + + +@pytest.mark.asyncio +async def test_start_shadow_eval_concurrent_unique_violation_is_a_409(monkeypatch: pytest.MonkeyPatch): + import litellm.proxy.proxy_server as proxy_server + from prisma.errors import UniqueViolationError + + prisma = _shadow_prisma() + prisma.db.litellm_shadowevaljob.create_many = AsyncMock( + side_effect=UniqueViolationError(MagicMock(message="unique constraint")) + ) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) + + with pytest.raises(HTTPException) as exc: + await start_shadow_eval(_start_request(), ADMIN) + assert exc.value.status_code == 409 @pytest.mark.parametrize( @@ -657,97 +858,25 @@ def test_start_request_pins_baseline_model_to_reverse(overrides): @pytest.mark.asyncio -async def test_start_shadow_eval_reverse_records_its_arms_and_holds_its_own_slot(monkeypatch: pytest.MonkeyPatch): - """The two directions ask opposite questions of the same key, so a forward job holding - the slot must not block a reverse one. The second reverse start still 409s.""" - import litellm.proxy.proxy_server as proxy_server - - prisma = _shadow_prisma() - active = {"forward": _job_record()} - prisma.db.litellm_shadowevaljob.find_first = AsyncMock( - side_effect=lambda where, **_: active.get(str(where.get("direction"))) - ) - prisma.db.litellm_shadowevaljob.create = AsyncMock( - return_value=_job_record(direction="reverse", baseline_model="openai/gpt-4o") - ) - monkeypatch.setattr(proxy_server, "prisma_client", prisma) - monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) - - reverse = _start_request(direction="reverse", baseline_model="openai/gpt-4o") - response = await start_shadow_eval(reverse, ADMIN) - - assert (response.direction, response.baseline_model) == ("reverse", "openai/gpt-4o") - create_data = prisma.db.litellm_shadowevaljob.create.call_args.kwargs["data"] - assert create_data["direction"] == "reverse" - assert create_data["baseline_model"] == "openai/gpt-4o" - - active["reverse"] = _job_record(id="job-2", direction="reverse") - with pytest.raises(HTTPException) as exc: - await start_shadow_eval(reverse, ADMIN) - assert exc.value.status_code == 409 - - -@pytest.mark.asyncio -async def test_start_shadow_eval_forward_leaves_the_baseline_column_empty(monkeypatch: pytest.MonkeyPatch): - import litellm.proxy.proxy_server as proxy_server - - prisma = _shadow_prisma() - monkeypatch.setattr(proxy_server, "prisma_client", prisma) - monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) - - await start_shadow_eval(_start_request(), ADMIN) - - create_data = prisma.db.litellm_shadowevaljob.create.call_args.kwargs["data"] - assert create_data["direction"] == "forward" - assert create_data["baseline_model"] is None - - -@pytest.mark.asyncio -async def test_start_shadow_eval_rejects_a_key_this_proxy_does_not_know(monkeypatch: pytest.MonkeyPatch): - """A typo'd api_key_id would otherwise create a job no traffic can ever match.""" - import litellm.proxy.proxy_server as proxy_server - - prisma = _shadow_prisma() - prisma.db.litellm_verificationtoken.find_unique = AsyncMock(return_value=None) - monkeypatch.setattr(proxy_server, "prisma_client", prisma) - monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) - - with pytest.raises(HTTPException) as exc: - await start_shadow_eval(_start_request(), ADMIN) - assert exc.value.status_code == 400 - assert "not a key on this proxy" in exc.value.detail - - -@pytest.mark.asyncio -async def test_start_shadow_eval_concurrent_unique_violation_is_a_409(monkeypatch: pytest.MonkeyPatch): - import litellm.proxy.proxy_server as proxy_server - from prisma.errors import UniqueViolationError - - prisma = _shadow_prisma() - prisma.db.litellm_shadowevaljob.create = AsyncMock( - side_effect=UniqueViolationError(MagicMock(message="unique constraint")) - ) - monkeypatch.setattr(proxy_server, "prisma_client", prisma) - monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) - - with pytest.raises(HTTPException) as exc: - await start_shadow_eval(_start_request(), ADMIN) - assert exc.value.status_code == 409 - - -@pytest.mark.asyncio -async def test_get_shadow_eval_job_derives_counts_spend_and_stratified_results(monkeypatch: pytest.MonkeyPatch): +async def test_get_shadow_eval_job_pools_counts_and_slices_results_per_key(monkeypatch: pytest.MonkeyPatch): + """One read answers for every leg: totals and stratifications aggregate over the + group's leg ids, and the by-key slice maps each leg id back to its key hash.""" import litellm.proxy.proxy_server as proxy_server tier_rows = [ {"grp": "SIMPLE", "turn_count": 8, "real_wins": 2, "shadow_wins": 4, "ties": 2, "avg_confidence": 0.8}, {"grp": "REASONING", "turn_count": 2, "real_wins": 2, "shadow_wins": 0, "ties": 0, "avg_confidence": 0.9}, ] - prisma = _shadow_prisma(agg_rows=tier_rows) - prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=_job_record()) - prisma.db.litellm_shadowevalattempt.find_first = AsyncMock( - return_value=MagicMock(error="judge call failed: boom") + leg_rows = [ + {"grp": "leg-1", "turn_count": 6, "real_wins": 1, "shadow_wins": 4, "ties": 1, "avg_confidence": 0.7}, + {"grp": "leg-2", "turn_count": 4, "real_wins": 3, "shadow_wins": 0, "ties": 1, "avg_confidence": 0.6}, + ] + prisma = _shadow_prisma( + legs=[_leg_record(), _leg_record(id="leg-2", api_key_id="key-hash-2", max_turns=50)], + agg_rows=tier_rows, + by_leg_rows=leg_rows, ) + prisma.db.litellm_shadowevalattempt.find_first = AsyncMock(return_value=MagicMock(error="judge call failed: boom")) monkeypatch.setattr(proxy_server, "prisma_client", prisma) response = await get_shadow_eval_job("job-1", VIEWER) @@ -762,6 +891,13 @@ async def test_get_shadow_eval_job_derives_counts_spend_and_stratified_results(m assert response.results.by_tier[0].shadow_win_rate_pct == 50.0 assert response.results.overall_shadow_win_rate_pct == 40.0 assert response.results.overall_tie_rate_pct == 20.0 + assert [(s.group, s.turn_count) for s in response.results.by_key] == [("key-hash", 6), ("key-hash-2", 4)] + assert response.results.by_key[0].shadow_win_rate_pct == 66.7 + assert [(key.api_key_id, key.max_turns) for key in response.keys] == [("key-hash", 200), ("key-hash-2", 50)] + totals_args = [call.args for call in prisma.db.query_raw.await_args_list if "judged_count" in call.args[0]] + assert totals_args == [(totals_args[0][0], ["leg-1", "leg-2"])] + error_where = prisma.db.litellm_shadowevalattempt.find_first.call_args.kwargs["where"] + assert error_where == {"job_id": {"in": ["leg-1", "leg-2"]}, "outcome": "error"} @pytest.mark.asyncio @@ -780,79 +916,326 @@ async def test_get_shadow_eval_job_404s_and_gates_on_role(monkeypatch: pytest.Mo @pytest.mark.asyncio -async def test_list_shadow_eval_jobs_returns_derived_status_without_aggregates(monkeypatch: pytest.MonkeyPatch): +async def test_list_shadow_eval_jobs_collapses_legs_into_jobs_newest_first(monkeypatch: pytest.MonkeyPatch): + """A job over two keys is one list entry with both keys, not two entries, and a job + whose keys all stopped reads stopped while a half-stopped one still runs.""" import litellm.proxy.proxy_server as proxy_server - prisma = _shadow_prisma() - prisma.db.litellm_shadowevaljob.find_many = AsyncMock( - return_value=[ - _job_record(), - _job_record(id="job-2", ends_at=datetime.now(timezone.utc) - timedelta(days=1)), - _job_record(id="job-3", stopped_at=datetime.now(timezone.utc)), + stamp = datetime.now(timezone.utc) + prisma = _shadow_prisma( + legs=[ + _leg_record(created_at=datetime(2026, 8, 13, tzinfo=timezone.utc)), + _leg_record( + id="leg-2", + api_key_id="key-hash-2", + stopped_at=stamp, + created_at=datetime(2026, 8, 13, tzinfo=timezone.utc), + ), + _leg_record( + id="leg-3", + group_id="job-2", + stopped_at=stamp, + created_at=datetime(2026, 8, 12, tzinfo=timezone.utc), + ), + _leg_record( + id="leg-4", + group_id="job-3", + ends_at=datetime.now(timezone.utc) - timedelta(days=1), + created_at=datetime(2026, 8, 11, tzinfo=timezone.utc), + ), ] ) monkeypatch.setattr(proxy_server, "prisma_client", prisma) jobs = await list_shadow_eval_jobs(VIEWER, api_key_id=None, limit=50) - assert [job.status for job in jobs] == ["running", "completed", "stopped"] - swept = ShadowEvalJobResponse.model_validate( - _job_record( - id="job-4", - ends_at=datetime.now(timezone.utc) - timedelta(days=1), - stopped_at=datetime.now(timezone.utc), - ), - from_attributes=True, - ) - assert swept.status == "completed" + assert [(job.job_id, job.status) for job in jobs] == [ + ("job-1", "running"), + ("job-2", "stopped"), + ("job-3", "completed"), + ] + assert [key.api_key_id for key in jobs[0].keys] == ["key-hash", "key-hash-2"] assert all(job.judged_count is None and job.results is None for job in jobs) - assert prisma.db.query_raw.await_count == 0 + legs_sql, legs_limit = prisma.db.query_raw.await_args_list[0].args + assert "GROUP BY group_id ORDER BY MAX(created_at) DESC LIMIT $1::int" in legs_sql + assert legs_limit == 50 + counts_sql, _ = prisma.db.query_raw.await_args_list[1].args + assert "AS attempt_count" in counts_sql + assert "j.stopped_at IS NULL OR a.created_at <= j.stopped_at" in counts_sql + assert prisma.db.query_raw.await_count == 2 + prisma.db.litellm_shadowevaljob.find_many.assert_not_called() @pytest.mark.asyncio -async def test_shadow_eval_responses_name_the_shadowed_key(monkeypatch: pytest.MonkeyPatch): +async def test_list_shadow_eval_jobs_filters_to_jobs_containing_the_key(monkeypatch: pytest.MonkeyPatch): + """The filter matches a key anywhere in a job's key set and still returns the whole + job, sibling keys included.""" import litellm.proxy.proxy_server as proxy_server - prisma = _shadow_prisma() - prisma.db.litellm_shadowevaljob.find_many = AsyncMock( - return_value=[_job_record(), _job_record(id="job-2", api_key_id="deleted-key-hash")] + prisma = _shadow_prisma( + legs=[ + _leg_record(), + _leg_record(id="leg-2", api_key_id="key-hash-2"), + _leg_record(id="leg-3", group_id="job-2", api_key_id="key-hash-2"), + _leg_record(id="leg-4", group_id="job-3"), + ] + ) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + jobs = await list_shadow_eval_jobs(VIEWER, api_key_id="key-hash-2", limit=50) + + assert [job.job_id for job in jobs] == ["job-1", "job-2"] + assert [key.api_key_id for key in jobs[0].keys] == ["key-hash", "key-hash-2"] + + +@pytest.mark.parametrize( + ("stopped_flags", "days_left", "expected"), + [ + ((False, False), 7, "running"), + ((True, False), 7, "running"), + ((True, True), 7, "stopped"), + ((True, True), -1, "completed"), + ((False, False), -1, "completed"), + ], +) +@pytest.mark.asyncio +async def test_job_status_runs_until_every_key_stops_and_completed_outranks_stopped( + monkeypatch: pytest.MonkeyPatch, stopped_flags: tuple[bool, ...], days_left: int, expected: str +): + import litellm.proxy.proxy_server as proxy_server + + stamp = datetime.now(timezone.utc) + prisma = _shadow_prisma( + legs=[ + _leg_record( + id=f"leg-{index}", + api_key_id=f"key-{index}", + stopped_at=stamp if stopped else None, + ends_at=datetime.now(timezone.utc) + timedelta(days=days_left), + ) + for index, stopped in enumerate(stopped_flags) + ] + ) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + jobs = await list_shadow_eval_jobs(VIEWER, api_key_id=None, limit=50) + + assert [job.status for job in jobs] == [expected] + + +@pytest.mark.asyncio +async def test_list_reads_completed_once_every_key_spends_its_budget(monkeypatch: pytest.MonkeyPatch): + """A job whose keys all exhausted their turn budgets stopped sampling on its own, so + it must read completed on the very next list, before any sweep stamps its legs; one + key under budget keeps the whole job running. An operator starting an unrelated eval + must never look like it terminated a finished one.""" + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma( + legs=[ + _leg_record(max_turns=5), + _leg_record(id="leg-2", api_key_id="key-hash-2", max_turns=5), + _leg_record(id="leg-3", group_id="job-2", api_key_id="key-hash", max_turns=5), + _leg_record(id="leg-4", group_id="job-2", api_key_id="key-hash-2", max_turns=5), + ] + ) + prisma.attempt_rows = [ + {"job_id": "leg-1", "attempt_count": 5}, + {"job_id": "leg-2", "attempt_count": 6}, + {"job_id": "leg-3", "attempt_count": 5}, + {"job_id": "leg-4", "attempt_count": 3}, + ] + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + jobs = await list_shadow_eval_jobs(VIEWER, api_key_id=None, limit=50) + + by_id = {job.job_id: job for job in jobs} + assert by_id["job-1"].status == "completed" + assert all(key.stopped_at is None for key in by_id["job-1"].keys) + assert by_id["job-2"].status == "running" + assert {key.api_key_id: key.attempt_count for key in by_id["job-2"].keys} == {"key-hash": 5, "key-hash-2": 3} + + +@pytest.mark.asyncio +async def test_recorded_operator_stop_outranks_budget_arithmetic(monkeypatch: pytest.MonkeyPatch): + """A detached attempt can land around the stop and push the raw count past the + budget; the recorded stopped_by must keep the job reading stopped regardless.""" + import litellm.proxy.proxy_server as proxy_server + + stamp = datetime.now(timezone.utc) + prisma = _shadow_prisma(legs=[_leg_record(max_turns=5, stopped_at=stamp, stopped_by="admin")]) + prisma.attempt_rows = [{"job_id": "leg-1", "attempt_count": 6}] + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + jobs = await list_shadow_eval_jobs(VIEWER, api_key_id=None, limit=50) + assert jobs[0].status == "stopped" + assert jobs[0].stopped_by == "admin" + + detail = await get_shadow_eval_job("job-1", VIEWER) + assert detail.status == "stopped" + + +@pytest.mark.asyncio +async def test_backfilled_legacy_stop_never_reads_as_completion(monkeypatch: pytest.MonkeyPatch): + """Jobs stopped before stopped_by existed are backfilled with 'unknown' by the + migration, so even one whose stray attempts crossed the budget stays stopped.""" + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma( + legs=[_leg_record(max_turns=5, stopped_at=datetime.now(timezone.utc), stopped_by="unknown")] + ) + prisma.attempt_rows = [{"job_id": "leg-1", "attempt_count": 6}] + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + jobs = await list_shadow_eval_jobs(VIEWER, api_key_id=None, limit=50) + assert jobs[0].status == "stopped" + + +def test_stopped_by_migration_backfills_every_job_that_displayed_stopped(): + """The migration must close the pre-column population: without the backfill, a + legacy stop whose stray attempts crossed the budget would read completed.""" + import litellm_proxy_extras + + sql = ( + Path(litellm_proxy_extras.__file__).parent + / "migrations" + / "20260818224500_add_shadow_eval_stopped_by" + / "migration.sql" + ).read_text() + assert 'ADD COLUMN "stopped_by" TEXT' in sql + assert "SET stopped_by = 'unknown'" in sql + assert "WHERE stopped_at IS NOT NULL AND ends_at > (NOW() AT TIME ZONE 'utc')" in sql + + +@pytest.mark.asyncio +async def test_stop_rejects_a_job_that_already_spent_its_budget(monkeypatch: pytest.MonkeyPatch): + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma(legs=[_leg_record(max_turns=3)]) + prisma.attempt_rows = [{"job_id": "leg-1", "attempt_count": 3}] + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + with pytest.raises(HTTPException) as exhausted: + await stop_shadow_eval_job("job-1", ADMIN) + assert exhausted.value.status_code == 400 + assert "completed" in exhausted.value.detail + prisma.db.litellm_shadowevaljob.update_many.assert_not_called() + + +@pytest.mark.asyncio +async def test_shadow_eval_responses_name_every_shadowed_key(monkeypatch: pytest.MonkeyPatch): + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma( + legs=[_leg_record(), _leg_record(id="leg-2", api_key_id="deleted-key-hash")], + known_keys=("key-hash", "key-hash-2"), ) monkeypatch.setattr(proxy_server, "llm_router", _shadow_router()) monkeypatch.setattr(proxy_server, "prisma_client", prisma) - started = await start_shadow_eval(_start_request(), ADMIN) - assert (started.key_alias, started.key_name) == ("prod-alpha", "sk-...lpha") - jobs = await list_shadow_eval_jobs(VIEWER, api_key_id=None, limit=50) - assert [(job.key_alias, job.key_name) for job in jobs] == [("prod-alpha", "sk-...lpha"), (None, None)] + assert [(key.key_alias, key.key_name) for key in jobs[0].keys] == [ + (None, None), + ("prod-alpha", "sk-...lpha"), + ] batched_where = prisma.db.litellm_verificationtoken.find_many.call_args.kwargs["where"] assert batched_where == {"token": {"in": ["deleted-key-hash", "key-hash"]}} - prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=_job_record()) detail = await get_shadow_eval_job("job-1", VIEWER) - assert detail.key_alias == "prod-alpha" + assert [key.key_alias for key in detail.keys] == [None, "prod-alpha"] @pytest.mark.asyncio -async def test_stop_shadow_eval_sets_stopped_at_and_rejects_non_running(monkeypatch: pytest.MonkeyPatch): +async def test_stop_shadow_eval_stops_every_unstopped_leg_and_rejects_non_running( + monkeypatch: pytest.MonkeyPatch, +): + """One stop ends sampling for the whole job, while a leg that already stopped on its + own budget keeps the stopped_at it earned.""" import litellm.proxy.proxy_server as proxy_server - prisma = _shadow_prisma() - prisma.db.litellm_shadowevaljob.find_unique = AsyncMock(return_value=_job_record()) + earned = datetime.now(timezone.utc) - timedelta(hours=1) + prisma = _shadow_prisma(legs=[_leg_record(), _leg_record(id="leg-2", api_key_id="key-hash-2", stopped_at=earned)]) monkeypatch.setattr(proxy_server, "prisma_client", prisma) stopped = await stop_shadow_eval_job("job-1", ADMIN) - assert stopped.status == "stopped" - update = prisma.db.litellm_shadowevaljob.update.call_args.kwargs - assert set(update["data"]) == {"stopped_at"} - prisma.db.litellm_shadowevaljob.find_unique = AsyncMock( - return_value=_job_record(ends_at=datetime.now(timezone.utc) - timedelta(days=1)) - ) + assert stopped.status == "stopped" + assert stopped.stopped_by == "admin" + stop_sql, stop_group, stop_operator, stop_stamp = prisma.db.execute_raw.call_args.args + assert "SET stopped_by = $2, stopped_at = COALESCE(stopped_at, $3::timestamp)" in stop_sql + assert "WHERE group_id = $1 AND stopped_by IS NULL" in stop_sql + assert "ends_at > (NOW() AT TIME ZONE 'utc')" in stop_sql + assert ") < k.max_turns" in stop_sql + assert (stop_group, stop_operator) == ("job-1", "admin") + assert datetime.fromisoformat(stop_stamp).tzinfo is None + assert prisma.db.execute_raw.await_count == 1 + prisma.db.litellm_shadowevaljob.update_many.assert_not_called() + by_key = {key.api_key_id: key.stopped_at for key in stopped.keys} + assert by_key["key-hash-2"] == earned + assert by_key["key-hash"] is not None and by_key["key-hash"] != earned + + done_leg = _leg_record(ends_at=datetime.now(timezone.utc) - timedelta(days=1)) + prisma_done = _shadow_prisma(legs=[done_leg]) + monkeypatch.setattr(proxy_server, "prisma_client", prisma_done) with pytest.raises(HTTPException) as exc: await stop_shadow_eval_job("job-1", ADMIN) assert exc.value.status_code == 400 + assert "already completed" in exc.value.detail + assert done_leg.stopped_by is None with pytest.raises(HTTPException) as forbidden: await stop_shadow_eval_job("job-1", VIEWER) assert forbidden.value.status_code == 403 + + +def test_every_shadow_eval_sql_constant_speaks_naive_utc(): + """The tables store naive UTC wall time (prisma's convention), so SQL-side time must be + NOW() AT TIME ZONE 'utc' and python-side params must cast ::timestamp; a bare NOW() or a + timestamptz cast writes session-local wall time into the naive column and skews every + comparison against prisma-written stamps.""" + import litellm.proxy.management_endpoints.auto_router_endpoints as module + + sql_constants = {name: value for name, value in vars(module).items() if name.endswith("_SQL")} + assert sql_constants + for name, sql in sql_constants.items(): + assert "::timestamptz" not in sql, name + for occurrence in sql.split("NOW()")[1:]: + assert occurrence.startswith(" AT TIME ZONE 'utc'"), name + + +@pytest.mark.asyncio +async def test_a_stop_racing_the_last_budgeted_attempt_reports_completed_not_stopped( + monkeypatch: pytest.MonkeyPatch, +): + """The statement claims the job only while a leg still samples, so a stop landing in + the same instant the budget spends records nothing and the job keeps reading + completed; stamping it would misreport a self-ended job as operator-stopped forever.""" + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma(legs=[_leg_record(max_turns=2)]) + prisma.attempt_rows = [{"job_id": "leg-1", "attempt_count": 2}] + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + with pytest.raises(HTTPException) as exc: + await stop_shadow_eval_job("job-1", ADMIN) + assert exc.value.status_code == 400 + assert "already completed" in exc.value.detail + assert prisma.db.litellm_shadowevaljob.find_many.await_args.kwargs["where"] == {"group_id": "job-1"} + + +@pytest.mark.asyncio +async def test_two_racing_stops_produce_exactly_one_winner(monkeypatch: pytest.MonkeyPatch): + """The statement's stopped_by IS NULL predicate lets only one racer claim rows; the + loser reads the stamped state and gets the same answer a late caller gets.""" + import litellm.proxy.proxy_server as proxy_server + + prisma = _shadow_prisma(legs=[_leg_record()]) + monkeypatch.setattr(proxy_server, "prisma_client", prisma) + + first = await stop_shadow_eval_job("job-1", ADMIN) + assert first.status == "stopped" + + with pytest.raises(HTTPException) as exc: + await stop_shadow_eval_job("job-1", ADMIN) + assert exc.value.status_code == 400 + assert "already stopped" in exc.value.detail 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 7bb550f729e..b307e3d0f2a 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 @@ -77,7 +77,15 @@ const job = (overrides: Partial = {}): ShadowEvalJob => ({ baseline_model: null, judge_model: "anthropic/claude-sonnet-5", shadow_percentage: 10, - max_turns: 200, + keys: [ + { + api_key_id: "hashed-key-abc", + max_turns: 200, + stopped_at: null, + key_alias: "prod-alpha", + key_name: "sk-...alpha", + }, + ], judged_count: 42, error_count: 1, judge_spend: 3.21, @@ -110,19 +118,28 @@ const job = (overrides: Partial = {}): ShadowEvalJob => ({ avg_judge_confidence: 0.8, }, ], + by_key: [], overall_shadow_win_rate_pct: 48.0, overall_tie_rate_pct: 22.0, }, created_at: "2026-08-07T00:00:00Z", ends_at: "2026-09-07T00:00:00Z", - stopped_at: null, - api_key_id: "hashed-key-abc", - key_alias: "prod-alpha", - key_name: "sk-...alpha", last_error: null, ...overrides, }); +const keyEntry = ( + api_key_id: string, + overrides: Partial = {}, +): ShadowEvalJob["keys"][number] => ({ + api_key_id, + max_turns: 200, + stopped_at: null, + key_alias: null, + key_name: null, + ...overrides, +}); + const mockHooks = ({ jobs = [], detailsById = {}, @@ -199,8 +216,8 @@ describe("ShadowEvalSection", () => { it("gives every active job its own card with a stop button, with the form still offered", () => { mockHooks({ jobs: [ - job({ job_id: "job-a", status: "running", api_key_id: "key-a" }), - job({ job_id: "job-b", status: "running", api_key_id: "key-b" }), + job({ job_id: "job-a", status: "running", keys: [keyEntry("key-a")] }), + job({ job_id: "job-b", status: "running", keys: [keyEntry("key-b")] }), ], }); render(); @@ -342,7 +359,7 @@ describe("ShadowEvalSection", () => { expect(container).toBeEmptyDOMElement(); }); - it("keeps the start button disabled until key, router, and judge model are picked, then submits them", async () => { + it("keeps the start button disabled until key, router, and judge model are picked, then submits the key as a list", async () => { const user = userEvent.setup(); const { start } = mockHooks({}); render(); @@ -361,7 +378,7 @@ describe("ShadowEvalSection", () => { await user.click(screen.getByText("Start shadow eval")); const expectedBody = { - api_key_id: "hash-alpha", + api_key_ids: ["hash-alpha"], router_name: "gpt-auto", direction: "forward", shadow_percentage: 10, @@ -396,7 +413,7 @@ describe("ShadowEvalSection", () => { await user.click(screen.getByText("Start shadow eval")); const expectedBody = { - api_key_id: "hash-alpha", + api_key_ids: ["hash-alpha"], router_name: "gpt-auto", direction: "reverse", baseline_model: "prod-claude", @@ -429,9 +446,9 @@ describe("ShadowEvalSection", () => { }); it("labels the shadowed key by alias, then masked name, then truncated hash", () => { - expect(shadowedKeyLabel(job())).toBe("prod-alpha"); - expect(shadowedKeyLabel(job({ key_alias: null }))).toBe("sk-...alpha"); - expect(shadowedKeyLabel(job({ key_alias: null, key_name: null }))).toBe("hashed-key…"); + expect(shadowedKeyLabel(job().keys[0])).toBe("prod-alpha"); + expect(shadowedKeyLabel(keyEntry("hashed-key-abc", { key_name: "sk-...alpha" }))).toBe("sk-...alpha"); + expect(shadowedKeyLabel(keyEntry("hashed-key-abc"))).toBe("hashed-key…"); }); it("keeps an older job's verdicts reachable through the previous evaluations list", async () => { 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 005636615b1..6d240a84c98 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 @@ -24,6 +24,7 @@ import { useStartShadowEval, useStopShadowEval, type ShadowEvalJob, + type ShadowEvalJobKey, type ShadowEvalSlice, } from "./useShadowEval"; @@ -50,19 +51,24 @@ const routerMatchedOrBeatPct = ( ? 100 - results.overall_shadow_win_rate_pct : results.overall_shadow_win_rate_pct + results.overall_tie_rate_pct; -export const shadowedKeyLabel = (job: ShadowEvalJob): string => - job.key_alias || job.key_name || `${job.api_key_id.slice(0, 10)}…`; +export const shadowedKeyLabel = (key: ShadowEvalJobKey): string => + key.key_alias || key.key_name || `${key.api_key_id.slice(0, 10)}…`; + +const shadowedKeysLabel = (job: ShadowEvalJob): string => + job.keys.length === 1 ? shadowedKeyLabel(job.keys[0]) : `${job.keys.length} keys`; + +const totalBudget = (job: ShadowEvalJob): number => job.keys.reduce((sum, key) => sum + key.max_turns, 0); const jobHeadline = (job: ShadowEvalJob): React.ReactNode => job.direction === "reverse" ? ( <> Comparing {job.router_name} to{" "} {job.baseline_model} on {job.shadow_percentage}% of{" "} - {shadowedKeyLabel(job)} traffic + {shadowedKeysLabel(job)} traffic ) : ( <> - Shadowing {job.shadow_percentage}% of {shadowedKeyLabel(job)} traffic + Shadowing {job.shadow_percentage}% of {shadowedKeysLabel(job)} traffic via {job.router_name} ); @@ -178,7 +184,8 @@ const emptyResultsText = (job: ShadowEvalJob, resultsError: boolean): string => const ResultsBody: React.FC<{ job: ShadowEvalJob; resultsError?: boolean }> = ({ job, resultsError = false }) => { const results = job.results; - if (!results || (results.by_tier.length === 0 && results.by_current_model.length === 0)) { + const stratifications = results ? [results.by_tier, results.by_current_model, results.by_key] : []; + if (!results || stratifications.every((slices) => slices.length === 0)) { return

{emptyResultsText(job, resultsError)}

; } return ( @@ -224,7 +231,7 @@ const JobResults: React.FC<{

{jobHeadline(job)}

- {(job.judged_count ?? 0).toLocaleString()} of {job.max_turns.toLocaleString()} turns judged ·{" "} + {(job.judged_count ?? 0).toLocaleString()} of {totalBudget(job).toLocaleString()} turns judged ·{" "} {(job.error_count ?? 0).toLocaleString()} errored · {usd(job.judge_spend ?? 0)} judge spend {active && remaining ? ` · ${remaining}` : ""}

@@ -386,14 +393,13 @@ const StartForm: React.FC = () => { const percentageValid = parsedPct >= 0.1 && parsedPct <= 100; const parsedMaxTurns = Number.parseInt(maxTurns, 10); const maxTurnsValid = parsedMaxTurns >= 1 && parsedMaxTurns <= 2000; - const filled = - [apiKeyId, routerName, judgeModel].every((field) => field !== "") && - (direction === "forward" || baselineModel !== ""); + const baselinePicked = direction === "forward" || baselineModel !== ""; + const filled = [apiKeyId, routerName, judgeModel].every((field) => field !== "") && baselinePicked; const boundsValid = percentageValid && maxTurnsValid; const valid = Boolean(accessToken) && filled && boundsValid; const handleStart = () => { const startBody = { - api_key_id: apiKeyId, + api_key_ids: [apiKeyId], router_name: routerName, direction, ...(direction === "reverse" ? { baseline_model: baselineModel } : {}), diff --git a/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/useShadowEval.ts b/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/useShadowEval.ts index 7645fcc3346..eef98320e67 100644 --- a/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/useShadowEval.ts +++ b/ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/useShadowEval.ts @@ -7,6 +7,7 @@ import { $api, fetchClient } from "@/lib/http/api"; import type { components } from "@/lib/http/schema"; export type ShadowEvalJob = components["schemas"]["ShadowEvalJobResponse"]; +export type ShadowEvalJobKey = components["schemas"]["ShadowEvalJobKeyResponse"]; export type ShadowEvalSlice = components["schemas"]["ShadowEvalSlice"]; export type StartShadowEvalRequest = components["schemas"]["StartShadowEvalRequest"]; diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 5794c60e97b..4c63586d4d6 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -816,7 +816,8 @@ export interface paths { }; /** * List Shadow Eval Jobs - * @description List shadow eval jobs, newest first. Counts and results ride the detail endpoint only. + * @description List shadow eval jobs, newest first, each key with its attempt count so status is + * accurate. Judged counts, spend, and results ride the detail endpoint only. */ get: operations["list_shadow_eval_jobs_auto_router_shadow_eval_get"]; put?: never; @@ -838,20 +839,21 @@ export interface paths { put?: never; /** * Start Shadow Eval - * @description Start a shadow eval: duplicate a sampled slice of a key's live traffic against a second - * arm, judge the two responses blind, and stratify win rates by tier and by the model that - * served the real arm. + * @description Start a shadow eval: duplicate a sampled slice of one or more keys' live traffic against + * a second arm, judge the two responses blind, and stratify win rates by tier, by the model + * that served the real arm, and by key. * - * A forward job answers whether the key should adopt router_name: it samples the requests + * A forward job answers whether the keys should adopt router_name: it samples the requests * the router did not serve and duplicates them through it. A reverse job answers whether a * key already on the router still gains from it: it samples the requests the router did * serve and duplicates them against baseline_model. A key can hold one active job per * direction, so both questions can run at once. * - * Shadow responses are never served to users. The job samples until it has judged - * max_turns turns, reaches the end of its window, or is stopped; sampling changes - * propagate to pods within about 10 seconds. Shadow and judge calls bill to the - * shadowed key but are excluded from request counts and auto-router adoption metrics. + * Shadow responses are never served to users. Each key samples until it has judged + * max_turns turns of its own traffic, the job's window ends, or the job is stopped, so one + * key running out of budget does not end sampling for the others; sampling changes + * propagate to pods within about 10 seconds. Shadow and judge calls bill to the shadowed + * key but are excluded from request counts and auto-router adoption metrics. */ post: operations["start_shadow_eval_auto_router_shadow_eval_start_post"]; delete?: never; @@ -891,7 +893,12 @@ export interface paths { put?: never; /** * Stop Shadow Eval Job - * @description Stop an active shadow eval job. Attempts are kept; sampling halts within ~10s. + * @description Stop an active shadow eval job, every key it scopes at once. Attempts are kept; + * sampling halts within ~10s. Keys that already stopped on their own budget keep the + * stopped_at they earned. The statement is the whole state machine: it claims the job + * only while a leg still samples inside the window with no stop recorded, so a racing + * operator, a same-instant budget spend, and a repeat stop all read the same 400 with + * the status the job actually holds. */ post: operations["stop_shadow_eval_job_auto_router_shadow_eval__job_id__stop_post"]; delete?: never; @@ -33201,18 +33208,49 @@ export interface components { timeout?: number | null; }; /** - * ShadowEvalJobResponse - * @description A shadow-eval job. Validates directly from the prisma record (job_id reads the - * row's id); status is derived from stopped_at and ends_at, never stored, so no writer - * anywhere can produce an inconsistent one. Aggregate fields are populated by the - * detail endpoint only and stay None on list responses. + * ShadowEvalJobKeyResponse + * @description One key a job shadows, with its own budget and stop state. */ - ShadowEvalJobResponse: { + ShadowEvalJobKeyResponse: { /** * Api Key Id - * @description The hashed virtual key whose traffic this job evaluates, and only that key's + * @description The hashed virtual key whose traffic this entry scopes */ api_key_id: string; + /** + * Attempt Count + * @description This key's sampled attempts so far, judged and errored alike, the same count the sampler budgets against max_turns; populated on list and detail responses. Frozen at stopped_at once the key is stamped, so in-flight attempts landing after a stop never reclassify it + */ + attempt_count?: number | null; + /** + * Key Alias + * @description Alias of the shadowed key, resolved from the key row at read time; None when unset or deleted + */ + key_alias?: string | null; + /** + * Key Name + * @description Masked display name (sk-...) of the shadowed key, resolved at read time like key_alias + */ + key_name?: string | null; + /** + * Max Turns + * @description This key's own sample budget, independent of its siblings' + */ + max_turns: number; + /** + * Stopped At + * @description When this key's slot was stamped free, whether its own budget ran out, the window closed, or an operator stopped the job; status is derived, so a spent budget reads completed even while this is still unset + */ + stopped_at?: string | null; + }; + /** + * ShadowEvalJobResponse + * @description A shadow-eval job over one or more keys, each with its own budget and stop state; + * status is derived from stopped_by, the keys' stop and budget state, and ends_at, + * never stored, so no writer anywhere can produce an inconsistent one. Aggregate + * fields are populated by the detail endpoint only and stay None on list responses. + */ + ShadowEvalJobResponse: { /** Baseline Model */ baseline_model?: string | null; /** @@ -33251,22 +33289,15 @@ export interface components { */ judged_count?: number | null; /** - * Key Alias - * @description Alias of the shadowed key, resolved from the key row at read time; None when unset or deleted + * Keys + * @description The keys whose traffic this job evaluates, and only those keys', each with its own budget */ - key_alias?: string | null; - /** - * Key Name - * @description Masked display name (sk-...) of the shadowed key, resolved at read time like key_alias - */ - key_name?: string | null; + keys: components["schemas"]["ShadowEvalJobKeyResponse"][]; /** * Last Error * @description Most recent attempt error; detail endpoint only */ last_error?: string | null; - /** Max Turns */ - max_turns: number; /** @description Stratified verdicts; detail endpoint only */ results?: components["schemas"]["ShadowEvalResult"] | null; /** Router Name */ @@ -33275,13 +33306,19 @@ export interface components { shadow_percentage: number; /** * Status - * @description A job whose window has passed reads completed even if a later sweep stamped - * stopped_at; stopped means sampling ended before the window did. + * @description Three recorded facts, no history-guessing: a stop is stopped_by (the migration + * backfills it for every job that displayed stopped when the column arrived, so the + * pre-column population is closed), completion is the window passing or every key + * spending its budget, and anything else is running. The all-keys-stamped fallback + * covers only stops written by pre-column pods during a rolling deploy. * @enum {string} */ readonly status: "running" | "completed" | "stopped"; - /** Stopped At */ - stopped_at?: string | null; + /** + * Stopped By + * @description The operator who stopped the job early, recorded by the stop endpoint; 'unknown' backfilled by migration for jobs that displayed stopped when the column arrived; None when the job ended on its own. Its presence is what makes a job read stopped rather than completed + */ + stopped_by?: string | null; }; /** * ShadowEvalResult @@ -33290,9 +33327,14 @@ export interface components { ShadowEvalResult: { /** * By Current Model - * @description Sliced by the model that served the real arm: the key's incumbent models in forward mode, and in reverse the models the router itself picked + * @description Sliced by the model that served the real arm: the keys' incumbent models in forward mode, and in reverse the models the router itself picked */ by_current_model: components["schemas"]["ShadowEvalSlice"][]; + /** + * By Key + * @description One slice per scoped key that has judged verdicts, grouped on the raw key hash. Keys the job scopes but has not judged a turn for yet are absent rather than reported as zero + */ + by_key: components["schemas"]["ShadowEvalSlice"][]; /** By Tier */ by_tier: components["schemas"]["ShadowEvalSlice"][]; /** Overall Shadow Win Rate Pct */ @@ -33500,14 +33542,14 @@ export interface components { }; /** * StartShadowEvalRequest - * @description Start duplicating a key's traffic for blind comparison against an auto-router. + * @description Start duplicating one or more keys' traffic for blind comparison against an auto-router. */ StartShadowEvalRequest: { /** - * Api Key Id - * @description The hashed virtual key whose traffic will be shadowed. Shadow evaluation runs ONLY on this key's traffic; requests made with any other key are not sampled. + * Api Key Ids + * @description The hashed virtual keys whose traffic will be shadowed. Shadow evaluation runs ONLY on these keys' traffic; requests made with any other key are not sampled. Each key carries its own max_turns budget, so one key exhausting its budget leaves the others sampling. At most 100 keys per job, which also bounds every read the job's endpoints make. */ - api_key_id: string; + api_key_ids: string[]; /** * Baseline Model * @description Required when direction is reverse and rejected otherwise: the fixed model the router's own responses are judged against. Must be a plain model rather than another auto-router @@ -33534,7 +33576,7 @@ export interface components { judge_model: string; /** * Max Turns - * @description Sample budget: the job judges at most this many turns, then completes. This is also the spend bound; expected judge cost is roughly max_turns times one judge call + * @description Per-key sample budget: the job judges at most this many turns of EACH scoped key's traffic, so a job over N keys judges at most N times max_turns turns. This is also the spend bound; expected judge cost is roughly that turn ceiling times one judge call * @default 200 */ max_turns: number; @@ -37931,7 +37973,7 @@ export interface operations { list_shadow_eval_jobs_auto_router_shadow_eval_get: { parameters: { query?: { - /** @description Filter to jobs shadowing this key */ + /** @description Filter to jobs that shadow this key, alone or alongside others */ api_key_id?: string | null; /** @description Newest jobs to return */ limit?: number;