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

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);

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):

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)

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,

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

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) == []

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

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)

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