diff --git a/config/config.yaml b/config/config.yaml index adca780..aad0f33 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -644,6 +644,37 @@ classifier: fallback_tier: 2 fallback_category: general_chat + + # --- fallback cascade --------------------------------------------------- + # When the local classifier fails, the router walks: stale session cache -> + # this session's history in route_decisions -> the optional cloud + # classifier below -> fallback_tier/fallback_category above. The first two + # steps are free and local. + # + # Global backoff after a classifier failure. This is what bounds cloud + # spend during a sustained outage: at most one cloud attempt per window + # across ALL requests, not one per request. It is also reused as the window + # in which an account-level provider refusal suppresses the cloud step, + # since an out-of-credit account makes that call a guaranteed waste. + cooldown_seconds: 30 + # Degradation warning on /metrics. Silent below degraded_warn_min + # decisions in 24h, because a share computed over a handful of requests is + # noise and a flapping warning is one nobody reads. + degraded_warn_min: 20 + degraded_warn_threshold: 0.5 + # Optional. ABSENT BY DEFAULT, which is what makes the cascade cost + # nothing: with no cloud_fallback block the router degrades straight to the + # static guess. Uncomment and point it at any OpenAI-compatible endpoint to + # trade a little money for a real classification during a local outage. + # A configured api_key_env whose variable is missing degrades to the static + # guess rather than failing the request -- unlike the primary classifier, + # which raises, because this one is only reached when things already broke. + # cloud_fallback: + # base_url: https://api.neuralwatt.com/v1 + # model: deepseek-v4-flash + # api_key_env: NEURALWATT_API_KEY + # timeout_seconds: 2 + # max_output_tokens: 1024 response_format: "json" # ask Ollama to constrain output to valid JSON # The dispatcher appends the authoritative category list from # proficiency.categories to this prompt at call time. Do not enumerate the diff --git a/config/schema.sql b/config/schema.sql index 82f5bfc..de9b6ae 100644 --- a/config/schema.sql +++ b/config/schema.sql @@ -218,7 +218,11 @@ CREATE TABLE IF NOT EXISTS route_decisions ( required_context_tokens INTEGER, confidence REAL, classifier_ms INTEGER, - classification_source TEXT, -- 'classifier' | 'override' | 'fallback' + -- 'classifier' | 'override' | 'fallback' | 'cached' | 'session_stale' + -- | 'session_history' | 'classifier_cloud'. The last three are the + -- degraded cascade steps; 'cached' predates the cascade and was + -- already written by the code but missing from this comment. + classification_source TEXT, latency_tolerance TEXT, -- 'interactive' | 'batch' candidates_considered INTEGER, selected_model TEXT, -- null when nothing was selected @@ -262,6 +266,9 @@ CREATE INDEX IF NOT EXISTS idx_proficiency_category ON proficiency (category); CREATE INDEX IF NOT EXISTS idx_energy_model ON energy_observations (model_id, provider); -- recent_decisions orders by id, but a time-window query benefits from this. CREATE INDEX IF NOT EXISTS idx_route_decisions_observed ON route_decisions (observed_at); +-- The cascade's session-history step looks up a session's last real +-- classification by session_key; without this it scans the table. +CREATE INDEX IF NOT EXISTS idx_route_decisions_session_key ON route_decisions (session_key); -- Guarded — re-applying a fresh schema is a no-op. CREATE INDEX IF NOT EXISTS idx_energy_observed ON energy_observations (observed_at); CREATE INDEX IF NOT EXISTS idx_verifications_observed ON verifications (observed_at); diff --git a/src/config.py b/src/config.py index 5c2b886..cc20b3d 100644 --- a/src/config.py +++ b/src/config.py @@ -673,6 +673,26 @@ class DatabaseConfig(StrictModel): path: str +class CloudFallbackConfig(StrictModel): + """A cloud classifier used ONLY when the local one is unreachable. + + Deliberately its own block rather than reusing ``ClassifierConfig``: the + two differ in the ways that matter under failure. This one has a short + timeout because it sits on the latency floor of every request, and a + missing ``api_key_env`` degrades to the next cascade step instead of + raising — an optional fallback that fails the request would be worse than + the outage it exists to soften. + """ + + base_url: str + model: str + # Short by design: this runs after the local attempt has already spent + # its own timeout, so it adds to a latency budget that is already over. + timeout_seconds: int = 2 + api_key_env: Optional[str] = None + max_output_tokens: int = 1024 + + class ClassifierConfig(StrictModel): provider: str base_url: str @@ -695,7 +715,19 @@ class ClassifierConfig(StrictModel): fallback_category: str = "general_chat" response_format: str system_prompt: str - + # Optional cloud classifier, tried only after the local one and the two + # free session-derived steps have all failed. Absent — the default — + # means the cascade degrades straight to the static fallback and the + # whole feature costs nothing. + cloud_fallback: Optional["CloudFallbackConfig"] = None + # Global backoff after a classifier failure. This is what bounds cloud + # spend during a sustained local outage: at most one cloud attempt per + # window across ALL requests, not one per request. + cooldown_seconds: int = 30 + # Degradation warning: how many classifications the window needs before + # the degraded share means anything, and the share that trips it. + degraded_warn_min: int = 20 + degraded_warn_threshold: float = 0.5 class LocalEnergyConfig(StrictModel): diff --git a/src/dispatcher.py b/src/dispatcher.py index 75b5d7c..37acc2e 100644 --- a/src/dispatcher.py +++ b/src/dispatcher.py @@ -35,6 +35,7 @@ there, still logged. from __future__ import annotations import asyncio +import contextvars import hashlib import json import os @@ -46,7 +47,7 @@ import time from datetime import datetime, timezone from pathlib import Path from statistics import median -from typing import Any, Literal, Optional +from typing import Any, Final, Literal, Optional import requests from dotenv import load_dotenv @@ -238,7 +239,18 @@ class Classification(BaseModel): required_context_tokens: int confidence: float escalated: bool = False - source: Literal["classifier", "override", "fallback", "cached"] = "classifier" + source: Literal[ + "classifier", + "override", + "fallback", + "cached", + # Degraded cascade steps, in the order they are tried. These are + # guesses or borrowed answers, and `report_outcome` refuses to + # attribute their outcomes to a model's proficiency. + "session_stale", + "session_history", + "classifier_cloud", + ] = "classifier" class Candidate(BaseModel): @@ -408,6 +420,14 @@ def ensure_route_decisions(conn: sqlite3.Connection) -> None: "CREATE INDEX IF NOT EXISTS idx_route_decisions_observed " "ON route_decisions (observed_at)" ) + # Mirrors config/schema.sql so a LIVE database gains the index too. The + # cascade's session-history step would otherwise scan route_decisions on + # every local-classifier failure — i.e. exactly when the router is already + # degraded and least able to afford it. IF NOT EXISTS keeps both idempotent. + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_route_decisions_session_key " + "ON route_decisions (session_key)" + ) # A live router.db that predates the flex-preference knob has the table # (from an earlier schema) but not its columns: CREATE TABLE IF NOT EXISTS # no-ops against it, so the new columns must be added by ALTER. PRAGMA @@ -483,6 +503,187 @@ def _classifier_client() -> OpenAI: ) +# --- classifier fallback cascade ------------------------------------------ +# +# When the local classifier fails, the router used to jump straight to a fixed +# tier/category guess. That is cheap and predictable but throws away two free +# signals it already holds, and stamps unrelated work with `general_chat` — +# which is a fully scored category, so the outcomes of that mislabelled +# traffic would otherwise pollute real proficiency (see `report_outcome`). +# +# The cascade is: local -> stale session cache -> session history -> optional +# cloud -> static fallback. Steps 2 and 3 are free and local; step 4 costs +# money and is disabled by default. + +# Timestamp of the last classifier failure. Module-level, so the backoff is +# GLOBAL across requests rather than per-request — that is what bounds cloud +# spend during a sustained outage to one attempt per cooldown window. +_last_classifier_failure: float = 0.0 +# Timestamp of the last account-level provider refusal. If the account is out +# of credit, a cloud classification is guaranteed to fail AND to cost a +# request; skip step 4 entirely for the cooldown window. +_last_account_refusal: float = 0.0 + +# The in-flight request's session key, for the two cascade steps that need it. +# +# A ContextVar rather than a parameter threaded dispatch -> route -> classify, +# for the same reason logs.py carries the trace id this way: request-scoped +# ambient data that only the innermost frame reads. Widening route()'s +# signature for it broke 78 tests that legitimately stub route with a lambda, +# which is the signal that a signature change here is a public API change and +# this is not one. +_current_session_key: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar( + "session_key", default=None +) + + +def _record_failure() -> None: + """Open the circuit after a classifier failure.""" + global _last_classifier_failure + _last_classifier_failure = time.time() + + +def _record_success_cooldown() -> None: + """Close the circuit — the classifier answered, so stop backing off.""" + global _last_classifier_failure + _last_classifier_failure = 0.0 + + +def _record_account_refusal() -> None: + """Note an account-level refusal so the cascade stops paying for cloud.""" + global _last_account_refusal + _last_account_refusal = time.time() + + +def _cloud_classifier_client(cf) -> Optional[OpenAI]: + """Client for the optional cloud classifier, or None if unusable. + + Mirrors `_classifier_client` with one deliberate divergence: a configured + but missing ``api_key_env`` returns None instead of raising 503. The + primary classifier raising is right — a misconfigured primary should fail + loudly. But this one is an OPTIONAL fallback reached only when things are + already broken, and failing the request outright would make the outage + worse than the dumb guess it replaces. + """ + api_key = "ollama" + if cf.api_key_env: + api_key = os.environ.get(cf.api_key_env) or "" + if not api_key: + logs.warning("cloud_classifier_unconfigured", key_env=cf.api_key_env) + return None + return OpenAI(base_url=cf.base_url, api_key=api_key, max_retries=0) + + +def _session_history_lookup(conn, session_key: str) -> Optional[tuple[str, int]]: + """The session's most recent NON-degraded classification, if any. + + Filters degraded sources out deliberately: reading back a previous + fallback would let one outage's guess propagate through every later turn + of the session and look like a real classification forever. + """ + row = conn.execute( + """ + SELECT task_category, task_tier + FROM route_decisions + WHERE session_key = ? + AND task_category IS NOT NULL + AND classification_source IN ('classifier', 'cached', 'override') + ORDER BY observed_at DESC + LIMIT 1 + """, + (session_key,), + ).fetchone() + if row is None: + return None + return (row["task_category"], int(row["task_tier"])) + + +def _classify_cascade( + session_key: Optional[str], + system_prompt: str, + user_content: str, +) -> Optional[Classification]: + """Try every degraded source in order; None means "use the static guess". + + Ordered cheapest-and-most-trustworthy first. Steps 1 and 2 cost nothing + and reuse a real classification; step 3 costs money and is skipped unless + configured, outside the failure cooldown, and the account is not refusing. + """ + _record_failure() + + # 1. This session's own last classification, staleness ignored. A stale + # real answer beats a generic guess. + if session_key: + cached = session_cache.stale_read(session_key) + if cached is not None: + logs.warning("classify_degraded", source="session_stale") + return Classification( + task_category=cached.task_category, + task_tier=cached.task_tier, + required_context_tokens=0, + confidence=0.0, + source="session_stale", + ) + + # 2. The session's history in route_decisions — survives a restart, + # which the in-memory cache does not. + try: + conn = _db() + try: + found = _session_history_lookup(conn, session_key) + finally: + conn.close() + except Exception as exc: # noqa: BLE001 — a DB hiccup falls through + logs.warning("classify_history_lookup_failed", error=str(exc)[:200]) + found = None + if found is not None: + category, tier = found + logs.warning("classify_degraded", source="session_history") + return Classification( + task_category=category, + task_tier=tier, + required_context_tokens=0, + confidence=0.0, + source="session_history", + ) + + # 3. The optional cloud classifier. + cf = getattr(cfg.classifier, "cloud_fallback", None) + if cf is None: + return None + cooldown = cfg.classifier.cooldown_seconds + if time.time() - _last_account_refusal < cooldown: + logs.warning("classify_cloud_skipped", reason="account_refusal") + return None + client = _cloud_classifier_client(cf) + if client is None: + return None + parsed = session_cache.classify_one( + client, + cf.model, + system_prompt, + user_content, + timeout_seconds=cf.timeout_seconds, + max_output_tokens=cf.max_output_tokens, + ) + if not parsed: + logs.warning("classify_cloud_failed") + return None + try: + result = Classification( + task_category=str(parsed["task_category"]), + task_tier=int(parsed["task_tier"]), + required_context_tokens=int(parsed.get("required_context_tokens") or 0), + confidence=float(parsed.get("confidence") or 0.0), + source="classifier_cloud", + ) + except (KeyError, TypeError, ValueError) as exc: + logs.warning("classify_cloud_unparseable", error=str(exc)[:200]) + return None + logs.warning("classify_degraded", source="classifier_cloud") + return result + + def _provider_client(provider: str) -> OpenAI: try: settings = cfg.dispatch_providers[provider] @@ -579,6 +780,11 @@ def classify(task: str, context: Optional[str]) -> Classification: cat=cfg.classifier.fallback_category, ms=_ms(started), ) + degraded = _classify_cascade( + _current_session_key.get(), system_prompt, user_content + ) + if degraded is not None: + return degraded return Classification( task_category=cfg.classifier.fallback_category, task_tier=cfg.classifier.fallback_tier, @@ -603,6 +809,11 @@ def classify(task: str, context: Optional[str]) -> Classification: cat=cfg.classifier.fallback_category, ms=_ms(started), ) + degraded = _classify_cascade( + _current_session_key.get(), system_prompt, user_content + ) + if degraded is not None: + return degraded return Classification( task_category=cfg.classifier.fallback_category, task_tier=cfg.classifier.fallback_tier, @@ -622,6 +833,9 @@ def classify(task: str, context: Optional[str]) -> Classification: chars=len(user_content), ms=_ms(started), ) + # The classifier answered: stop backing off so the next failure gets a + # fresh cooldown window rather than inheriting a stale one. + _record_success_cooldown() return resp @@ -1811,6 +2025,55 @@ def _most_recent_if_unambiguous(conn: sqlite3.Connection): return recent[0] +# Classification sources whose category is a GUESS or a borrowed answer, +# not a measurement of this request. Outcomes on these decisions are recorded +# but must not move proficiency. +DEGRADED_CLASSIFICATION_SOURCES: Final = frozenset( + {"fallback", "session_stale", "session_history"} +) + + +def _outcome_is_attributable(conn, request_id: Optional[str]) -> bool: + """Whether a client outcome may train proficiency for its category. + + The trap this closes: a fallback-classified request is recorded with + task_category=general_chat, and general_chat is a FULLY SCORED category + (13 proficiency rows on the live DB) as well as the configured + fallback_category. So without this, POST /outcome reports on mislabelled + outage traffic fold into proficiency(model, general_chat) via feedback.py + and drag real scores toward whatever happened to be flowing while the + classifier was down. + + Attributable: classifier, cached, override, classifier_cloud — the + category was measured on THIS request, or declared by a client who knows. + Not attributable: fallback (a guess), session_stale and session_history (a + real classification of a DIFFERENT request, already past the staleness + bound the design itself set). + + The asymmetry is deliberate: a degraded source is good enough to route one + visibly-flagged request, but a proficiency score is consulted by every + future request, so attribution must not trust more than routing does. + + FAILS OPEN. An unknown decision row counts as attributable, because + withholding ground truth needs positive knowledge that the category was a + guess. Over-applying an exclusion already starved feedback once — the + client_capped lesson. + """ + if not request_id: + return True + try: + row = conn.execute( + "SELECT classification_source FROM route_decisions " + "WHERE request_id = ? ORDER BY observed_at DESC LIMIT 1", + (request_id,), + ).fetchone() + except Exception: # noqa: BLE001 — a lookup failure must not lose the report + return True + if row is None or row["classification_source"] is None: + return True + return row["classification_source"] not in DEGRADED_CLASSIFICATION_SOURCES + + def _find_outcome_row( conn: sqlite3.Connection, request_id: Optional[str], @@ -1946,6 +2209,7 @@ def report_outcome(report: OutcomeReport): "report on.", ) + attributable = _outcome_is_attributable(conn, row["request_id"]) log_verification( row["model_id"], row["provider"], @@ -1954,6 +2218,7 @@ def report_outcome(report: OutcomeReport): kind="client_outcome", verdict="succeeded" if report.ok else "failed", detail=" | ".join(x for x in (report.detail, report.source) if x)[:300], + model_attributable=attributable, ) logs.info( "outcome", @@ -2992,6 +3257,9 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks): # local-vision paths that never reach the dispatch section, so they are # computed up here rather than late (where they previously lived). session_key = session_fingerprint(messages) + # Publish for the classifier fallback cascade, which needs it two + # frames down without route() having to carry it. + _current_session_key.set(session_key) streamed = bool(body.get("stream")) # Pinch stats are filled by the routed and passthrough prune paths (when @@ -3151,10 +3419,17 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks): profile=profile_name, profile_obj=profile_obj, ) + # `classifier_cloud` is admitted alongside `classifier` because it + # is a REAL classification by a real model — and caching it is what + # bounds cloud cost to one call per session per staleness window + # instead of one per turn. `session_stale` and `session_history` + # are deliberately NOT admitted: they are borrowed answers, and + # writing them back would renew a staleness clock they never + # earned, keeping a degraded guess alive indefinitely. if ( cfg.session_cache.enabled and session_key is not None - and classified_src == "classifier" + and classified_src in ("classifier", "classifier_cloud") ): session_cache.put( session_key, @@ -3467,6 +3742,11 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks): cfg.circuit_breaker.max_cooldown_seconds, cfg.circuit_breaker.backoff_multiplier, ) + if _account_level_refusal(resp.status_code): + # The account is refusing. A cloud CLASSIFICATION would + # fail the same way and still cost a request, so tell the + # cascade to skip step 3 for the cooldown window. + _record_account_refusal() if _account_level_refusal(resp.status_code) or not alternatives: # Both spec §1c triggers: every cloud candidate exhausted, or a # non-retryable account-level refusal — the remaining cloud rows @@ -3767,6 +4047,9 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks): upstream_text = attempt.text[:500] attempt.close() if _account_level_refusal(attempt.status_code): + # Same reason as the streaming path: an account-level refusal + # makes the cloud classifier a guaranteed waste of a request. + _record_account_refusal() break if upstream is None: fallback = try_local_fallback(target, upstream_status) diff --git a/src/metrics.py b/src/metrics.py index 3f0054a..a33a29b 100644 --- a/src/metrics.py +++ b/src/metrics.py @@ -413,6 +413,7 @@ def scoring_coverage( cap_ctx = capability_ceilings(conn, cfg) warnings.extend(capability_demand_warnings(conn, cfg, cap_ctx)) warnings.extend(rejection_warnings(conn, cfg)) + warnings.extend(classifier_degradation_warning(conn, cfg)) return { "routable_models": total, @@ -773,6 +774,51 @@ NOVEL_GROUP_MIN_COUNT: Final = 2 # window; widen objective.rejection_warning_baseline_hours to suppress it. + +def classifier_degradation_warning(conn: sqlite3.Connection, cfg: Any) -> list[str]: + """Warn when most recent classifications came from a degraded source. + + The cascade makes a local-classifier outage survivable, which is the + point — but survivable failures are the ones that go unnoticed for weeks. + Routing keeps working on borrowed categories while nothing says the + classifier has been down since Tuesday. + + Silent below ``degraded_warn_min`` decisions: a share computed over three + requests is noise, and a warning that flaps on low traffic is one nobody + reads. + """ + min_total = getattr(cfg.classifier, "degraded_warn_min", 20) + threshold = getattr(cfg.classifier, "degraded_warn_threshold", 0.5) + row = conn.execute( + """ + SELECT + COUNT(*) AS total, + SUM( + CASE WHEN classification_source IN + ('fallback', 'session_stale', 'session_history', + 'classifier_cloud') + THEN 1 ELSE 0 END + ) AS degraded + FROM route_decisions + WHERE classification_source IS NOT NULL + AND observed_at >= datetime('now', '-24 hours') + """ + ).fetchone() + total = (row["total"] if row else 0) or 0 + degraded = (row["degraded"] if row else 0) or 0 + if total < min_total: + return [] + share = degraded / total + if share < threshold: + return [] + return [ + f"{share:.0%} of the last {total} classifications came from a degraded " + f"source ({degraded} of {total}) — the local classifier has been " + f"failing. Routing still works on borrowed categories, but their " + f"outcomes are excluded from proficiency." + ] + + def rejection_warnings( conn: sqlite3.Connection, cfg: Any, diff --git a/src/session_cache.py b/src/session_cache.py index e9e826e..58044ad 100644 --- a/src/session_cache.py +++ b/src/session_cache.py @@ -5,9 +5,10 @@ lifetime matches the dispatcher process (a restart just reclassifies each session once — a safe failure mode). Never imports dispatcher or config. """ +import json import time from dataclasses import dataclass -from typing import Final, Optional +from typing import Any, Final, Optional # session_key -> CachedClassification _store: Final[dict[str, "CachedClassification"]] = {} @@ -48,6 +49,57 @@ def put(session_key: str, task_category: str, task_tier: int) -> None: ) +def stale_read(session_key: str) -> Optional[CachedClassification]: + """Return a session's cached classification IGNORING staleness. + + Deliberately separate from ``get`` rather than a flag on it, because the + two answer different questions. ``get`` asks "may I reuse this instead of + classifying?", where staleness is the whole point. ``stale_read`` asks + "the classifier is DOWN — do we know anything at all about this session?", + and there a stale real classification beats a generic guess. + + Never refreshes ``cached_at``: a degraded read must not renew the window + that a healthy write earned. + """ + return _store.get(session_key) + + +def classify_one( + client: Any, + model: str, + system_prompt: str, + user_content: str, + *, + timeout_seconds: float, + max_output_tokens: int, +) -> Optional[dict]: + """One classification attempt against an OpenAI-compatible client. + + Returns the parsed JSON object, or ``None`` on ANY failure — network, + timeout, refusal, unparseable body. Returning None rather than raising is + what lets the caller keep walking its cascade instead of turning a + fallback into an error of its own. + + Returns a plain dict so this module stays what its docstring claims: no + project imports. The caller builds its own typed object from the result. + """ + try: + resp = client.chat.completions.create( + model=model, + messages=[ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": user_content}, + ], + temperature=0.0, + max_tokens=max_output_tokens, + timeout=timeout_seconds, + ) + parsed = json.loads(resp.choices[0].message.content) + except Exception: # noqa: BLE001 — any failure means "try the next step" + return None + return parsed if isinstance(parsed, dict) else None + + def clear() -> None: """Drop all cached classifications (test isolation / reset helper).""" _store.clear() diff --git a/tests/test_classifier_cascade.py b/tests/test_classifier_cascade.py new file mode 100644 index 0000000..5a0accc --- /dev/null +++ b/tests/test_classifier_cascade.py @@ -0,0 +1,271 @@ +"""The classifier fallback cascade: local -> stale -> history -> cloud -> guess. + +Each step is asserted by making the ones before it fail, so a test that +passes proves that step actually ran rather than that some earlier step +happened to produce the same answer. +""" +from __future__ import annotations + +import sqlite3 +from datetime import datetime, timedelta, timezone +from pathlib import Path +from types import SimpleNamespace + +import pytest + +import dispatcher +import session_cache + +ROOT = Path(__file__).resolve().parent.parent +SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text() + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +@pytest.fixture(autouse=True) +def _clean_state(monkeypatch): + """Each test starts with an empty cache and a closed circuit.""" + session_cache.clear() + monkeypatch.setattr(dispatcher, "_last_classifier_failure", 0.0) + monkeypatch.setattr(dispatcher, "_last_account_refusal", 0.0) + yield + session_cache.clear() + + +@pytest.fixture +def db(tmp_path, monkeypatch): + path = tmp_path / "cascade.db" + conn = sqlite3.connect(path) + conn.row_factory = sqlite3.Row + conn.executescript(SCHEMA_SQL) + conn.commit() + + def _open(): + c = sqlite3.connect(path) + c.row_factory = sqlite3.Row + return c + + monkeypatch.setattr(dispatcher, "_db", _open) + yield conn + conn.close() + + +def test_step1_stale_session_beats_a_generic_guess(): + """A stale REAL classification is used before falling back to a guess.""" + session_cache.put("sess-1", task_category="coding_refactor", task_tier=3) + got = dispatcher._classify_cascade("sess-1", "sys", "user") + assert got is not None + assert got.source == "session_stale" + assert got.task_category == "coding_refactor" + assert got.task_tier == 3 + + +def test_step1_ignores_staleness_but_never_renews_it(): + """stale_read must not refresh cached_at — a degraded read earns nothing.""" + session_cache.put("sess-2", task_category="debugging", task_tier=2) + before = session_cache.stale_read("sess-2").cached_at + dispatcher._classify_cascade("sess-2", "sys", "user") + assert session_cache.stale_read("sess-2").cached_at == before + + +def test_step2_session_history_survives_a_restart(db): + """With an empty cache, the session's own history is read from the DB. + + This is the step that matters after a router restart, when the in-memory + cache is gone but route_decisions still remembers the session. + """ + db.execute( + """ + INSERT INTO route_decisions + (kind, task_category, task_tier, required_context_tokens, + selected_model, selected_provider, classification_source, + session_key, observed_at) + VALUES ('chat','translation',1,50,'m','neuralwatt','classifier',?,?) + """, + ("sess-3", _now().isoformat()), + ) + db.commit() + got = dispatcher._classify_cascade("sess-3", "sys", "user") + assert got is not None + assert got.source == "session_history" + assert got.task_category == "translation" + + +def test_step2_refuses_to_reuse_a_previous_degraded_answer(db): + """A prior fallback must not be read back as if it were real. + + Otherwise one outage's guess propagates through every later turn of the + session and looks like a genuine classification forever. + """ + for src in ("fallback", "session_stale", "session_history"): + db.execute( + """ + INSERT INTO route_decisions + (kind, task_category, task_tier, required_context_tokens, + selected_model, selected_provider, classification_source, + session_key, observed_at) + VALUES ('chat','general_chat',2,0,'m','neuralwatt',?,?,?) + """, + (src, "sess-4", _now().isoformat()), + ) + db.commit() + # No cloud configured -> None means "use the static fallback". + assert dispatcher._classify_cascade("sess-4", "sys", "user") is None + + +def test_cloud_step_is_skipped_when_unconfigured(db): + """Zero cost unless someone explicitly configures cloud_fallback.""" + assert getattr(dispatcher.cfg.classifier, "cloud_fallback", None) is None + assert dispatcher._classify_cascade("sess-none", "sys", "user") is None + + +def test_cloud_step_used_when_configured(db, monkeypatch): + """A configured cloud classifier answers when everything free has failed.""" + monkeypatch.setattr( + dispatcher.cfg.classifier, + "cloud_fallback", + SimpleNamespace( + base_url="https://cloud.example/v1", + model="cloud-classifier", + timeout_seconds=2, + api_key_env=None, + max_output_tokens=1024, + ), + ) + monkeypatch.setattr(dispatcher, "_cloud_classifier_client", lambda cf: object()) + monkeypatch.setattr( + session_cache, + "classify_one", + lambda *a, **k: { + "task_category": "reasoning_math", + "task_tier": 3, + "required_context_tokens": 1234, + "confidence": 0.8, + }, + ) + got = dispatcher._classify_cascade("sess-5", "sys", "user") + assert got is not None + assert got.source == "classifier_cloud" + assert got.task_category == "reasoning_math" + assert got.required_context_tokens == 1234 + + +def test_account_refusal_skips_the_cloud_step(db, monkeypatch): + """Out of credit -> a cloud classification is a guaranteed wasted request.""" + monkeypatch.setattr( + dispatcher.cfg.classifier, + "cloud_fallback", + SimpleNamespace( + base_url="https://cloud.example/v1", + model="cloud-classifier", + timeout_seconds=2, + api_key_env=None, + max_output_tokens=1024, + ), + ) + called = [] + monkeypatch.setattr( + dispatcher, + "_cloud_classifier_client", + lambda cf: called.append(1) or object(), + ) + dispatcher._record_account_refusal() + assert dispatcher._classify_cascade("sess-6", "sys", "user") is None + assert called == [], "cloud client built despite an account-level refusal" + + +def test_cloud_failure_degrades_to_the_static_guess(db, monkeypatch): + """An unreachable cloud classifier must not raise — it returns None.""" + monkeypatch.setattr( + dispatcher.cfg.classifier, + "cloud_fallback", + SimpleNamespace( + base_url="https://cloud.example/v1", + model="cloud-classifier", + timeout_seconds=2, + api_key_env=None, + max_output_tokens=1024, + ), + ) + monkeypatch.setattr(dispatcher, "_cloud_classifier_client", lambda cf: object()) + monkeypatch.setattr(session_cache, "classify_one", lambda *a, **k: None) + assert dispatcher._classify_cascade("sess-7", "sys", "user") is None + + +def test_missing_api_key_degrades_rather_than_raising(): + """Unlike the primary classifier, a keyless cloud fallback must not 503. + + The primary raising is right — a misconfigured primary should fail loudly. + This one is reached only when things are already broken, so failing the + request would make the outage worse than the guess it replaces. + """ + cf = SimpleNamespace( + base_url="https://cloud.example/v1", + model="cloud-classifier", + timeout_seconds=2, + api_key_env="DEFINITELY_NOT_SET_ANYWHERE", + max_output_tokens=1024, + ) + assert dispatcher._cloud_classifier_client(cf) is None + + +def test_cascade_with_no_session_key_falls_straight_through(db): + """No session -> steps 1 and 2 are impossible, and that is not an error.""" + assert dispatcher._classify_cascade(None, "sys", "user") is None + + +# --------------------------------------------------------------------------- +# The degradation warning: a survivable failure is the kind that goes unnoticed. +# --------------------------------------------------------------------------- + + +def _seed_sources(conn, pairs): + for source, count in pairs: + for _ in range(count): + conn.execute( + """ + INSERT INTO route_decisions + (kind, task_category, task_tier, required_context_tokens, + selected_model, selected_provider, classification_source, + observed_at) + VALUES ('chat','general_chat',2,0,'m','neuralwatt',?,?) + """, + (source, _now().isoformat()), + ) + conn.commit() + + +def test_degradation_warning_fires_above_threshold(db): + from metrics import classifier_degradation_warning + + cfg = SimpleNamespace( + classifier=SimpleNamespace(degraded_warn_min=20, degraded_warn_threshold=0.5) + ) + _seed_sources(db, [("fallback", 15), ("classifier", 10)]) + out = classifier_degradation_warning(db, cfg) + assert len(out) == 1 + assert "60%" in out[0] + assert "degraded source" in out[0] + + +def test_degradation_warning_silent_on_thin_traffic(db): + """A share over three requests is noise; a flapping warning goes unread.""" + from metrics import classifier_degradation_warning + + cfg = SimpleNamespace( + classifier=SimpleNamespace(degraded_warn_min=20, degraded_warn_threshold=0.5) + ) + _seed_sources(db, [("fallback", 3)]) + assert classifier_degradation_warning(db, cfg) == [] + + +def test_degradation_warning_silent_when_healthy(db): + from metrics import classifier_degradation_warning + + cfg = SimpleNamespace( + classifier=SimpleNamespace(degraded_warn_min=20, degraded_warn_threshold=0.5) + ) + _seed_sources(db, [("classifier", 24), ("fallback", 1)]) + assert classifier_degradation_warning(db, cfg) == [] diff --git a/tests/test_outcome_attribution.py b/tests/test_outcome_attribution.py index c456517..cccf05a 100644 --- a/tests/test_outcome_attribution.py +++ b/tests/test_outcome_attribution.py @@ -154,3 +154,75 @@ def test_quota_burn_counts_only_the_last_thirty_days(db, tmp_path, monkeypatch): db.commit() assert quota_burn(db, dispatcher.cfg)["metered_kwh_30d"] == pytest.approx(0.25) + + +# --------------------------------------------------------------------------- +# Degraded classifications must not train proficiency. +# --------------------------------------------------------------------------- +# +# A fallback-classified request is recorded with task_category=general_chat, +# and general_chat is a fully scored category AND the configured +# fallback_category. Without this guard, outcome reports on mislabelled outage +# traffic fold into proficiency(model, general_chat) via feedback.py and drag +# real scores toward whatever was flowing while the classifier was down. + + +def _decision_with_source(conn, request_id: str, source: str) -> None: + conn.execute( + """ + INSERT INTO route_decisions + (kind, task_category, task_tier, required_context_tokens, + selected_model, selected_provider, classification_source, + request_id, observed_at) + VALUES ('chat','general_chat',2,100,'m','neuralwatt',?,?,?) + """, + (source, request_id, _now().isoformat()), + ) + conn.commit() + + +@pytest.mark.parametrize("source", ["fallback", "session_stale", "session_history"]) +def test_degraded_source_outcome_is_not_attributable(db, source): + """A guessed or borrowed category must not move proficiency.""" + _decision_with_source(db, "rid-degraded", source) + assert dispatcher._outcome_is_attributable(db, "rid-degraded") is False + + +@pytest.mark.parametrize( + "source", ["classifier", "cached", "override", "classifier_cloud"] +) +def test_trusted_source_outcome_is_attributable(db, source): + """A measured category still trains proficiency, cloud included. + + classifier_cloud is a real classification by a real model — degrading its + attribution too would throw away good ground truth. + """ + _decision_with_source(db, "rid-trusted", source) + assert dispatcher._outcome_is_attributable(db, "rid-trusted") is True + + +def test_unknown_decision_row_fails_open(db): + """No matching decision -> attributable. + + Withholding ground truth needs positive knowledge the category was a + guess. Over-applying an exclusion already starved feedback once (the + client_capped lesson), so the default direction is to keep the signal. + + NOTE: this test passes on unmodified main by design — it pins the + fail-open rule against a future over-correction, not a fixed defect. + """ + assert dispatcher._outcome_is_attributable(db, "rid-does-not-exist") is True + assert dispatcher._outcome_is_attributable(db, None) is True + + +def test_degraded_sources_match_the_cascade_steps(): + """The exclusion set and the cascade's degraded steps cannot drift apart. + + If a new degraded source is added to Classification without adding it + here, its outcomes silently start training proficiency again. + """ + literal_sources = set( + dispatcher.Classification.model_fields["source"].annotation.__args__ + ) + trusted = {"classifier", "override", "cached", "classifier_cloud"} + assert dispatcher.DEGRADED_CLASSIFICATION_SOURCES == literal_sources - trusted diff --git a/tests/test_route_decisions.py b/tests/test_route_decisions.py index d22cec1..db928fa 100644 --- a/tests/test_route_decisions.py +++ b/tests/test_route_decisions.py @@ -100,7 +100,10 @@ def _schema_minus_route_decisions() -> str: if stripped == ");": skipping = False continue - if "idx_route_decisions_observed" in line: + # Drop EVERY index on the table, not one named index: a hand-kept + # list of names drifts the moment another index is added, and the + # failure is an opaque "no such table" from inside executescript. + if "ON route_decisions" in line or "idx_route_decisions" in line: continue lines.append(line) return "\n".join(lines) diff --git a/tests/test_tui_warnings.py b/tests/test_tui_warnings.py index be8416d..965d129 100644 --- a/tests/test_tui_warnings.py +++ b/tests/test_tui_warnings.py @@ -53,6 +53,7 @@ WARN_CLASS_MATCHERS = { "capability-subceiling": r"(vision|json_mode)-capable tier", "rejection-new-pattern": r"new rejection pattern:", "rejection-rate": r"rejection rate:", + "classifier-degraded": r"came from a degraded source", } CFG = SimpleNamespace( @@ -64,6 +65,7 @@ CFG = SimpleNamespace( ), objective=SimpleNamespace(plan_kwh_per_period=6.25, billing_reset_day=None), escalation=SimpleNamespace(enabled=True), + classifier=SimpleNamespace(degraded_warn_min=20, degraded_warn_threshold=0.5), ) @@ -191,6 +193,14 @@ def _seed_chernobyl(conn: sqlite3.Connection) -> None: selected_provider=None, rejected_reason=familiar, ) + # 20 fallback-classified decisions: 100% degraded over the minimum count, + # firing the classifier-degradation class. Two details keep them from + # disturbing the classes seeded above. selected_model is set, so they do + # not read as rejections. And their context MATCHES the demand row rather + # than being small — 20 small rows drag the tier-1 p95 down far enough to + # silence the escalation hazard, which is how this fixture first broke. + for _ in range(20): + _decision(classification_source="fallback", required_context_tokens=999999) conn.commit()