6krrt/baseline_report.py

#!/usr/bin/env python3
"""Read-only retrospective comparison of routing against two trivial baselines.

Replays recent ``route_decisions`` rows against two counterfactuals that skip
routing's scoring entirely:

- **always_cheapest** — the eligible candidate with the lowest
  ``routing.estimated_cost`` for that request's shape.
- **always_best_proficiency** — the eligible candidate with the highest
  ``proficiency.blended_score`` for the decision's ``task_category``, ties
  broken by lowest cost.

For each decision the eligible set is reconstructed from the current catalog by
re-running ``routing.select_candidates`` with the constraints that row actually
carried (``task_tier``, ``required_context_tokens``, ``latency_tolerance``,
and its ``tools``/``images``/``json_mode`` flags). This automates the "check for dominance" step in docs/routing.md: a high dominance share with a near-zero proficiency
delta means the real scoring is not earning its complexity for that slice of
traffic.

A third counterfactual is **incumbent-free dial**: for each decision the
session's prior selection (resolved via the incumbent lookup query below) is
fed to ``routing.rank_candidates`` with a ``challenger_cache_rate`` dial,
producing a rank order where the incumbent retains its measured pricing but
challengers are priced at ``min(dial, inc_rate)``.  Two dials are always
reported:

- **neutral** — dial equals ``assumed_cache_rate``: the ranking is
  byte-identical to the pre-incumbency ranking, so it reproduces the real
  router's choice whenever the router ran at its neutral setting.
- **--incumbent-challenger (default 0.0)** — the "cold challenger" setting:
  challengers lose every token of cache discount (full list price) while the
  incumbent keeps its measured rate. Any dial at which the ranking would
  have chosen a different model than the real one is an incumbency premium
  the operator can now see.

Read-only, adds no schema, reuses ``routing.select_candidates`` and
``routing.estimated_cost``. Baselines are recomputed against the CURRENT
catalog and proficiency table, not a historical snapshot.

Run directly:
    python baseline_report.py --since 2026-08-01
    python baseline_report.py --since 2026-08-01 --category coding_refactor
    python baseline_report.py --since 2026-08-01 --csv
    python baseline_report.py --since 2026-08-01 --incumbent-challenger 0.5
"""

from __future__ import annotations

import argparse
import csv
import sqlite3
import sys
from collections.abc import Sequence

from config import RouterConfig, load_config
from routing import estimated_cost, rank_candidates, select_candidates


def load_candidates(
    conn: sqlite3.Connection, category: str, cfg: RouterConfig,
) -> list[dict]:
    """Current catalog rows joined with proficiency for ``category``.

    Mirrors dispatcher's candidate-join shape (but without the eco/energy
    columns this report does not need): each row gains ``proficiency`` (the
    ``blended_score`` for ``category``, or None when unmeasured) and
    ``tool_proficiency`` (the blend for the tool gate's category, so the
    tool filter reconstructs the same way the router gates it).
    """
    rows = conn.execute(
        """
        SELECT m.*,
               p.blended_score  AS proficiency,
               tp.blended_score AS tool_proficiency
          FROM models m
          LEFT JOIN proficiency p
            ON p.model_id = m.model_id
           AND p.provider = m.provider
           AND p.category = ?
          LEFT JOIN proficiency tp
            ON tp.model_id = m.model_id
           AND tp.provider = m.provider
           AND tp.category = ?
        """,
        (category, cfg.routing.tool_use_category),
    ).fetchall()
    return [dict(r) for r in rows]


def load_decisions(
    conn: sqlite3.Connection,
    since: str | None,
    category: str | None,
) -> list[sqlite3.Row]:
    """Recent routed decisions in the window, newest first.

    ``observed_at`` is ISO8601 UTC text, so a lexicographic comparison with a
    ``YYYY-MM-DD`` prefix is a valid time filter — a decision whose timestamp
    starts at or after ``since`` is inside the window.
    """
    clauses: list[str] = [
        "kind IN ('route', 'chat', 'dispatch')",
        "selected_model IS NOT NULL",
    ]
    params: list[str] = []
    if since is not None:
        clauses.append("observed_at >= ?")
        params.append(since)
    if category is not None:
        clauses.append("task_category = ?")
        params.append(category)
    sql = (
        "SELECT * FROM route_decisions WHERE "
        + " AND ".join(clauses)
        + " ORDER BY observed_at DESC"
    )
    return conn.execute(sql, params).fetchall()


