From 2b91f114dc399dd69f58f9e8096803828458bb60 Mon Sep 17 00:00:00 2001 From: adlee-was-taken Date: Sat, 5 Sep 2026 02:40:07 -0400 Subject: [PATCH 1/4] feat(classify): cascading fallback instead of a dumb static guess When the local classifier failed, the router jumped straight to a fixed tier/category. Cheap and predictable, but it discards two signals the router already holds, and stamps unrelated work with general_chat. 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, so the feature is zero-cost until someone configures cloud_fallback. Three things bound the cost of step 4: - a global cooldown (default 30s) after any classifier failure, so a sustained outage buys at most one cloud attempt per window across ALL requests rather than one per request; - an account-refusal gate wired at both _account_level_refusal sites: if the account is out of credit, a cloud classification is a guaranteed waste of a request, so the step is skipped entirely; - the session_cache.put() gate now admits classifier_cloud, so a session pays for at most one cloud call per staleness window instead of one per turn. session_stale/session_history stay OUT of the cache: they are borrowed answers and must not renew a staleness clock they never earned. _session_history_lookup filters degraded sources, so a previous outage's guess cannot propagate through every later turn and masquerade as a real classification. Two deviations from the plan, both forced by the code: session_key travels in a ContextVar rather than threaded through dispatch -> route -> classify. Adding the kwarg to route() broke 78 tests that legitimately stub it with a lambda -- that is the signal that a signature change here is a public API change, and this is not one. logs.py already carries the trace id exactly this way. session_cache.classify_one returns a plain dict rather than a Classification. The plan asked for it in session_cache AND for that module to stay import-free; returning a dispatcher type would violate the second. The caller builds the typed object. Also makes _schema_minus_route_decisions strip every index on the table rather than one by name -- the hand-kept name list drifted the moment this change added an index, failing with an opaque "no such table". Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01VRQXz5SYZYVWscxS1QqF6U --- config/schema.sql | 9 +- src/config.py | 34 ++++- src/dispatcher.py | 236 +++++++++++++++++++++++++++++++++- src/session_cache.py | 54 +++++++- tests/test_route_decisions.py | 5 +- 5 files changed, 332 insertions(+), 6 deletions(-) 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..ea982ab 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 @@ -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 @@ -2992,6 +3206,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 +3368,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 +3691,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 +3996,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/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_route_decisions.py b/tests/test_route_decisions.py index 3588f59..b1639c1 100644 --- a/tests/test_route_decisions.py +++ b/tests/test_route_decisions.py @@ -97,7 +97,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) -- 2.49.1 From 2afdb40e2b58d57991e49d36fa7292640f8202b0 Mon Sep 17 00:00:00 2001 From: adlee-was-taken Date: Sat, 5 Sep 2026 02:43:39 -0400 Subject: [PATCH 2/4] fix(outcome): degraded classifications must not train proficiency A fallback-classified request is recorded with task_category=general_chat. general_chat is a FULLY SCORED category -- 13 proficiency rows on the live DB, kimi-k2.7-code-fast at 0.95 -- and it is also the configured fallback_category. So without a guard, POST /outcome reports on mislabelled outage traffic fold into proficiency(model, general_chat) through feedback.py and drag real scores toward whatever traffic happened to be flowing while the classifier was down. That pollutes the one table the router treats as ground truth. report_outcome now resolves the decision's classification_source and passes model_attributable accordingly. feedback.py is NOT touched: its existing `AND model_attributable = 1` filter already implements keep-the-record, don't-steer-routing -- the same treatment the tool-call false-failures got. 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; a proficiency score is consulted by every future request, so attribution must not trust more than routing does. FAILS OPEN on an unknown decision row. Withholding ground truth requires positive knowledge the category was a guess, and an over-applied exclusion already starved feedback once (client_capped). The test pinning that passes on main by design -- it guards a future over-correction, not a fixed defect. Also adds a drift test: the exclusion set must equal the Classification source Literal minus the trusted four, so a new degraded source cannot be added without deciding its attribution. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01VRQXz5SYZYVWscxS1QqF6U --- src/dispatcher.py | 53 ++++++++++++++++++++++- tests/test_outcome_attribution.py | 72 +++++++++++++++++++++++++++++++ 2 files changed, 124 insertions(+), 1 deletion(-) diff --git a/src/dispatcher.py b/src/dispatcher.py index ea982ab..37acc2e 100644 --- a/src/dispatcher.py +++ b/src/dispatcher.py @@ -47,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 @@ -2025,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], @@ -2160,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"], @@ -2168,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", 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 -- 2.49.1 From 9e4ca526022b5f18307e2fe281ae3d5b9d21fc86 Mon Sep 17 00:00:00 2001 From: adlee-was-taken Date: Sat, 5 Sep 2026 02:48:39 -0400 Subject: [PATCH 3/4] feat(metrics): classifier degradation warning, and document the knobs The cascade makes a local-classifier outage survivable, which is the point -- but survivable failures are exactly the ones that go unnoticed. Routing keeps working on borrowed categories while nothing says the classifier has been down since Tuesday. classifier_degradation_warning reports the share of the last 24h that came from a degraded source. Silent below degraded_warn_min decisions: a share computed over a handful of requests is noise, and a warning that flaps on low traffic is one nobody reads. Also documents cooldown_seconds, degraded_warn_min, degraded_warn_threshold and a commented cloud_fallback block in config/config.yaml. CLAUDE.md is explicit that a knob living only as a Pydantic default is invisible to anyone tuning it -- and cloud_fallback in particular needs to be visible precisely because it is absent by default. Adds tests/test_classifier_cascade.py: each cascade step asserted by making the earlier ones fail, so a passing test proves that step ran rather than that an earlier one produced the same answer. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01VRQXz5SYZYVWscxS1QqF6U --- config/config.yaml | 31 ++++ src/metrics.py | 46 ++++++ tests/test_classifier_cascade.py | 271 +++++++++++++++++++++++++++++++ 3 files changed, 348 insertions(+) create mode 100644 tests/test_classifier_cascade.py 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/src/metrics.py b/src/metrics.py index 0d948b9..ff62d56 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/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) == [] -- 2.49.1 From 9b0712f12eacae94e3041671cb405474fd2bba6d Mon Sep 17 00:00:00 2001 From: adlee-was-taken Date: Sat, 5 Sep 2026 11:41:19 -0400 Subject: [PATCH 4/4] test(tui): register the classifier-degradation warning class PR #33's tripwire fails on any warning class scoring_coverage emits that nothing displays. This branch adds one, so the tripwire fired on merge -- working exactly as designed. The fix was three parts, not the one line predicted: 1. the matcher, "came from a degraded source"; 2. the CFG stub gains a `classifier` attribute. It never had one, so the failure was an AttributeError rather than an unregistered-warning report. The real RouterConfig always has .classifier -- the minimal stub was simply unfaithful; 3. the chernobyl fixture seeds 20 fallback-classified decisions so the class actually fires. Step 3 needed a correction of its own. Seeding those rows with a SMALL context silenced the escalation-hazard class, because 20 low rows drag the tier-1 p95 below the tier-2 ceiling. They now carry the same context as the demand row, leaving the p95 unmoved. Worth recording: the chernobyl fixture is a coupled system. A new seed can silence an existing class, and only the "every registered class fires" direction catches it -- which is the argument for keeping both directions. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01VRQXz5SYZYVWscxS1QqF6U --- tests/test_tui_warnings.py | 10 ++++++++++ 1 file changed, 10 insertions(+) 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() -- 2.49.1