From b84b4839dc77b47ab5be2a55f1d071343cd03fe8 Mon Sep 17 00:00:00 2001 From: adlee-was-taken Date: Sun, 23 Aug 2026 20:19:46 -0400 Subject: [PATCH 1/3] feat(cli): router_cli one-shot routing-decision printer Post a task to POST /route and print the full decision tree (classification, candidates, selected model, est cost, rejections) without dispatching a provider call. Supports --category/--tier/--context overrides and --json. Route-only and no-spend. --- router_cli.py | 233 +++++++++++++++++++++++++++++++++++++++ tests/test_router_cli.py | 225 +++++++++++++++++++++++++++++++++++++ 2 files changed, 458 insertions(+) create mode 100644 router_cli.py create mode 100644 tests/test_router_cli.py diff --git a/router_cli.py b/router_cli.py new file mode 100644 index 0000000..49c78ad --- /dev/null +++ b/router_cli.py @@ -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 "" [--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()) diff --git a/tests/test_router_cli.py b/tests/test_router_cli.py new file mode 100644 index 0000000..143ff9f --- /dev/null +++ b/tests/test_router_cli.py @@ -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 -- 2.49.1 From 8ef8c401d61a2d93f94267715abf8235d876391b Mon Sep 17 00:00:00 2001 From: adlee-was-taken Date: Sun, 23 Aug 2026 20:19:51 -0400 Subject: [PATCH 2/3] feat(tui): Textual monitoring dashboard over GET /metrics A terminal dashboard that polls the router's GET /metrics and renders quota burn vs plan, per-model cost/energy/carbon, verdict mix, recent routing decisions, and health warnings. Auto-refresh with last-good-data error resilience and keyboard controls (q/Q/Ctrl+C quit, r refresh, 1-5 focus panels). Pins textual (deliberate UI-only dependency, imported only by tui.py, never by the service). --- requirements.txt | 5 + tests/test_tui.py | 485 ++++++++++++++++++++++++++++++++++++++++++++++ tui.py | 365 ++++++++++++++++++++++++++++++++++ 3 files changed, 855 insertions(+) create mode 100644 tests/test_tui.py create mode 100644 tui.py diff --git a/requirements.txt b/requirements.txt index ef766fd..3468c4a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/tests/test_tui.py b/tests/test_tui.py new file mode 100644 index 0000000..52ec584 --- /dev/null +++ b/tests/test_tui.py @@ -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) diff --git a/tui.py b/tui.py new file mode 100644 index 0000000..9480f42 --- /dev/null +++ b/tui.py @@ -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 ``/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 ``/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() -- 2.49.1 From bfb0ff7c00c1a041246425d95f53a17c5c52830a Mon Sep 17 00:00:00 2001 From: adlee-was-taken Date: Sun, 23 Aug 2026 20:20:09 -0400 Subject: [PATCH 3/3] feat(metrics): persist routing decisions and expose GET /metrics The router previously recorded only completions (energy_observations), not the routing decisions behind them, so 'how routing is performing' was not answerable from data. This adds: - route_decisions table + idempotent ensure_route_decisions (guarded CREATE TABLE IF NOT EXISTS, never regenerates a live DB) gated by logging.log_route_decisions; every decision kind (route/dispatch/chat/ passthrough/local-vision) is persisted best-effort via persist_route_decision (never fails a request; only session_key, never session_dir). The table is ensured on the write path (mirroring proficiency_store._write -> ensure_columns) so a live DB that predates the feature migrates safely. - metrics.py aggregator moved quota_burn/scoring_coverage in from the dispatcher (breaking a would-be circular import) and adds recent_decisions/per_model/verdict_mix/top_proficiency; /health now imports them and GET /metrics exposes the 7-key JSON (window-bounded, loopback-only, no auth). - observed_at indexes on energy_observations/verifications. --- config.py | 5 + config.yaml | 5 + dispatcher.py | 423 +++++++++++++------ metrics.py | 210 ++++++++++ schema.sql | 50 +++ tests/test_metrics.py | 538 ++++++++++++++++++++++++ tests/test_metrics_endpoint.py | 201 +++++++++ tests/test_outcome_attribution.py | 4 +- tests/test_route_decisions.py | 660 ++++++++++++++++++++++++++++++ 9 files changed, 1978 insertions(+), 118 deletions(-) create mode 100644 metrics.py create mode 100644 tests/test_metrics.py create mode 100644 tests/test_metrics_endpoint.py create mode 100644 tests/test_route_decisions.py diff --git a/config.py b/config.py index 3979190..fcacee5 100644 --- a/config.py +++ b/config.py @@ -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" diff --git a/config.yaml b/config.yaml index 70b016a..c60665f 100644 --- a/config.yaml +++ b/config.yaml @@ -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, diff --git a/dispatcher.py b/dispatcher.py index b870121..2523dae 100644 --- a/dispatcher.py +++ b/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, diff --git a/metrics.py b/metrics.py new file mode 100644 index 0000000..2f51ce4 --- /dev/null +++ b/metrics.py @@ -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() + ] diff --git a/schema.sql b/schema.sql index b9c9a74..e1f9a44 100644 --- a/schema.sql +++ b/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); diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..fa3a390 --- /dev/null +++ b/tests/test_metrics.py @@ -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"] diff --git a/tests/test_metrics_endpoint.py b/tests/test_metrics_endpoint.py new file mode 100644 index 0000000..100a093 --- /dev/null +++ b/tests/test_metrics_endpoint.py @@ -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 diff --git a/tests/test_outcome_attribution.py b/tests/test_outcome_attribution.py index 599dfd1..0e0ad02 100644 --- a/tests/test_outcome_attribution.py +++ b/tests/test_outcome_attribution.py @@ -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) diff --git a/tests/test_route_decisions.py b/tests/test_route_decisions.py new file mode 100644 index 0000000..46ecfff --- /dev/null +++ b/tests/test_route_decisions.py @@ -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 -- 2.49.1