config.py

@@ -314,6 +314,11 @@ class LoggingConfig(StrictModel):
     # stderr and systemd captures that to the journal, so the setting named a
     # destination that did not exist.
     log_energy_observations: bool
+    # Whether to write a row to route_decisions for every routing decision
+    # (kind route | dispatch | chat | passthrough | local_vision). Off means
+    # the monitoring TUI's decision history is empty; it does not affect
+    # routing itself.
+    log_route_decisions: bool = True
     # LLM_ROUTER_LOG_LEVEL overrides this at runtime — see logs.resolve_level.
     level: str = "info"
 

config.yaml

@@ -385,6 +385,11 @@ logging:
   # -u llm-router -f`), so the setting named a destination that did not exist.
   log_energy_observations: true
 
+  # Whether to write a row to route_decisions for every routing decision
+  # (route | dispatch | chat | passthrough | local-vision). Off leaves the
+  # monitoring TUI's decision history empty; it does not change what routes.
+  log_route_decisions: true
+
   # debug | info | warning | error.
   #
   # info gives one line per request: what it was classified as, which model won,

dispatcher.py

@@ -72,6 +72,14 @@ from verification import (
     verify_response,
     worth_local_check,
 )
+from metrics import (
+    per_model,
+    quota_burn,
+    recent_decisions,
+    scoring_coverage,
+    top_proficiency,
+    verdict_mix,
+)
 
 # Virtual model names that mean "you pick". Anything else is taken as a real
 # model id and dispatched as asked.
@@ -240,6 +248,73 @@ def _db() -> sqlite3.Connection:
     return conn
 
 
+def ensure_route_decisions(conn: sqlite3.Connection) -> None:
+    """Idempotently create the route_decisions observability table.
+
+    schema.sql is CREATE TABLE IF NOT EXISTS, so it defines a NEW database and
+    silently does nothing to an existing one — the same reason
+    proficiency_store.ensure_columns exists. A live router.db predating this
+    table therefore never gets it from re-running schema.sql, so the table is
+    created here, from code, with a guard. This is the ONLY way it appears on
+    a live database; the live DB is never recreated or dropped.
+
+    Safe to call any number of times against an existing connection: the
+    CREATE TABLE and CREATE INDEX are both IF NOT EXISTS, so a table that is
+    already present is left fully intact (rows included) and a second call
+    no-ops.
+    """
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS route_decisions (
+            id                       INTEGER PRIMARY KEY AUTOINCREMENT,
+            observed_at              TEXT NOT NULL,
+            kind                     TEXT NOT NULL,
+            task_category            TEXT,
+            task_tier                INTEGER,
+            required_context_tokens  INTEGER,
+            confidence               REAL,
+            classifier_ms            INTEGER,
+            classification_source    TEXT,
+            latency_tolerance        TEXT,
+            candidates_considered    INTEGER,
+            selected_model           TEXT,
+            selected_provider       TEXT,
+            runner_up_models         TEXT,
+            est_cost_usd             REAL,
+            est_proficiency          REAL,
+            rejected_reason          TEXT,
+            session_key              TEXT,
+            tools                    INTEGER,
+            images                   INTEGER,
+            json_mode                INTEGER,
+            streamed                 INTEGER
+        )
+        """
+    )
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS idx_route_decisions_observed "
+        "ON route_decisions (observed_at)"
+    )
+    conn.commit()
+
+
+def _ensure_route_decisions_table() -> None:
+    """Create route_decisions on the live DB if it predates this feature."""
+    try:
+        conn = sqlite3.connect(cfg.database.path)
+        try:
+            ensure_route_decisions(conn)
+        finally:
+            conn.close()
+    except Exception as exc:  # noqa: BLE001
+        logs.warning(
+            "startup_route_decisions_migration_failed", error=str(exc)
+        )
+
+
+_ensure_route_decisions_table()
+
+
 def _classifier_client() -> OpenAI:
     """The classification endpoint, which need not be local.
 
@@ -697,6 +772,144 @@ def log_decision(
     )
 
 
+def persist_route_decision(
+    decision_kind: str,
+    *,
+    classification=None,
+    latency_tolerance=None,
+    selected_model=None,
+    selected_provider=None,
+    runners_up=None,
+    rejected_reason=None,
+    session_key=None,
+    tools=0,
+    images=0,
+    json_mode=0,
+    streamed=0,
+    classification_source=None,
+    classifier_ms=None,
+) -> None:
+    """Record one routing decision to route_decisions, best-effort and gated.
+
+    Every decision path calls this exactly once per request. A write that
+    fails for any reason (read-only or locked DB, missing table) is logged at
+    warning and swallowed, because a decision record is worth having but never
+    worth failing or slowing a request for. Gated by
+    ``cfg.logging.log_route_decisions`` so the monitoring history can be left
+    empty where it is not wanted.
+
+    ``classification`` may be a RouteResponse (deriving category/tier/context/…
+    from its Classification and the selected model from its ``selected``) or a
+    bare Classification; the explicit arguments cover the paths with no
+    RouteResponse (passthrough, local-vision) and let a caller override what
+    the RouteResponse would otherwise say — the chat re-route's
+    Classification reads ``source='override'`` even though the classifier did
+    decide the request, so the caller passes the first route's actual source.
+    """
+    # The gate is read before anything can fail: turning the table off must be
+    # a guaranteed no-op even on a broken DB.
+    if not cfg.logging.log_route_decisions:
+        return
+
+    clf = None
+    selected = None
+    derived_runners = None
+    candidates = None
+    derived_latency = latency_tolerance
+
+    if isinstance(classification, RouteResponse):
+        clf = classification.classification
+        selected = classification.selected
+        derived_runners = classification.runners_up
+        candidates = classification.candidates_considered
+        derived_latency = classification.latency_tolerance
+    elif isinstance(classification, Classification):
+        clf = classification
+
+    if clf is not None and classification_source is None:
+        classification_source = clf.source
+    # An override-created Classification never consulted the classifier, so it
+    # has no classifier latency worth recording — even when a caller passed a
+    # timing value (route_endpoint's whole-route ms would otherwise leak in).
+    if classification_source == "override":
+        classifier_ms = None
+
+    if selected is not None:
+        model = selected.model_id
+        provider = selected.provider
+        est_cost = selected.cost
+        est_prof = selected.proficiency_score
+        runner_json = (
+            json.dumps(
+                [{"model_id": c.model_id, "provider": c.provider}
+                 for c in derived_runners[:3]]
+            )
+            if derived_runners
+            else None
+        )
+    else:
+        model = selected_model
+        provider = selected_provider
+        est_cost = None
+        est_prof = None
+        runner_json = json.dumps(runners_up[:3]) if runners_up else None
+
+    conn = _db()
+    try:
+        # Guarantee the table on the WRITE path, mirroring
+        # proficiency_store._write -> ensure_columns. schema.sql is CREATE
+        # TABLE IF NOT EXISTS so it never adds the table to a live router.db
+        # that predates it; the module-load hook covers the normal boot, but
+        # any other path that calls persist first (a test harness, a lazy
+        # import, a future refactor) must still get the table here or the
+        # INSERT raises no-such-table and the decision is silently lost.
+        ensure_route_decisions(conn)
+        conn.execute(
+            """
+            INSERT INTO route_decisions (
+                observed_at, kind, task_category, task_tier,
+                required_context_tokens, confidence, classifier_ms,
+                classification_source, latency_tolerance, candidates_considered,
+                selected_model, selected_provider, runner_up_models,
+                est_cost_usd, est_proficiency, rejected_reason, session_key,
+                tools, images, json_mode, streamed
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                datetime.now(timezone.utc).isoformat(),
+                decision_kind,
+                clf.task_category if clf is not None else None,
+                clf.task_tier if clf is not None else None,
+                clf.required_context_tokens if clf is not None else None,
+                clf.confidence if clf is not None else None,
+                classifier_ms,
+                classification_source,
+                derived_latency,
+                candidates,
+                model,
+                provider,
+                runner_json,
+                est_cost,
+                est_prof,
+                rejected_reason,
+                session_key,
+                int(bool(tools)),
+                int(bool(images)),
+                int(bool(json_mode)),
+                int(bool(streamed)),
+            ),
+        )
+        conn.commit()
+    except Exception as e:  # noqa: BLE001 - best-effort must never raise
+        logs.warning(
+            "route_decision_persist",
+            kind=decision_kind,
+            error=type(e).__name__,
+        )
+    finally:
+        conn.close()
+
+
 # --- energy / cost accounting --------------------------------------------
 
 class Telemetry(BaseModel):
@@ -938,6 +1151,7 @@ def health():
                 """
             ).fetchall()
         )
+        scoring = scoring_coverage(conn, cfg)
     finally:
         conn.close()
 
@@ -950,7 +1164,7 @@ def health():
     return {
         "status": "ok",
         "counts": counts,
-        "scoring": scoring_coverage(),
+        "scoring": scoring,
         "classifier_reachable": classifier_ok,
         "classifier_model": cfg.classifier.model,
         # The tier labels, documented as being for logging and dashboards.
@@ -964,128 +1178,36 @@ def health():
     }
 
 
