From 0ed21a08b820ef7560a764aa5fabe3dc0058d44e Mon Sep 17 00:00:00 2001 From: Yash-Chindam Date: Sat, 3 Oct 2026 19:57:06 +0530 Subject: [PATCH] feat: canary models, adapters, and policies separately with automatic rollback Section 13 canaries models, adapters, and router policies separately, and the design targets require rollback to be automatic once readiness or canary criteria fail. There was one combined plan that listed rollback triggers as text, and nothing acted on them. A staged adapter with the larger measured gain simply took all of the traffic. Each track now has its own plan naming what it rolls back to, and one rule decides them all: failed readiness rolls back at once; error rate, latency, and quality roll back once enough requests have been seen. Error rate and latency only count against a canary when the stable baseline does not share the problem, so an engine outage that degrades both arms is not blamed on the change. The adapter track runs inside the gateway. A staged adapter is offered a fixed share of eligible requests, bucketed by tenant and prompt so a retry cannot flip between adapters, and is suspended the moment it fails. Its responses are never cached, so nothing it produced outlives a rollback. Model and policy canaries are rollouts, so the plan is evaluated through the same rule by a command whose exit code says promote, hold, or roll back. Promotion is never automatic on any track. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/cd.yml | 16 +- README.md | 50 +++- src/llm_router/app.py | 63 ++++- src/llm_router/canary.py | 383 +++++++++++++++++++++++++++ src/llm_router/models.py | 3 + src/llm_router/observability.py | 20 ++ src/llm_router/registry.py | 41 ++- src/llm_router/routing.py | 57 +++- src/llm_router/serving.py | 30 --- src/llm_router/tracing.py | 2 + tests/integration/test_canary_api.py | 141 ++++++++++ tests/unit/test_canary.py | 315 ++++++++++++++++++++++ tests/unit/test_serving.py | 30 --- 13 files changed, 1063 insertions(+), 88 deletions(-) create mode 100644 src/llm_router/canary.py create mode 100644 tests/integration/test_canary_api.py create mode 100644 tests/unit/test_canary.py 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`" - )