diff --git a/.github/workflows/cd.yml b/.github/workflows/cd.yml index 756e6d6..3895472 100644 --- a/.github/workflows/cd.yml +++ b/.github/workflows/cd.yml @@ -60,18 +60,10 @@ jobs: - run: python -m pip install -e ".[dev]" - name: Verify the serving configuration matches the catalog run: python -m llm_router.serving > /tmp/ray-serve.yaml && diff -u config/ray-serve.yaml /tmp/ray-serve.yaml - - name: Render the canary and rollback plan - run: | - python - <<'PY' > canary-plan.json - import json - from llm_router.registry import load_registry - from llm_router.serving import canary_config - registry = load_registry("config/registry.yaml") - production = [item for item in registry.deployments if item.stage.value == "production"] - plan = canary_config(registry, production[0].id) - plan["release_ref"] = "${{ env.RELEASE_REF }}" - print(json.dumps(plan, indent=2)) - PY + - name: Render the canary and rollback plans + # One plan per track (model, adapter, policy), each naming what it + # rolls back to and the criteria that trigger it. + run: python -m llm_router.canary > canary-plan.json - name: Validate the Kubernetes manifests run: | curl -sSLo kubeconform.tar.gz https://github.com/yannh/kubeconform/releases/download/v0.7.0/kubeconform-linux-amd64.tar.gz diff --git a/README.md b/README.md index e79ccc9..e0ac7dd 100644 --- a/README.md +++ b/README.md @@ -187,7 +187,7 @@ in-process and correct for a single replica only. Install the client with the ex python -m pip install -e ".[redis]" ``` -CD renders the canary plan (with its rollback target and triggers), verifies +CD renders the canary plans (one per track, each with its rollback target), verifies `config/ray-serve.yaml` against the catalog, and validates the manifests with kubeconform. Applying to a cluster stays disabled until a deployment destination is configured. @@ -211,6 +211,7 @@ be served. A request can never introduce a model path, revision, or adapter. | `GET /v1/registry/adapters` | Promoted LoRA and QLoRA adapters. | | `GET /v1/registry/deployments` | Deployment revisions and rollback targets. | | `GET /v1/registry/variants` | Optimization variants with their measured deltas. | +| `GET /v1/registry/canaries` | Canary plans by track, with live adapter state. | Send `routing.domain` to request a domain adapter; the router applies the promoted adapter with the largest measured quality gain for that base revision and task, or none at all. @@ -238,6 +239,53 @@ warm replica never pay it on the request path. The high-capability tier scales t first request after an idle period waits for a full model load: budget for it, or raise its `min_replicas`. +## Canaries and rollback + +Models, adapters, and router policies are canaried on separate tracks, each with its own plan +naming exactly what it rolls back to: + +```bash +python -m llm_router.canary +``` + +| Track | Canary | Rolls back to | +|---|---|---| +| `model` | A deployment revision | The revision in `previous_revision_id`. | +| `adapter` | A `staging` adapter | The `production` adapter for the same base revision and domain, or the base model alone. | +| `policy` | The policy `version` | `previous_version`. | + +One rule decides every track. Failed readiness rolls back at once. Once 50 requests have been +observed, the canary rolls back if its error rate exceeds 1%, its p95 latency exceeds the +strictest tier objective among the models it serves, or its observed quality falls more than +0.05 below their benchmarked quality. Error rate and latency only count against the canary +when the stable baseline does not share the problem, so an engine outage that degrades both is +not blamed on the change. A canary that stays clean for 500 requests is reported ready. + +**The adapter track runs inside the gateway.** A staged adapter is offered +`canary_traffic_percent` of eligible requests, bucketed by tenant and prompt so a retry cannot +flip between adapters. It is suspended automatically the moment it fails, and the production +adapter serves everything again. A canary's responses are never cached, so nothing it produced +outlives a rollback. Before this, a staged adapter with the larger measured gain took all of the +traffic. + +`GET /v1/registry/canaries` lists every plan with live state for adapters (`in-progress`, +`ready-to-promote`, or `rolled-back`, with the reasons). `router_canary_requests_total` and +`router_canary_rollbacks_total` report it to Prometheus. State is per gateway replica: each +reaches the same verdict from its own share of traffic. + +**Model and policy canaries are rollouts** of the serving pool or the gateway, so whatever +controls that rollout evaluates the plan through the same rule: + +```bash +python -m llm_router.canary --plan model:deploy-0002 --observation observed.json --baseline stable.json +``` + +It exits `0` to promote, `2` to hold, and `3` to roll back, and prints the rollback target. No +rollout controller is wired to it yet, so on those two tracks the decision is automatic and the +action is not. + +Promotion is never automatic on any track. It is a catalog change and goes through review. + ## Tenants What a caller may use is governance and lives in the catalog; the credential that proves which diff --git a/src/llm_router/app.py b/src/llm_router/app.py index bf8a93a..95aa55f 100644 --- a/src/llm_router/app.py +++ b/src/llm_router/app.py @@ -40,6 +40,7 @@ exact_cache_eligible, semantic_cache_eligible, ) +from llm_router.canary import CanaryMonitor, canary_plans from llm_router.classifier import TaskClassifier, load_classifier from llm_router.config import Settings, get_settings from llm_router.engine_stats import ColdStartTracker, EngineStatsCollector @@ -125,12 +126,15 @@ def create_app( catalog.policy.version if catalog is not None else runtime_settings.routing_policy_version ) load = LoadTracker() + plans = canary_plans(catalog) if catalog is not None else () + canary = CanaryMonitor(plans) router = Router( profiles=profiles, external_fallback_enabled=runtime_settings.external_fallback_enabled, registry=catalog, classifier=_load_task_classifier(runtime_settings.task_classifier_path), load=load, + canary=canary, ) admission = AdmissionController( runtime_settings.max_concurrency, @@ -400,6 +404,23 @@ async def variants() -> dict[str, object]: ) return {"object": "list", "data": data} + @app.get("/v1/registry/canaries", dependencies=[Depends(authenticate)]) + async def canaries() -> dict[str, object]: + """Every canary plan by track, with live state for the adapter track.""" + + return { + "object": "list", + "data": [ + { + **plan.model_dump(mode="json"), + "live": ( + canary.status(plan.subject) if plan.subject in canary.subjects else None + ), + } + for plan in plans + ], + } + @app.get("/v1/registry/deployments", dependencies=[Depends(authenticate)]) async def deployments() -> dict[str, object]: if catalog is None: @@ -496,7 +517,9 @@ async def _store_cache( decision: RouteDecision, result: BackendResult, ) -> None: - if not runtime_settings.cache_enabled: + # A canary's responses are never cached: if it is rolled back, nothing + # it produced may keep being served from the cache afterwards. + if not runtime_settings.cache_enabled or decision.canary_arm == "canary": return entry = CachedCompletion( text=result.text, @@ -513,6 +536,25 @@ async def _store_cache( scope = semantic_cache.scope(payload, tenant=tenant, model_revision=catalog_version) semantic_cache.store(scope, prompt, entry) + def _record_canary( + decision: RouteDecision, *, ok: bool, started: float, quality: float | None = None + ) -> None: + """Feed a canaried route's outcome to the monitor, suspending on failure.""" + + if decision.canary_subject is None or decision.canary_arm is None: + return + on_canary = decision.canary_arm == "canary" + telemetry.record_canary(decision.canary_subject, arm=decision.canary_arm, ok=ok) + verdict = canary.record( + decision.canary_subject, + canary=on_canary, + ok=ok, + latency_ms=(time.perf_counter() - started) * 1000, + quality=quality, + ) + if verdict is not None and verdict.action == "rollback": + telemetry.record_canary_rollback(decision.canary_subject) + async def _consume_quota(subject: str, tenant: TenantRecord | None) -> None: # A tenant may carry its own limit; absent one the platform default # applies, which is why None means "defer" rather than "unlimited". @@ -608,8 +650,12 @@ async def _stream_completion( prompt_tokens = max(1, len(prompt) // 4) completion_tokens = max(1, len(text) // 4) span.set_usage(prompt_tokens=prompt_tokens, completion_tokens=completion_tokens) + validity: float | None = None if payload.routing.structured: - telemetry.record_structured_output(decision, valid=structured_output_valid(text)) + valid = structured_output_valid(text) + validity = 1.0 if valid else 0.0 + telemetry.record_structured_output(decision, valid=valid) + _record_canary(decision, ok=True, started=started, quality=validity) telemetry.record_completion( decision, latency_seconds=time.perf_counter() - started, @@ -629,6 +675,8 @@ async def _stream_completion( ) except Exception as error: span.fail(error) + if isinstance(error, BackendUnavailableError): + _record_canary(decision, ok=False, started=started) raise finally: lease.close() @@ -731,6 +779,7 @@ async def _complete( tenant.allow_external_fallback is not False if tenant is not None else True ), privacy_raised_from=privacy_raised_from, + canary_key=f"{subject}|{prompt}", ) span.set_route(decision) if runtime_settings.cache_enabled: @@ -779,6 +828,9 @@ async def _complete( try: result = await inference_backend.generate(payload, decision) + except BackendUnavailableError: + _record_canary(decision, ok=False, started=started) + raise finally: telemetry.inflight_requests.dec() admission.release() @@ -786,8 +838,12 @@ async def _complete( span.set_usage( prompt_tokens=result.prompt_tokens, completion_tokens=result.completion_tokens ) + validity: float | None = None if payload.routing.structured: - telemetry.record_structured_output(decision, valid=structured_output_valid(result.text)) + valid = structured_output_valid(result.text) + validity = 1.0 if valid else 0.0 + telemetry.record_structured_output(decision, valid=valid) + _record_canary(decision, ok=True, started=started, quality=validity) telemetry.record_completion( decision, latency_seconds=time.perf_counter() - started, @@ -827,6 +883,7 @@ async def _complete( "task_source": decision.task_source, "task_confidence": decision.task_confidence, "complexity": decision.complexity, + "canary_arm": decision.canary_arm, "reason": decision.reason, "score": decision.score, "candidate_count": decision.candidate_count, diff --git a/src/llm_router/canary.py b/src/llm_router/canary.py new file mode 100644 index 0000000..5bbd71e --- /dev/null +++ b/src/llm_router/canary.py @@ -0,0 +1,383 @@ +"""Canary tracks and automatic rollback (section 13 and the design targets). + +Section 13 canaries models, adapters, and router policies separately and keeps +the previous revision for rollback; the design targets require rollback to be +automatic once readiness or canary criteria fail. Each track therefore gets its +own plan naming exactly what it rolls back to, and one evaluation rule decides +between holding, promoting, and rolling back. + +The adapter track runs inside the gateway: a staged adapter takes a fixed share +of eligible traffic and is suspended the moment it fails its criteria. Model +and policy canaries are rollouts of the gateway or the serving pool, so their +plans are evaluated by whatever controls that rollout, through the same rule. + +Promotion is never automatic. A canary that passes is reported as ready, and +promoting it is a catalog change that goes through review like any other. +""" + +import hashlib +import json +from collections import deque +from collections.abc import Sequence +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Literal + +from pydantic import BaseModel, Field + +from llm_router.registry import LifecycleStage, Registry + +CanaryAction = Literal["promote", "hold", "rollback"] +LATENCY_WINDOW = 1000 + + +class CanaryTrack(StrEnum): + MODEL = "model" + ADAPTER = "adapter" + POLICY = "policy" + + +class CanaryCriteria(BaseModel): + max_error_rate: float = Field(default=0.01, ge=0.0, le=1.0) + max_p95_latency_ms: float = Field(gt=0.0) + min_quality: float = Field(default=0.0, ge=0.0, le=1.0) + # Requests the canary must serve cleanly before it is ready to promote. + min_requests: int = Field(default=500, ge=1) + # Requests needed before a rate or percentile is trusted enough to act on, + # so one early failure out of two requests does not end a canary. + min_sample: int = Field(default=50, ge=1) + + +class CanaryPlan(BaseModel): + """One canary: what is being tried, what it reverts to, and when.""" + + id: str + track: CanaryTrack + subject: str + rollback_to: str | None + rollback_action: str + traffic_percent: int = Field(ge=1, le=100) + criteria: CanaryCriteria + + +class CanaryObservation(BaseModel): + ready: bool = True + requests: int = Field(default=0, ge=0) + errors: int = Field(default=0, ge=0) + p95_latency_ms: float | None = None + quality: float | None = None + + @property + def error_rate(self) -> float: + return self.errors / self.requests if self.requests else 0.0 + + +class CanaryDecision(BaseModel): + action: CanaryAction + reasons: tuple[str, ...] = () + + +def evaluate( + criteria: CanaryCriteria, + observed: CanaryObservation, + baseline: CanaryObservation | None = None, +) -> CanaryDecision: + """Decide whether a canary rolls back, holds, or is ready to promote. + + Failed readiness rolls back at once, with no sample required. Error rate + and latency roll back only when the stable baseline does not share the + problem: an engine outage degrades both arms, and blaming the canary for + it would roll back a change that did nothing wrong. + """ + + if not observed.ready: + return CanaryDecision(action="rollback", reasons=("readiness probe failure",)) + + control = ( + baseline if baseline is not None and baseline.requests >= criteria.min_sample else None + ) + reasons: list[str] = [] + if observed.requests >= criteria.min_sample: + if observed.error_rate > criteria.max_error_rate and ( + control is None or observed.error_rate > control.error_rate + ): + reasons.append( + f"error rate {observed.error_rate:.3f} above {criteria.max_error_rate:.3f}" + ) + if ( + observed.p95_latency_ms is not None + and observed.p95_latency_ms > criteria.max_p95_latency_ms + and ( + control is None + or control.p95_latency_ms is None + or observed.p95_latency_ms > control.p95_latency_ms + ) + ): + reasons.append( + f"p95 latency {observed.p95_latency_ms:.0f} ms above the " + f"{criteria.max_p95_latency_ms:.0f} ms objective" + ) + if observed.quality is not None and observed.quality < criteria.min_quality: + reasons.append( + f"quality {observed.quality:.3f} below the {criteria.min_quality:.3f} floor" + ) + if reasons: + return CanaryDecision(action="rollback", reasons=tuple(reasons)) + if observed.requests < criteria.min_requests: + return CanaryDecision( + action="hold", + reasons=(f"{observed.requests} of {criteria.min_requests} requests observed",), + ) + return CanaryDecision(action="promote") + + +def canary_plans(registry: Registry) -> tuple[CanaryPlan, ...]: + """Build one plan per canary, on the track it belongs to.""" + + policy = registry.policy + cards = {card.id: card for card in registry.models} + + def criteria(model_ids: Sequence[str]) -> CanaryCriteria: + deployed = [cards[model_id] for model_id in model_ids if model_id in cards] + objectives = [policy.latency_objectives_ms[card.tier] for card in deployed] + qualities = [card.quality for card in deployed] + return CanaryCriteria( + max_error_rate=policy.canary_max_error_rate, + # The strictest objective among what the canary serves. + max_p95_latency_ms=min(objectives, default=max(policy.latency_objectives_ms.values())), + min_quality=round( + max( + policy.quality_floor, + min(qualities, default=0.0) - policy.canary_quality_tolerance, + 0.0, + ), + 4, + ), + min_requests=policy.canary_min_requests, + ) + + plans: list[CanaryPlan] = [] + for deployment in registry.deployments: + if deployment.stage is LifecycleStage.DEPRECATED: + continue + target = registry.rollback_target(deployment.id) + plans.append( + CanaryPlan( + id=f"model:{deployment.id}", + track=CanaryTrack.MODEL, + subject=deployment.id, + rollback_to=None if target is None else target.id, + rollback_action=( + "no previous revision; stop the rollout" + if target is None + else f"redeploy {target.id} ({target.container_digest})" + ), + traffic_percent=policy.canary_traffic_percent, + criteria=criteria(list(deployment.model_checksums)), + ) + ) + + for adapter in registry.servable_adapters(): + if adapter.stage is not LifecycleStage.STAGING: + continue + stable = next( + ( + item + for item in registry.servable_adapters() + if item.stage is LifecycleStage.PRODUCTION + and item.base_model_id == adapter.base_model_id + and item.base_revision == adapter.base_revision + and item.domain == adapter.domain + ), + None, + ) + plans.append( + CanaryPlan( + id=f"adapter:{adapter.id}", + track=CanaryTrack.ADAPTER, + subject=adapter.id, + rollback_to=None if stable is None else stable.id, + rollback_action=( + f"serve {adapter.base_model_id} without an adapter" + if stable is None + else f"serve {stable.id} for all {adapter.domain} traffic" + ), + traffic_percent=policy.canary_traffic_percent, + criteria=criteria([adapter.base_model_id]), + ) + ) + + plans.append( + CanaryPlan( + id=f"policy:{policy.version}", + track=CanaryTrack.POLICY, + subject=policy.version, + rollback_to=policy.previous_version, + rollback_action=( + "no previous policy version; stop the rollout" + if policy.previous_version is None + else f"redeploy the gateway with policy {policy.previous_version}" + ), + traffic_percent=policy.canary_traffic_percent, + criteria=criteria([card.id for card in registry.servable_models()]), + ) + ) + return tuple(plans) + + +@dataclass +class _Arm: + requests: int = 0 + errors: int = 0 + latencies: deque[float] = field(default_factory=lambda: deque(maxlen=LATENCY_WINDOW)) + quality_total: float = 0.0 + quality_samples: int = 0 + + def record(self, *, ok: bool, latency_ms: float, quality: float | None) -> None: + self.requests += 1 + self.errors += 0 if ok else 1 + self.latencies.append(latency_ms) + if quality is not None: + self.quality_total += quality + self.quality_samples += 1 + + def observation(self) -> CanaryObservation: + ordered = sorted(self.latencies) + return CanaryObservation( + requests=self.requests, + errors=self.errors, + p95_latency_ms=(ordered[max(0, -(-len(ordered) * 95 // 100) - 1)] if ordered else None), + quality=(self.quality_total / self.quality_samples if self.quality_samples else None), + ) + + +class CanaryMonitor: + """Splits traffic to a staged adapter and suspends it when it fails. + + State is held per gateway replica. Each replica reaches the same verdict + from its own share of traffic, so a failing adapter is suspended everywhere + without the replicas having to agree first. + """ + + def __init__(self, plans: Sequence[CanaryPlan]) -> None: + self._plans = {plan.subject: plan for plan in plans if plan.track is CanaryTrack.ADAPTER} + self._canary: dict[str, _Arm] = {subject: _Arm() for subject in self._plans} + self._stable: dict[str, _Arm] = {subject: _Arm() for subject in self._plans} + self._rolled_back: dict[str, tuple[str, ...]] = {} + + def takes(self, subject: str, key: str) -> bool: + """Whether this request is served by the canary. + + The bucket is derived from the request, so the same request always + lands on the same arm and a retry cannot flip between adapters. + """ + + plan = self._plans.get(subject) + if plan is None or subject in self._rolled_back: + return False + digest = hashlib.sha256(f"{subject}|{key}".encode()).hexdigest() + return int(digest[:8], 16) % 100 < plan.traffic_percent + + def record( + self, + subject: str, + *, + canary: bool, + ok: bool, + latency_ms: float, + quality: float | None = None, + ) -> CanaryDecision | None: + """Record an outcome; returns the decision when it was a canary request.""" + + plan = self._plans.get(subject) + if plan is None: + return None + arms = self._canary if canary else self._stable + arms[subject].record(ok=ok, latency_ms=latency_ms, quality=quality) + if not canary or subject in self._rolled_back: + return None + decision = evaluate( + plan.criteria, + self._canary[subject].observation(), + self._stable[subject].observation(), + ) + if decision.action == "rollback": + self._rolled_back[subject] = decision.reasons + return decision + + def status(self, subject: str) -> dict[str, object]: + """Live state of one adapter canary, for the registry endpoint.""" + + plan = self._plans[subject] + observed = self._canary[subject].observation() + if subject in self._rolled_back: + state, reasons = "rolled-back", self._rolled_back[subject] + else: + decision = evaluate(plan.criteria, observed, self._stable[subject].observation()) + state = "ready-to-promote" if decision.action == "promote" else "in-progress" + reasons = decision.reasons + return { + "state": state, + "reasons": list(reasons), + "canary": observed.model_dump(mode="json"), + "stable": self._stable[subject].observation().model_dump(mode="json"), + } + + @property + def subjects(self) -> tuple[str, ...]: + return tuple(self._plans) + + +def main(argv: Sequence[str] | None = None) -> int: + """Print every canary plan, or evaluate one against an observation. + + Evaluation exits 0 to promote, 2 to hold, and 3 to roll back, so a rollout + controller can act on the result without parsing it. + """ + + import argparse + from pathlib import Path + + from llm_router.registry import load_registry + + parser = argparse.ArgumentParser(description=main.__doc__) + parser.add_argument("--catalog", default="config/registry.yaml") + parser.add_argument("--plan", help="plan id to evaluate, for example model:deploy-0002") + parser.add_argument("--observation", help="JSON file holding the observed canary metrics") + parser.add_argument("--baseline", help="JSON file holding the stable baseline metrics") + arguments = parser.parse_args(argv) + + plans = canary_plans(load_registry(arguments.catalog)) + if arguments.plan is None: + print(json.dumps([plan.model_dump(mode="json") for plan in plans], indent=2)) + return 0 + + plan = next((item for item in plans if item.id == arguments.plan), None) + if plan is None or arguments.observation is None: + parser.error("--plan must name a known plan and be given with --observation") + + def read(path: str) -> CanaryObservation: + return CanaryObservation.model_validate_json(Path(path).read_text(encoding="utf-8")) + + decision = evaluate( + plan.criteria, + read(arguments.observation), + read(arguments.baseline) if arguments.baseline else None, + ) + print( + json.dumps( + { + "plan": plan.id, + "action": decision.action, + "reasons": list(decision.reasons), + "rollback_to": plan.rollback_to, + "rollback_action": plan.rollback_action, + }, + indent=2, + ) + ) + return {"promote": 0, "hold": 2, "rollback": 3}[decision.action] + + +if __name__ == "__main__": # pragma: no cover - command-line entry point + raise SystemExit(main()) diff --git a/src/llm_router/models.py b/src/llm_router/models.py index effa86e..f2a621e 100644 --- a/src/llm_router/models.py +++ b/src/llm_router/models.py @@ -82,6 +82,9 @@ class RouteDecision(BaseModel): task_source: str = "declared" task_confidence: float | None = None complexity: str | None = None + # Set when a staged adapter exists for this route: which arm served it. + canary_arm: str | None = None + canary_subject: str | None = None class ChatCompletionChoice(BaseModel): diff --git a/src/llm_router/observability.py b/src/llm_router/observability.py index 48904fb..3c55f67 100644 --- a/src/llm_router/observability.py +++ b/src/llm_router/observability.py @@ -129,6 +129,18 @@ def __init__(self, registry: CollectorRegistry | None = None) -> None: buckets=LATENCY_BUCKETS, registry=self.registry, ) + self.canary_requests_total = Counter( + "router_canary_requests_total", + "Requests on a canaried route by subject, arm, and outcome.", + ["subject", "arm", "outcome"], + registry=self.registry, + ) + self.canary_rollbacks_total = Counter( + "router_canary_rollbacks_total", + "Canaries suspended automatically after failing their criteria.", + ["subject"], + registry=self.registry, + ) self.engine_circuit_open = Gauge( "router_engine_circuit_open", "Engine circuit state: 0 closed, 0.5 half-open, 1 open.", @@ -243,6 +255,14 @@ def record_structured_output(self, decision: RouteDecision, *, valid: bool) -> N abs(decision.profile.quality - observed) ) + def record_canary(self, subject: str, *, arm: str, ok: bool) -> None: + self.canary_requests_total.labels( + subject=subject, arm=arm, outcome="success" if ok else "error" + ).inc() + + def record_canary_rollback(self, subject: str) -> None: + self.canary_rollbacks_total.labels(subject=subject).inc() + def record_circuit_state(self, state: str, *, engine: str) -> None: value = {"closed": 0.0, "half-open": 0.5, "open": 1.0}[state] self.engine_circuit_open.labels(engine=engine).set(value) diff --git a/src/llm_router/registry.py b/src/llm_router/registry.py index 640af84..227829a 100644 --- a/src/llm_router/registry.py +++ b/src/llm_router/registry.py @@ -272,6 +272,32 @@ class RoutePolicy(BaseModel): quality_floor: float = Field(default=0.0, ge=0.0, le=1.0) resource_ceiling: float = Field(default=10.0, gt=0.0) fallback_order: tuple[str, ...] = () + # The version this policy replaced, kept so a policy canary has + # something to roll back to. + previous_version: str | None = None + canary_traffic_percent: int = Field(default=10, ge=1, le=100) + canary_min_requests: int = Field(default=500, ge=1) + canary_max_error_rate: float = Field(default=0.01, ge=0.0, le=1.0) + # How far below a model's benchmarked quality a canary may fall. + canary_quality_tolerance: float = Field(default=0.05, ge=0.0, le=1.0) + # p95 latency objective per tier; a canary is held to the strictest + # objective among the models it serves. + latency_objectives_ms: dict[ModelTier, float] = Field( + default_factory=lambda: { + ModelTier.SMALL_SPECIALIST: 1500.0, + ModelTier.GENERAL_LOCAL: 4000.0, + ModelTier.HIGH_CAPABILITY: 12000.0, + ModelTier.EXTERNAL_FALLBACK: 12000.0, + } + ) + + @model_validator(mode="after") + def cover_every_tier_with_an_objective(self) -> "RoutePolicy": + missing = set(ModelTier) - set(self.latency_objectives_ms) + if missing: + names = ", ".join(sorted(tier.value for tier in missing)) + raise ValueError(f"latency objectives are missing for: {names}") + return self class RegistryError(RuntimeError): @@ -504,9 +530,19 @@ def servable_adapters(self) -> tuple[AdapterProfile, ...]: ) def select_adapter( - self, *, model_id: str, revision: str, domain: str | None, task: TaskClass + self, + *, + model_id: str, + revision: str, + domain: str | None, + task: TaskClass, + stage: LifecycleStage | None = None, ) -> AdapterProfile | None: - """Pick the best-scoring approved adapter for a base revision and domain.""" + """Pick the best-scoring approved adapter for a base revision and domain. + + A stage narrows the choice to that stage, which is how routing keeps + the production adapter and its staged canary apart. + """ if domain is None: return None @@ -517,6 +553,7 @@ def select_adapter( and adapter.base_revision == revision and adapter.domain == domain and task in adapter.intended_tasks + and (stage is None or adapter.stage is stage) ] if not candidates: return None diff --git a/src/llm_router/routing.py b/src/llm_router/routing.py index d3abc7e..f3ab257 100644 --- a/src/llm_router/routing.py +++ b/src/llm_router/routing.py @@ -12,7 +12,8 @@ ) if TYPE_CHECKING: # pragma: no cover - import cycle guard for type checking only - from llm_router.registry import Registry + from llm_router.canary import CanaryMonitor + from llm_router.registry import AdapterProfile, Registry # How strongly predicted complexity pulls a request toward higher measured @@ -95,6 +96,8 @@ class Router: # keeps the router usable where no training data is deployed. classifier: TaskClassifier | None = None load: LoadTracker | None = None + # Without a monitor a staged adapter takes no traffic at all. + canary: "CanaryMonitor | None" = None def classify_task(self, request: ChatCompletionRequest) -> TaskClass: if request.routing.task is not None: @@ -128,6 +131,7 @@ def select( quality_floor: float = 0.0, tenant_allows_external: bool = True, privacy_raised_from: PrivacyClass | None = None, + canary_key: str = "", ) -> RouteDecision: prediction = ( self.classifier.predict(request.prompt) if self.classifier is not None else None @@ -213,15 +217,8 @@ def score(profile: ModelProfile) -> float: ) selected = max(candidates, key=score) - adapter = ( - self.registry.select_adapter( - model_id=selected.id, - revision=selected.revision, - domain=request.routing.domain, - task=task, - ) - if self.registry is not None - else None + adapter, canary_arm, canary_subject = self._select_adapter( + selected, request.routing.domain, task, canary_key ) reason = ( f"selected highest policy score among {len(candidates)} eligible model(s); " @@ -250,6 +247,10 @@ def score(profile: ModelProfile) -> float: f"; applied adapter {adapter.id} for domain {adapter.domain} " f"(measured quality delta {adapter.benchmark.quality_delta:+.3f})" ) + if canary_arm == "canary": + reason += f"; canary arm of {canary_subject}" + elif canary_arm == "stable": + reason += f"; stable arm while {canary_subject} is canaried" return RouteDecision( profile=selected, task=task, @@ -261,7 +262,43 @@ def score(profile: ModelProfile) -> float: task_source=task_source, task_confidence=task_confidence, complexity=None if complexity is None else complexity.value, + canary_arm=canary_arm, + canary_subject=canary_subject, + ) + + def _select_adapter( + self, selected: ModelProfile, domain: str | None, task: TaskClass, canary_key: str + ) -> "tuple[AdapterProfile | None, str | None, str | None]": + """Choose the production adapter, or its staged canary for a share of traffic. + + A staged adapter never replaces the production one outright. It is + offered a fixed share of eligible requests while the monitor still + trusts it, and none once it has been rolled back. + """ + + if self.registry is None: + return None, None, None + from llm_router.registry import LifecycleStage + + stable = self.registry.select_adapter( + model_id=selected.id, + revision=selected.revision, + domain=domain, + task=task, + stage=LifecycleStage.PRODUCTION, + ) + staged = self.registry.select_adapter( + model_id=selected.id, + revision=selected.revision, + domain=domain, + task=task, + stage=LifecycleStage.STAGING, ) + if staged is None or self.canary is None or staged.id not in self.canary.subjects: + return stable, None, None + if self.canary.takes(staged.id, canary_key): + return staged, "canary", staged.id + return stable, "stable", staged.id @staticmethod def _privacy_allows(profile: ModelProfile, privacy: PrivacyClass) -> bool: diff --git a/src/llm_router/serving.py b/src/llm_router/serving.py index 7ce6ec5..ef62de0 100644 --- a/src/llm_router/serving.py +++ b/src/llm_router/serving.py @@ -11,7 +11,6 @@ from llm_router.registry import ( EngineVariant, - LifecycleStage, ModelCard, Quantization, Registry, @@ -168,35 +167,6 @@ def build_serving_config(registry: Registry) -> dict[str, Any]: return config -def canary_config(registry: Registry, deployment_id: str) -> dict[str, Any]: - """Describe a canary and the revision it rolls back to (section 13).""" - - current = next((item for item in registry.deployments if item.id == deployment_id), None) - if current is None: - raise ServingConfigError(f"unknown deployment {deployment_id}") - target = registry.rollback_target(deployment_id) - staged_adapters = [ - adapter.id - for adapter in registry.servable_adapters() - if adapter.stage is LifecycleStage.STAGING - ] - return { - "deployment_id": current.id, - "container_digest": current.container_digest, - "gpu_pool": current.gpu_pool, - "canary_traffic_percent": 10, - "promote_after_successful_requests": 500, - "rollback_to": None if target is None else target.id, - "rollback_triggers": [ - "readiness probe failure", - "p95 latency above the tier objective", - "quality below the benchmark floor", - "error rate above one percent", - ], - "staged_adapters": sorted(staged_adapters), - } - - def render_serving_config(registry: Registry) -> str: return yaml.safe_dump(build_serving_config(registry), sort_keys=False) diff --git a/src/llm_router/tracing.py b/src/llm_router/tracing.py index ae4b356..9fc20f0 100644 --- a/src/llm_router/tracing.py +++ b/src/llm_router/tracing.py @@ -106,6 +106,8 @@ def set_route(self, decision: RouteDecision) -> None: "router.task.source": decision.task_source, "router.task.confidence": decision.task_confidence, "router.complexity": decision.complexity, + "router.canary.arm": decision.canary_arm, + "router.canary.subject": decision.canary_subject, "router.route.reason": decision.reason, "router.route.score": decision.score, "router.route.candidates": decision.candidate_count, diff --git a/tests/integration/test_canary_api.py b/tests/integration/test_canary_api.py new file mode 100644 index 0000000..573e6bf --- /dev/null +++ b/tests/integration/test_canary_api.py @@ -0,0 +1,141 @@ +from typing import Any + +from fastapi.testclient import TestClient + +from llm_router.app import create_app +from llm_router.backends import BackendResult, BackendUnavailableError, MockInferenceBackend +from llm_router.config import Settings +from llm_router.models import ChatCompletionRequest, RouteDecision +from llm_router.registry import Registry, load_registry + +CATALOG = load_registry("config/registry.yaml") +HEADERS = {"Authorization": "Bearer canary-key"} +STAGED = "claims-extraction-lora-next" +STABLE = "claims-extraction-lora" + + +class FailsOnCanary(MockInferenceBackend): + """Serves the production adapter and fails whenever the staged one is used.""" + + async def generate( + self, request: ChatCompletionRequest, decision: RouteDecision + ) -> BackendResult: + if decision.adapter_id == STAGED: + raise BackendUnavailableError("inference engine returned 500 for the staged adapter") + return await super().generate(request, decision) + + +def catalog(**policy: Any) -> Registry: + document = CATALOG.model_dump(mode="json") + document["policy"] = {**document["policy"], **policy} + return Registry.model_validate(document) + + +def build(registry: Registry, backend: MockInferenceBackend | None = None) -> TestClient: + # A high circuit threshold keeps the engine circuit out of the picture, so + # the only thing that can stop the staged adapter is the canary monitor. + settings = Settings( + api_keys="canary-key", quota_requests_per_minute=10_000, engine_failure_threshold=10_000 + ) + return TestClient(create_app(settings, registry=registry, backend=backend)) + + +def claims(client: TestClient, index: int) -> Any: + return client.post( + "/v1/chat/completions", + headers=HEADERS, + json={ + "model": "auto", + "messages": [ + {"role": "user", "content": f"Extract the claim fields from note {index}"} + ], + "routing": {"privacy": "private", "domain": "claims", "task": "extraction"}, + }, + ) + + +def test_a_staged_adapter_takes_only_its_share_of_traffic() -> None: + with build(catalog(canary_traffic_percent=20)) as client: + served = [claims(client, index).json()["routing"] for index in range(300)] + + on_canary = [item for item in served if item["adapter_id"] == STAGED] + assert 30 <= len(on_canary) <= 90 + assert all(item["canary_arm"] == "canary" for item in on_canary) + assert all( + item["adapter_id"] == STABLE and item["canary_arm"] == "stable" + for item in served + if item["adapter_id"] != STAGED + ) + + +def test_a_failing_staged_adapter_is_rolled_back_automatically() -> None: + registry = catalog(canary_traffic_percent=100, canary_min_requests=500) + with build(registry, FailsOnCanary()) as client: + statuses = [claims(client, index).status_code for index in range(60)] + after = claims(client, 1000) + listing = client.get("/v1/registry/canaries", headers=HEADERS).json()["data"] + metrics = client.get("/metrics").text + + live = next(item for item in listing if item["subject"] == STAGED)["live"] + # Every request went to the staged adapter and failed until the sample was + # large enough to act on; from then on the production adapter served. + assert statuses[:50] == [502] * 50 + assert statuses[50:] == [200] * 10 + assert after.json()["routing"]["adapter_id"] == STABLE + assert live["state"] == "rolled-back" + assert "error rate 1.000 above 0.010" in live["reasons"] + assert f'router_canary_rollbacks_total{{subject="{STAGED}"}} 1.0' in metrics + + +def test_a_canary_response_is_never_served_from_the_cache() -> None: + with build(catalog(canary_traffic_percent=100)) as client: + first = client.post( + "/v1/chat/completions", + headers=HEADERS, + json={ + "model": "auto", + "messages": [{"role": "user", "content": "Extract the claim fields"}], + "routing": {"privacy": "public", "domain": "claims", "task": "extraction"}, + }, + ) + second = client.post( + "/v1/chat/completions", + headers=HEADERS, + json={ + "model": "auto", + "messages": [{"role": "user", "content": "Extract the claim fields"}], + "routing": {"privacy": "public", "domain": "claims", "task": "extraction"}, + }, + ) + + assert first.json()["routing"]["canary_arm"] == "canary" + # Cacheable in every other respect, yet generated again. + assert second.headers["X-Cache"] == "miss" + + +def test_the_registry_lists_every_track_with_live_state_for_adapters() -> None: + with build(CATALOG) as client: + claims(client, 1) + listing = client.get("/v1/registry/canaries", headers=HEADERS).json()["data"] + + by_id = {item["id"]: item for item in listing} + assert by_id["model:deploy-0002"]["live"] is None + assert by_id["policy:v1"]["live"] is None + assert by_id[f"adapter:{STAGED}"]["live"]["state"] == "in-progress" + assert by_id[f"adapter:{STAGED}"]["rollback_to"] == STABLE + + +def test_a_route_with_no_staged_adapter_reports_no_canary() -> None: + with build(CATALOG) as client: + response = client.post( + "/v1/chat/completions", + headers=HEADERS, + json={ + "model": "auto", + "messages": [{"role": "user", "content": "Classify this ticket"}], + "routing": {"domain": "support", "task": "classification"}, + }, + ) + + assert response.json()["routing"]["adapter_id"] == "support-classification-lora" + assert response.json()["routing"]["canary_arm"] is None diff --git a/tests/unit/test_canary.py b/tests/unit/test_canary.py new file mode 100644 index 0000000..3f17898 --- /dev/null +++ b/tests/unit/test_canary.py @@ -0,0 +1,315 @@ +import json +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from llm_router.canary import ( + CanaryCriteria, + CanaryMonitor, + CanaryObservation, + CanaryPlan, + CanaryTrack, + canary_plans, + evaluate, + main, +) +from llm_router.models import ChatCompletionRequest, TaskClass +from llm_router.registry import LifecycleStage, Registry, RoutePolicy, load_registry +from llm_router.routing import Router + +CATALOG = load_registry("config/registry.yaml") +CRITERIA = CanaryCriteria(max_p95_latency_ms=1500, min_quality=0.8, min_requests=500) +STAGED = "claims-extraction-lora-next" + + +def observed(**values: object) -> CanaryObservation: + return CanaryObservation.model_validate({"requests": 100, **values}) + + +def plans_by_id() -> dict[str, CanaryPlan]: + return {plan.id: plan for plan in canary_plans(CATALOG)} + + +def test_each_track_gets_its_own_plan_and_rollback_target() -> None: + plans = plans_by_id() + + assert {plan.track for plan in plans.values()} == set(CanaryTrack) + assert plans["model:deploy-0002"].rollback_to == "deploy-0001" + assert plans[f"adapter:{STAGED}"].rollback_to == "claims-extraction-lora" + assert "claims-extraction-lora for all claims traffic" in ( + plans[f"adapter:{STAGED}"].rollback_action + ) + assert plans["policy:v1"].rollback_to is None + # A deprecated deployment is history, not a canary. + assert "model:deploy-0001" not in plans + + +def test_a_canary_is_held_to_the_strictest_objective_it_serves() -> None: + plans = plans_by_id() + + # deploy-0002 serves the small specialist and the general tier. + assert plans["model:deploy-0002"].criteria.max_p95_latency_ms == 1500.0 + assert plans["model:deploy-0002"].criteria.min_quality == pytest.approx(0.77) + + +def test_a_staged_adapter_with_no_production_sibling_rolls_back_to_the_base_model() -> None: + document = CATALOG.model_dump(mode="json") + document["adapters"] = [item for item in document["adapters"] if item["id"] == STAGED] + document["deployments"] = [] + + plans = {plan.id: plan for plan in canary_plans(Registry.model_validate(document))} + + assert plans[f"adapter:{STAGED}"].rollback_to is None + assert plans[f"adapter:{STAGED}"].rollback_action == "serve small-specialist without an adapter" + + +def test_a_policy_canary_rolls_back_to_the_version_it_replaced() -> None: + document = CATALOG.model_dump(mode="json") + document["policy"] = {**document["policy"], "version": "v2", "previous_version": "v1"} + + plans = {plan.id: plan for plan in canary_plans(Registry.model_validate(document))} + + assert plans["policy:v2"].rollback_to == "v1" + assert plans["policy:v2"].rollback_action == "redeploy the gateway with policy v1" + + +def test_a_policy_must_give_every_tier_a_latency_objective() -> None: + with pytest.raises(ValidationError, match="latency objectives are missing"): + RoutePolicy(latency_objectives_ms={"small-specialist": 1000.0}) # type: ignore[dict-item] + + +def test_failed_readiness_rolls_back_at_once_with_no_sample() -> None: + decision = evaluate(CRITERIA, CanaryObservation(ready=False, requests=1)) + + assert decision.action == "rollback" + assert decision.reasons == ("readiness probe failure",) + + +def test_one_early_failure_does_not_end_a_canary() -> None: + decision = evaluate(CRITERIA, CanaryObservation(requests=2, errors=1)) + + assert decision.action == "hold" + + +def test_an_error_rate_over_the_limit_rolls_back() -> None: + decision = evaluate(CRITERIA, observed(errors=5)) + + assert decision.action == "rollback" + assert "error rate 0.050 above 0.010" in decision.reasons + + +def test_latency_over_the_objective_rolls_back() -> None: + decision = evaluate(CRITERIA, observed(p95_latency_ms=2400.0)) + + assert decision.action == "rollback" + assert "p95 latency 2400 ms above the 1500 ms objective" in decision.reasons + + +def test_quality_under_the_floor_rolls_back() -> None: + decision = evaluate(CRITERIA, observed(quality=0.6)) + + assert decision.action == "rollback" + assert "quality 0.600 below the 0.800 floor" in decision.reasons + + +def test_every_failed_criterion_is_named() -> None: + decision = evaluate(CRITERIA, observed(errors=10, p95_latency_ms=3000.0, quality=0.1)) + + assert len(decision.reasons) == 3 + + +def test_an_outage_shared_by_the_stable_arm_is_not_blamed_on_the_canary() -> None: + canary = observed(errors=40, p95_latency_ms=9000.0) + suffering_too = observed(errors=45, p95_latency_ms=9500.0) + healthy = observed(errors=0, p95_latency_ms=400.0) + + assert evaluate(CRITERIA, canary, suffering_too).action == "hold" + assert evaluate(CRITERIA, canary, healthy).action == "rollback" + + +def test_a_baseline_too_small_to_trust_is_ignored() -> None: + canary = observed(errors=40) + thin = CanaryObservation(requests=3, errors=3) + + assert evaluate(CRITERIA, canary, thin).action == "rollback" + + +def test_a_clean_canary_holds_until_it_has_served_enough_then_is_ready() -> None: + holding = evaluate(CRITERIA, observed(requests=499)) + ready = evaluate(CRITERIA, observed(requests=500, p95_latency_ms=300.0, quality=0.95)) + + assert holding.action == "hold" + assert holding.reasons == ("499 of 500 requests observed",) + assert ready.action == "promote" + + +def monitor(percent: int = 10, **criteria: object) -> CanaryMonitor: + plan = plans_by_id()[f"adapter:{STAGED}"] + return CanaryMonitor( + [ + plan.model_copy( + update={ + "traffic_percent": percent, + "criteria": plan.criteria.model_copy(update=criteria), + } + ) + ] + ) + + +def test_the_canary_takes_about_its_share_and_a_request_keeps_its_arm() -> None: + tracker = monitor(percent=10) + keys = [f"tenant|prompt {index}" for index in range(2000)] + + taken = [key for key in keys if tracker.takes(STAGED, key)] + + assert 150 <= len(taken) <= 250 + # The same request always lands on the same arm. + assert all(tracker.takes(STAGED, key) for key in taken) + assert tracker.takes("not-a-canary", keys[0]) is False + + +def test_a_failing_canary_is_suspended_and_takes_no_more_traffic() -> None: + tracker = monitor(percent=100, min_sample=10) + + decisions = [tracker.record(STAGED, canary=True, ok=False, latency_ms=100.0) for _ in range(10)] + + assert decisions[-1] is not None and decisions[-1].action == "rollback" + assert tracker.takes(STAGED, "any request") is False + assert tracker.status(STAGED)["state"] == "rolled-back" + # Nothing recorded after the rollback reopens it. + assert tracker.record(STAGED, canary=True, ok=True, latency_ms=1.0) is None + + +def test_a_passing_canary_is_reported_ready_but_keeps_its_share() -> None: + tracker = monitor(percent=100, min_sample=5, min_requests=20) + + for _ in range(20): + tracker.record(STAGED, canary=True, ok=True, latency_ms=50.0, quality=1.0) + + status = tracker.status(STAGED) + assert status["state"] == "ready-to-promote" + # Promotion is a catalog change; the monitor never widens traffic itself. + assert tracker.takes(STAGED, "any request") is True + + +def test_stable_arm_outcomes_are_the_control_and_never_a_verdict() -> None: + tracker = monitor(percent=100, min_sample=5) + + assert tracker.record(STAGED, canary=False, ok=False, latency_ms=10.0) is None + assert tracker.record("unknown", canary=True, ok=False, latency_ms=10.0) is None + assert tracker.status(STAGED)["stable"]["errors"] == 1 # type: ignore[index] + + +def claims_request(prompt: str = "Extract the claim fields") -> ChatCompletionRequest: + return ChatCompletionRequest.model_validate( + { + "messages": [{"role": "user", "content": prompt}], + "routing": {"domain": "claims", "task": "extraction"}, + } + ) + + +def test_without_a_monitor_a_staged_adapter_takes_no_traffic() -> None: + router = Router(CATALOG.profiles(), registry=CATALOG) + + decision = router.select(claims_request()) + + assert decision.adapter_id == "claims-extraction-lora" + assert decision.canary_arm is None + + +def test_the_router_serves_the_canary_arm_and_says_so() -> None: + router = Router(CATALOG.profiles(), registry=CATALOG, canary=monitor(percent=100)) + + decision = router.select(claims_request(), canary_key="tenant|prompt") + + assert decision.adapter_id == STAGED + assert decision.canary_arm == "canary" + assert f"canary arm of {STAGED}" in decision.reason + + +def test_the_router_keeps_most_traffic_on_the_production_adapter() -> None: + tracker = monitor(percent=10) + router = Router(CATALOG.profiles(), registry=CATALOG, canary=tracker) + key = next(f"t|{index}" for index in range(100) if not tracker.takes(STAGED, f"t|{index}")) + + decision = router.select(claims_request(), canary_key=key) + + assert decision.adapter_id == "claims-extraction-lora" + assert decision.canary_arm == "stable" + assert decision.canary_subject == STAGED + + +def test_a_rolled_back_canary_returns_all_traffic_to_production() -> None: + tracker = monitor(percent=100, min_sample=5) + router = Router(CATALOG.profiles(), registry=CATALOG, canary=tracker) + for _ in range(5): + tracker.record(STAGED, canary=True, ok=False, latency_ms=10.0) + + decision = router.select(claims_request(), canary_key="tenant|prompt") + + assert decision.adapter_id == "claims-extraction-lora" + assert decision.canary_arm == "stable" + + +def test_select_adapter_can_be_narrowed_to_one_stage() -> None: + arguments = { + "model_id": "small-specialist", + "revision": CATALOG.model_card("small-specialist").revision, + "domain": "claims", + "task": TaskClass.EXTRACTION, + } + + production = CATALOG.select_adapter(**arguments, stage=LifecycleStage.PRODUCTION) # type: ignore[arg-type] + staging = CATALOG.select_adapter(**arguments, stage=LifecycleStage.STAGING) # type: ignore[arg-type] + + assert production is not None and production.id == "claims-extraction-lora" + assert staging is not None and staging.id == STAGED + + +def test_cli_prints_every_plan(capsys: pytest.CaptureFixture[str]) -> None: + assert main([]) == 0 + + printed = json.loads(capsys.readouterr().out) + assert {plan["track"] for plan in printed} == {"model", "adapter", "policy"} + + +@pytest.mark.parametrize( + ("observation", "baseline", "code", "action"), + [ + ({"requests": 600, "errors": 0, "p95_latency_ms": 300.0}, None, 0, "promote"), + ({"requests": 120, "errors": 0}, None, 2, "hold"), + ({"requests": 120, "errors": 30}, None, 3, "rollback"), + ({"requests": 120, "errors": 30}, {"requests": 120, "errors": 40}, 2, "hold"), + ({"ready": False, "requests": 0}, None, 3, "rollback"), + ], +) +def test_cli_exit_code_tells_a_rollout_controller_what_to_do( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + observation: dict[str, object], + baseline: dict[str, object] | None, + code: int, + action: str, +) -> None: + observed_path = tmp_path / "observed.json" + observed_path.write_text(json.dumps(observation), encoding="utf-8") + arguments = ["--plan", "model:deploy-0002", "--observation", str(observed_path)] + if baseline is not None: + baseline_path = tmp_path / "baseline.json" + baseline_path.write_text(json.dumps(baseline), encoding="utf-8") + arguments += ["--baseline", str(baseline_path)] + + assert main(arguments) == code + + printed = json.loads(capsys.readouterr().out) + assert printed["action"] == action + assert printed["rollback_to"] == "deploy-0001" + + +def test_cli_rejects_an_unknown_plan(capsys: pytest.CaptureFixture[str]) -> None: + with pytest.raises(SystemExit): + main(["--plan", "model:nope", "--observation", "x.json"]) + capsys.readouterr() diff --git a/tests/unit/test_serving.py b/tests/unit/test_serving.py index faa6533..3f1ac58 100644 --- a/tests/unit/test_serving.py +++ b/tests/unit/test_serving.py @@ -1,5 +1,3 @@ -from pathlib import Path - import pytest import yaml @@ -7,7 +5,6 @@ from llm_router.serving import ( ServingConfigError, build_serving_config, - canary_config, render_serving_config, ) @@ -87,30 +84,3 @@ def test_rendered_configuration_is_valid_yaml(registry: Registry) -> None: def test_empty_catalog_is_rejected() -> None: with pytest.raises(ServingConfigError, match="no servable local models"): build_serving_config(Registry(models=())) - - -def test_canary_config_names_its_rollback_target(registry: Registry) -> None: - canary = canary_config(registry, "deploy-0002") - - assert canary["rollback_to"] == "deploy-0001" - assert canary["canary_traffic_percent"] == 10 - assert canary["staged_adapters"] == ["claims-extraction-lora-next"] - assert "readiness probe failure" in canary["rollback_triggers"] - - -def test_canary_config_reports_an_unknown_deployment(registry: Registry) -> None: - with pytest.raises(ServingConfigError, match="unknown deployment"): - canary_config(registry, "deploy-9999") - - -def test_first_deployment_has_no_rollback_target(registry: Registry) -> None: - assert canary_config(registry, "deploy-0001")["rollback_to"] is None - - -def test_committed_deployment_configuration_matches_the_catalog(registry: Registry) -> None: - committed = Path("config/ray-serve.yaml").read_text(encoding="utf-8") - - assert committed == render_serving_config(registry), ( - "config/ray-serve.yaml is stale; regenerate with " - "`python -m llm_router.serving > config/ray-serve.yaml`" - )