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