-def scoring_coverage() -> dict:
+def _quota_burn(conn: "sqlite3.Connection") -> Optional[dict]:
-    """Report which scoring axes actually have data behind them.
+    """Thin wrapper so /health can call quota_burn with the SAME connection."""
+    return quota_burn(conn, cfg)
 
-    An axis with no data is not an error — every candidate takes the neutral
-    0.5 and the router still works. It is worse than an error: it is silent.
-    A weight of 0.4 can be contributing exactly nothing while `/health`
-    cheerfully says "ok", and the only symptom is that routing stops
-    discriminating in a way nobody notices.
 
-    This has already happened twice: recreating the DB for a schema change
+@app.get("/metrics")
-    dropped every reference-workload observation, and cost and eco scored
+def metrics_endpoint():
-    neutral for days afterwards. So the state is surfaced rather than
+    """Aggregated read-only view over the router's observability tables.
-    inferred.
+
+    Mirrors the metrics.py helpers into one JSON payload for a dashboard.
+    Unauthenticated and loopback-bound exactly like /health — it contains no
+    conversation text, prompt, or session_dir.
     """
     conn = _db()
     try:
-        placeholders = ",".join("?" * len(cfg.routing.allowed_access_levels))
+        # per_model aggregates only energy_observations (cloud completions);
-        routable = [
+        # local-vision fallback decisions appear in recent_decisions but not
-            (r["model_id"], r["provider"])
+        # per_model (no cloud call), so the two views legitimately differ.
-            for r in conn.execute(
+        return {
-                f"""
+            "quota": quota_burn(conn, cfg),
-                SELECT model_id, provider FROM models
+            "coverage": scoring_coverage(conn, cfg),
-                 WHERE availability = 'active'
+            "recent_decisions": recent_decisions(conn),
-                   AND access_level IN ({placeholders})
+            "per_model": per_model(conn),
-                """,
+            "verdict_mix": verdict_mix(conn),
-                tuple(cfg.routing.allowed_access_levels),
+            "top_proficiency": top_proficiency(conn, "coding_general"),
-            )
+            "generated_at": datetime.now(timezone.utc).isoformat(),
-        ]
-        with_energy = {
-            (r["model_id"], r["provider"])
-            for r in conn.execute(
-                "SELECT DISTINCT model_id, provider FROM energy_observations "
-                "WHERE task_category = ?",
-                (SEED_CATEGORY,),
-            )
-        }
-        with_proficiency = {
-            (r["model_id"], r["provider"])
-            for r in conn.execute(
-                "SELECT DISTINCT model_id, provider FROM proficiency "
-                "WHERE blended_score IS NOT NULL"
-            )
         }
     finally:
         conn.close()
 
-    total = len(routable)
-    missing_energy = [m for m, p in routable if (m, p) not in with_energy]
-    missing_proficiency = [m for m, p in routable if (m, p) not in with_proficiency]
-
-    warnings = []
-    quota = quota_burn()
-    if quota and quota["metered_fraction_of_plan"] > 0.8:
-        warnings.append(
-            f"metered usage is {quota['metered_fraction_of_plan']*100:.0f}% of the "
-            f"{quota['plan_kwh']} kWh plan allowance. A quota is a wall, not a bill — "
-            "requests fail rather than costing more."
-        )
-    if missing_energy:
-        warnings.append(
-            f"{len(missing_energy)}/{total} routable models have no reference-workload "
-            f"observations — eco scores the neutral 0.5 for them and "
-            f"objective.max_energy_per_request cannot bound them. (Cost is "
-            f"unaffected: it is priced per request from catalog prices.) "
-            f"Run: python seed_energy.py --samples 7"
-        )
-    if missing_proficiency:
-        warnings.append(
-            f"{len(missing_proficiency)}/{total} routable models have no proficiency "
-            f"data — task_category cannot influence their ranking. "
-            f"Run: python eval_proficiency.py"
-        )
-
-    return {
-        "routable_models": total,
-        "with_energy_data": total - len(missing_energy),
-        "with_proficiency_data": total - len(missing_proficiency),
-        "quota": quota,
-        "warnings": warnings,
-    }
-
-
-def quota_burn() -> Optional[dict]:
-    """Energy this router has metered, against the plan's allowance.
-
-    Reported because the plan is a fixed kWh quota rather than a bill: you do
-    not overspend it, you hit it, mid-task.
-
-    This counts only what the ROUTER saw. Traffic that bypasses it — the eval
-    harness, ad-hoc scripts, a client pointed straight at the provider — is
-    invisible here, so treat this as a floor on real consumption.
-
-    How good a floor, measured against the provider's own 24h session view:
-    **96.9%** on 2026-08-23 (0.1543 kWh metered against 0.1593 kWh charged,
-    567 calls, 40.7M tokens). An earlier check put it at ~86%; the gap closed
-    as more traffic moved through the router rather than around it. The
-    remainder is mostly `eval_proficiency.py`, which calls the provider
-    directly and writes no observation.
-    """
-    if not cfg.objective.plan_kwh_per_period:
-        return None
-    conn = _db()
-    try:
-        row = conn.execute(
-            """
-            SELECT COALESCE(SUM(energy_kwh), 0) kwh, COUNT(*) n
-              FROM energy_observations
-             WHERE julianday(observed_at) > julianday('now', '-30 days')
-            """
-        ).fetchone()
-    finally:
-        conn.close()
-
-    plan = cfg.objective.plan_kwh_per_period
-    return {
-        "plan_kwh": plan,
-        "metered_kwh_30d": round(row["kwh"], 5),
-        "metered_fraction_of_plan": round(row["kwh"] / plan, 4),
-        "metered_calls_30d": row["n"],
-        "note": "router-metered only; traffic bypassing the router is not counted",
-    }
-
 
 class OutcomeReport(BaseModel):
     """A client telling the router whether an answer actually worked."""
@@ -1289,6 +1411,14 @@ def route_endpoint(req: TaskRequest):
         ms=_ms(started),
         ctx_src="caller" if req.required_context_tokens is not None else "classifier",
     )
+    persist_route_decision(
+        "route",
+        classification=decision,
+        tools=req.tools_present,
+        images=req.has_images,
+        json_mode=req.require_json_mode,
+        classifier_ms=_ms(started),
+    )
     return decision
 
 
@@ -1686,6 +1816,12 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
     if not messages:
         raise HTTPException(400, "messages is required")
 
+    # Needed by every decision branch below, including the passthrough and
+    # 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)
+    streamed = bool(body.get("stream"))
+
     # What this request needs, read from the body once and used by both the
     # routed and pass-through branches. Stated by the caller, not guessed at:
     # a tools array, an image_url part and a response_format each say exactly
@@ -1722,6 +1858,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
     if wants_routing:
         latency = BATCH if requested == ROUTER_MODEL_BATCH else INTERACTIVE
         tools_present = caps.tools_present
+        classify_started = time.perf_counter()
         decision = route(
             TaskRequest(
                 task=_last_user_text(messages),
@@ -1737,6 +1874,7 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
         # The classifier's own verdict, before the re-route below rewrites the
         # Classification's source to 'override'.
         classified_src = decision.classification.source
+        classifier_ms = _ms(classify_started)
         ctx_src = "classifier"
         measured = estimate_prompt_tokens(messages, tools=body.get("tools"))
         if measured > decision.classification.required_context_tokens:
@@ -1761,6 +1899,18 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
             ):
                 fallback = _run_local_vision(messages, cfg.local_vision)
                 if fallback is not None:
+                    persist_route_decision(
+                        "local_vision",
+                        latency_tolerance=decision.latency_tolerance,
+                        selected_model=cfg.local_vision.model,
+                        selected_provider="local",
+                        rejected_reason=None,
+                        session_key=session_key,
+                        tools=tools_present,
+                        images=int(caps.has_images),
+                        json_mode=int(caps.require_json_mode),
+                        streamed=streamed,
+                    )
                     return _local_vision_response(
                         fallback, streaming=bool(body.get("stream"))
                     )
@@ -1787,6 +1937,16 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
                     "json-mode-capable model (response_format requires it)"
                 )
             logs.warning("no_candidate", limits="; ".join(limits))
+            persist_route_decision(
+                "chat",
+                classification=decision,
+                rejected_reason="; ".join(limits),
+                session_key=session_key,
+                tools=tools_present,
+                images=int(caps.has_images),
+                json_mode=int(caps.require_json_mode),
+                streamed=streamed,
+            )
             raise HTTPException(
                 422,
                 "No model satisfies the hard filters for this request ("
@@ -1800,6 +1960,17 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
             src=classified_src,
             ctx_src=ctx_src,
         )
+        persist_route_decision(
+            "chat",
+            classification=decision,
+            session_key=session_key,
+            tools=tools_present,
+            images=int(caps.has_images),
+            json_mode=int(caps.require_json_mode),
+            streamed=streamed,
+            classification_source=classified_src,
+            classifier_ms=classifier_ms,
+        )
         target = decision.selected.model_id
         provider = decision.selected.provider
         category = decision.classification.task_category
@@ -1808,6 +1979,17 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
         # Not a routing decision at all, and worth saying so plainly: a client
         # pinned to one model gets none of the filtering or ranking below.
         logs.info("passthrough", model=target, stream=bool(body.get("stream")))
+        persist_route_decision(
+            "passthrough",
+            selected_model=target,
+            selected_provider=provider,
+            rejected_reason=None,
+            session_key=session_key,
+            tools=int(caps.tools_present),
+            images=int(caps.has_images),
+            json_mode=int(caps.require_json_mode),
+            streamed=streamed,
+        )
 
         # A pinned model id is dispatched as asked, but a pin that cannot
         # possibly satisfy the request should fail with a clear 422 instead of
@@ -1823,8 +2005,9 @@ def chat_completions(body: dict[str, Any], background: BackgroundTasks):
 
     # A truncated answer under a cap the CLIENT chose is not the model failing.
     client_capped = body.get("max_tokens") is not None
-    # Who is talking to us, so concurrent clients stay distinguishable.
+    # What session this dispatch belongs to; the hashed key is computed at the
-    session_key = session_fingerprint(messages)
+    # top (needed by the passthrough/local-vision branches), only the working
+    # directory is derived here because only the observation path uses it.
     session_dir = session_directory(messages)
     upstream_body = {**body, "model": target}
     streaming = bool(body.get("stream"))
@@ -2133,6 +2316,14 @@ def dispatch_endpoint(req: TaskRequest):
         ms=_ms(started),
         ctx_src="caller" if req.required_context_tokens is not None else "classifier",
     )
+    persist_route_decision(
+        "dispatch",
+        classification=decision,
+        tools=req.tools_present,
+        images=req.has_images,
+        json_mode=req.require_json_mode,
+        classifier_ms=_ms(started),
+    )
     if decision.selected is None:
         raise HTTPException(
             422,

metrics.py

@@ -0,0 +1,210 @@
+"""Read-only aggregation helpers for the router dashboard / /health.
+
+This module MUST NOT import ``dispatcher`` — it exists specifically to break
+what would otherwise be a circular import (dispatcher wants /health metrics,
+metrics wants the config and DB path that dispatcher already knows).
+
+All functions take an ``sqlite3.Connection`` (with ``row_factory`` set) and
+optionally a ``RouterConfig`` instance; none rely on module-level globals.
+
+Functions
+---------
+quota_burn      — kWh metered in the last 30 d, against the plan allowance
+scoring_coverage — which scoring axes actually have data
+recent_decisions — last N rows from the route_decisions observability table
+per_model        — per-model aggregates over energy_observations (last 30 d)
+verdict_mix      — counts by verdict from verifications (last N days)
+top_proficiency  — top models by blended_score for a category
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from typing import Any, List, Optional
+
+SEED_CATEGORY = "seed_reference"
+FALLBACK_CARBON_SOURCE = "static_fallback"
+
+
+def quota_burn(
+    conn: sqlite3.Connection,
+    cfg: Any,
+) -> Optional[dict]:
+    """Energy this router has metered, against the plan's allowance.
+
+    Accepts ``(conn, cfg)`` so the caller owns the connection and the config
+    — metrics.py never touches dispatcher's module-level ``cfg`` or its
+    ``_db()`` helper, which is exactly why this module must never import
+    dispatcher.
+    """
+    if not cfg.objective.plan_kwh_per_period:
+        return None
+    row = conn.execute(
+        """
+        SELECT COALESCE(SUM(energy_kwh), 0) kwh, COUNT(*) n
+          FROM energy_observations
+         WHERE julianday(observed_at) > julianday('now', '-30 days')
+        """
+    ).fetchone()
+
+    plan = cfg.objective.plan_kwh_per_period
+    return {
+        "plan_kwh": plan,
+        "metered_kwh_30d": round(float(row["kwh"]), 5),
+        "metered_fraction_of_plan": round(float(row["kwh"]) / plan, 4),
+        "metered_calls_30d": row["n"],
+        "note": "router-metered only; traffic bypassing the router is not counted",
+    }
+
+
+def scoring_coverage(
+    conn: sqlite3.Connection,
+    cfg: Any,
+) -> dict:
+    """Report which scoring axes actually have data behind them."""
+    placeholders = ",".join("?" * len(cfg.routing.allowed_access_levels))
+    routable = [
+        (r["model_id"], r["provider"])
+        for r in conn.execute(
+            f"""
+            SELECT model_id, provider FROM models
+             WHERE availability = 'active'
+               AND access_level IN ({placeholders})
+            """,
+            tuple(cfg.routing.allowed_access_levels),
+        )
+    ]
+    with_energy = {
+        (r["model_id"], r["provider"])
+        for r in conn.execute(
+            "SELECT DISTINCT model_id, provider FROM energy_observations "
+            "WHERE task_category = ?",
+            (SEED_CATEGORY,),
+        )
+    }
+    with_proficiency = {
+        (r["model_id"], r["provider"])
+        for r in conn.execute(
+            "SELECT DISTINCT model_id, provider FROM proficiency "
+            "WHERE blended_score IS NOT NULL"
+        )
+    }
+
+    total = len(routable)
+    missing_energy = [m for m, p in routable if (m, p) not in with_energy]
+    missing_proficiency = [m for m, p in routable if (m, p) not in with_proficiency]
+
+    warnings: List[str] = []
+    quota = quota_burn(conn, cfg)
+    if quota and quota["metered_fraction_of_plan"] > 0.8:
+        warnings.append(
+            f"metered usage is {quota['metered_fraction_of_plan']*100:.0f}% of the "
+            f"{quota['plan_kwh']} kWh plan allowance. A quota is a wall, not a bill — "
+            "requests fail rather than costing more."
+        )
+    if missing_energy:
+        warnings.append(
+            f"{len(missing_energy)}/{total} routable models have no reference-workload "
+            f"observations — eco scores the neutral 0.5 for them and "
+            f"objective.max_energy_per_request cannot bound them. (Cost is "
+            f"unaffected: it is priced per request from catalog prices.) "
+            f"Run: python seed_energy.py --samples 7"
+        )
+    if missing_proficiency:
+        warnings.append(
+            f"{len(missing_proficiency)}/{total} routable models have no proficiency "
+            f"data — task_category cannot influence their ranking. "
+            f"Run: python eval_proficiency.py"
+        )
+
+    return {
+        "routable_models": total,
+        "with_energy_data": total - len(missing_energy),
+        "with_proficiency_data": total - len(missing_proficiency),
+        "quota": quota,
+        "warnings": warnings,
+    }
+
+
+def recent_decisions(
+    conn: sqlite3.Connection,
+    limit: int = 50,
+) -> List[dict]:
+    """Last *N* rows from route_decisions, ordered by id DESC."""
+    return [
+        dict(row)
+        for row in conn.execute(
+            """
+            SELECT id, observed_at, kind, task_category, task_tier,
+                   required_context_tokens, confidence, classifier_ms,
+                   classification_source, latency_tolerance,
+                   candidates_considered, selected_model, selected_provider,
+                   runner_up_models, est_cost_usd, est_proficiency,
+                   rejected_reason, session_key, tools, images, json_mode, streamed
+              FROM route_decisions
+             ORDER BY id DESC
+             LIMIT ?
+            """,
+            (limit,),
+        ).fetchall()
+    ]
+
+
+def per_model(conn: sqlite3.Connection) -> List[dict]:
+    """Per-model aggregates over the last 30 d of energy_observations."""
+    return [
+        dict(row)
+        for row in conn.execute(
+            """
+            SELECT model_id,
+                   provider,
+                   COUNT(*)                                            AS calls,
+                   COALESCE(SUM(cost_usd), 0)                          AS sum_cost_usd,
+                   COALESCE(SUM(energy_kwh), 0)                        AS sum_energy_kwh,
+                   COALESCE(SUM(carbon_g_co2eq), 0)                    AS sum_carbon_g_co2eq,
+                   AVG(completion_tokens)                              AS avg_completion_tokens,
+                   AVG(attribution_ratio)                              AS avg_attribution_ratio
+              FROM energy_observations
+             WHERE julianday(observed_at) > julianday('now', '-30 days')
+             GROUP BY model_id, provider
+            """
+        ).fetchall()
+    ]
+
+
+def verdict_mix(
+    conn: sqlite3.Connection,
+    since_days: int = 7,
+) -> dict:
+    """Counts by verdict from verifications in the last *since_days*."""
+    rows = conn.execute(
+        """
+        SELECT verdict, COUNT(*) n
+          FROM verifications
+         WHERE julianday(observed_at) > julianday('now', '-' || ? || ' days')
+         GROUP BY verdict
+        """,
+        (str(since_days),),
+    ).fetchall()
+    return {row["verdict"]: row["n"] for row in rows}
+
+
+def top_proficiency(
+    conn: sqlite3.Connection,
+    category: str,
+) -> List[dict]:
+    """Top models by blended_score for *category*, ordered DESC."""
+    return [
+        dict(row)
+        for row in conn.execute(
+            """
+            SELECT model_id, provider, blended_score, source,
+                   self_eval_samples
+              FROM proficiency
+             WHERE category = ?
+               AND blended_score IS NOT NULL
+             ORDER BY blended_score DESC
+            """,
+            (category,),
+        ).fetchall()
+    ]

requirements.txt

@@ -10,3 +10,8 @@ openai==3.0.0
 python-dotenv==1.2.2
 pytest==9.1.1
 pytest-cov==7.1.0
+# Deliberate UI-bring-your-own-tool dependency: textual is the terminal
+# dashboard framework for the monitoring TUI (tui.py), a SEPARATE entrypoint.
+# It is only imported by tui.py — never by the service dispatch path, so the
+# router itself has no UI dependency and dispatcher's import path is unchanged.
+textual==8.2.8

router_cli.py