def baseline_selection(
    candidates: Sequence[dict],
    decision: sqlite3.Row,
    cfg: RouterConfig,
) -> tuple[dict | None, dict | None]:
    """Pick the always_cheapest and always_best_proficiency candidates.

    ``candidates`` is the request's reconstructed eligible set (already run
    through ``routing.select_candidates``). Both baselines are pure selections
    over that set:

    - cheapest: lowest ``estimated_cost`` priced for the request's own shape.
    - best proficiency: highest ``proficiency`` for the decision's category,
      ties broken by lowest cost, then model_id for determinism.

    Returns ``(cheapest, best)``; either may be None when the set is empty.
    """
    if not candidates:
        return None, None

    prompt_tokens = decision["required_context_tokens"] or 0
    cache_rate = cfg.objective.assumed_cache_rate
    completion_tokens = cfg.objective.assumed_completion_tokens

    priced = []
    for row in candidates:
        cost = estimated_cost(row, prompt_tokens, completion_tokens, cache_rate)
        priced.append((row, cost))

    cheapest = min(
        priced,
        key=lambda rc: (rc[1] if rc[1] is not None else float("inf"), rc[0]["model_id"]),
    )[0]

    def prof_key(rc: tuple) -> tuple:
        row, cost = rc
        prof = row.get("proficiency")
        return (
            -(prof if prof is not None else 0.5),
            cost if cost is not None else float("inf"),
            row["model_id"],
        )

    best = min(priced, key=prof_key)[0]
    return cheapest, best


def reconstruct_decision(
    conn: sqlite3.Connection,
    decision: sqlite3.Row,
    cfg: RouterConfig,
) -> tuple[dict | None, dict | None]:
    """Re-run the decision's hard filters against the current catalog.

    Returns ``(always_cheapest, always_best_proficiency)`` candidate rows the
    real decision could have picked (or None when no candidate is eligible).
    """
    category = decision["task_category"] or cfg.classifier.fallback_category
    filters = {
        "required_context_tokens": decision["required_context_tokens"] or 0,
        "required_tier": decision["task_tier"] or 1,
        "latency_tolerance": decision["latency_tolerance"]
        or cfg.routing.default_latency_tolerance,
        "allowed_access_levels": cfg.routing.allowed_access_levels,
        "exclude_stale": cfg.freshness.exclude_stale,
        "exclude_deprecated": cfg.freshness.exclude_deprecated,
    }
    if decision["tools"]:
        filters["min_tool_proficiency"] = cfg.routing.min_tool_proficiency
    if decision["images"]:
        filters["require_vision"] = cfg.routing.require_vision
    if decision["json_mode"]:
        filters["require_json_mode"] = cfg.routing.require_json_mode

    candidates = select_candidates(
        load_candidates(conn, category, cfg), **filters,
    )
    return baseline_selection(candidates, decision, cfg)


def _reconstruct_candidates(
    conn: sqlite3.Connection,
    decision: sqlite3.Row,
    cfg: RouterConfig,
) -> list[dict]:
    """Hard-filter the catalog using the decision's constraints.

    Like ``reconstruct_decision`` but returns the eligible set without
    running the baseline selectors — the caller picks the winners.
    """
    category = decision["task_category"] or cfg.classifier.fallback_category
    filters: dict[str, object] = {
        "required_context_tokens": decision["required_context_tokens"] or 0,
        "required_tier": decision["task_tier"] or 1,
        "latency_tolerance": decision["latency_tolerance"]
        or cfg.routing.default_latency_tolerance,
        "allowed_access_levels": cfg.routing.allowed_access_levels,
        "exclude_stale": cfg.freshness.exclude_stale,
        "exclude_deprecated": cfg.freshness.exclude_deprecated,
    }
    if decision["tools"]:
        filters["min_tool_proficiency"] = cfg.routing.min_tool_proficiency
    if decision["images"]:
        filters["require_vision"] = cfg.routing.require_vision
    if decision["json_mode"]:
        filters["require_json_mode"] = cfg.routing.require_json_mode
    return select_candidates(
        load_candidates(conn, category, cfg), **filters,
    )


def _model_cost(row: dict | None, decision: sqlite3.Row, cfg: RouterConfig) -> float | None:
    """Estimated cost of a baseline row priced for the decision's shape."""
    if row is None:
        return None
    return estimated_cost(
        row,
        decision["required_context_tokens"] or 0,
        cfg.objective.assumed_completion_tokens,
        cfg.objective.assumed_cache_rate,
    )


