6krrt/metrics.py

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