@@ -0,0 +1,233 @@
+#!/usr/bin/env python3
+"""router_cli.py -- one-shot routing-decision printer.
+
+POSTs a task to the running router's ``POST /route`` endpoint and prints the
+full decision tree: classification, candidates considered, the selected model
+(plus its estimated cost and proficiency), runners-up, and -- when nothing was
+selected -- the rejection reason.
+
+This is a **route-only, no-spend probe**. It never dispatches to a provider and
+never calls a model. It is not a daemon; it runs once in the foreground and
+exits.
+
+Usage:
+    python router_cli.py "<task>" [--category CAT] [--tier N] [--context N] [--json]
+
+The router base URL comes from the ROUTER_URL env var, defaulting to
+``http://127.0.0.1:8080``. ``/route`` needs no auth (loopback-only bind).
+"""
+
+import argparse
+import json
+import os
+import sys
+
+import requests
+
+DEFAULT_BASE_URL = "http://127.0.0.1:8080"
+ROUTE_PATH = "/route"
+
+
+# --- pure rendering helpers -----------------------------------------------
+
+
+def _fmt_cost(cost):
+    """A compact human-readable US$ for a cost figure, or a placeholder."""
+    if cost is None:
+        return "n/a"
+    if cost < 0.001:
+        return f"US${cost:.3e}"
+    return f"US${cost:.6f}"
+
+
+def _fmt_proficiency(prof):
+    if prof is None:
+        return "n/a"
+    return f"{prof:.3f}"
+
+
+def render_decision(data: dict) -> str:
+    """Render a RouteResponse-shaped dict into a human decision tree."""
+    lines = []
+    cls = data.get("classification") or {}
+    lines.append("classification:")
+    lines.append(f"  category        : {cls.get('task_category', 'n/a')}")
+    lines.append(f"  tier            : {cls.get('task_tier', 'n/a')}")
+    lines.append(
+        f"  required_context: {cls.get('required_context_tokens', 'n/a')}"
+    )
+    lines.append(f"  confidence      : {cls.get('confidence', 'n/a')}")
+    lines.append(f"  source          : {cls.get('source', 'n/a')}")
+    lines.append(f"latency_tolerance : {data.get('latency_tolerance', 'n/a')}")
+    lines.append(f"candidates_considered: {data.get('candidates_considered', 'n/a')}")
+
+    selected = data.get("selected")
+    if selected:
+        lines.append("selected:")
+        lines.append(f"  model      : {selected.get('model_id', 'n/a')}")
+        lines.append(f"  provider   : {selected.get('provider', 'n/a')}")
+        lines.append(f"  tier       : {selected.get('tier', 'n/a')}")
+        lines.append(f"  est cost   : {_fmt_cost(selected.get('cost'))}")
+        lines.append(
+            f"  proficiency: {_fmt_proficiency(selected.get('proficiency_score'))}"
+        )
+    else:
+        lines.append("selected: (none)")
+
+    runners_up = data.get("runners_up") or []
+    if runners_up:
+        lines.append("runners-up:")
+        for ru in runners_up[:3]:
+            lines.append(
+                f"  {ru.get('model_id', 'n/a')} "
+                f"(est cost {_fmt_cost(ru.get('cost'))}, "
+                f"proficiency {_fmt_proficiency(ru.get('proficiency_score'))})"
+            )
+    else:
+        lines.append("runners-up: (none)")
+
+    if not selected:
+        reason = data.get("detail") or data.get("rejected_reason")
+        lines.append("rejected_reason: " + (reason or "no model selected"))
+
+    return "\n".join(lines) + "\n"
+
+
+def _validate_overrides(tier, context):
+    """Return (tier, context) validated, or raise ValueError with a message."""
+    if tier is not None:
+        if isinstance(tier, bool):
+            raise ValueError("--tier must be an integer between 1 and 3")
+        try:
+            tier = int(tier)
+        except (TypeError, ValueError):
+            raise ValueError("--tier must be an integer between 1 and 3")
+        if not (1 <= tier <= 3):
+            raise ValueError("--tier must be between 1 and 3")
+    if context is not None:
+        if isinstance(context, bool):
+            raise ValueError("--context must be a non-negative integer")
+        try:
+            context = int(context)
+        except (TypeError, ValueError):
+            raise ValueError("--context must be a non-negative integer")
+        if context < 0:
+            raise ValueError("--context must be a non-negative integer")
+    return tier, context
+
+
+# --- the callable decision logic ------------------------------------------
+
+
+def run(
+    task,
+    category=None,
+    tier=None,
+    context=None,
+    as_json=False,
+    post=None,
+    base_url=None,
+):
+    """Route ``task`` and print the decision tree.
+
+    Returns a process exit code (0 on success). ``post`` is an injectable
+    callback ``post(url, json=..., timeout=...) -> response`` replacing
+    ``requests.post`` for testing; ``base_url`` defaults to the ROUTER_URL env
+    var or the loopback default. Route-only: never dispatches to a provider.
+    """
+    try:
+        tier, context = _validate_overrides(tier, context)
+    except ValueError as e:
+        print(f"router-cli: {e}", file=sys.stderr)
+        return 2
+
+    if base_url is None:
+        base_url = os.environ.get("ROUTER_URL", DEFAULT_BASE_URL)
+    url = base_url.rstrip("/") + ROUTE_PATH
+
+    body = {"task": task}
+    if category is not None:
+        body["task_category"] = category
+    if tier is not None:
+        body["task_tier"] = tier
+    if context is not None:
+        body["required_context_tokens"] = context
+
+    if post is None:
+        post = lambda u, json=None, **kw: requests.post(u, json=json, timeout=30, **kw)
+
+    try:
+        resp = post(url, json=body)
+    except requests.exceptions.RequestException as e:
+        # requests' own ConnectionError/Timeout/HTTPError family.
+        print(f"router-cli: request to {url} failed: {e}", file=sys.stderr)
+        return 1
+    except ConnectionError as e:
+        # The injectable callback (and tests) may raise the builtin instead.
+        print(
+            f"router-cli: could not reach the router at {url} "
+            f"(connection error: {e}). Is the dispatcher running?",
+            file=sys.stderr,
+        )
+        return 1
+
+    if resp.status_code != 200:
+        reason = None
+        try:
+            reason = resp.json().get("detail")
+        except (ValueError, TypeError, AttributeError):
+            if getattr(resp, "text", None):
+                reason = resp.text
+        print(
+            f"router-cli: router returned {resp.status_code}: "
+            f"{reason or 'no model satisfies the routing filters'}",
+            file=sys.stderr,
+        )
+        return 1
+
+    data = resp.json()
+    if as_json:
+        print(json.dumps(data, indent=2))
+    else:
+        sys.stdout.write(render_decision(data))
+    return 0
+
+
+# --- CLI entry ------------------------------------------------------------
+
+def _build_parser():
+    p = argparse.ArgumentParser(
+        prog="router_cli.py",
+        description=(
+            "Route a task through the running router and print the decision "
+            "tree. Route-only: never dispatches to a model or spends quota."
+        ),
+    )
+    p.add_argument("task", help="The task text to route.")
+    p.add_argument("--category", help="Override task_category (skips classifier).")
+    p.add_argument("--tier", type=int, help="Override task_tier (1-3).")
+    p.add_argument(
+        "--context", type=int, help="Override required_context_tokens."
+    )
+    p.add_argument(
+        "--json",
+        action="store_true",
+        help="Emit the raw RouteResponse JSON instead of the decision tree.",
+    )
+    return p
+
+
+def main(argv=None):
+    parser = _build_parser()
+    args = parser.parse_args(argv)
+    return run(
+        args.task,
+        category=args.category,
+        tier=args.tier,
+        context=args.context,
+        as_json=args.json,
+    )
+
+
+if __name__ == "__main__":
+    sys.exit(main())

schema.sql

@@ -188,6 +188,51 @@ CREATE TABLE IF NOT EXISTS verifications (
     model_attributable INTEGER DEFAULT 1
 );
 
+-- One row per routing decision, logged by the dispatcher on every route that
+-- is made (see todo #2 of router-monitoring-tui.md for the writes; this table
+-- and its inline-create helper are todo #1). It makes "how is routing
+-- performing" answerable: which model was picked, for what category/tier, how
+-- long classification took, and — when nothing was selected — which hard
+-- filter shut it out.
+--
+-- This is an OBSERVABILITY table, not a scoring input: nothing in routing.py
+-- reads it. It stores only the hashed session fingerprint in `session_key`;
+-- never `session_dir` and never any prompt or answer text. A test enforces
+-- that the write path stores only the hash.
+--
+-- Like `energy_observations`, `observed_at` uses
+-- datetime.now(timezone.utc).isoformat().
+CREATE TABLE IF NOT EXISTS route_decisions (
+    id                       INTEGER PRIMARY KEY AUTOINCREMENT,
+    observed_at              TEXT NOT NULL,          -- ISO8601, UTC
+    kind                     TEXT NOT NULL,          -- 'route' | 'dispatch' | 'chat'
+                                                     -- | 'passthrough' | 'local_vision'
+    task_category            TEXT,
+    task_tier                INTEGER,                -- 1-3
+    required_context_tokens  INTEGER,
+    confidence               REAL,
+    classifier_ms            INTEGER,
+    classification_source    TEXT,                   -- 'classifier' | 'override' | 'fallback'
+    latency_tolerance        TEXT,                   -- 'interactive' | 'batch'
+    candidates_considered    INTEGER,
+    selected_model           TEXT,                   -- null when nothing was selected
+    -- 'neuralwatt' for routed/dispatch/passthrough, 'local' for the local-vision
+    -- fallback. Kept so a decision row can join to energy_observations on the
+    -- (model_id, provider) key that table uses.
+    selected_provider        TEXT,
+    runner_up_models         TEXT,                   -- JSON array of {"model_id":..,
+                                                     -- "provider":..}, <=3; nullable
+    est_cost_usd             REAL,
+    est_proficiency          REAL,
+    rejected_reason          TEXT,                   -- the 422 limits when no selection
+    session_key              TEXT,                   -- hashed session fingerprint ONLY,
+                                                     -- never session_dir or prompt text
+    tools                    INTEGER,                -- 0/1
+    images                   INTEGER,                -- 0/1
+    json_mode                INTEGER,                -- 0/1
+    streamed                 INTEGER                 -- 0/1
+);
+
 CREATE INDEX IF NOT EXISTS idx_verifications_model ON verifications (model_id, provider);
 CREATE INDEX IF NOT EXISTS idx_verifications_verdict ON verifications (verdict);
 CREATE INDEX IF NOT EXISTS idx_observations_request ON energy_observations (request_id);
@@ -197,3 +242,8 @@ CREATE INDEX IF NOT EXISTS idx_models_availability ON models (availability);
 CREATE INDEX IF NOT EXISTS idx_models_routing ON models (access_level, latency_class, tier);
 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);
+-- 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);

tests/test_metrics.py