def _model_proficiency(row: dict | None) -> float | None:
    if row is None:
        return None
    prof = row.get("proficiency")
    return prof if prof is not None else 0.5


# ---------------------------------------------------------------------------
# Incumbent-free counterfactual helpers
# ---------------------------------------------------------------------------

# The canonical session-incumbent query lives in dispatcher._session_incumbent_
# lookup.  The baseline_report module must not import dispatcher (pulls in
# FastAPI, requests, openai, etc.), so we replicate the single SELECT here
# with a mirror docstring.
_SESS_INCUMBENT_SQL = """
    SELECT selected_provider, selected_model
      FROM route_decisions
     WHERE session_key = ?
       AND selected_model IS NOT NULL
       AND kind IN ('chat')
     ORDER BY id DESC
     LIMIT 1
"""


def _session_incumbent_lookup(
    conn: sqlite3.Connection, session_key: str,
) -> tuple[str, str] | None:
    """Return the session's most recent (provider, model_id) pair.

    Mirrors ``dispatcher._session_incumbent_lookup``: an allowlist on ``kind``
    means only explicit chat turns set the incumbent (not route probes or
    dispatch-fallback rows).  Returns ``None`` when the session has no chat
    turns — in that case the caller should pass ``incumbent=None`` to
    ``rank_candidates``, restoring the pre-feature ranking.
    """
    row = conn.execute(_SESS_INCUMBENT_SQL, (session_key,)).fetchone()
    if row is None:
        return None
    return (row["selected_provider"], row["selected_model"])


def _load_cache_rates(conn: sqlite3.Connection) -> dict[tuple[str, str], float]:
    """Per-(provider, model_id) measured cache rates from the DB.

    Mirrors ``dispatcher._measured_cache_rates`` without the periodic-cache
    wrapper or dispatcher imports.  Reads the same source table as
    ``metrics.cache_rate_series`` (``energy_observations``, reported
    ``cached_prompt_tokens``), token-weighted per (provider, model_id).
    Unlike the dispatcher helper it applies no observation-count floor:
    this is a retrospective report over a bounded window, not a live
    routing input, and the caller (``incumbent_free_selection``) falls
    back to the assumed rate for incumbents absent from the map, which
    is the same behavior the floor produces.

    Returns ``{}`` when no row reports cached tokens.
    """
    rows = conn.execute(
        """
        SELECT provider, model_id,
               SUM(cached_prompt_tokens) * 1.0
               / SUM(prompt_tokens) AS cache_rate
          FROM energy_observations
         WHERE cached_prompt_tokens IS NOT NULL
           AND prompt_tokens IS NOT NULL
           AND prompt_tokens > 0
         GROUP BY provider, model_id
        """,
    ).fetchall()

    rates: dict[tuple[str, str], float] = {}
    for row in rows:
        cr = row["cache_rate"]
        if cr is not None and cr > 0:
            key = (row["provider"], row["model_id"])
            rates[key] = cr
    return rates


def incumbent_free_selection(
    conn: sqlite3.Connection,
    candidates: Sequence[dict],
    decision: sqlite3.Row,
    cfg: RouterConfig,
    measured_rates: dict[tuple[str, str], float] | None,
    challenger_dial: float = 0.0,
) -> dict[str, dict | None]:
    """Run the incumbent-counterfactual ranking for every dial.

    Returns ``{dial_str: {winner, matched, cost}}`` for ``"neutral"`` and
    ``"challenger"``.
    """
    if not candidates:
        return {"neutral": None, "challenger": None}

    pt = decision["required_context_tokens"] or 0
    ct = cfg.objective.assumed_completion_tokens
    cr = cfg.objective.assumed_cache_rate
    qt = cfg.objective.quality_tolerance
    result: dict[str, dict | None] = {}

    # Neutral dial: incumbent=None → pre-feature ranking.
    ranked = rank_candidates(
        [dict(c) for c in candidates],
        quality_tolerance=qt, prompt_tokens=pt, completion_tokens=ct,
        cache_rate=cr, incumbent=None,
    )
    if ranked:
        w = ranked[0]
        wc = estimated_cost(w, pt, ct, cr)
        result["neutral"] = {
            "winner": w["model_id"],
            "matched": w["model_id"] == (decision["selected_model"] or ""),
            "cost": wc,
        }
    else:
        result["neutral"] = {"winner": None, "matched": False, "cost": None}

    # Challenger dial: incumbent priced at its measured rate, challengers
    # clamped to min(dial, inc_rate). At dial == cr this is byte-identical
    # to the neutral ranking.
    incumbent = _session_incumbent_lookup(
        conn, decision["session_key"] or "",
    )
    ranked = rank_candidates(
        [dict(c) for c in candidates],
        quality_tolerance=qt, prompt_tokens=pt, completion_tokens=ct,
        cache_rate=cr, incumbent=incumbent,
        measured_cache_rates=measured_rates,
        challenger_cache_rate=challenger_dial,
    )
    if ranked:
        w = ranked[0]
        wc = estimated_cost(w, pt, ct, cr)
        result["challenger"] = {
            "winner": w["model_id"],
            "matched": w["model_id"] == (decision["selected_model"] or ""),
            "cost": wc,
        }
    else:
        result["challenger"] = {"winner": None, "matched": False, "cost": None}

    return result


