feat(classify): cascading fallback instead of a dumb static guess #34
@@ -644,6 +644,37 @@ classifier:
|
|||||||
|
|
||||||
fallback_tier: 2
|
fallback_tier: 2
|
||||||
fallback_category: general_chat
|
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
|
response_format: "json" # ask Ollama to constrain output to valid JSON
|
||||||
# The dispatcher appends the authoritative category list from
|
# The dispatcher appends the authoritative category list from
|
||||||
# proficiency.categories to this prompt at call time. Do not enumerate the
|
# proficiency.categories to this prompt at call time. Do not enumerate the
|
||||||
|
|||||||
@@ -218,7 +218,11 @@ CREATE TABLE IF NOT EXISTS route_decisions (
|
|||||||
required_context_tokens INTEGER,
|
required_context_tokens INTEGER,
|
||||||
confidence REAL,
|
confidence REAL,
|
||||||
classifier_ms INTEGER,
|
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'
|
latency_tolerance TEXT, -- 'interactive' | 'batch'
|
||||||
candidates_considered INTEGER,
|
candidates_considered INTEGER,
|
||||||
selected_model TEXT, -- null when nothing was selected
|
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);
|
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.
|
-- 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);
|
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.
|
-- 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_energy_observed ON energy_observations (observed_at);
|
||||||
CREATE INDEX IF NOT EXISTS idx_verifications_observed ON verifications (observed_at);
|
CREATE INDEX IF NOT EXISTS idx_verifications_observed ON verifications (observed_at);
|
||||||
|
|||||||
@@ -673,6 +673,26 @@ class DatabaseConfig(StrictModel):
|
|||||||
path: str
|
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):
|
class ClassifierConfig(StrictModel):
|
||||||
provider: str
|
provider: str
|
||||||
base_url: str
|
base_url: str
|
||||||
@@ -695,7 +715,19 @@ class ClassifierConfig(StrictModel):
|
|||||||
fallback_category: str = "general_chat"
|
fallback_category: str = "general_chat"
|
||||||
response_format: str
|
response_format: str
|
||||||
system_prompt: 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):
|
class LocalEnergyConfig(StrictModel):
|
||||||
|
|||||||
@@ -35,6 +35,7 @@ there, still logged.
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import contextvars
|
||||||
import hashlib
|
import hashlib
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
@@ -46,7 +47,7 @@ import time
|
|||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from statistics import median
|
from statistics import median
|
||||||
from typing import Any, Literal, Optional
|
from typing import Any, Final, Literal, Optional
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
from dotenv import load_dotenv
|
from dotenv import load_dotenv
|
||||||
@@ -238,7 +239,18 @@ class Classification(BaseModel):
|
|||||||
required_context_tokens: int
|
required_context_tokens: int
|
||||||
confidence: float
|
confidence: float
|
||||||
escalated: bool = False
|
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):
|
class Candidate(BaseModel):
|
||||||
@@ -408,6 +420,14 @@ def ensure_route_decisions(conn: sqlite3.Connection) -> None:
|
|||||||
"CREATE INDEX IF NOT EXISTS idx_route_decisions_observed "
|
"CREATE INDEX IF NOT EXISTS idx_route_decisions_observed "
|
||||||
"ON route_decisions (observed_at)"
|
"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
|
# 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
|
# (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
|
# 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:
|
def _provider_client(provider: str) -> OpenAI:
|
||||||
try:
|
try:
|
||||||
settings = cfg.dispatch_providers[provider]
|
settings = cfg.dispatch_providers[provider]
|
||||||
@@ -579,6 +780,11 @@ def classify(task: str, context: Optional[str]) -> Classification:
|
|||||||
cat=cfg.classifier.fallback_category,
|
cat=cfg.classifier.fallback_category,
|
||||||
ms=_ms(started),
|
ms=_ms(started),
|
||||||
)
|
)
|
||||||
|
degraded = _classify_cascade(
|
||||||
|
_current_session_key.get(), system_prompt, user_content
|
||||||
|
)
|
||||||
|
if degraded is not None:
|
||||||
|
return degraded
|
||||||
return Classification(
|
return Classification(
|
||||||
task_category=cfg.classifier.fallback_category,
|
task_category=cfg.classifier.fallback_category,
|
||||||
task_tier=cfg.classifier.fallback_tier,
|
task_tier=cfg.classifier.fallback_tier,
|
||||||
@@ -603,6 +809,11 @@ def classify(task: str, context: Optional[str]) -> Classification:
|
|||||||
cat=cfg.classifier.fallback_category,
|
cat=cfg.classifier.fallback_category,
|
||||||
ms=_ms(started),
|
ms=_ms(started),
|
||||||
)
|
)
|
||||||
|
degraded = _classify_cascade(
|
||||||
|
_current_session_key.get(), system_prompt, user_content
|
||||||
|
)
|
||||||
|
if degraded is not None:
|
||||||
|
return degraded
|
||||||
return Classification(
|
return Classification(
|
||||||
task_category=cfg.classifier.fallback_category,
|
task_category=cfg.classifier.fallback_category,
|
||||||
task_tier=cfg.classifier.fallback_tier,
|
task_tier=cfg.classifier.fallback_tier,
|
||||||
@@ -622,6 +833,9 @@ def classify(task: str, context: Optional[str]) -> Classification:
|
|||||||
chars=len(user_content),
|
chars=len(user_content),
|
||||||
ms=_ms(started),
|
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
|
return resp
|
||||||
|
|
||||||
|
|
||||||
@@ -1811,6 +2025,55 @@ def _most_recent_if_unambiguous(conn: sqlite3.Connection):
|
|||||||
return recent[0]
|
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(
|
def _find_outcome_row(
|
||||||
conn: sqlite3.Connection,
|
conn: sqlite3.Connection,
|
||||||
request_id: Optional[str],
|
request_id: Optional[str],
|
||||||
@@ -1946,6 +2209,7 @@ def report_outcome(report: OutcomeReport):
|
|||||||
"report on.",
|
"report on.",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
attributable = _outcome_is_attributable(conn, row["request_id"])
|
||||||
log_verification(
|
log_verification(
|
||||||
row["model_id"],
|
row["model_id"],
|
||||||
row["provider"],
|
row["provider"],
|
||||||
@@ -1954,6 +2218,7 @@ def report_outcome(report: OutcomeReport):
|
|||||||
kind="client_outcome",
|
kind="client_outcome",
|
||||||
verdict="succeeded" if report.ok else "failed",
|
verdict="succeeded" if report.ok else "failed",
|
||||||
detail=" | ".join(x for x in (report.detail, report.source) if x)[:300],
|
detail=" | ".join(x for x in (report.detail, report.source) if x)[:300],
|
||||||
|
model_attributable=attributable,
|
||||||
)
|
)
|
||||||
logs.info(
|
logs.info(
|
||||||
"outcome",
|
"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
|
# local-vision paths that never reach the dispatch section, so they are
|
||||||
# computed up here rather than late (where they previously lived).
|
# computed up here rather than late (where they previously lived).
|
||||||
session_key = session_fingerprint(messages)
|
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"))
|
streamed = bool(body.get("stream"))
|
||||||
|
|
||||||
# Pinch stats are filled by the routed and passthrough prune paths (when
|
# 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=profile_name,
|
||||||
profile_obj=profile_obj,
|
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 (
|
if (
|
||||||
cfg.session_cache.enabled
|
cfg.session_cache.enabled
|
||||||
and session_key is not None
|
and session_key is not None
|
||||||
and classified_src == "classifier"
|
and classified_src in ("classifier", "classifier_cloud")
|
||||||
):
|
):
|
||||||
session_cache.put(
|
session_cache.put(
|
||||||
session_key,
|
session_key,
|
||||||
@@ -3467,6 +3742,11 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
|
|||||||
cfg.circuit_breaker.max_cooldown_seconds,
|
cfg.circuit_breaker.max_cooldown_seconds,
|
||||||
cfg.circuit_breaker.backoff_multiplier,
|
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:
|
if _account_level_refusal(resp.status_code) or not alternatives:
|
||||||
# Both spec §1c triggers: every cloud candidate exhausted, or a
|
# Both spec §1c triggers: every cloud candidate exhausted, or a
|
||||||
# non-retryable account-level refusal — the remaining cloud rows
|
# 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]
|
upstream_text = attempt.text[:500]
|
||||||
attempt.close()
|
attempt.close()
|
||||||
if _account_level_refusal(attempt.status_code):
|
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
|
break
|
||||||
if upstream is None:
|
if upstream is None:
|
||||||
fallback = try_local_fallback(target, upstream_status)
|
fallback = try_local_fallback(target, upstream_status)
|
||||||
|
|||||||
@@ -413,6 +413,7 @@ def scoring_coverage(
|
|||||||
cap_ctx = capability_ceilings(conn, cfg)
|
cap_ctx = capability_ceilings(conn, cfg)
|
||||||
warnings.extend(capability_demand_warnings(conn, cfg, cap_ctx))
|
warnings.extend(capability_demand_warnings(conn, cfg, cap_ctx))
|
||||||
warnings.extend(rejection_warnings(conn, cfg))
|
warnings.extend(rejection_warnings(conn, cfg))
|
||||||
|
warnings.extend(classifier_degradation_warning(conn, cfg))
|
||||||
|
|
||||||
return {
|
return {
|
||||||
"routable_models": total,
|
"routable_models": total,
|
||||||
@@ -773,6 +774,51 @@ NOVEL_GROUP_MIN_COUNT: Final = 2
|
|||||||
# window; widen objective.rejection_warning_baseline_hours to suppress it.
|
# 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(
|
def rejection_warnings(
|
||||||
conn: sqlite3.Connection,
|
conn: sqlite3.Connection,
|
||||||
cfg: Any,
|
cfg: Any,
|
||||||
|
|||||||
@@ -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.
|
session once — a safe failure mode). Never imports dispatcher or config.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
import json
|
||||||
import time
|
import time
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Final, Optional
|
from typing import Any, Final, Optional
|
||||||
|
|
||||||
# session_key -> CachedClassification
|
# session_key -> CachedClassification
|
||||||
_store: Final[dict[str, "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:
|
def clear() -> None:
|
||||||
"""Drop all cached classifications (test isolation / reset helper)."""
|
"""Drop all cached classifications (test isolation / reset helper)."""
|
||||||
_store.clear()
|
_store.clear()
|
||||||
|
|||||||
271
tests/test_classifier_cascade.py
Normal file
271
tests/test_classifier_cascade.py
Normal file
@@ -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) == []
|
||||||
@@ -154,3 +154,75 @@ def test_quota_burn_counts_only_the_last_thirty_days(db, tmp_path, monkeypatch):
|
|||||||
db.commit()
|
db.commit()
|
||||||
|
|
||||||
assert quota_burn(db, dispatcher.cfg)["metered_kwh_30d"] == pytest.approx(0.25)
|
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
|
||||||
|
|||||||
@@ -100,7 +100,10 @@ def _schema_minus_route_decisions() -> str:
|
|||||||
if stripped == ");":
|
if stripped == ");":
|
||||||
skipping = False
|
skipping = False
|
||||||
continue
|
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
|
continue
|
||||||
lines.append(line)
|
lines.append(line)
|
||||||
return "\n".join(lines)
|
return "\n".join(lines)
|
||||||
|
|||||||
@@ -53,6 +53,7 @@ WARN_CLASS_MATCHERS = {
|
|||||||
"capability-subceiling": r"(vision|json_mode)-capable tier",
|
"capability-subceiling": r"(vision|json_mode)-capable tier",
|
||||||
"rejection-new-pattern": r"new rejection pattern:",
|
"rejection-new-pattern": r"new rejection pattern:",
|
||||||
"rejection-rate": r"rejection rate:",
|
"rejection-rate": r"rejection rate:",
|
||||||
|
"classifier-degraded": r"came from a degraded source",
|
||||||
}
|
}
|
||||||
|
|
||||||
CFG = SimpleNamespace(
|
CFG = SimpleNamespace(
|
||||||
@@ -64,6 +65,7 @@ CFG = SimpleNamespace(
|
|||||||
),
|
),
|
||||||
objective=SimpleNamespace(plan_kwh_per_period=6.25, billing_reset_day=None),
|
objective=SimpleNamespace(plan_kwh_per_period=6.25, billing_reset_day=None),
|
||||||
escalation=SimpleNamespace(enabled=True),
|
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,
|
selected_provider=None,
|
||||||
rejected_reason=familiar,
|
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()
|
conn.commit()
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user