@@ -0,0 +1,538 @@
+"""Tests for metrics.py — read-only aggregation helpers.
+
+Every test seeds a throwaway SQLite DB directly from schema.sql, never writes
+to the live ``router.db``, and asserts on *actual queried aggregates* rather
+than mock-call assertions (to defeat ``misleading_success_output``).
+
+Functions tested:
+- quota_burn
+- scoring_coverage
+- recent_decisions
+- per_model
+- verdict_mix
+- top_proficiency
+
+Also verifies that ``import dispatcher`` and ``/health`` still work after the
+move, and that ``import metrics`` alone succeeds (no circular import).
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from starlette.testclient import TestClient
+
+import dispatcher
+from config import load_config
+from metrics import (
+    quota_burn,
+    scoring_coverage,
+    recent_decisions,
+    per_model,
+    verdict_mix,
+    top_proficiency,
+)
+
+ROOT = Path(__file__).resolve().parent.parent
+SCHEMA_SQL = (ROOT / "schema.sql").read_text()
+CFG = load_config(str(ROOT / "config.yaml"))
+
+
+# =============================================================================
+# Helpers
+# =============================================================================
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _make_db(tmp_path: Path, extra_sql: str = "") -> sqlite3.Connection:
+    """Create a clean DB seeded from schema.sql, returning a Row-backed conn."""
+    conn = sqlite3.connect(str(tmp_path / "test.db"))
+    conn.row_factory = sqlite3.Row
+    conn.executescript(SCHEMA_SQL + extra_sql)
+    return conn
+
+
+def _seed_models(conn: sqlite3.Connection) -> None:
+    """Insert routable model rows into an (empty) DB."""
+    for model_id, tier, context, cost, vision in (
+        ("cheap", 2, 262128, 0.30, 1),
+        ("dear", 2, 262128, 9.00, 0),
+        ("tiny", 1, 131072, 0.10, 1),
+    ):
+        conn.execute(
+            """
+            INSERT INTO models (
+                model_id, provider, base_model_id, tier, context_window,
+                effective_context_window, max_output_tokens,
+                cost_per_1m_prompt, cost_per_1m_completion,
+                supports_vision, supports_json_mode,
+                latency_class, reasoning_mode, context_variant,
+                access_level, availability, last_updated
+            ) VALUES (?, 'neuralwatt', ?, ?, ?, 192500, 16384, ?, ?,
+                      ?, 1, 'standard', 'default', 'full', 'public', 'active',
+                      '2026-08-22T00:00:00+00:00')
+            """,
+            (model_id, model_id, tier, context, cost, cost / 3, vision),
+        )
+    conn.commit()
+
+
+def _seed_proficiency(conn: sqlite3.Connection) -> None:
+    """Insert proficiency rows for the seeded models."""
+    for model_id, score in (
+        ("cheap", 0.90),
+        ("dear", 0.95),
+        ("tiny", 0.70),
+    ):
+        conn.execute(
+            """
+            INSERT INTO proficiency (
+                model_id, provider, category, blended_score, source, last_updated
+            ) VALUES (?, 'neuralwatt', 'coding_general', ?, 'self_eval_thin', '2026-01-01T00:00:00+00:00')
+            """,
+            (model_id, score),
+        )
+    conn.commit()
+
+
+def _seed_energy(conn: sqlite3.Connection) -> None:
+    now = _now()
+    recent_rows = [
+        ("cheap", 5.0e-05, 100, 0.25),       # 2 days ago
+        ("cheap", 3.0e-05, 200, 0.50),       # 2 days ago
+        ("dear", 1.0e-04, 150, 0.75),        # 5 days ago
+        ("tiny", 1.0e-05, 50, 1.00),         # 10 days ago
+    ]
+    for (model_id, kwh, tokens, attr) in recent_rows:
+        conn.execute(
+            """
+            INSERT INTO energy_observations (
+                model_id, provider, task_category, prompt_tokens,
+                completion_tokens, energy_kwh, attribution_ratio,
+                observed_at
+            ) VALUES (?, 'neuralwatt', 'coding_general', 1000, ?, ?, ?, ?)
+            """,
+            (model_id, tokens, kwh, attr, (now - timedelta(days=2)).isoformat()),
+        )
+    conn.commit()
+
+
+# --- Import / no-circular-import smoke tests -----------------------------------
+
+
+def test_metrics_can_be_imported_alone():
+    """metrics.py must not require dispatcher — it *is* the cycle-breaker."""
+    # If this import raises ImportError (circular), we fail.
+    import metrics  # noqa: F401
+
+
+def test_dispatcher_imports_after_metrics():
+    """importing metrics first, then dispatcher, must not raise."""
+    # This test runs *after* metrics has already been imported above.
+    # The import chain is: dispatcher → metrics (one-way).
+    assert hasattr(dispatcher, "app")
+
+
+# --- quota_burn tests ---------------------------------------------------------
+
+
+def test_quota_burn_returns_none_when_no_plan(tmp_path):
+    """When plan_kwh_per_period is falsy, quota_burn returns None."""
+    conn = _make_db(tmp_path)
+    no_plan_cfg = SimpleNamespace(
+        objective=SimpleNamespace(plan_kwh_per_period=None)
+    )
+    assert quota_burn(conn, no_plan_cfg) is None
+
+
+def test_quota_burn_returns_none_when_plan_is_zero(tmp_path):
+    """plan_kwh_per_period == 0 is treated the same as None."""
+    conn = _make_db(tmp_path)
+    zero_plan_cfg = SimpleNamespace(
+        objective=SimpleNamespace(plan_kwh_per_period=0)
+    )
+    assert quota_burn(conn, zero_plan_cfg) is None
+
+
+def test_quota_burn_aggregates_last_30_days(tmp_path):
+    """Two recent rows and one old row — only the recent ones count."""
+    cfg = SimpleNamespace(
+        objective=SimpleNamespace(plan_kwh_per_period=6.25)
+    )
+    conn = _make_db(tmp_path)
+    now = _now()
+    conn.execute(
+        "INSERT INTO energy_observations "
+        "(model_id, provider, energy_kwh, completion_tokens, observed_at) "
+        "VALUES ('m', 'neuralwatt', 0.10, 100, ?)",
+        ((now - timedelta(days=1)).isoformat(),),
+    )
+    conn.execute(
+        "INSERT INTO energy_observations "
+        "(model_id, provider, energy_kwh, completion_tokens, observed_at) "
+        "VALUES ('m', 'neuralwatt', 0.15, 200, ?)",
+        ((now - timedelta(days=5)).isoformat(),),
+    )
+    conn.execute(
+        "INSERT INTO energy_observations "
+        "(model_id, provider, energy_kwh, completion_tokens, observed_at) "
+        "VALUES ('m', 'neuralwatt', 0.90, 300, ?)",
+        ((now - timedelta(days=60)).isoformat(),),
+    )
+    conn.commit()
+
+    result = quota_burn(conn, cfg)
+    assert result is not None
+    assert result["metered_kwh_30d"] == pytest.approx(0.25)
+    assert result["metered_calls_30d"] == 2
+    assert result["plan_kwh"] == 6.25
+    assert result["metered_fraction_of_plan"] == pytest.approx(0.25 / 6.25)
+
+
+def test_quota_burn_empty_db(tmp_path):
+    """Zero rows → kwh=0, calls=0, not an error."""
+    cfg = SimpleNamespace(
+        objective=SimpleNamespace(plan_kwh_per_period=1.0)
+    )
+    conn = _make_db(tmp_path)
+    result = quota_burn(conn, cfg)
+    assert result["metered_kwh_30d"] == 0.0
+    assert result["metered_calls_30d"] == 0
+
+
+# --- scoring_coverage tests ---------------------------------------------------
+
+
+def test_scoring_coverage_has_all_keys(tmp_path):
+    """The return dict always has these keys, even when empty."""
+    conn = _make_db(tmp_path)
+    # Seed routable models
+    _seed_models(conn)
+    result = scoring_coverage(conn, CFG)
+    assert "routable_models" in result
+    assert "with_energy_data" in result
+    assert "with_proficiency_data" in result
+    assert "quota" in result
+    assert "warnings" in result
+
+
+def test_scoring_coverage_warning_when_no_energy(tmp_path):
+    """Models with no seed_reference observations produce a warning."""
+    conn = _make_db(tmp_path)
+    _seed_models(conn)
+    # No energy_observations rows at all
+    result = scoring_coverage(conn, CFG)
+    warning_texts = result["warnings"]
+    assert any("no reference-workload observations" in w for w in warning_texts)
+    assert result["with_energy_data"] == 0
+
+
+def test_scoring_coverage_no_warning_when_full_coverage(tmp_path):
+    """When every routable model has energy + proficiency, no warnings."""
+    conn = _make_db(tmp_path)
+    _seed_models(conn)
+    # Seed SEED_CATEGORY energy observations
+    now = _now()
+    for model in ["cheap", "dear"]:
+        conn.execute(
+            "INSERT INTO energy_observations "
+            "(model_id, provider, task_category, prompt_tokens, "
+            "completion_tokens, energy_kwh, attribution_ratio, observed_at) "
+            "VALUES (?, 'neuralwatt', 'seed_reference', 1000, 100, 0.001, 0.25, ?)",
+            (model, now.isoformat()),
+        )
+        conn.execute(
+            "INSERT INTO proficiency "
+            "(model_id, provider, category, blended_score, source, last_updated) "
+            "VALUES (?, 'neuralwatt', 'coding_general', 0.9, 'self_eval', ?)",
+            (model, now.isoformat()),
+        )
+    conn.commit()
+    result = scoring_coverage(conn, CFG)
+    # 'tiny' has no data so we expect warnings. Let's also add 'tiny'.
+    conn.execute(
+        "INSERT INTO energy_observations "
+        "(model_id, provider, task_category, prompt_tokens, "
+        "completion_tokens, energy_kwh, attribution_ratio, observed_at) "
+        "VALUES ('tiny', 'neuralwatt', 'seed_reference', 1000, 100, 0.001, 0.25, ?)",
+        (now.isoformat(),),
+    )
+    conn.execute(
+        "INSERT INTO proficiency "
+        "(model_id, provider, category, blended_score, source, last_updated) "
+        "VALUES ('tiny', 'neuralwatt', 'coding_general', 0.7, 'self_eval', ?)",
+        (now.isoformat(),),
+    )
+    conn.commit()
+    result = scoring_coverage(conn, CFG)
+    assert result["with_energy_data"] == 3
+    assert result["with_proficiency_data"] == 3
+    assert len(result["warnings"]) == 0
+
+
+def test_scoring_coverage_empty_db(tmp_path):
+    """Empty DB: 0 routable, no warnings, quota=None."""
+    conn = _make_db(tmp_path)
+    result = scoring_coverage(conn, CFG)
+    assert result["routable_models"] == 0
+    assert result["with_energy_data"] == 0
+    assert result["with_proficiency_data"] == 0
+    # quota returns None because plan_kwh_per_period may be None in default cfg
+    # (it is 6.25 by default, but let's just check the structure)
+    assert result["warnings"] == []
+
+
+# --- recent_decisions tests ---------------------------------------------------
+
+
+def test_recent_decisions_returns_rows_in_desc_order(tmp_path, monkeypatch):
+    """Rows come back ordered by id DESC, and selected_provider is included."""
+    conn = _make_db(tmp_path)
+    # Seed two models
+    _seed_models(conn)
+
+    # Create fake dispatcher state to use TestClient, but we'll insert
+    # route_decisions rows directly and call recent_decisions(conn).
+    conn.execute(
+        """
+        INSERT INTO route_decisions (
+            observed_at, kind, task_category, task_tier, required_context_tokens,
+            confidence, classifier_ms, classification_source, latency_tolerance,
+            candidates_considered, selected_model, selected_provider,
+            runner_up_models, est_cost_usd, est_proficiency,
+            session_key, tools, images, json_mode, streamed
+        ) VALUES (?, 'route', 'coding_general', 2, 100, 0.95, 200,
+                  'classifier', 'interactive', 5, 'cheap', 'neuralwatt',
+                  '[{"model_id":"dear","provider":"neuralwatt"}]',
+                  0.001, 0.9, 'abc123', 0, 0, 0, 0)
+        """,
+        (datetime.now(timezone.utc).isoformat(),),
+    )
+    conn.execute(
+        """
+        INSERT INTO route_decisions (
+            observed_at, kind, task_category, task_tier, required_context_tokens,
+            confidence, classifier_ms, classification_source, latency_tolerance,
+            candidates_considered, selected_model, selected_provider,
+            runner_up_models, est_cost_usd, est_proficiency,
+            session_key, tools, images, json_mode, streamed
+        ) VALUES (?, 'route', 'docs_writing', 1, 50, 0.88, 150,
+                  'classifier', 'interactive', 3, 'dear', 'neuralwatt',
+                  NULL, 0.005, 0.95, 'def456', 0, 0, 0, 0)
+        """,
+        (datetime.now(timezone.utc).isoformat(),),
+    )
+    conn.commit()
+
+    rows = recent_decisions(conn, limit=50)
+    assert len(rows) == 2
+    # DESC order: dear (id=2) first
+    assert rows[0]["selected_model"] == "dear"
+    assert rows[0]["kind"] == "route"
+    assert rows[0]["selected_provider"] == "neuralwatt"
+    assert rows[1]["selected_model"] == "cheap"
+
+
+def test_recent_decisions_respects_limit(tmp_path):
+    """limit=1 should return only one row regardless of DB content."""
+    conn = _make_db(tmp_path)
+    _seed_models(conn)
+    now = _now().isoformat()
+    for i in range(5):
+        conn.execute(
+            "INSERT INTO route_decisions "
+            "(observed_at, kind, selected_model, selected_provider) "
+            "VALUES (?, 'route', 'm', 'neuralwatt')",
+            (now,),
+        )
+    conn.commit()
+    rows = recent_decisions(conn, limit=1)
+    assert len(rows) == 1
+
+
+def test_recent_decisions_empty_db(tmp_path):
+    """No rows: returns an empty list, not an error."""
+    conn = _make_db(tmp_path)
+    assert recent_decisions(conn) == []
+
+
+# --- per_model tests ----------------------------------------------------------
+
+
+def test_per_model_aggregates_correctly(tmp_path):
+    """Sum/cost/energy/carbon/tokens match hand-computed values."""
+    conn = _make_db(tmp_path)
+    now = _now()
+    rows = [
+        ("cheap", 0.001, 5.0e-05, 2.4e-03, 100, 0.25),
+        ("cheap", 0.002, 3.0e-05, 1.2e-03, 200, 0.50),
+        ("dear", 0.010, 1.0e-04, 5.0e-03, 150, 0.75),
+    ]
+    for (model_id, cost, kwh, carbon, tokens, attr) in rows:
+        conn.execute(
+            "INSERT INTO energy_observations "
+            "(model_id, provider, cost_usd, energy_kwh, carbon_g_co2eq, "
+            "completion_tokens, attribution_ratio, observed_at) "
+            "VALUES (?, 'neuralwatt', ?, ?, ?, ?, ?, ?)",
+            (model_id, cost, kwh, carbon, tokens, attr, now.isoformat()),
+        )
+    conn.commit()
+
+    results = per_model(conn)
+    by_model = {r["model_id"]: r for r in results}
+
+    cheap = by_model["cheap"]
+    assert cheap["calls"] == 2
+    assert cheap["sum_cost_usd"] == pytest.approx(0.003)
+    assert cheap["sum_energy_kwh"] == pytest.approx(8.0e-05)
+    assert cheap["sum_carbon_g_co2eq"] == pytest.approx(3.6e-03)
+    assert cheap["avg_completion_tokens"] == pytest.approx(150.0)
+    assert cheap["avg_attribution_ratio"] == pytest.approx(0.375)
+
+    dear = by_model["dear"]
+    assert dear["calls"] == 1
+    assert dear["sum_cost_usd"] == pytest.approx(0.010)
+
+
+def test_per_model_empty_db(tmp_path):
+    """No energy rows: returns empty list."""
+    conn = _make_db(tmp_path)
+    assert per_model(conn) == []
+
+
+# --- verdict_mix tests --------------------------------------------------------
+
+
+def test_verdict_mix_counts_by_verdict(tmp_path):
+    """Counts match the inserted rows."""
+    conn = _make_db(tmp_path)
+    now = _now()
+    conn.execute(
+        "INSERT INTO verifications (model_id, provider, kind, verdict, observed_at) "
+        "VALUES ('m', 'neuralwatt', 'structural', 'ok', ?)",
+        (now.isoformat(),),
+    )
+    conn.execute(
+        "INSERT INTO verifications (model_id, provider, kind, verdict, observed_at) "
+        "VALUES ('m', 'neuralwatt', 'structural', 'ok', ?)",
+        (now.isoformat(),),
+    )
+    conn.execute(
+        "INSERT INTO verifications (model_id, provider, kind, verdict, observed_at) "
+        "VALUES ('m', 'neuralwatt', 'local_llm', 'malformed', ?)",
+        (now.isoformat(),),
+    )
+    conn.execute(
+        "INSERT INTO verifications (model_id, provider, kind, verdict, observed_at) "
+        "VALUES ('m', 'neuralwatt', 'structural', 'unverifiable', ?)",
+        (now.isoformat(),),
+    )
+    # Stale row outside the window
+    conn.execute(
+        "INSERT INTO verifications (model_id, provider, kind, verdict, observed_at) "
+        "VALUES ('m', 'neuralwatt', 'structural', 'truncated', ?)",
+        ((now - timedelta(days=30)).isoformat(),),
+    )
+    conn.commit()
+
+    result = verdict_mix(conn, since_days=7)
+    assert result["ok"] == 2
+    assert result["malformed"] == 1
+    assert result["unverifiable"] == 1
+    # truncated is > 7 days ago, so excluded
+    assert "truncated" not in result or result["truncated"] == 0
+
+
+def test_verdict_mix_empty_db(tmp_path):
+    """No rows: returns empty dict."""
+    conn = _make_db(tmp_path)
+    assert verdict_mix(conn) == {}
+
+
+# --- top_proficiency tests ----------------------------------------------------
+
+
+def test_top_proficiency_ordered_correctly(tmp_path):
+    """Models are returned ordered by blended_score DESC."""
+    conn = _make_db(tmp_path)
+    _seed_models(conn)
+    _seed_proficiency(conn)
+    conn.commit()
+
+    results = top_proficiency(conn, "coding_general")
+    assert len(results) == 3
+    assert results[0]["model_id"] == "dear"     # 0.95
+    assert results[1]["model_id"] == "cheap"    # 0.90
+    assert results[2]["model_id"] == "tiny"     # 0.70
+
+
+def test_top_proficiency_filter_by_category(tmp_path):
+    """Requests for one category exclude models that have scores only for another."""
+    conn = _make_db(tmp_path)
+    _seed_models(conn)
+    # Only code categories
+    conn.execute(
+        """
+        INSERT INTO proficiency (
+            model_id, provider, category, blended_score, source, last_updated
+        ) VALUES ('cheap', 'neuralwatt', 'coding_general', 0.90, 'self_eval_thin', '2026-01-01T00:00:00+00:00')
+        """,
+    )
+    conn.execute(
+        """
+        INSERT INTO proficiency (
+            model_id, provider, category, blended_score, source, last_updated
+        ) VALUES ('cheap', 'neuralwatt', 'docs_writing', 0.85, 'self_eval_thin', '2026-01-01T00:00:00+00:00')
+        """,
+    )
+    conn.commit()
+
+    coding = top_proficiency(conn, "coding_general")
+    docs = top_proficiency(conn, "docs_writing")
+    assert len(coding) == 1
+    assert coding[0]["model_id"] == "cheap"
+    assert len(docs) == 1
+    assert docs[0]["model_id"] == "cheap"
+
+
+def test_top_proficiency_empty_for_missing_category(tmp_path):
+    """No proficiency rows for category → empty list."""
+    conn = _make_db(tmp_path)
+    _seed_models(conn)
+    assert top_proficiency(conn, "nonexistent_category") == []
+
+
+# --- /health endpoint compatibility -------------------------------------------
+
+
+def test_health_endpoint_returns_scoring_key(tmp_path, monkeypatch):
+    """/health still returns the same SHAPE after moving functions to metrics."""
+    db_path = tmp_path / "test.db"
+    conn = _make_db(tmp_path)
+    _seed_models(conn)
+    conn.commit()
+    conn.close()
+
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
+    # Disable local verification to avoid Ollama dependency
+    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
+    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
+
+    with TestClient(dispatcher.app) as client:
+        resp = client.get("/health")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert "scoring" in data
+    assert "routable_models" in data["scoring"]
+    assert "with_energy_data" in data["scoring"]
+    assert "with_proficiency_data" in data["scoring"]
+    assert "quota" in data["scoring"]
+    assert "warnings" in data["scoring"]

tests/test_metrics_endpoint.py

@@ -0,0 +1,201 @@
+"""Tests for the GET /metrics endpoint on dispatcher.
+
+Seeds a throwaway temp DB and asserts on the actual JSON returned by a real
+TestClient GET (never a mock-call assertion), to defeat
+``misleading_success_output``.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+from starlette.testclient import TestClient
+
+import dispatcher
+from config import load_config
+
+ROOT = Path(__file__).resolve().parent.parent
+SCHEMA_SQL = (ROOT / "schema.sql").read_text()
+CFG = load_config(str(ROOT / "config.yaml"))
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _make_db(tmp_path: Path) -> sqlite3.Connection:
+    conn = sqlite3.connect(str(tmp_path / "test.db"))
+    conn.row_factory = sqlite3.Row
+    conn.executescript(SCHEMA_SQL)
+    return conn
+
+
+def _seed_models(conn: sqlite3.Connection) -> None:
+    for model_id, tier, context, cost, vision in (
+        ("cheap", 2, 262128, 0.30, 1),
+        ("dear", 2, 262128, 9.00, 0),
+        ("tiny", 1, 131072, 0.10, 1),
+    ):
+        conn.execute(
+            """
+            INSERT INTO models (
+                model_id, provider, base_model_id, tier, context_window,
+                effective_context_window, max_output_tokens,
+                cost_per_1m_prompt, cost_per_1m_completion,
+                supports_vision, supports_json_mode,
+                latency_class, reasoning_mode, context_variant,
+                access_level, availability, last_updated
+            ) VALUES (?, 'neuralwatt', ?, ?, ?, 192500, 16384, ?, ?,
+                      ?, 1, 'standard', 'default', 'full', 'public', 'active',
+                      '2026-08-22T00:00:00+00:00')
+            """,
+            (model_id, model_id, tier, context, cost, cost / 3, vision),
+        )
+    conn.commit()
+
+
+def _seed_decision(conn: sqlite3.Connection) -> None:
+    conn.execute(
+        """
+        INSERT INTO route_decisions (
+            observed_at, kind, task_category, task_tier, required_context_tokens,
+            confidence, classifier_ms, classification_source, latency_tolerance,
+            candidates_considered, selected_model, selected_provider,
+            runner_up_models, est_cost_usd, est_proficiency,
+            session_key, tools, images, json_mode, streamed
+        ) VALUES (?, 'route', 'coding_general', 2, 100, 0.95, 200,
+                  'classifier', 'interactive', 5, 'cheap', 'neuralwatt',
+                  '[{"model_id":"dear","provider":"neuralwatt"}]',
+                  0.001, 0.9, 'abc123', 0, 0, 0, 0)
+        """,
+        (_now().isoformat(),),
+    )
+    conn.commit()
+
+
+def _seed_energy(conn: sqlite3.Connection) -> None:
+    now = _now()
+    conn.execute(
+        "INSERT INTO energy_observations "
+        "(model_id, provider, task_category, completion_tokens, energy_kwh, "
+        "cost_usd, carbon_g_co2eq, attribution_ratio, observed_at) "
+        "VALUES ('cheap', 'neuralwatt', 'coding_general', 100, 5.0e-05, 0.001, "
+        "2.4e-03, 0.25, ?)",
+        ((now - timedelta(days=2)).isoformat(),),
+    )
+    conn.commit()
+
+
+def _seed_verification(conn: sqlite3.Connection) -> None:
+    conn.execute(
+        "INSERT INTO verifications (model_id, provider, kind, verdict, observed_at) "
+        "VALUES ('cheap', 'neuralwatt', 'structural', 'ok', ?)",
+        (_now().isoformat(),),
+    )
+    conn.commit()
+
+
+def _seed_proficiency(conn: sqlite3.Connection) -> None:
+    conn.execute(
+        "INSERT INTO proficiency (model_id, provider, category, blended_score, "
+        "source, last_updated) "
+        "VALUES ('cheap', 'neuralwatt', 'coding_general', 0.9, "
+        "'self_eval_thin', '2026-01-01T00:00:00+00:00')",
+    )
+    conn.commit()
+
+
+@pytest.fixture
+def seeded_client(tmp_path, monkeypatch):
+    """A TestClient wired to a seeded temp DB, at /metrics."""
+    conn = _make_db(tmp_path)
+    _seed_models(conn)
+    for i in range(60):  # exceed the 50 cap
+        _seed_decision(conn)
+    _seed_energy(conn)
+    _seed_verification(conn)
+    _seed_proficiency(conn)
+    conn.close()
+
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
+    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
+    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
+
+    with TestClient(dispatcher.app) as client:
+        yield client
+
+
+def test_metrics_endpoint_has_all_top_level_keys(seeded_client):
+    """GET /metrics returns 200 with every required top-level key."""
+    resp = seeded_client.get("/metrics")
+    assert resp.status_code == 200
+    data = resp.json()
+    for key in (
+        "quota",
+        "coverage",
+        "recent_decisions",
+        "per_model",
+        "verdict_mix",
+        "top_proficiency",
+        "generated_at",
+    ):
+        assert key in data, f"missing top-level key {key!r}"
+
+
+def test_metrics_recent_decisions_capped_and_carries_provider(seeded_client):
+    """recent_decisions is capped at 50 and each row has selected_provider."""
+    resp = seeded_client.get("/metrics")
+    data = resp.json()
+    decisions = data["recent_decisions"]
+    assert isinstance(decisions, list)
+    assert len(decisions) <= 50
+    assert len(decisions) > 0
+    for row in decisions:
+        assert "selected_provider" in row
+    # DESC order: the first decision has the largest id.
+    first_id = decisions[0]["id"]
+    for row in decisions[1:]:
+        assert row["id"] <= first_id
+
+
+def test_metrics_aggregations_are_populated(seeded_client):
+    """per_model / verdict_mix / top_proficiency reflect seeded data."""
+    resp = seeded_client.get("/metrics")
+    data = resp.json()
+    assert isinstance(data["per_model"], list)
+    assert any(r["model_id"] == "cheap" for r in data["per_model"])
+    assert data["per_model"][0]["calls"] == 1
+    assert data["verdict_mix"]["ok"] == 1
+    assert isinstance(data["top_proficiency"], list)
+    assert data["top_proficiency"][0]["model_id"] == "cheap"
+    assert data["quota"] is not None
+
+
+def test_metrics_contains_no_session_dir(seeded_client):
+    """The JSON must never name session_dir or expose conversation text."""
+    body = seeded_client.get("/metrics").text
+    assert "session_dir" not in body
+
+
+def test_metrics_empty_db_returns_200(monkeypatch, tmp_path):
+    """Fresh empty temp DB: 200 with empty arrays, no exception."""
+    conn = _make_db(tmp_path)
+    conn.close()
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
+    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.routing, "require_vision", False)
+    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
+
+    with TestClient(dispatcher.app) as client:
+        resp = client.get("/metrics")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["recent_decisions"] == []
+    assert data["per_model"] == []
+    assert data["verdict_mix"] == {}
+    assert data["top_proficiency"] == []
+    assert "generated_at" in data

