From e45cf4e0618b4de02608f68c502ac4cf1270eb84 Mon Sep 17 00:00:00 2001 From: Ashish Date: Wed, 17 Jun 2026 21:28:23 -0700 Subject: [PATCH] feat(cli): add headroom doctor setup diagnostics (#926) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description Headroom fails silently: a client not routed through the proxy (or a proxy running stale code) keeps working โ€” it just stops saving tokens. State that determines whether you are actually saving lives in five places nothing reconciles. `headroom doctor` correlates them in one command (the diagnostic idiom of `claude doctor` / `pnpm doctor`, and the repo's own `headroom tools doctor`). Closes # ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [x] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - `headroom/cli/doctor.py`: new command with 8 pure checks (proxy liveness, version drift, claude/codex routing, shell env, savings flow, budget, deployments); exit codes 0/1/2; `--json`; `--port`/`HEADROOM_PORT`. - `headroom/proxy/cost.py`: expose `budget_limit_usd`/`budget_period` in `CostTracker.stats()` so the budget check can read it (older proxies degrade to a warning). - `headroom/cli/main.py`: register the command. - `tests/test_cli_doctor.py`: 41 tests, zero network (probed payloads / paths / env injected). ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [x] Manual testing performed ### Test Output ```text $ pytest tests/test_cli_doctor.py -q 41 passed ``` ## Real Behavior Proof - Environment: local macOS, Python 3.11, against a real proxy running for 3 days, branch `feat/doctor-command`. - Exact command / steps: `headroom doctor` (live), plus `pytest tests/test_cli_doctor.py -q`. - Observed result: Correctly flagged real version drift (proxy 0.25.0 vs installed 0.26.0), an unrouted claude client, and a shell `OPENAI_BASE_URL` pointed at a non-Headroom gateway; savings check showed 17.6M tokens / $7.82 saved; exit code 1 (warnings). - Not tested: Windows path handling for client config files (logic is OS-agnostic via pathlib). ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Screenshots (if applicable) Terminal output of `headroom doctor` (rich table) can be attached; the rendered table is reproduced in the live-proof bullet above. ## Additional Notes Branched fresh from main. The budget check connects to the enforcement fix in #885. ๐Ÿค– Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Fable 5 Co-authored-by: JD Davis Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- headroom/cli/doctor.py | 412 +++++++++++++++++++++++++++++++++++++++ headroom/cli/main.py | 1 + headroom/proxy/cost.py | 4 + tests/test_cli_doctor.py | 349 +++++++++++++++++++++++++++++++++ 4 files changed, 766 insertions(+) create mode 100644 headroom/cli/doctor.py create mode 100644 tests/test_cli_doctor.py diff --git a/headroom/cli/doctor.py b/headroom/cli/doctor.py new file mode 100644 index 000000000..08d6c9064 --- /dev/null +++ b/headroom/cli/doctor.py @@ -0,0 +1,412 @@ +"""`headroom doctor` โ€” diagnose whether the local Headroom setup is working. + +Headroom's failure mode is silent: when a client is not routed through the +proxy (or the proxy runs stale code), everything still works โ€” you just +stop saving tokens. This command correlates the state nothing else +reconciles: the proxy process, per-client wrap configs, the current shell +environment, savings flow, and budget configuration. + +Exit codes: 0 = all checks pass, 1 = warnings only, 2 = any failure. +""" + +from __future__ import annotations + +import json +import os +import re +from collections.abc import Mapping +from dataclasses import asdict, dataclass +from datetime import datetime +from pathlib import Path +from typing import Any + +import click + +from headroom.install.health import probe_json +from headroom.install.paths import claude_settings_path, codex_config_path +from headroom.install.state import list_manifests +from headroom.paths import savings_path + +from .main import get_version, main + +PASS = "pass" +WARN = "warn" +FAIL = "fail" +SKIP = "skip" + +_LOOPBACK_URL_RE = re.compile(r"https?://(?:127\.0\.0\.1|localhost):(\d+)") +_CODEX_BASE_URL_RE = re.compile(r'base_url\s*=\s*"https?://(?:127\.0\.0\.1|localhost):(\d+)') + + +@dataclass +class CheckResult: + """One diagnostic outcome.""" + + name: str + status: str # pass | warn | fail | skip + summary: str + hint: str | None = None + + +def _format_uptime(seconds: float) -> str: + total = int(seconds) + days, rem = divmod(total, 86400) + hours, rem = divmod(rem, 3600) + minutes = rem // 60 + if days: + return f"{days}d {hours}h" + if hours: + return f"{hours}h {minutes}m" + return f"{minutes}m" + + +def _format_since(iso_ts: str) -> str | None: + try: + then = datetime.fromisoformat(iso_ts.replace("Z", "+00:00")) + except (ValueError, AttributeError): + return None + delta = datetime.now(then.tzinfo) - then + seconds = max(0, int(delta.total_seconds())) + if seconds < 60: + return "just now" + if seconds < 3600: + return f"{seconds // 60}m ago" + if seconds < 86400: + return f"{seconds // 3600}h ago" + return f"{seconds // 86400}d ago" + + +def check_proxy_liveness(livez: dict[str, Any] | None, base_url: str) -> CheckResult: + """Is the proxy process up and answering /livez?""" + if livez is None: + return CheckResult( + name="proxy", + status=FAIL, + summary=f"not reachable at {base_url}", + hint="start it with: headroom proxy", + ) + version = livez.get("version", "unknown") + uptime = livez.get("uptime_seconds") + uptime_text = f"up {_format_uptime(uptime)}" if isinstance(uptime, (int, float)) else "up" + return CheckResult( + name="proxy", + status=PASS, + summary=f"running at {base_url} ({uptime_text}, v{version})", + ) + + +def check_version_drift(livez: dict[str, Any] | None, installed: str) -> CheckResult: + """Does the running proxy match the installed package version?""" + if livez is None: + return CheckResult(name="version", status=SKIP, summary="proxy not reachable") + running = str(livez.get("version") or "unknown") + if "unknown" in (running, installed): + return CheckResult( + name="version", + status=WARN, + summary=f"cannot compare versions (proxy {running}, installed {installed})", + ) + if running != installed: + return CheckResult( + name="version", + status=WARN, + summary=f"version drift: proxy {running}, installed {installed}", + hint="restart the proxy to pick up new code: headroom proxy", + ) + return CheckResult(name="version", status=PASS, summary=f"proxy matches installed v{installed}") + + +def check_claude_routing(settings_path: Path, port: int) -> CheckResult: + """Is Claude Code configured to route through the proxy?""" + name = "claude" + if not settings_path.exists(): + return CheckResult( + name=name, + status=WARN, + summary="not routed (no ~/.claude/settings.json)", + hint="wrap it: headroom wrap claude", + ) + try: + payload = json.loads(settings_path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + return CheckResult( + name=name, + status=WARN, + summary=f"could not parse {settings_path}: {exc}", + ) + base_url = "" + env_block = payload.get("env") + if isinstance(env_block, dict): + base_url = str(env_block.get("ANTHROPIC_BASE_URL", "") or "") + if not base_url: + return CheckResult( + name=name, + status=WARN, + summary="not routed (no ANTHROPIC_BASE_URL in settings env)", + hint="wrap it: headroom wrap claude", + ) + return _classify_routing_url(name, base_url, port, source=str(settings_path)) + + +def check_codex_routing(config_path: Path, port: int) -> CheckResult: + """Is Codex configured to route through the proxy? + + Detection keys on the ``[model_providers.headroom]`` section, which both + writers emit (install's persistent block and wrap's auto-injected block). + Substring matching keeps malformed TOML a WARN instead of a crash. + """ + name = "codex" + if not config_path.exists(): + return CheckResult( + name=name, + status=WARN, + summary="not routed (no ~/.codex/config.toml)", + hint="wrap it: headroom wrap codex", + ) + try: + text = config_path.read_text(encoding="utf-8", errors="replace") + except OSError as exc: + return CheckResult(name=name, status=WARN, summary=f"could not read {config_path}: {exc}") + if "[model_providers.headroom]" not in text: + return CheckResult( + name=name, + status=WARN, + summary="not routed (no Headroom provider in config.toml)", + hint="wrap it: headroom wrap codex", + ) + match = _CODEX_BASE_URL_RE.search(text) + if match and int(match.group(1)) != port: + return CheckResult( + name=name, + status=WARN, + summary=f"routed to port {match.group(1)}, but doctor probed port {port}", + hint=f"re-run with: headroom doctor --port {match.group(1)}", + ) + return CheckResult(name=name, status=PASS, summary=f"routed ({config_path})") + + +def check_shell_env(environ: Mapping[str, str], port: int) -> CheckResult: + """Is the *current shell* pointed at the proxy for ad-hoc runs?""" + name = "shell env" + for var in ("ANTHROPIC_BASE_URL", "OPENAI_BASE_URL"): + value = environ.get(var, "") + if value: + return _classify_routing_url(name, value, port, source=var) + return CheckResult( + name=name, + status=WARN, + summary="ANTHROPIC_BASE_URL / OPENAI_BASE_URL unset โ€” this shell bypasses the proxy", + hint=f"export ANTHROPIC_BASE_URL=http://127.0.0.1:{port} (or launch via headroom wrap)", + ) + + +def _classify_routing_url(name: str, url: str, port: int, *, source: str) -> CheckResult: + match = _LOOPBACK_URL_RE.match(url.strip()) + if match is None: + return CheckResult( + name=name, + status=WARN, + summary=f"points at {url}, not the local Headroom proxy ({source})", + ) + found_port = int(match.group(1)) + if found_port != port: + return CheckResult( + name=name, + status=WARN, + summary=f"routed to port {found_port}, but doctor probed port {port} ({source})", + hint=f"re-run with: headroom doctor --port {found_port}", + ) + return CheckResult(name=name, status=PASS, summary=f"routed via {source}") + + +def check_savings(stats: dict[str, Any] | None, savings_file: Path) -> CheckResult: + """Are savings actually flowing? Lifetime totals + last activity.""" + name = "savings" + payload: dict[str, Any] | None = None + source = "proxy /stats" + if stats is not None and isinstance(stats.get("persistent_savings"), dict): + payload = stats["persistent_savings"] + elif savings_file.exists(): + source = str(savings_file) + try: + payload = json.loads(savings_file.read_text(encoding="utf-8")) + except (OSError, ValueError): + return CheckResult( + name=name, status=WARN, summary=f"could not read savings file {savings_file}" + ) + if payload is None: + return CheckResult( + name=name, + status=WARN, + summary="no savings recorded yet", + hint="route a client through the proxy and make a request", + ) + + lifetime = payload.get("lifetime") or {} + tokens = lifetime.get("tokens_saved", 0) or 0 + usd = lifetime.get("compression_savings_usd", 0.0) or 0.0 + if not tokens: + return CheckResult( + name=name, + status=WARN, + summary="no tokens saved yet", + hint="route a client through the proxy and make a request", + ) + + session = payload.get("display_session") or {} + freshness = None + last_activity = session.get("last_activity_at") + if isinstance(last_activity, str): + freshness = _format_since(last_activity) + summary = f"{tokens:,} tokens / ${usd:,.2f} saved lifetime" + if freshness: + summary += f" โ€” last request {freshness}" + return CheckResult(name=name, status=PASS, summary=f"{summary} ({source})") + + +def check_budget(stats: dict[str, Any] | None) -> CheckResult: + """Is a spend budget configured on the proxy?""" + name = "budget" + if stats is None: + return CheckResult(name=name, status=SKIP, summary="proxy not reachable") + cost = stats.get("cost") + if not isinstance(cost, dict): + return CheckResult(name=name, status=WARN, summary="cost tracking disabled (--no-cost)") + if "budget_limit_usd" not in cost: + return CheckResult( + name=name, + status=WARN, + summary="proxy does not report budget config (older version?)", + hint="restart the proxy on the current version", + ) + limit = cost.get("budget_limit_usd") + if limit is None: + return CheckResult( + name=name, + status=WARN, + summary="no budget configured โ€” spend is unlimited", + hint="set one: headroom proxy --budget 10 (env: HEADROOM_BUDGET)", + ) + period = cost.get("budget_period", "daily") + return CheckResult(name=name, status=PASS, summary=f"${limit}/{period} budget enforced") + + +def check_deployments(manifests: list[Any], probe: Any = probe_json) -> CheckResult | None: + """Probe persistent deployment health URLs. None when no deployments.""" + if not manifests: + return None + down = [] + for manifest in manifests: + payload = probe(manifest.health_url) + ready = bool(payload and (payload.get("ready") or payload.get("status") == "healthy")) + if not ready: + down.append(manifest.profile) + if down: + return CheckResult( + name="deployments", + status=FAIL, + summary=f"{len(down)} of {len(manifests)} deployment(s) down: {', '.join(down)}", + hint="inspect with: headroom install status --profile ", + ) + return CheckResult( + name="deployments", + status=PASS, + summary=f"{len(manifests)} deployment(s) healthy", + ) + + +_STATUS_STYLE = {PASS: "green", WARN: "yellow", FAIL: "red", SKIP: "dim"} +_STATUS_GLYPH = {PASS: "โœ“", WARN: "โš ", FAIL: "โœ—", SKIP: "ยท"} + + +def _render(checks: list[CheckResult], port: int, installed: str) -> None: + from rich.console import Console + from rich.markup import escape + from rich.table import Table + + console = Console() + console.print(f"[bold]Headroom Doctor[/bold] [dim]v{installed} ยท port {port}[/dim]\n") + table = Table(show_header=True, header_style="bold") + table.add_column("check") + table.add_column("status") + table.add_column("summary") + for check in checks: + style = _STATUS_STYLE.get(check.status, "white") + glyph = _STATUS_GLYPH.get(check.status, "?") + table.add_row( + check.name, + f"[{style}]{glyph} {check.status}[/{style}]", + escape(check.summary), + ) + console.print(table) + for check in checks: + if check.hint: + console.print(f"[dim]{check.name}:[/dim] {escape(check.hint)}") + + fails = sum(1 for c in checks if c.status == FAIL) + warns = sum(1 for c in checks if c.status == WARN) + if fails or warns: + console.print(f"\n[bold]{fails} failure(s), {warns} warning(s)[/bold]") + else: + console.print("\n[green bold]all checks passed[/green bold]") + + +@main.command() +@click.option( + "--port", + "-p", + default=8787, + type=int, + envvar="HEADROOM_PORT", + help="Proxy port to check (default: 8787, env: HEADROOM_PORT)", +) +@click.option("--json", "emit_json", is_flag=True, help="Emit JSON instead of formatted output.") +def doctor(port: int, emit_json: bool) -> None: + """Check that the Headroom proxy and client routing are working. + + \b + Exit codes: + 0 everything healthy + 1 warnings only (working, but not optimally wired) + 2 at least one failure (proxy down / deployment down) + """ + base_url = f"http://127.0.0.1:{port}" + livez = probe_json(f"{base_url}/livez") + stats = probe_json(f"{base_url}/stats", timeout=5.0) if livez else None + installed = get_version() + + checks = [ + check_proxy_liveness(livez, base_url), + check_version_drift(livez, installed), + check_claude_routing(claude_settings_path(), port), + check_codex_routing(codex_config_path(), port), + check_shell_env(os.environ, port), + check_savings(stats, savings_path()), + check_budget(stats), + ] + deployments = check_deployments(list_manifests()) + if deployments is not None: + checks.append(deployments) + + if any(c.status == FAIL for c in checks): + exit_code = 2 + elif any(c.status == WARN for c in checks): + exit_code = 1 + else: + exit_code = 0 + + if emit_json: + click.echo( + json.dumps( + { + "port": port, + "installed_version": installed, + "exit_code": exit_code, + "checks": [asdict(c) for c in checks], + }, + indent=2, + ) + ) + else: + _render(checks, port, installed) + raise SystemExit(exit_code) diff --git a/headroom/cli/main.py b/headroom/cli/main.py index 861ea51d6..429d0b2a1 100644 --- a/headroom/cli/main.py +++ b/headroom/cli/main.py @@ -40,6 +40,7 @@ def _register_commands() -> None: audit, # noqa: F401 capture, # noqa: F401 copilot_auth, # noqa: F401 + doctor, # noqa: F401 evals, # noqa: F401 init, # noqa: F401 install, # noqa: F401 diff --git a/headroom/proxy/cost.py b/headroom/proxy/cost.py index 938d2217f..1128f1f3b 100644 --- a/headroom/proxy/cost.py +++ b/headroom/proxy/cost.py @@ -853,4 +853,8 @@ class CostTracker: "per_model": per_model, "cost_with_headroom_usd": round(cost_with_headroom, 4), "savings_usd": round(savings_usd, 4), + # Budget config passthrough โ€” surfaces in /stats["cost"] so + # `headroom doctor` can report whether a budget is set. + "budget_limit_usd": self.budget_limit_usd, + "budget_period": self.budget_period, } diff --git a/tests/test_cli_doctor.py b/tests/test_cli_doctor.py new file mode 100644 index 000000000..040053e03 --- /dev/null +++ b/tests/test_cli_doctor.py @@ -0,0 +1,349 @@ +"""Tests for `headroom doctor`.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass + +import pytest +from click.testing import CliRunner + +import headroom.cli.doctor as doctor_mod +from headroom.cli.doctor import ( + FAIL, + PASS, + SKIP, + WARN, + check_budget, + check_claude_routing, + check_codex_routing, + check_deployments, + check_proxy_liveness, + check_savings, + check_shell_env, + check_version_drift, +) +from headroom.cli.main import main + +LIVEZ_OK = { + "service": "headroom-proxy", + "status": "healthy", + "alive": True, + "version": "0.26.0", + "uptime_seconds": 260135.0, +} + +STATS_OK = { + "persistent_savings": { + "lifetime": {"tokens_saved": 17_583_102, "compression_savings_usd": 7.81701}, + "display_session": {"last_activity_at": "2026-06-12T12:00:00Z"}, + }, + "cost": {"budget_limit_usd": 10.0, "budget_period": "daily"}, +} + + +class TestProxyLiveness: + def test_down_is_fail_with_hint(self): + result = check_proxy_liveness(None, "http://127.0.0.1:8787") + assert result.status == FAIL + assert "headroom proxy" in (result.hint or "") + + def test_up_mentions_version_and_uptime(self): + result = check_proxy_liveness(LIVEZ_OK, "http://127.0.0.1:8787") + assert result.status == PASS + assert "v0.26.0" in result.summary + assert "3d" in result.summary + + +class TestVersionDrift: + def test_match_passes(self): + assert check_version_drift(LIVEZ_OK, "0.26.0").status == PASS + + def test_mismatch_warns_with_restart_hint(self): + result = check_version_drift(LIVEZ_OK, "0.27.0") + assert result.status == WARN + assert "drift" in result.summary + assert "restart" in (result.hint or "") + + def test_proxy_down_skips(self): + assert check_version_drift(None, "0.26.0").status == SKIP + + def test_unknown_version_warns(self): + assert check_version_drift({"version": "unknown"}, "0.26.0").status == WARN + assert check_version_drift(LIVEZ_OK, "unknown").status == WARN + + +class TestClaudeRouting: + def test_missing_file_warns(self, tmp_path): + result = check_claude_routing(tmp_path / "settings.json", 8787) + assert result.status == WARN + assert "wrap claude" in (result.hint or "") + + def test_malformed_json_warns(self, tmp_path): + path = tmp_path / "settings.json" + path.write_text("{not json", encoding="utf-8") + assert check_claude_routing(path, 8787).status == WARN + + def test_no_env_key_warns(self, tmp_path): + path = tmp_path / "settings.json" + path.write_text(json.dumps({"env": {}}), encoding="utf-8") + assert check_claude_routing(path, 8787).status == WARN + + def test_correct_url_passes(self, tmp_path): + path = tmp_path / "settings.json" + path.write_text( + json.dumps({"env": {"ANTHROPIC_BASE_URL": "http://127.0.0.1:8787"}}), + encoding="utf-8", + ) + assert check_claude_routing(path, 8787).status == PASS + + def test_port_mismatch_warns(self, tmp_path): + path = tmp_path / "settings.json" + path.write_text( + json.dumps({"env": {"ANTHROPIC_BASE_URL": "http://127.0.0.1:8788"}}), + encoding="utf-8", + ) + result = check_claude_routing(path, 8787) + assert result.status == WARN + assert "8788" in result.summary + + def test_non_headroom_url_warns(self, tmp_path): + path = tmp_path / "settings.json" + path.write_text( + json.dumps({"env": {"ANTHROPIC_BASE_URL": "https://gateway.corp.example/v1"}}), + encoding="utf-8", + ) + result = check_claude_routing(path, 8787) + assert result.status == WARN + assert "gateway.corp.example" in result.summary + + +class TestCodexRouting: + def test_missing_file_warns(self, tmp_path): + assert check_codex_routing(tmp_path / "config.toml", 8787).status == WARN + + def test_marker_block_right_port_passes(self, tmp_path): + path = tmp_path / "config.toml" + path.write_text( + 'model_provider = "headroom"\n' + "[model_providers.headroom]\n" + 'base_url = "http://127.0.0.1:8787/v1"\n', + encoding="utf-8", + ) + assert check_codex_routing(path, 8787).status == PASS + + def test_port_mismatch_warns(self, tmp_path): + path = tmp_path / "config.toml" + path.write_text( + '[model_providers.headroom]\nbase_url = "http://127.0.0.1:9999/v1"\n', + encoding="utf-8", + ) + result = check_codex_routing(path, 8787) + assert result.status == WARN + assert "9999" in result.summary + + def test_no_marker_warns(self, tmp_path): + path = tmp_path / "config.toml" + path.write_text('model = "gpt-5"\n', encoding="utf-8") + assert check_codex_routing(path, 8787).status == WARN + + def test_garbage_bytes_warn_not_crash(self, tmp_path): + path = tmp_path / "config.toml" + path.write_bytes(b"\xff\xfe garbage \x00") + assert check_codex_routing(path, 8787).status == WARN + + +class TestShellEnv: + def test_unset_warns(self): + result = check_shell_env({}, 8787) + assert result.status == WARN + assert "bypasses" in result.summary + + def test_matching_anthropic_url_passes(self): + env = {"ANTHROPIC_BASE_URL": "http://127.0.0.1:8787"} + assert check_shell_env(env, 8787).status == PASS + + def test_localhost_also_passes(self): + env = {"OPENAI_BASE_URL": "http://localhost:8787/v1"} + assert check_shell_env(env, 8787).status == PASS + + def test_other_url_warns(self): + env = {"ANTHROPIC_BASE_URL": "https://api.anthropic.com"} + assert check_shell_env(env, 8787).status == WARN + + +class TestSavings: + def test_from_stats_passes_with_totals(self, tmp_path): + result = check_savings(STATS_OK, tmp_path / "missing.json") + assert result.status == PASS + assert "17,583,102" in result.summary + assert "$7.82" in result.summary + + def test_falls_back_to_file_when_proxy_down(self, tmp_path): + savings_file = tmp_path / "proxy_savings.json" + savings_file.write_text( + json.dumps( + { + "lifetime": {"tokens_saved": 500, "compression_savings_usd": 0.01}, + "display_session": {"last_activity_at": "2026-06-12T11:00:00Z"}, + } + ), + encoding="utf-8", + ) + result = check_savings(None, savings_file) + assert result.status == PASS + assert "500" in result.summary + assert str(savings_file) in result.summary + + def test_no_data_warns(self, tmp_path): + assert check_savings(None, tmp_path / "missing.json").status == WARN + + def test_zero_tokens_warns(self, tmp_path): + stats = {"persistent_savings": {"lifetime": {"tokens_saved": 0}}} + assert check_savings(stats, tmp_path / "missing.json").status == WARN + + +class TestBudget: + def test_proxy_down_skips(self): + assert check_budget(None).status == SKIP + + def test_cost_tracking_disabled_warns(self): + assert check_budget({"cost": None}).status == WARN + + def test_old_proxy_without_keys_warns(self): + result = check_budget({"cost": {"savings_usd": 1.0}}) + assert result.status == WARN + assert "older version" in result.summary + + def test_unset_budget_warns_with_hint(self): + result = check_budget({"cost": {"budget_limit_usd": None}}) + assert result.status == WARN + assert "--budget" in (result.hint or "") + + def test_configured_budget_passes(self): + result = check_budget(STATS_OK) + assert result.status == PASS + assert "$10.0/daily" in result.summary + + +@dataclass +class _FakeManifest: + profile: str + health_url: str + + +class TestDeployments: + def test_no_manifests_omits_section(self): + assert check_deployments([]) is None + + def test_all_healthy_passes(self): + manifests = [_FakeManifest("default", "http://127.0.0.1:8787/readyz")] + result = check_deployments(manifests, probe=lambda url: {"ready": True}) + assert result is not None and result.status == PASS + + def test_unhealthy_fails_naming_profile(self): + manifests = [_FakeManifest("prod", "http://127.0.0.1:9999/readyz")] + result = check_deployments(manifests, probe=lambda url: None) + assert result is not None and result.status == FAIL + assert "prod" in result.summary + + +class TestDoctorCommand: + @pytest.fixture + def runner(self): + return CliRunner() + + @pytest.fixture + def isolated(self, tmp_path, monkeypatch): + """Point all filesystem/network surfaces at controlled fakes.""" + monkeypatch.setattr(doctor_mod, "claude_settings_path", lambda: tmp_path / "settings.json") + monkeypatch.setattr(doctor_mod, "codex_config_path", lambda: tmp_path / "config.toml") + monkeypatch.setattr(doctor_mod, "savings_path", lambda: tmp_path / "savings.json") + monkeypatch.setattr(doctor_mod, "list_manifests", lambda: []) + for var in ("ANTHROPIC_BASE_URL", "OPENAI_BASE_URL", "HEADROOM_PORT"): + monkeypatch.delenv(var, raising=False) + return tmp_path + + def _probe(self, livez, stats): + def fake_probe(url, timeout=2.0): + if url.endswith("/livez"): + return livez + if url.endswith("/stats"): + return stats + return None + + return fake_probe + + def test_proxy_down_exits_2(self, runner, isolated, monkeypatch): + monkeypatch.setattr(doctor_mod, "probe_json", self._probe(None, None)) + result = runner.invoke(main, ["doctor"]) + assert result.exit_code == 2 + assert "not reachable" in result.output + + def test_warnings_only_exits_1(self, runner, isolated, monkeypatch): + monkeypatch.setattr(doctor_mod, "probe_json", self._probe(LIVEZ_OK, STATS_OK)) + monkeypatch.setattr(doctor_mod, "get_version", lambda: "0.26.0") + # proxy healthy, but clients unwrapped + shell env unset -> warns + result = runner.invoke(main, ["doctor"]) + assert result.exit_code == 1 + + def test_all_pass_exits_0(self, runner, isolated, monkeypatch): + monkeypatch.setattr(doctor_mod, "probe_json", self._probe(LIVEZ_OK, STATS_OK)) + monkeypatch.setattr(doctor_mod, "get_version", lambda: "0.26.0") + (isolated / "settings.json").write_text( + json.dumps({"env": {"ANTHROPIC_BASE_URL": "http://127.0.0.1:8787"}}), + encoding="utf-8", + ) + (isolated / "config.toml").write_text( + '[model_providers.headroom]\nbase_url = "http://127.0.0.1:8787/v1"\n', + encoding="utf-8", + ) + result = runner.invoke( + main, ["doctor"], env={"ANTHROPIC_BASE_URL": "http://127.0.0.1:8787"} + ) + assert result.exit_code == 0, result.output + assert "all checks passed" in result.output + + def test_json_output_parses(self, runner, isolated, monkeypatch): + monkeypatch.setattr(doctor_mod, "probe_json", self._probe(LIVEZ_OK, STATS_OK)) + result = runner.invoke(main, ["doctor", "--json"]) + payload = json.loads(result.output) + assert payload["port"] == 8787 + assert {c["name"] for c in payload["checks"]} >= {"proxy", "version", "budget"} + assert all(c["status"] in ("pass", "warn", "fail", "skip") for c in payload["checks"]) + + def test_port_option_changes_probe_url(self, runner, isolated, monkeypatch): + seen: list[str] = [] + + def recording_probe(url, timeout=2.0): + seen.append(url) + return None + + monkeypatch.setattr(doctor_mod, "probe_json", recording_probe) + runner.invoke(main, ["doctor", "--port", "9999"]) + assert "http://127.0.0.1:9999/livez" in seen + + def test_port_env_var_respected(self, runner, isolated, monkeypatch): + seen: list[str] = [] + + def recording_probe(url, timeout=2.0): + seen.append(url) + return None + + monkeypatch.setattr(doctor_mod, "probe_json", recording_probe) + runner.invoke(main, ["doctor"], env={"HEADROOM_PORT": "9999"}) + assert "http://127.0.0.1:9999/livez" in seen + + +class TestCostTrackerBudgetKeys: + def test_stats_exposes_budget_config(self): + from headroom.proxy.cost import CostTracker + + stats = CostTracker(budget_limit_usd=5.0, budget_period="monthly").stats() + assert stats["budget_limit_usd"] == 5.0 + assert stats["budget_period"] == "monthly" + + def test_stats_budget_none_when_unset(self): + from headroom.proxy.cost import CostTracker + + assert CostTracker().stats()["budget_limit_usd"] is None