#!/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())