tests/test_outcome_attribution.py

@@ -25,6 +25,7 @@ import pytest
 
 import dispatcher
 from dispatcher import AMBIGUOUS, SEED_CATEGORY, _most_recent_if_unambiguous
+from metrics import quota_burn
 
 ROOT = Path(__file__).resolve().parent.parent
 SCHEMA_SQL = (ROOT / "schema.sql").read_text()
@@ -132,7 +133,6 @@ def test_quota_burn_counts_only_the_last_thirty_days(db, tmp_path, monkeypatch):
     gets wrong. A row from 45 days ago would be excluded either way and would
     pin nothing.
     """
-    monkeypatch.setattr(dispatcher.cfg.database, "path", str(tmp_path / "test.db"))
     monkeypatch.setattr(dispatcher.cfg.objective, "plan_kwh_per_period", 6.25)
     just_outside = (_now() - timedelta(days=30)).replace(
         hour=0, minute=0, second=0, microsecond=1
@@ -153,4 +153,4 @@ def test_quota_burn_counts_only_the_last_thirty_days(db, tmp_path, monkeypatch):
     )
     db.commit()
 
-    assert dispatcher.quota_burn()["metered_kwh_30d"] == pytest.approx(0.25)
+    assert quota_burn(db, dispatcher.cfg)["metered_kwh_30d"] == pytest.approx(0.25)

tests/test_route_decisions.py

@@ -0,0 +1,660 @@
+"""Tests for the route_decisions table, its inline-create helper, and the gate.
+
+The monitoring TUI (see .omo/plans/router-monitoring-tui.md) needs a record of
+every routing decision — which model was picked and why — that survives in the
+datbase rather than only in the journal. This file pins the three pieces todo #1
+adds:
+
+- the `route_decisions` table in schema.sql (columns and the guarded index),
+- `dispatcher.ensure_route_decisions(conn)` — the idempotent inline-create
+  helper that is the *only* way the table appears on a live router.db (the
+  live DB is never recreated; schema.sql alone is CREATE TABLE IF NOT EXISTS
+  and silently does nothing to an existing DB),
+- the `logging.log_route_decisions` config gate.
+
+The database tests follow the same offline temp-DB pattern as
+tests/test_chat_completions.py: a throwaway SQLite file seeded from schema.sql,
+never the live router.db.
+"""
+
+import json
+import sqlite3
+from pathlib import Path
+
+import pytest
+from starlette.testclient import TestClient
+
+import dispatcher
+from dispatcher import Classification, app
+import config
+import metrics
+
+ROOT = Path(__file__).resolve().parent.parent
+SCHEMA_SQL = (ROOT / "schema.sql").read_text()
+
+ROUTE_DECISIONS_COLUMNS = [
+    "id",
+    "observed_at",
+    "kind",
+    "task_category",
+    "task_tier",
+    "required_context_tokens",
+    "confidence",
+    "classifier_ms",
+    "classification_source",
+    "latency_tolerance",
+    "candidates_considered",
+    "selected_model",
+    "selected_provider",
+    "runner_up_models",
+    "est_cost_usd",
+    "est_proficiency",
+    "rejected_reason",
+    "session_key",
+    "tools",
+    "images",
+    "json_mode",
+    "streamed",
+]
+
+
+def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
+    row = conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
+        (table,),
+    ).fetchone()
+    return row is not None
+
+
+def _index_exists(conn: sqlite3.Connection, index: str) -> bool:
+    row = conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='index' AND name=?",
+        (index,),
+    ).fetchone()
+    return row is not None
+
+
+def _schema_minus_route_decisions() -> str:
+    """schema.sql with the route_decisions block removed, for the failure case."""
+    lines = []
+    skipping = False
+    for line in SCHEMA_SQL.splitlines():
+        stripped = line.strip()
+        if stripped.startswith("CREATE TABLE IF NOT EXISTS route_decisions"):
+            skipping = True
+            continue
+        if skipping:
+            # The create ends at the closing paren + semicolon of the table
+            # statement. Anything still in the table body is skipped.
+            if stripped == ");":
+                skipping = False
+            continue
+        if "idx_route_decisions_observed" in line:
+            continue
+        lines.append(line)
+    return "\n".join(lines)
+
+
+# --- schema round-trips -----------------------------------------------------
+
+
+def test_schema_defines_route_decisions_table():
+    """schema.sql declares the table, so a fresh DB from it has it already."""
+    assert "CREATE TABLE IF NOT EXISTS route_decisions" in SCHEMA_SQL
+
+
+def test_schema_index_is_guarded():
+    """The observed_at index must be IF NOT EXISTS so re-applying is a no-op."""
+    assert "idx_route_decisions_observed" in SCHEMA_SQL
+    assert (
+        "CREATE INDEX IF NOT EXISTS idx_route_decisions_observed "
+        "ON route_decisions (observed_at)" in SCHEMA_SQL
+    )
+
+
+# --- happy path: fresh DB from full schema ----------------------------------
+
+
+def test_fresh_schema_already_has_table(tmp_path):
+    conn = sqlite3.connect(tmp_path / "fresh.db")
+    conn.executescript(SCHEMA_SQL)
+    assert _table_exists(conn, "route_decisions")
+    for col in ROUTE_DECISIONS_COLUMNS:
+        assert col in {r[1] for r in conn.execute("PRAGMA table_info(route_decisions)")}
+    assert _index_exists(conn, "idx_route_decisions_observed")
+    conn.close()
+
+
+def test_ensure_route_decisions_is_idempotent(tmp_path):
+    """Fresh DB already has the table; calling the helper twice no-ops."""
+    conn = sqlite3.connect(tmp_path / "idem.db")
+    conn.executescript(SCHEMA_SQL)
+    # Seed some rows in another table so we can prove nothing is dropped.
+    conn.execute(
+        "INSERT INTO models (model_id, provider, last_updated) "
+        "VALUES ('m1', 'neuralwatt', '2026-01-01T00:00:00+00:00')"
+    )
+    conn.commit()
+
+    dispatcher.ensure_route_decisions(conn)  # first call
+    dispatcher.ensure_route_decisions(conn)  # second call: must no-op cleanly
+
+    assert _table_exists(conn, "route_decisions")
+    count = conn.execute("SELECT COUNT(*) FROM models").fetchone()[0]
+    assert count == 1  # pre-existing rows survived
+    conn.close()
+
+
+# --- failure path: pre-existing DB WITHOUT the table ------------------------
+
+
+def test_ensure_route_decisions_adds_table_without_dropping_rows(tmp_path):
+    """A DB that predates the table gets it added; existing rows survive."""
+    conn = sqlite3.connect(tmp_path / "old.db")
+    conn.executescript(_schema_minus_route_decisions())
+    assert not _table_exists(conn, "route_decisions")
+    # A row in a genuinely existing table, to prove it survives the upgrade.
+    conn.execute(
+        "INSERT INTO models (model_id, provider, last_updated) "
+        "VALUES ('legacy', 'neuralwatt', '2026-01-01T00:00:00+00:00')"
+    )
+    conn.commit()
+
+    dispatcher.ensure_route_decisions(conn)
+
+    assert _table_exists(conn, "route_decisions")
+    assert _index_exists(conn, "idx_route_decisions_observed")
+    legacy = conn.execute(
+        "SELECT model_id FROM models WHERE model_id='legacy'"
+    ).fetchone()
+    assert legacy is not None  # the existing row was not dropped
+    conn.close()
+
+
+def test_ensure_route_decisions_allows_insert(tmp_path):
+    """After the helper runs, the table actually accepts the documented shape."""
+    conn = sqlite3.connect(tmp_path / "insert.db")
+    conn.executescript(_schema_minus_route_decisions())
+    dispatcher.ensure_route_decisions(conn)
+    conn.execute(
+        """
+        INSERT INTO route_decisions (
+            observed_at, kind, task_category, task_tier,
+            required_context_tokens, confidence, classifier_ms,
+            classification_source, latency_tolerance, candidates_considered,
+            selected_model, selected_provider, runner_up_models,
+            est_cost_usd, est_proficiency, rejected_reason, session_key,
+            tools, images, json_mode, streamed
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        """,
+        (
+            "2026-01-01T00:00:00+00:00", "route", "coding_general", 2, 500,
+            0.95, 1868, "classifier", "interactive", 8, "deepseek-v4-flash",
+            "neuralwatt", '[{"model_id": "gemma-4-31b", "provider": "neuralwatt"}]',
+            0.00016296, 1.0, None, "sess-hash", 0, 0, 0, 1,
+        ),
+    )
+    conn.commit()
+    kind = conn.execute(
+        "SELECT kind FROM route_decisions WHERE selected_model='deepseek-v4-flash'"
+    ).fetchone()
+    assert kind is not None and kind[0] == "route"
+    conn.close()
+
+
+def test_persist_ensure_on_write_fixes_live_db_missing_table(tmp_path, monkeypatch):
+    """A live router.db without route_decisions gets it on the WRITE path.
+
+    F4 scope-fidelity regression: ``persist_route_decision`` INSERTs without
+    ever calling ``ensure_route_decisions``, so if a live DB lacks the table
+    and the module-load startup hook did not run (a test harness, a process
+    that calls persist first, a future lazy-import refactor), every decision
+    row is silently swallowed (the INSERT raises no-such-table) and /metrics'
+    recent_decisions 500s. Mirroring proficiency_store._write -> ensure_columns,
+    the migration must be guaranteed on the write path too, not only at module
+    load. The temp DB is deliberately the schema-minus-route_decisions shape —
+    a DB that predates the feature.
+    """
+    db_path = tmp_path / "live-no-table.db"
+    conn = sqlite3.connect(db_path)
+    conn.executescript(_schema_minus_route_decisions())
+    assert not _table_exists(conn, "route_decisions")
+    conn.close()
+
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
+    monkeypatch.setattr(dispatcher.cfg.logging, "log_route_decisions", True)
+
+    # The write itself. Best-effort means it must not raise even on a missing
+    # table; then the row must actually land and the table must exist.
+    dispatcher.persist_route_decision(
+        "route",
+        classification=Classification(
+            task_category="coding_general", task_tier=2,
+            required_context_tokens=100, confidence=0.9,
+        ),
+        latency_tolerance="interactive",
+    )
+
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    assert _table_exists(conn, "route_decisions"), \
+        "the migration must run on the write path so the table exists"
+    rows = conn.execute(
+        "SELECT kind FROM route_decisions WHERE kind='route'"
+    ).fetchall()
+    assert len(rows) == 1, "the decision row must persist once the table exists"
+
+    # /metrics / recent_decisions against the same live-DB shape must not 500:
+    # reading must succeed now that the table exists.
+    rec = metrics.recent_decisions(conn)
+    assert len(rec) == 1
+    assert rec[0]["kind"] == "route"
+    conn.close()
+
+
+# --- config gate ------------------------------------------------------------
+
+
+def test_log_route_decisions_gate_defaults_on():
+    """The key is declared and defaults to on, matching the config.yaml value."""
+    cfg = config.load_config(str(ROOT / "config.yaml"))
+    assert cfg.logging.log_route_decisions is True
+
+
+def test_config_has_log_route_decisions_key():
+    """The strict config accepts the key — it must be declared or load fails."""
+    assert "log_route_decisions" in (ROOT / "config.yaml").read_text()
+
+
+# =============================================================================
+# Todo #2: persist_route_decision wired into every decision path.
+# =============================================================================
+
+CHEAP = "cheap-model"
+DEAR = "dear-model"
+
+
+class FakeResponse:
+    """Just enough of requests.Response for the dispatcher's provider calls."""
+
+    def __init__(self, payload=None, *, status_code=200, lines=None):
+        self.status_code = status_code
+        self._payload = payload or {}
+        self._lines = lines or []
+        self.text = json.dumps(self._payload)
+        self.closed = False
+
+    def json(self):
+        return self._payload
+
+    def iter_lines(self, decode_unicode=False):
+        yield from self._lines
+
+    def close(self):
+        self.closed = True
+
+
+def _completion(model, content="hello there"):
+    return {
+        "id": "chatcmpl-dec-1",
+        "model": model,
+        "choices": [
+            {"message": {"role": "assistant", "content": content},
+             "finish_reason": "stop"}
+        ],
+        "usage": {"prompt_tokens": 31, "completion_tokens": 12},
+        "energy": {"energy_kwh": 5.0e-05, "carbon_g_co2eq": 2.4e-03},
+        "cost": {"request_cost_usd": 4.0e-04},
+    }
+
+
+_STREAM_LINES = [
+    'data: {"id":"chatcmpl-stream-dec","choices":[{"delta":{"content":"hel"}}]}',
+    "",
+    'data: {"id":"chatcmpl-stream-dec","choices":[{"delta":{"content":"lo"},'
+    '"finish_reason":"stop"}],"usage":{"prompt_tokens":31,'
+    '"completion_tokens":9}}',
+    "",
+    "data: [DONE]",
+    "",
+]
+
+
+def _messages(text="write me a function"):
+    return [{"role": "user", "content": text}]
+
+
+def _image_messages():
+    return [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "what is in this image?"},
+                {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
+            ],
+        }
+    ]
+
+
+@pytest.fixture
+def decision_router(tmp_path, monkeypatch):
+    """A routable dispatcher over a throwaway DB; nothing dials out."""
+    db_path = tmp_path / "decisions.db"
+    conn = sqlite3.connect(db_path)
+    conn.executescript(SCHEMA_SQL)
+    for model_id, completion_price, vision in (
+        (CHEAP, 0.30, 1),
+        (DEAR, 9.00, 0),
+    ):
+        conn.execute(
+            """
+            INSERT INTO models (
+                model_id, provider, base_model_id, tier, context_window,
+                effective_context_window, max_output_tokens,
+                cost_per_1m_prompt, cost_per_1m_completion,
+                supports_vision, supports_json_mode,
+                latency_class, reasoning_mode, context_variant,
+                access_level, availability, last_updated
+            ) VALUES (?, 'neuralwatt', ?, 2, 262128, 192500, 16384, ?, ?,
+                      ?, 1, 'standard', 'default', 'full', 'public', 'active',
+                      '2026-08-22T00:00:00+00:00')
+            """,
+            (model_id, model_id, completion_price / 3, completion_price, vision),
+        )
+    conn.commit()
+    conn.close()
+
+    monkeypatch.setattr(dispatcher.cfg.database, "path", str(db_path))
+    monkeypatch.setattr(dispatcher.cfg.verification, "local_llm_enabled", False)
+    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", False)
+    monkeypatch.setenv("NEURALWATT_API_KEY", "test-key")
+    # ensure the gate is on for the happy-path tests (default, but pin it).
+    monkeypatch.setattr(dispatcher.cfg.logging, "log_route_decisions", True)
+
+    calls = []
+
+    def fake_post(url, headers=None, json=None, stream=False, timeout=None):
+        calls.append({"url": url, "body": json, "stream": stream})
+        if stream:
+            return FakeResponse(lines=_STREAM_LINES)
+        return FakeResponse(_completion(json["model"]))
+
+    monkeypatch.setattr(dispatcher.requests, "post", fake_post)
+    monkeypatch.setattr(
+        dispatcher, "classify",
+        lambda task, context: Classification(
+            task_category="coding_general", task_tier=2,
+            required_context_tokens=100, confidence=0.9,
+        ),
+    )
+
+    class _Raw:
+        text = json.dumps(_completion(CHEAP))
+
+    class _Completions:
+        @property
+        def with_raw_response(self):
+            return self
+
+        def create(self, **kwargs):
+            return _Raw()
+
+    class _FakeClient:
+        chat = type("_Chat", (), {"completions": _Completions()})()
+
+    monkeypatch.setattr(dispatcher, "_provider_client", lambda provider: _FakeClient())
+    yield TestClient(app), db_path
+
+
+def _rows(db_path):
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    rows = conn.execute(
+        "SELECT * FROM route_decisions ORDER BY id ASC"
+    ).fetchall()
+    conn.close()
+    return rows
+
+
+def _drop_cheap(db_path):
+    """Make CHEAP ineligible so DEAR (or nothing) remains."""
+    conn = sqlite3.connect(db_path)
+    conn.execute("UPDATE models SET tier = 1 WHERE model_id = ?", (CHEAP,))
+    conn.commit()
+    conn.close()
+
+
+# --- happy paths ------------------------------------------------------------
+
+
+def test_route_endpoint_persists_one_row(decision_router):
+    client, db_path = decision_router
+    resp = client.post(
+        "/route", json={"task": "write me a function"}
+    )
+    assert resp.status_code == 200
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    r = rows[0]
+    assert r["kind"] == "route"
+    assert r["selected_model"] == CHEAP
+    assert r["selected_provider"] == "neuralwatt"
+    assert r["classification_source"] == "classifier"
+    assert r["task_category"] == "coding_general"
+    assert r["task_tier"] == 2
+    assert r["latency_tolerance"] == "interactive"
+    assert r["session_key"] is None
+    assert "session_dir" not in r.keys()
+
+
+def test_route_endpoint_override_has_classifier_ms_null(decision_router):
+    client, db_path = decision_router
+    resp = client.post(
+        "/route",
+        json={
+            "task": "x",
+            "task_category": "coding_refactor",
+            "task_tier": 3,
+            "required_context_tokens": 5000,
+        },
+    )
+    assert resp.status_code == 200
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    r = rows[0]
+    assert r["classification_source"] == "override"
+    assert r["classifier_ms"] is None, "an override never consulted the classifier"
+    assert r["task_category"] == "coding_refactor"
+
+
+def test_dispatch_endpoint_persists_one_row(decision_router):
+    client, db_path = decision_router
+    resp = client.post(
+        "/dispatch",
+        json={"task": "write me a function",
+              "task_category": "coding_general",
+              "task_tier": 2,
+              "required_context_tokens": 100},
+    )
+    assert resp.status_code == 200
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    r = rows[0]
+    assert r["kind"] == "dispatch"
+    assert r["selected_model"] == CHEAP
+    assert r["classification_source"] == "override"
+
+
+def test_routed_chat_persists_one_row(decision_router):
+    client, db_path = decision_router
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto", "messages": _messages()},
+    )
+    assert resp.status_code == 200
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    r = rows[0]
+    assert r["kind"] == "chat"
+    assert r["selected_model"] == CHEAP
+    assert r["selected_provider"] == "neuralwatt"
+    assert r["classification_source"] == "classifier"
+    assert r["session_key"] is not None
+    # The session key is a hash — never a directory, never content.
+    assert len(r["session_key"]) == 16
+    assert "/" not in (r["session_key"] or "")
+
+
+def test_routed_chat_reroute_keeps_classifier_source(decision_router):
+    """Re-routing for measured context must not persist source='override'."""
+    client, db_path = decision_router
+    # >100 measured tokens (measured = chars/3), so chat_completions reroutes.
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto", "messages": _messages("refactor " + "x " * 600)},
+    )
+    assert resp.status_code == 200
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    assert rows[0]["classification_source"] == "classifier", \
+        "the re-route's source='override' must not overwrite the classifier's"
+    assert rows[0]["classifier_ms"] is not None
+
+
+def test_streamed_routed_chat_persists_one_row(decision_router):
+    client, db_path = decision_router
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto", "messages": _messages(), "stream": True},
+    )
+    assert resp.status_code == 200
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    assert rows[0]["kind"] == "chat"
+    assert rows[0]["selected_model"] == CHEAP
+    assert rows[0]["streamed"] == 1
+
+
+def test_passthrough_persists_one_row_with_no_nameerror(decision_router):
+    """The pre-existing pass-through NameError must stay gone, and a row lands."""
+    client, db_path = decision_router
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": DEAR, "messages": _messages()},
+    )
+    assert resp.status_code == 200
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    r = rows[0]
+    assert r["kind"] == "passthrough"
+    assert r["selected_model"] == DEAR
+    assert r["selected_provider"] == "neuralwatt"
+    assert r["classification_source"] is None
+    assert r["session_key"] is not None
+
+
+def test_local_vision_success_persists_one_local_row(decision_router, monkeypatch):
+    client, db_path = decision_router
+    _drop_cheap(db_path)
+    monkeypatch.setattr(dispatcher.cfg.local_vision, "enabled", True)
+
+    def fake_post(url, headers=None, json=None, stream=False, timeout=None):
+        return FakeResponse(
+            {"choices": [{"message": {"role": "assistant",
+                                      "content": "local caption"},
+                          "finish_reason": "stop"}]}
+        )
+
+    monkeypatch.setattr(dispatcher.requests, "post", fake_post)
+
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto", "messages": _image_messages()},
+    )
+    assert resp.status_code == 200
+
+    rows = _rows(db_path)
+    assert len(rows) == 1, "one decision, and only one: the local_vision row"
+    r = rows[0]
+    assert r["kind"] == "local_vision"
+    assert r["selected_model"] == dispatcher.cfg.local_vision.model
+    assert r["selected_provider"] == "local"
+    assert r["rejected_reason"] is None
+    assert r["images"] == 1
+
+
+def test_no_candidate_422_still_persists_a_rejection_row(decision_router):
+    client, db_path = decision_router
+    # No vision cloud candidate, local fallback disabled -> 422.
+    _drop_cheap(db_path)
+    resp = client.post(
+        "/v1/chat/completions",
+        json={"model": "auto", "messages": _image_messages()},
+    )
+    assert resp.status_code == 422
+
+    rows = _rows(db_path)
+    assert len(rows) == 1
+    r = rows[0]
+    assert r["kind"] == "chat"
+    assert r["selected_model"] is None
+    assert r["rejected_reason"] is not None
+    assert "vision" in r["rejected_reason"]
+
+
+# --- failure modes: best-effort, config-gated -------------------------------
+
+
+def test_gate_off_writes_nothing_but_routing_still_200(decision_router):
+    client, db_path = decision_router
+    dispatcher.cfg.logging.log_route_decisions = False
+    try:
+        resp = client.post("/route", json={"task": "write me a function"})
+        assert resp.status_code == 200
+    finally:
+        dispatcher.cfg.logging.log_route_decisions = True
+
+    assert _rows(db_path) == [], "the gate off must leave the table untouched"
+
+
+def _raise_on_route_decisions_insert(db_path):
+    real_db = dispatcher._db
+
+    class GuardingConn:
+        def __init__(self, conn):
+            self._conn = conn
+
+        def __getattr__(self, name):
+            return getattr(self._conn, name)
+
+        def execute(self, sql, parameters=()):
+            if isinstance(sql, str) and "INSERT INTO route_decisions" in sql:
+                raise sqlite3.OperationalError("database is locked")
+            return self._conn.execute(sql, parameters)
+
+    def wrapped_db():
+        return GuardingConn(real_db())
+
+    return wrapped_db
+
+
+def test_db_write_failure_never_fails_routing(decision_router, monkeypatch):
+    """A locked/read-only DB must not error the request; persistence is best-effort."""
+    client, db_path = decision_router
+    monkeypatch.setattr(dispatcher, "_db", _raise_on_route_decisions_insert(db_path))
+
+    resp = client.post("/route", json={"task": "write me a function"})
+    assert resp.status_code == 200, "a failed decision write must never fail routing"
+
+    # And the same holds for a routed completion.
+    resp2 = client.post(
+        "/v1/chat/completions", json={"model": "auto", "messages": _messages()}
+    )
+    assert resp2.status_code == 200