def analyze(
    conn: sqlite3.Connection,
    cfg: RouterConfig,
    since: str | None = None,
    category: str | None = None,
    challenger_dial: float = 0.0,
) -> tuple[dict, list[dict]]:
    """Compute aggregate and per-category baseline comparison.

    Returns ``(aggregate_row, category_rows)`` where each row is a plain dict
    keyed by column name, ready for both the human table and the CSV writer.
    """
    decisions = load_decisions(conn, since, category)
    measured_rates = _load_cache_rates(conn)

    def empty_row(cat: str | None) -> dict:
        return {
            "category": cat if cat is not None else "total",
            "count": 0,
            "actual_cost": 0.0,
            "cheapest_cost": 0.0,
            "best_prof_cost": 0.0,
            "actual_proficiency": 0.0,
            "cheapest_proficiency": 0.0,
            "best_prof_proficiency": 0.0,
            "dominance_count": 0,
            "dominance_pct": None,
            # Incumbent-free counters:
            "incumbent_neutral_match": 0,
            "incumbent_challenger_match": 0,
            "incumbent_challenger_winner": "",
        }

    by_cat: dict[str, dict] = {}
    total = empty_row(None)

    for d in decisions:
        cat = d["task_category"] or cfg.classifier.fallback_category
        row = by_cat.setdefault(cat, empty_row(cat))
        candidates = _reconstruct_candidates(conn, d, cfg)
        cheapest, best = baseline_selection(candidates, d, cfg)

        for agg in (total, row):
            agg["count"] += 1
            agg["actual_cost"] += d["est_cost_usd"] or 0.0
            cc = _model_cost(cheapest, d, cfg)
            bc = _model_cost(best, d, cfg)
            if cc is not None:
                agg["cheapest_cost"] += cc
            if bc is not None:
                agg["best_prof_cost"] += bc
            ap = d["est_proficiency"]
            agg["actual_proficiency"] += ap if ap is not None else 0.0
            cp = _model_proficiency(cheapest)
            bp = _model_proficiency(best)
            if cp is not None:
                agg["cheapest_proficiency"] += cp
            if bp is not None:
                agg["best_prof_proficiency"] += bp
            if cheapest is not None and cheapest["model_id"] == d["selected_model"]:
                agg["dominance_count"] += 1

        # Incumbent-free counterfactual for this decision
        cf = incumbent_free_selection(
            conn, candidates, d, cfg, measured_rates,
            challenger_dial=challenger_dial,
        )
        for agg in (total, row):
            if cf["neutral"] is not None and cf["neutral"]["matched"]:
                agg["incumbent_neutral_match"] += 1
            if cf["challenger"] is not None and cf["challenger"]["matched"]:
                agg["incumbent_challenger_match"] += 1
            # Track the most frequent challenger winner
            cw = (cf["challenger"] or {}).get("winner")
            if cw:
                agg.setdefault("_challenger_counts", {}).setdefault(cw, 0)
                agg["_challenger_counts"][cw] += 1

    def finalize(r: dict) -> None:
        if r["count"] == 0:
            r["dominance_pct"] = None
            return
        r["actual_proficiency"] /= r["count"]
        r["cheapest_proficiency"] /= r["count"]
        r["best_prof_proficiency"] /= r["count"]
        r["dominance_pct"] = 100.0 * r["dominance_count"] / r["count"]
        # Pick the most common challenger winner
        counts = r.pop("_challenger_counts", {})
        if counts:
            r["incumbent_challenger_winner"] = max(counts, key=counts.get)
        else:
            r["incumbent_challenger_winner"] = ""

    finalize(total)
    for row in by_cat.values():
        finalize(row)

    cat_rows = [by_cat[c] for c in sorted(by_cat)]
    return total, cat_rows


