feat(monitoring): routing-decision persistence, /metrics, monitoring TUI, router-cli #4
@@ -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"
|
||||
|
||||
|
||||
@@ -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
423
dispatcher.py
@@ -72,6 +72,14 @@ from verification import (
|
||||
verify_response,
|
||||
worth_local_check,
|
||||
)
|
||||
from metrics import (
|
||||
per_model,
|
||||
quota_burn,
|
||||
recent_decisions,
|
||||
scoring_coverage,
|
||||
top_proficiency,
|
||||
verdict_mix,
|
||||
)
|
||||
|
||||
# Virtual model names that mean "you pick". Anything else is taken as a real
|
||||
# model id and dispatched as asked.
|
||||
@@ -240,6 +248,73 @@ def _db() -> sqlite3.Connection:
|
||||
return conn
|
||||
|
||||
|
||||
def ensure_route_decisions(conn: sqlite3.Connection) -> None:
|
||||
"""Idempotently create the route_decisions observability table.
|
||||
|
||||
schema.sql is CREATE TABLE IF NOT EXISTS, so it defines a NEW database and
|
||||
silently does nothing to an existing one — the same reason
|
||||
proficiency_store.ensure_columns exists. A live router.db predating this
|
||||
table therefore never gets it from re-running schema.sql, so the table is
|
||||
created here, from code, with a guard. This is the ONLY way it appears on
|
||||
a live database; the live DB is never recreated or dropped.
|
||||
|
||||
Safe to call any number of times against an existing connection: the
|
||||
CREATE TABLE and CREATE INDEX are both IF NOT EXISTS, so a table that is
|
||||
already present is left fully intact (rows included) and a second call
|
||||
no-ops.
|
||||
"""
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS route_decisions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
observed_at TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
task_category TEXT,
|
||||
task_tier INTEGER,
|
||||
required_context_tokens INTEGER,
|
||||
confidence REAL,
|
||||
classifier_ms INTEGER,
|
||||
classification_source TEXT,
|
||||
latency_tolerance TEXT,
|
||||
candidates_considered INTEGER,
|
||||
selected_model TEXT,
|
||||
selected_provider TEXT,
|
||||
runner_up_models TEXT,
|
||||
est_cost_usd REAL,
|
||||
est_proficiency REAL,
|
||||
rejected_reason TEXT,
|
||||
session_key TEXT,
|
||||
tools INTEGER,
|
||||
images INTEGER,
|
||||
json_mode INTEGER,
|
||||
streamed INTEGER
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_route_decisions_observed "
|
||||
"ON route_decisions (observed_at)"
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def _ensure_route_decisions_table() -> None:
|
||||
"""Create route_decisions on the live DB if it predates this feature."""
|
||||
try:
|
||||
conn = sqlite3.connect(cfg.database.path)
|
||||
try:
|
||||
ensure_route_decisions(conn)
|
||||
finally:
|
||||
conn.close()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logs.warning(
|
||||
"startup_route_decisions_migration_failed", error=str(exc)
|
||||
)
|
||||
|
||||
|
||||
_ensure_route_decisions_table()
|
||||
|
||||
|
||||
def _classifier_client() -> OpenAI:
|
||||
"""The classification endpoint, which need not be local.
|
||||
|
||||
@@ -697,6 +772,144 @@ def log_decision(
|
||||
)
|
||||
|
||||
|
||||
def persist_route_decision(
|
||||
decision_kind: str,
|
||||
*,
|
||||
classification=None,
|
||||
latency_tolerance=None,
|
||||
selected_model=None,
|
||||
selected_provider=None,
|
||||
runners_up=None,
|
||||
rejected_reason=None,
|
||||
session_key=None,
|
||||
tools=0,
|
||||
images=0,
|
||||
json_mode=0,
|
||||
streamed=0,
|
||||
classification_source=None,
|
||||
classifier_ms=None,
|
||||
) -> None:
|
||||
"""Record one routing decision to route_decisions, best-effort and gated.
|
||||
|
||||
Every decision path calls this exactly once per request. A write that
|
||||
fails for any reason (read-only or locked DB, missing table) is logged at
|
||||
warning and swallowed, because a decision record is worth having but never
|
||||
worth failing or slowing a request for. Gated by
|
||||
``cfg.logging.log_route_decisions`` so the monitoring history can be left
|
||||
empty where it is not wanted.
|
||||
|
||||
``classification`` may be a RouteResponse (deriving category/tier/context/…
|
||||
from its Classification and the selected model from its ``selected``) or a
|
||||
bare Classification; the explicit arguments cover the paths with no
|
||||
RouteResponse (passthrough, local-vision) and let a caller override what
|
||||
the RouteResponse would otherwise say — the chat re-route's
|
||||
Classification reads ``source='override'`` even though the classifier did
|
||||
decide the request, so the caller passes the first route's actual source.
|
||||
"""
|
||||
# The gate is read before anything can fail: turning the table off must be
|
||||
# a guaranteed no-op even on a broken DB.
|
||||
if not cfg.logging.log_route_decisions:
|
||||
return
|
||||
|
||||
clf = None
|
||||
selected = None
|
||||
derived_runners = None
|
||||
candidates = None
|
||||
derived_latency = latency_tolerance
|
||||
|
||||
if isinstance(classification, RouteResponse):
|
||||
clf = classification.classification
|
||||
selected = classification.selected
|
||||
derived_runners = classification.runners_up
|
||||
candidates = classification.candidates_considered
|
||||
derived_latency = classification.latency_tolerance
|
||||
elif isinstance(classification, Classification):
|
||||
clf = classification
|
||||
|
||||
if clf is not None and classification_source is None:
|
||||
classification_source = clf.source
|
||||
# An override-created Classification never consulted the classifier, so it
|
||||
# has no classifier latency worth recording — even when a caller passed a
|
||||
# timing value (route_endpoint's whole-route ms would otherwise leak in).
|
||||
if classification_source == "override":
|
||||
classifier_ms = None
|
||||
|
||||
if selected is not None:
|
||||
model = selected.model_id
|
||||
provider = selected.provider
|
||||
est_cost = selected.cost
|
||||
est_prof = selected.proficiency_score
|
||||
runner_json = (
|
||||
json.dumps(
|
||||
[{"model_id": c.model_id, "provider": c.provider}
|
||||
for c in derived_runners[:3]]
|
||||
)
|
||||
if derived_runners
|
||||
else None
|
||||
)
|
||||
else:
|
||||
model = selected_model
|
||||
provider = selected_provider
|
||||
est_cost = None
|
||||
est_prof = None
|
||||
runner_json = json.dumps(runners_up[:3]) if runners_up else None
|
||||
|
||||
conn = _db()
|
||||
try:
|
||||
# Guarantee the table on the WRITE path, mirroring
|
||||
# proficiency_store._write -> ensure_columns. schema.sql is CREATE
|
||||
# TABLE IF NOT EXISTS so it never adds the table to a live router.db
|
||||
# that predates it; the module-load hook covers the normal boot, but
|
||||
# any other path that calls persist first (a test harness, a lazy
|
||||
# import, a future refactor) must still get the table here or the
|
||||
# INSERT raises no-such-table and the decision is silently lost.
|
||||
ensure_route_decisions(conn)
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO route_decisions (
|
||||
observed_at, kind, task_category, task_tier,
|
||||
required_context_tokens, confidence, classifier_ms,
|
||||
classification_source, latency_tolerance, candidates_considered,
|
||||
selected_model, selected_provider, runner_up_models,
|
||||
est_cost_usd, est_proficiency, rejected_reason, session_key,
|
||||
tools, images, json_mode, streamed
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
datetime.now(timezone.utc).isoformat(),
|
||||
decision_kind,
|
||||
clf.task_category if clf is not None else None,
|
||||
clf.task_tier if clf is not None else None,
|
||||
clf.required_context_tokens if clf is not None else None,
|
||||
clf.confidence if clf is not None else None,
|
||||
classifier_ms,
|
||||
classification_source,
|
||||
derived_latency,
|
||||
candidates,
|
||||
model,
|
||||
provider,
|
||||
runner_json,
|
||||
est_cost,
|
||||
est_prof,
|
||||
rejected_reason,
|
||||
session_key,
|
||||
int(bool(tools)),
|
||||
int(bool(images)),
|
||||
int(bool(json_mode)),
|
||||
int(bool(streamed)),
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
except Exception as e: # noqa: BLE001 - best-effort must never raise
|
||||
logs.warning(
|
||||
"route_decision_persist",
|
||||
kind=decision_kind,
|
||||
error=type(e).__name__,
|
||||
)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
# --- energy / cost accounting --------------------------------------------
|
||||
|
||||
class Telemetry(BaseModel):
|
||||
@@ -938,6 +1151,7 @@ def health():
|
||||
"""
|
||||
).fetchall()
|
||||
)
|
||||
scoring = scoring_coverage(conn, cfg)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
@@ -950,7 +1164,7 @@ def health():
|
||||
return {
|
||||
"status": "ok",
|
||||
"counts": counts,
|
||||
"scoring": scoring_coverage(),
|
||||
"scoring": scoring,
|
||||
"classifier_reachable": classifier_ok,
|
||||
"classifier_model": cfg.classifier.model,
|
||||
# The tier labels, documented as being for logging and dashboards.
|
||||
@@ -964,128 +1178,36 @@ def health():
|
||||
}
|
||||
|
||||
|
||||
def scoring_coverage() -> dict:
|
||||
"""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
210
metrics.py
Normal 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()
|
||||
]
|
||||
@@ -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
233
router_cli.py
Normal 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
50
schema.sql
@@ -188,6 +188,51 @@ CREATE TABLE IF NOT EXISTS verifications (
|
||||
model_attributable INTEGER DEFAULT 1
|
||||
);
|
||||
|
||||
-- One row per routing decision, logged by the dispatcher on every route that
|
||||
-- is made (see todo #2 of router-monitoring-tui.md for the writes; this table
|
||||
-- and its inline-create helper are todo #1). It makes "how is routing
|
||||
-- performing" answerable: which model was picked, for what category/tier, how
|
||||
-- long classification took, and — when nothing was selected — which hard
|
||||
-- filter shut it out.
|
||||
--
|
||||
-- This is an OBSERVABILITY table, not a scoring input: nothing in routing.py
|
||||
-- reads it. It stores only the hashed session fingerprint in `session_key`;
|
||||
-- never `session_dir` and never any prompt or answer text. A test enforces
|
||||
-- that the write path stores only the hash.
|
||||
--
|
||||
-- Like `energy_observations`, `observed_at` uses
|
||||
-- datetime.now(timezone.utc).isoformat().
|
||||
CREATE TABLE IF NOT EXISTS route_decisions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
observed_at TEXT NOT NULL, -- ISO8601, UTC
|
||||
kind TEXT NOT NULL, -- 'route' | 'dispatch' | 'chat'
|
||||
-- | 'passthrough' | 'local_vision'
|
||||
task_category TEXT,
|
||||
task_tier INTEGER, -- 1-3
|
||||
required_context_tokens INTEGER,
|
||||
confidence REAL,
|
||||
classifier_ms INTEGER,
|
||||
classification_source TEXT, -- 'classifier' | 'override' | 'fallback'
|
||||
latency_tolerance TEXT, -- 'interactive' | 'batch'
|
||||
candidates_considered INTEGER,
|
||||
selected_model TEXT, -- null when nothing was selected
|
||||
-- 'neuralwatt' for routed/dispatch/passthrough, 'local' for the local-vision
|
||||
-- fallback. Kept so a decision row can join to energy_observations on the
|
||||
-- (model_id, provider) key that table uses.
|
||||
selected_provider TEXT,
|
||||
runner_up_models TEXT, -- JSON array of {"model_id":..,
|
||||
-- "provider":..}, <=3; nullable
|
||||
est_cost_usd REAL,
|
||||
est_proficiency REAL,
|
||||
rejected_reason TEXT, -- the 422 limits when no selection
|
||||
session_key TEXT, -- hashed session fingerprint ONLY,
|
||||
-- never session_dir or prompt text
|
||||
tools INTEGER, -- 0/1
|
||||
images INTEGER, -- 0/1
|
||||
json_mode INTEGER, -- 0/1
|
||||
streamed INTEGER -- 0/1
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_verifications_model ON verifications (model_id, provider);
|
||||
CREATE INDEX IF NOT EXISTS idx_verifications_verdict ON verifications (verdict);
|
||||
CREATE INDEX IF NOT EXISTS idx_observations_request ON energy_observations (request_id);
|
||||
@@ -197,3 +242,8 @@ CREATE INDEX IF NOT EXISTS idx_models_availability ON models (availability);
|
||||
CREATE INDEX IF NOT EXISTS idx_models_routing ON models (access_level, latency_class, tier);
|
||||
CREATE INDEX IF NOT EXISTS idx_proficiency_category ON proficiency (category);
|
||||
CREATE INDEX IF NOT EXISTS idx_energy_model ON energy_observations (model_id, provider);
|
||||
-- recent_decisions orders by id, but a time-window query benefits from this.
|
||||
CREATE INDEX IF NOT EXISTS idx_route_decisions_observed ON route_decisions (observed_at);
|
||||
-- Guarded — re-applying a fresh schema is a no-op.
|
||||
CREATE INDEX IF NOT EXISTS idx_energy_observed ON energy_observations (observed_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_verifications_observed ON verifications (observed_at);
|
||||
|
||||
538
tests/test_metrics.py
Normal file
538
tests/test_metrics.py
Normal 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
201
tests/test_metrics_endpoint.py
Normal 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
|
||||
@@ -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
660
tests/test_route_decisions.py
Normal 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
225
tests/test_router_cli.py
Normal 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
485
tests/test_tui.py
Normal 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
365
tui.py
Normal 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()
|
||||
Reference in New Issue
Block a user