tests/test_router_cli.py

@@ -0,0 +1,225 @@
+"""Tests for router_cli.py -- the one-shot routing-decision printer.
+
+The CLI is route-only: it POSTs to the FastAPI ``POST /route`` endpoint and
+prints the decision tree without ever dispatching to a provider. These tests
+exercise the decision-rendering / posting logic through ``run()`` with an
+injected ``post`` callback so no live server or provider is involved.
+"""
+
+import json
+
+import router_cli
+
+
+def _candidate(model_id, cost=0.0005, proficiency=0.95):
+    return {
+        "model_id": model_id,
+        "provider": "neuralwatt",
+        "tier": 2,
+        "latency_class": "standard",
+        "reasoning_mode": "default",
+        "context_variant": "full",
+        "effective_context_window": 262128,
+        "cost": cost,
+        "energy": 1.23e-05,
+        "eco": 0.5,
+        "list_price_per_1m": 0.42,
+        "composite": 0.8,
+        "cost_score": 0.9,
+        "proficiency_score": proficiency,
+    }
+
+
+def _route_payload(selected=True, rejected_reason=None):
+    payload = {
+        "classification": {
+            "task_category": "coding_general",
+            "task_tier": 2,
+            "required_context_tokens": 5000,
+            "confidence": 0.9,
+            "escalated": False,
+            "source": "classifier",
+        },
+        "latency_tolerance": "interactive",
+        "candidates_considered": 4,
+        "selected": (
+            _candidate("deepseek-v4-flash", cost=1.6e-05, proficiency=0.97)
+            if selected
+            else None
+        ),
+        "runners_up": [
+            _candidate("gemma-4-31b", cost=5.02e-05, proficiency=0.91),
+            _candidate("kimi-k3", cost=2.2e-05, proficiency=0.9),
+        ],
+    }
+    # The 422 no-candidate body carries the limits/rejected reason in `detail`.
+    if rejected_reason is not None:
+        payload["detail"] = rejected_reason
+    return payload
+
+
+class _FakeResponse:
+    """minimal requests.Response stand-in with the fields router_cli reads."""
+
+    def __init__(self, status_code, payload):
+        self.status_code = status_code
+        self._payload = payload
+
+    def json(self):
+        return self._payload
+
+    @property
+    def text(self):
+        return json.dumps(self._payload)
+
+
+def _fake_post(route_payload, status_code=200):
+    captured = {}
+
+    def post(url, json=None, **kwargs):
+        captured["url"] = url
+        captured["body"] = json
+        return _FakeResponse(status_code, route_payload)
+
+    return post, captured
+
+
+# --- happy path -----------------------------------------------------------
+
+
+def test_happy_path_renders_decision_tree(capsys):
+    """The rendered text names the selected model, the category, and a cost."""
+    post, _ = _fake_post(_route_payload())
+    code = router_cli.run(
+        "Refactor this Django view", post=post
+    )
+    out = capsys.readouterr().out
+
+    assert code == 0
+    assert "selected" in out.lower()
+    assert "coding_general" in out
+    # A US$ cost figure for the selected model.
+    assert "1.6e-05" in out or "US$" in out
+
+
+def test_happy_path_includes_runners_up_and_candidates(capsys):
+    post, _ = _fake_post(_route_payload())
+    router_cli.run("Refactor this Django view", post=post)
+    out = capsys.readouterr().out
+
+    assert "candidates_considered" in out.lower() or "considered" in out.lower()
+    assert "gemma-4-31b" in out  # a runner-up
+    assert "kimi-k3" in out
+
+
+def test_json_flag_emits_raw_route_json(capsys):
+    payload = _route_payload()
+    post, _ = _fake_post(payload)
+    code = router_cli.run("hello", as_json=True, post=post)
+    out = capsys.readouterr().out.strip()
+
+    assert code == 0
+    parsed = json.loads(out)
+    assert parsed["classification"]["task_category"] == "coding_general"
+    assert parsed["selected"]["model_id"] == "deepseek-v4-flash"
+    assert parsed["candidates_considered"] == 4
+    assert len(parsed["runners_up"]) == 2
+
+
+def test_payload_includes_overrides_only_when_given(capsys):
+    post, captured = _fake_post(_route_payload())
+    router_cli.run(
+        "task",
+        category="coding_refactor",
+        tier=3,
+        context=12000,
+        post=post,
+    )
+    capsys.readouterr()
+
+    body = captured["body"]
+    assert body["task"] == "task"
+    assert body["task_category"] == "coding_refactor"
+    assert body["task_tier"] == 3
+    assert body["required_context_tokens"] == 12000
+
+
+def test_payload_omits_overrides_when_not_given(capsys):
+    post, captured = _fake_post(_route_payload())
+    router_cli.run("task", post=post)
+    capsys.readouterr()
+
+    body = captured["body"]
+    assert body["task"] == "task"
+    assert "task_category" not in body
+    assert "task_tier" not in body
+    assert "required_context_tokens" not in body
+
+
+# --- failure: 422 no candidate -------------------------------------------
+
+def test_422_no_candidate_prints_rejection_and_exits_nonzero(capsys):
+    """A no-candidate /route returns 422 with the rejected reason; the CLI
+    prints it and exits non-zero without a traceback."""
+    post, _ = _fake_post(
+        _route_payload(selected=False, rejected_reason="no model fits"),
+        status_code=422,
+    )
+    code = router_cli.run("task", post=post)
+    captured = capsys.readouterr()
+    out, err = captured.out, captured.err
+
+    assert code != 0
+    assert "no model fits" in out or "no model fits" in err
+    # No traceback leaked.
+    assert "Traceback" not in out
+    assert "Traceback" not in err
+
+
+# --- failure: connection error -------------------------------------------
+
+def test_connection_error_prints_clean_error_and_exits_nonzero(capsys):
+    def post(url, json=None, **kwargs):
+        raise ConnectionError("refused")
+
+    code = router_cli.run("task", post=post)
+    captured = capsys.readouterr()
+    out, err = captured.out, captured.err
+
+    assert code != 0
+    assert "refused" in out or "refused" in err
+    assert "Traceback" not in out
+    assert "Traceback" not in err
+
+
+# --- malformed input ------------------------------------------------------
+
+def test_tier_flag_rejects_out_of_range(capsys):
+    """A --tier outside 1..3 must be rejected cleanly, non-zero, no traceback."""
+    code = router_cli.run("task", tier=9)
+    out = capsys.readouterr().out
+    err = capsys.readouterr().err
+
+    assert code != 0
+    assert "Traceback" not in out
+    assert "Traceback" not in err
+
+
+def test_tier_flag_rejects_non_int(capsys):
+    code = router_cli.run("task", tier="abc")
+    out = capsys.readouterr().out
+    err = capsys.readouterr().err
+
+    assert code != 0
+    assert "Traceback" not in out
+    assert "Traceback" not in err
+
+
+# --- arg parsing / usage --------------------------------------------------
+
+def test_help_exits_zero():
+    try:
+        code = router_cli.main(["--help"])
+    except SystemExit as e:
+        code = e.code
+    assert code == 0