def format_summary(
    total: dict, cat_rows: list[dict], challenger_dial: float = 0.0,
) -> str:
    """Human-readable report text."""
    lines = ["baseline routing comparator"]
    lines.append(f"  decisions: {total['count']}")
    lines.append(
        f"  total cost:   actual ${total['actual_cost']:.4f}  "
        f"cheapest ${total['cheapest_cost']:.4f}  "
        f"best-prof ${total['best_prof_cost']:.4f}"
    )
    pct = total["dominance_pct"]
    dom = "n/a" if pct is None else f"{pct:.1f}%"
    lines.append(
        f"  dominance (selected == cheapest): {total['dominance_count']}/{total['count']} "
        f"({dom})"
    )
    lines.append(
        f"  mean proficiency: actual {total['actual_proficiency']:.3f}  "
        f"cheapest {total['cheapest_proficiency']:.3f}  "
        f"best-prof {total['best_prof_proficiency']:.3f}"
    )
    n = total["count"] or 1
    lines.append(
        f"  incumbent-free: neutral-matched {total['incumbent_neutral_match']}/{n}  "
        f"challenger@{challenger_dial:.2f}-matched "
        f"{total['incumbent_challenger_match']}/{n}  "
        f"challenger-top-choice {total['incumbent_challenger_winner'] or '(none)'}"
    )
    if cat_rows:
        lines.append("")
        lines.append(
            f"{'category':18s}{'n':>5}{'dom%':>7}{'act$/dec':>10}"
            f"{'cheap$/dec':>11}{'best$/dec':>11}{'act prof':>10}{'best prof':>10}"
        )
        for r in cat_rows:
            dom = "n/a" if r["dominance_pct"] is None else f"{r['dominance_pct']:.1f}"
            lines.append(
                f"{r['category']:18s}{r['count']:>5}{dom:>7}"
                f"{r['actual_cost'] / r['count']:>10.4f}"
                f"{r['cheapest_cost'] / r['count']:>11.4f}"
                f"{r['best_prof_cost'] / r['count']:>11.4f}"
                f"{r['actual_proficiency']:>10.3f}"
                f"{r['best_prof_proficiency']:>10.3f}"
            )
    return "\n".join(lines)


CSV_COLUMNS = [
    "category", "count", "actual_cost", "cheapest_cost", "best_prof_cost",
    "actual_proficiency", "cheapest_proficiency", "best_prof_proficiency",
    "dominance_count", "dominance_pct",
    "incumbent_neutral_match", "incumbent_challenger_match",
    "incumbent_challenger_winner",
]


def write_csv(total: dict, cat_rows: list[dict], out) -> None:
    """Write aggregate + per-category rows as CSV."""
    writer = csv.DictWriter(out, fieldnames=CSV_COLUMNS)
    writer.writeheader()
    writer.writerow(total)
    for r in cat_rows:
        writer.writerow(r)


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument(
        "--since",
        help="only consider decisions observed at/after this ISO date (YYYY-MM-DD)",
    )
    ap.add_argument("--category", help="only consider a single task_category")
    ap.add_argument("--csv", action="store_true", help="emit CSV instead of the table")
    ap.add_argument(
        "--incumbent-challenger",
        type=float,
        default=0.0,
        metavar="RATE",
        help="challenger_cache_rate dial for the incumbent-free "
        "counterfactual (default: 0.0, the cold-challenger setting). "
        "The neutral dial (assumed_cache_rate) is always reported alongside.",
    )
    args = ap.parse_args()

    cfg = load_config("config/config.yaml")
    conn = sqlite3.connect(cfg.database.path)
    conn.row_factory = sqlite3.Row

    total, cat_rows = analyze(
        conn, cfg, since=args.since, category=args.category,
        challenger_dial=args.incumbent_challenger,
    )

    if args.csv:
        write_csv(total, cat_rows, sys.stdout)
    else:
        print(format_summary(total, cat_rows, challenger_dial=args.incumbent_challenger))

    conn.close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())