tests/test_tui.py

@@ -0,0 +1,485 @@
+"""Tests for the Textual monitoring dashboard (tui.py).
+
+Offline, no real router / network. The data layer (``build_model`` and
+``fetch_metrics``) is importable without a running TUI, so nearly all
+assertions are on the rendered panel PAYLOADS (plain dicts of display rows),
+not on pixels. ``App.run_test`` drives the app itself with a stubbed fetcher.
+
+This file imports ``tui`` and ``textual`` deliberately — it is the ONE test
+file that may. No non-tui module imports textual.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+import tui
+from tui import build_model, fetch_metrics
+
+
+def _fixture() -> dict:
+    """A representative GET /metrics payload (matches dispatcher's shape)."""
+    return {
+        "quota": {
+            "plan_kwh": 6.25,
+            "metered_kwh_30d": 1.25,
+            "metered_fraction_of_plan": 0.2,
+            "metered_calls_30d": 18,
+            "note": "router-metered only",
+        },
+        "coverage": {
+            "routable_models": 13,
+            "with_energy_data": 10,
+            "with_proficiency_data": 12,
+            "quota": {
+                "plan_kwh": 6.25,
+                "metered_kwh_30d": 1.25,
+                "metered_fraction_of_plan": 0.2,
+                "metered_calls_30d": 18,
+                "note": "router-metered only",
+            },
+            "warnings": [
+                "3/13 routable models have no reference-workload observations",
+                "1/13 routable models have no proficiency data",
+            ],
+        },
+        "recent_decisions": [
+            {
+                "id": 42,
+                "observed_at": "2026-08-23T10:00:00+00:00",
+                "kind": "chat",
+                "task_category": "coding_general",
+                "task_tier": 2,
+                "selected_model": "deepseek-v4-flash",
+                "selected_provider": "neuralwatt",
+                "est_cost_usd": 0.00016,
+            },
+            {
+                "id": 41,
+                "observed_at": "2026-08-23T09:59:00+00:00",
+                "kind": "route",
+                "task_category": "docs_writing",
+                "task_tier": 3,
+                "selected_model": "kimi-k2.7-code",
+                "selected_provider": "neuralwatt",
+                "est_cost_usd": 0.0136,
+            },
+            {
+                "id": 40,
+                "observed_at": "2026-08-23T09:58:00+00:00",
+                "kind": "chat",
+                "task_category": "debugging",
+                "task_tier": 1,
+                "selected_model": None,
+                "selected_provider": None,
+                "est_cost_usd": None,
+            },
+        ],
+        "per_model": [
+            {
+                "model_id": "deepseek-v4-flash",
+                "provider": "neuralwatt",
+                "calls": 12,
+                "sum_cost_usd": 0.0012,
+                "sum_energy_kwh": 2.5e-05,
+                "sum_carbon_g_co2eq": 1.2e-04,
+            },
+            {
+                "model_id": "kimi-k3",
+                "provider": "neuralwatt",
+                "calls": 4,
+                "sum_cost_usd": 0.08,
+                "sum_energy_kwh": 1.0e-03,
+                "sum_carbon_g_co2eq": 3.0e-03,
+            },
+        ],
+        "verdict_mix": {"ok": 5, "unverifiable": 2, "truncated": 1},
+        "top_proficiency": [
+            {"model_id": "deepseek-v4-flash", "provider": "neuralwatt",
+             "blended_score": 1.0, "source": "self_eval_thin",
+             "self_eval_samples": 3}
+        ],
+        "generated_at": "2026-08-23T10:01:00+00:00",
+    }
+
+
+# --------------------------------------------------------------------------
+# Direct unit tests of the data layer (no TUI running).
+# --------------------------------------------------------------------------
+
+
+def test_build_model_quota_panel():
+    m = build_model(_fixture())
+    rows = m["quota"]
+    # plan number is surfaced as a display row
+    joined = " ".join(r["label"] + "=" + str(r["value"]) for r in rows)
+    assert "plan_kwh=6.25" in joined
+    assert "metered_kwh_30d=1.25" in joined
+    assert "fraction=0.2" in joined
+    assert "calls=18" in joined
+
+
+def test_build_model_per_model_lists_seeded_models():
+    m = build_model(_fixture())
+    rows = m["per_model"]
+    assert rows[0]["model"] == "deepseek-v4-flash"
+    assert rows[0]["calls"] == 12
+    assert rows[1]["model"] == "kimi-k3"
+    # every row keeps numeric cost/energy/carbon for display
+    assert rows[0]["cost_usd"] == 0.0012
+    assert rows[0]["energy_kwh"] == 2.5e-05
+    assert rows[0]["carbon_g_co2eq"] == 1.2e-04
+
+
+def test_build_model_verdict_mix():
+    m = build_model(_fixture())
+    mix = m["verdict_mix"]
+    by_verdict = {r["verdict"]: r["count"] for r in mix}
+    assert by_verdict == {"ok": 5, "unverifiable": 2, "truncated": 1}
+
+
+def test_build_model_recent_decisions_top_rows():
+    m = build_model(_fixture())
+    rows = m["recent_decisions"]
+    # DESC by id: first row is id 42
+    assert rows[0]["id"] == 42
+    assert rows[0]["kind"] == "chat"
+    assert rows[0]["category"] == "coding_general"
+    assert rows[0]["tier"] == 2
+    assert rows[0]["selected"] == "deepseek-v4-flash"
+    assert rows[1]["kind"] == "route"
+    assert rows[2]["selected"] == "none"  # no-candidate row renders "none"
+
+
+def test_build_model_warnings_from_coverage():
+    m = build_model(_fixture())
+    warnings = m["warnings"]
+    assert len(warnings) == 2
+    assert "reference-workload" in warnings[0]
+    assert "proficiency data" in warnings[1]
+
+
+def test_build_model_handles_missing_quota():
+    """Empty DB / null plan: the quota section degrades to a notice."""
+    data = _fixture()
+    data["quota"] = None
+    data["coverage"]["quota"] = None
+    m = build_model(data)
+    joined = " ".join(str(r) for r in m["quota"])
+    assert "unset" in joined.lower() or "no plan" in joined.lower() or "n/a" in joined.lower()
+
+
+def test_fetch_metrics_returns_parsed_dict(monkeypatch):
+    """fetch_metrics hits the right URL and returns parsed JSON."""
+    captured = {}
+
+    class _FakeResp:
+        def raise_for_status(self):
+            return None
+
+        def json(self):
+            return {"quota": None, "ok": True}
+
+    def _fake_get(url, timeout=None):
+        captured["url"] = url
+        captured["timeout"] = timeout
+        return _FakeResp()
+
+    monkeypatch.setattr(tui.requests, "get", _fake_get)
+    out = fetch_metrics("http://testhost:8081")
+    assert out == {"quota": None, "ok": True}
+    assert captured["url"] == "http://testhost:8081/metrics"
+
+
+def test_fetch_metrics_raises_on_http_error(monkeypatch):
+    class _Err:
+        def raise_for_status(self):
+            raise RuntimeError("500")
+
+    monkeypatch.setattr(tui.requests, "get", lambda *a, **k: _Err())
+    with pytest.raises(Exception):
+        fetch_metrics("http://x")
+
+
+def test_fetch_metrics_raises_on_network_error(monkeypatch):
+    def _boom(*a, **k):
+        raise ConnectionError("refused")
+
+    monkeypatch.setattr(tui.requests, "get", _boom)
+    with pytest.raises(ConnectionError):
+        fetch_metrics("http://x")
+
+
+# --------------------------------------------------------------------------
+# App-level tests via App.run_test with a stubbed fetch_metrics.
+# --------------------------------------------------------------------------
+
+
+class _StubFetcher:
+    """Swappable fake for fetch_metrics the App calls."""
+
+    def __init__(self):
+        self.payload = None
+        self.error = None
+        self.calls = 0
+
+    def __call__(self, base_url):
+        self.calls += 1
+        if self.error is not None:
+            raise self.error
+        return self.payload
+
+
+@pytest.mark.parametrize("fetcher_arg", ["callable", "subclass"])
+def test_app_run_test_populates_quota_and_model_panels(fetcher_arg):
+    stub = _StubFetcher()
+    stub.payload = _fixture()
+
+    app = tui.DashboardApp(fetcher=stub)
+
+    def _assert(a):
+        stash = getattr(a, "_last_model", None)
+        assert stash is not None, "build_model result was not stashed on the app"
+        # quota plan number surfaced
+        quota_text = " ".join(
+            f"{r['label']}={r['value']}" for r in stash["quota"]
+        )
+        assert "plan_kwh=6.25" in quota_text
+        # per-model lists the seeded models
+        models = {r["model"] for r in stash["per_model"]}
+        assert {"deepseek-v4-flash", "kimi-k3"} <= models
+        assert stub.calls == 1
+
+    _run_app(app, _assert)
+
+
+def test_app_run_test_recent_and_warnings_panels():
+    stub = _StubFetcher()
+    stub.payload = _fixture()
+
+    app = tui.DashboardApp(fetcher=stub)
+
+    def _assert(a):
+        stash = a._last_model
+        assert stash["recent_decisions"][0]["selected"] == "deepseek-v4-flash"
+        assert len(stash["warnings"]) == 2
+
+    _run_app(app, _assert)
+
+
+def test_app_run_test_quota_panel_static_shows_plan():
+    """The rendered Static widget carries the plan number after a good fetch."""
+    stub = _StubFetcher()
+    stub.payload = _fixture()
+
+    app = tui.DashboardApp(fetcher=stub)
+
+    def _assert(a):
+        quota_widget = a.query_one("#quota-panel")
+        assert "6.25" in str(quota_widget.content)
+
+    _run_app(app, _assert)
+
+
+def test_app_run_test_failure_shows_error_and_does_not_crash():
+    """fetch raising -> error panel visible, run_test completes without raising."""
+    stub = _StubFetcher()
+    stub.error = ConnectionError("cannot reach router")
+
+    app = tui.DashboardApp(fetcher=stub, base_url="http://127.0.0.1:8080")
+
+    def _assert(a):
+        error_widget = a.query_one("#error-panel")
+        assert "cannot reach router" in str(error_widget.content)
+        assert a._last_error is not None
+
+    _run_app(app, _assert)  # must not raise
+
+
+def _run_app(app: tui.DashboardApp, body) -> None:
+    """Drive the app via Textual App.run_test synchronously.
+
+    ``body(app)`` runs while the app is mounted, so queries and the data
+    model are live. Assertion failures inside propagate out of ``asyncio.run``
+    as normal test failures.
+    """
+    async def _go():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            body(app)
+
+    asyncio.run(_go())
+
+
+# --------------------------------------------------------------------------
+# Auto-refresh, error resilience and keyboard controls.
+#
+# These drive the app through App.run_test with a tiny REFRESH_SECONDS and no
+# real wall-clock sleep. Textual 8.2.8 schedules set_interval timers on the
+# asyncio event loop, so repeatedly awaiting ``pilot.pause()`` lets due ticks
+# fire without the test asserting on elapsed time or calling time.sleep().
+# Assertions are on the re-rendered panel payload (``app._last_model``), not
+# on mock-call counts.
+# --------------------------------------------------------------------------
+
+
+class _CountingFetcher:
+    """Hands back a metrics payload whose first per-model call count equals
+    the invocation number, so each fetch produces a distinct, inspectable
+    payload with no script to exhaust."""
+
+    def __init__(self):
+        self.calls = 0
+
+    def __call__(self, base_url):
+        self.calls += 1
+        return _variant(self.calls)
+
+
+def _variant(calls: int) -> dict:
+    """Return a metrics payload whose per-model call count is ``calls``."""
+    data = _fixture()
+    data["per_model"][0]["calls"] = calls
+    return data
+
+
+def _first_model_calls(app) -> int:
+    """Read the re-rendered per-model payload's first row call count."""
+    return app._last_model["per_model"][0]["calls"]
+
+
+def test_auto_refresh_rerenders_updated_payload():
+    """Interval ticks re-fetch and re-render: the re-rendered panel tracks the
+    fetcher's latest payload.
+
+    No real sleep: the interval is tiny and every tick fires while the event
+    loop is pumped through ``pilot.pause()``. The assertion reads the actual
+    re-rendered payload back, not a mock-call count.
+    """
+    fetcher = _CountingFetcher()
+    app = tui.DashboardApp(fetcher=fetcher, refresh_seconds=0.05)
+
+    async def _go():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            # >1 fetch means an auto-tick fired beyond the on_mount refresh.
+            for _ in range(30):
+                await pilot.pause()
+            assert fetcher.calls > 1
+            # The displayed panel reflects the fetcher's newest payload.
+            assert _first_model_calls(app) == fetcher.calls
+            assert app._refreshing is False
+
+    asyncio.run(_go())
+
+
+def test_error_resilience_keeps_last_good_data_and_recovers():
+    """A transient fetch failure keeps the app running and the last good data
+    displayed; a later successful refresh re-renders the new payload.
+
+    A large interval means no stray auto-ticks, so the 2nd fetch is exactly the
+    failing one, driven deterministically through the same ``_on_interval``
+    callback the timer invokes — no sleep, no timing race.
+    """
+    calls = {"n": 0}
+
+    def _scripted(base_url):
+        calls["n"] += 1
+        if calls["n"] == 2:
+            raise ConnectionError("transient blip")
+        return _variant(calls["n"])
+
+    app = tui.DashboardApp(fetcher=_scripted, refresh_seconds=60)
+
+    async def _go():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            assert _first_model_calls(app) == 1  # initial good fetch displayed
+            # advance one tick (the failing 2nd fetch)
+            app._on_interval()
+            await pilot.pause()
+            assert app._last_error is not None, "transient failure was not seen"
+            assert not app._exit, "app must not exit on a transient failure"
+            err_widget = app.query_one("#error-panel")
+            assert "cannot reach router" in str(err_widget.content)
+            assert "visible" in err_widget.classes
+            # last good data is still the displayed payload
+            assert _first_model_calls(app) == 1
+            # recover on the next refresh (3rd fetch, now successful)
+            app._on_interval()
+            await pilot.pause()
+            assert _first_model_calls(app) == 3
+            assert app._last_error is None
+
+    asyncio.run(_go())
+
+
+def test_force_refresh_binding_reloads_on_r():
+    """Pressing ``r`` immediately re-fetches and re-renders a new payload."""
+    fetcher = _CountingFetcher()
+    # A large interval ensures only the forced refresh advances the payload.
+    app = tui.DashboardApp(fetcher=fetcher, refresh_seconds=60)
+
+    async def _go():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            baseline = fetcher.calls
+            await pilot.press("r")
+            await pilot.pause()
+            assert fetcher.calls == baseline + 1
+            assert _first_model_calls(app) == fetcher.calls
+
+    asyncio.run(_go())
+
+
+@pytest.mark.parametrize("key", ["q", "Q", "ctrl+c"])
+def test_quit_bindings_exit_app(key):
+    """q, Q and Ctrl+C all quit the running app."""
+    fetcher = _CountingFetcher()
+    app = tui.DashboardApp(fetcher=fetcher, refresh_seconds=60)
+
+    async def _go():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            assert not app._exit
+            await pilot.press(key)
+            assert app._exit
+
+    asyncio.run(_go())
+
+
+@pytest.mark.parametrize(
+    "key,panel",
+    [
+        ("1", "model-table"),
+        ("2", "verdict-table"),
+        ("3", "decision-table"),
+        ("4", "quota-panel"),
+        ("5", "warnings-panel"),
+    ],
+)
+def test_number_bindings_focus_panel(key, panel):
+    """Number keys 1-5 focus the corresponding panel."""
+    fetcher = _CountingFetcher()
+    app = tui.DashboardApp(fetcher=fetcher, refresh_seconds=60)
+
+    async def _go():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            await pilot.press(key)
+            await pilot.pause()
+            widget = app.query_one(f"#{panel}")
+            assert widget.has_focus, f"{panel} should have focus after {key!r}"
+
+    asyncio.run(_go())
+
+
+@pytest.fixture(autouse=True)
+def _no_real_network(monkeypatch):
+    """Safety net: even if the fetcher is mis-wired, never hit a real router."""
+    def _guard(base_url):
+        raise AssertionError(f"real fetch_metrics called with {base_url!r}")
+
+    monkeypatch.setattr(tui, "fetch_metrics", _guard)

tui.py

@@ -0,0 +1,365 @@
+"""Textual terminal dashboard over the router's GET /metrics endpoint.
+
+This is a SEPARATE entrypoint from the FastAPI service: ``python tui.py`` runs
+the dashboard in the foreground. Importing this module never starts the app —
+``DashboardApp.run()``/``run_test()`` are what run it.
+
+Data layer vs rendering is split so the TUI is testable without a terminal:
+
+* ``fetch_metrics(base_url)`` — pure HTTP GET of ``<base_url>/metrics`` using
+  ``requests``; returns the parsed JSON as a plain dict. Raises on HTTP or
+  network error so the caller can surface it.
+* ``build_model(data)`` — pure: turns the raw /metrics JSON into a plain dict
+  of rendered panel payloads (lists of display rows). No textual involved.
+
+The ``DashboardApp`` class is a Textual ``App``; tests drive it via
+``App.run_test()`` with a stubbed fetcher and assert on the *data model*
+(``app._last_model``, the panel payloads) rather than pixels.
+
+Only this module imports ``textual`` anywhere in the repository; the service
+dispatch path never touches it.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Callable, Optional
+
+import requests
+from textual.app import App, ComposeResult
+from textual.containers import VerticalScroll
+from textual.widgets import DataTable, Footer, Header, Static
+
+__all__ = ["fetch_metrics", "build_model", "DashboardApp"]
+
+DEFAULT_BASE_URL = "http://127.0.0.1:8080"
+
+
+# --------------------------------------------------------------------------
+# Data layer — importable and testable without a running TUI.
+# --------------------------------------------------------------------------
+
+
+def fetch_metrics(base_url: str) -> dict:
+    """GET ``<base_url>/metrics`` and return the parsed JSON dict.
+
+    Raises on any HTTP or network error (via ``raise_for_status`` and the
+    underlying ``requests`` exception) so the caller can catch it and render
+    a "cannot reach router" state.
+    """
+    url = base_url.rstrip("/") + "/metrics"
+    resp = requests.get(url, timeout=10)
+    resp.raise_for_status()
+    return resp.json()
+
+
+def build_model(data: dict) -> dict:
+    """Turn raw /metrics JSON into a plain dict of rendered panel payloads.
+
+    Returns keys: ``quota`` (list of {label, value} rows), ``per_model``
+    (list of rows), ``verdict_mix`` (list of {verdict, count}),
+    ``recent_decisions`` (list of rows), ``warnings`` (list of strings).
+    """
+    quota = data.get("quota")
+    if quota:
+        quota_rows = [
+            {"label": "plan_kwh", "value": quota.get("plan_kwh")},
+            {"label": "metered_kwh_30d", "value": quota.get("metered_kwh_30d")},
+            {
+                "label": "fraction",
+                "value": quota.get("metered_fraction_of_plan"),
+            },
+            {"label": "calls", "value": quota.get("metered_calls_30d")},
+        ]
+    else:
+        quota_rows = [{"label": "quota", "value": "N/A (plan not set)"}]
+
+    per_model = [
+        {
+            "model": r.get("model_id"),
+            "calls": r.get("calls"),
+            "cost_usd": r.get("sum_cost_usd"),
+            "energy_kwh": r.get("sum_energy_kwh"),
+            "carbon_g_co2eq": r.get("sum_carbon_g_co2eq"),
+        }
+        for r in (data.get("per_model") or [])
+    ]
+
+    mix = data.get("verdict_mix") or {}
+    verdict_mix = [
+        {"verdict": k, "count": v} for k, v in sorted(mix.items())
+    ]
+
+    recent = []
+    for r in data.get("recent_decisions") or []:
+        recent.append(
+            {
+                "id": r.get("id"),
+                "kind": r.get("kind"),
+                "category": r.get("task_category"),
+                "tier": r.get("task_tier"),
+                "selected": r.get("selected_model") or "none",
+                "est_cost_usd": r.get("est_cost_usd"),
+            }
+        )
+
+    coverage = data.get("coverage") or {}
+    warnings = list(coverage.get("warnings") or [])
+
+    return {
+        "quota": quota_rows,
+        "per_model": per_model,
+        "verdict_mix": verdict_mix,
+        "recent_decisions": recent,
+        "warnings": warnings,
+    }
+
+
+# --------------------------------------------------------------------------
+# Textual App — the terminal dashboard.
+# --------------------------------------------------------------------------
+
+
+def _fmt_usd(v: Optional[float]) -> str:
+    if v is None:
+        return "n/a"
+    return f"{v:.6f}"
+
+
+class DashboardApp(App):
+    """A terminal dashboard that reads GET /metrics and renders panels."""
+
+    TITLE = "LLM Router Monitor"
+    # How often to re-fetch /metrics when running. A class attribute so tests
+    # can override it per-instance (via the ``refresh_seconds`` constructor
+    # arg) and drive ticks without real wall-clock sleeps.
+    REFRESH_SECONDS = 5.0
+    # Textual's binding table is a framework-read list; not mutated by us.
+    BINDINGS = [  # noqa: RUF012
+        ("q", "quit", "Quit"),
+        ("Q", "quit", "Quit"),
+        ("ctrl+c", "quit", "Quit"),
+        ("r", "refresh", "Refresh"),
+        ("1", "focus_panel(0)", "Model table"),
+        ("2", "focus_panel(1)", "Verdict table"),
+        ("3", "focus_panel(2)", "Decision table"),
+        ("4", "focus_panel(3)", "Quota panel"),
+        ("5", "focus_panel(4)", "Warnings panel"),
+    ]
+
+    CSS = """
+    #error-panel {
+        background: $error;
+        color: $text;
+        padding: 1;
+        margin: 1 0;
+        display: none;
+    }
+    #error-panel.visible {
+        display: block;
+    }
+    #loading-panel {
+        color: $text-muted;
+        text-style: italic;
+        display: none;
+    }
+    #loading-panel.visible {
+        display: block;
+    }
+    .panel-title {
+        text-style: bold;
+        color: $accent;
+        margin: 1 0 0 0;
+    }
+    DataTable {
+        height: auto;
+        max-height: 12;
+        border: round $primary;
+    }
+    """
+
+    def __init__(
+        self,
+        *,
+        base_url: Optional[str] = None,
+        fetcher: Optional[Callable[[str], dict]] = None,
+        refresh_seconds: Optional[float] = None,
+    ) -> None:
+        self.base_url = base_url or os.environ.get(
+            "ROUTER_METRICS_URL", DEFAULT_BASE_URL
+        )
+        self._fetcher: Callable[[str], dict] = fetcher or fetch_metrics
+        self.refresh_seconds = (
+            refresh_seconds if refresh_seconds is not None else self.REFRESH_SECONDS
+        )
+        self._last_model: Optional[dict] = None
+        self._last_error: Optional[Exception] = None
+        self._refreshing = False
+        # Ordered list of focusable panels, indexed by the 1-5 number keys.
+        self._panels = [
+            "model-table",
+            "verdict-table",
+            "decision-table",
+            "quota-panel",
+            "warnings-panel",
+        ]
+        super().__init__()
+
+    def compose(self) -> ComposeResult:
+        yield Header(show_clock=True)
+        with VerticalScroll():
+            yield Static("", id="error-panel")
+            yield Static("", id="loading-panel")
+            yield Static("Quota burn", classes="panel-title")
+            yield Static("—", id="quota-panel")
+            yield Static("Per-model", classes="panel-title")
+            yield DataTable(id="model-table")
+            yield Static("Verdict mix", classes="panel-title")
+            yield DataTable(id="verdict-table")
+            yield Static("Recent decisions", classes="panel-title")
+            yield DataTable(id="decision-table")
+            yield Static("Health / warnings", classes="panel-title")
+            yield Static("—", id="warnings-panel")
+        yield Footer()
+
+    def on_mount(self) -> None:
+        self._set_up_tables()
+        self._refresh()
+        # Poll on the configured interval. set_interval returns a Timer; we
+        # keep the reference so a test can stop it, but we only ever schedule
+        # from on_mount so the timer is started exactly once.
+        if self.refresh_seconds > 0:
+            self._interval_timer = self.set_interval(
+                self.refresh_seconds, self._on_interval
+            )
+
+    def _on_interval(self) -> None:
+        self._refresh()
+
+    def action_refresh(self) -> None:
+        """Force an immediate re-fetch (``r``)."""
+        self._refresh()
+
+    def _set_up_tables(self) -> None:
+        model_table = self.query_one("#model-table", DataTable)
+        model_table.add_columns("model", "calls", "cost $", "kWh", "gCO2eq")
+        verdict_table = self.query_one("#verdict-table", DataTable)
+        verdict_table.add_columns("verdict", "count")
+        decision_table = self.query_one("#decision-table", DataTable)
+        decision_table.add_columns(
+            "time", "kind", "category", "tier", "selected", "est $"
+        )
+        # The two Static panels are also number-key targets (4 and 5); make
+        # them focusable so 1-5 focus is uniform.
+        self.query_one("#quota-panel", Static).can_focus = True
+        self.query_one("#warnings-panel", Static).can_focus = True
+
+    def _refresh(self) -> None:
+        self._refreshing = True
+        self._set_loading(True)
+        try:
+            data = self._fetcher(self.base_url)
+        except Exception as exc:  # noqa: BLE001 — any fetch failure must degrade
+            self._last_error = exc
+            self._render_error(exc)
+            # _last_model and the rendered panels are deliberately untouched:
+            # a transient failure keeps displaying the last good data and the
+            # interval timer retries on the next tick.
+            return
+        finally:
+            self._refreshing = False
+            self._set_loading(False)
+        self._last_error = None
+        model = build_model(data)
+        self._last_model = model
+        self._render(model)
+
+    def _set_loading(self, visible: bool) -> None:
+        try:
+            panel = self.query_one("#loading-panel", Static)
+        except Exception:  # noqa: BLE001 — panel not composed yet (pre-mount)
+            return
+        panel.update("refreshing metrics…" if visible else "")
+        if visible:
+            panel.add_class("visible")
+        else:
+            panel.remove_class("visible")
+
+    def _render_error(self, exc: Exception) -> None:
+        panel = self.query_one("#error-panel", Static)
+        panel.update(
+            f"cannot reach router at {self.base_url}: {exc.__class__.__name__}: {exc}"
+        )
+        panel.add_class("visible")
+
+    def action_focus_panel(self, index: int) -> None:
+        """Move focus to one of the numbered panels (1-5)."""
+        if index < 0 or index >= len(self._panels):
+            return
+        widget = self.query_one(f"#{self._panels[index]}")
+        try:
+            widget.focus()
+        except Exception:  # noqa: BLE001 — a Static still takes focus harmlessly
+            pass
+
+    def _render(self, model: dict) -> None:
+        # Quota panel
+        quota_widget = self.query_one("#quota-panel", Static)
+        quota_widget.update(
+            "\n".join(
+                f"{r['label']}: {r['value']}" for r in model["quota"]
+            )
+        )
+
+        # Per-model table
+        mt = self.query_one("#model-table", DataTable)
+        mt.clear()
+        for r in model["per_model"]:
+            mt.add_row(
+                str(r["model"]),
+                str(r["calls"]),
+                _fmt_usd(r["cost_usd"]),
+                f"{r['energy_kwh']:.6g}",
+                f"{r['carbon_g_co2eq']:.4g}",
+            )
+
+        # Verdict mix
+        vt = self.query_one("#verdict-table", DataTable)
+        vt.clear()
+        for r in model["verdict_mix"]:
+            vt.add_row(r["verdict"], str(r["count"]))
+        if not model["verdict_mix"]:
+            vt.add_row("(no data)", "")
+
+        # Recent decisions
+        dt = self.query_one("#decision-table", DataTable)
+        dt.clear()
+        for r in model["recent_decisions"]:
+            time_s = str(r["id"])  # id is the stable ordering handle
+            dt.add_row(
+                str(time_s),
+                str(r["kind"]),
+                str(r["category"]),
+                str(r["tier"]),
+                str(r["selected"]),
+                _fmt_usd(r["est_cost_usd"]),
+            )
+        if not model["recent_decisions"]:
+            dt.add_row("(no decisions)", "", "", "", "", "")
+
+        # Warnings panel
+        warn_widget = self.query_one("#warnings-panel", Static)
+        if model["warnings"]:
+            warn_widget.update(
+                "\n".join(f"• {w}" for w in model["warnings"])
+            )
+        else:
+            warn_widget.update("No warnings.")
+
+
+def main() -> None:
+    DashboardApp().run()
+
+
+if __name__ == "__main__":
+    main()