From 11aea86920cbf650a8a6de76e90db017c34dca69 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 22 Aug 2026 14:33:03 +0200 Subject: [PATCH] Land the token_audit standing-cost checkup (116 tests, 100% branch coverage) Answers "what is billed on every turn regardless of use" by measurement instead of recall, which is the part earlier audits kept getting wrong. A surface is reported dead only when unused in BOTH windows. One earlier audit's seven-day dead list included six servers that had been used within thirty days, so a single window is not evidence. Shares are weighted with cache-read weight 0.1, so a cached surface is not billed as though it were re-read in full each turn. Verified by running it, not just by tests: the report renders over 44,311 turns, putting rules/ at 4,907 tokens (1.77%), MEMORY.md at 4,312 (1.55%), memories/ at 3,758 (1.35%), skill descriptions at 3,279 (1.18%) and CLAUDE.md at 2,057 (0.74%), with per-MCP 7d/30d call counts behind each verdict. The files were found untracked in the tree from an earlier session; every module is already under the 250-line cap, so nothing needed splitting, and the package was already at 100% branch coverage, so nothing needed new tests. This lands the instrument only -- no surface is unregistered on its say-so. Co-Authored-By: Claude Opus 5 (1M context) --- .../token-audit-standing-cost-2026-08-22.json | 16 ++ .../token-audit-standing-cost-2026-08-22.json | 47 ++++++ python_pkg/token_audit/checkup.py | 71 ++++++++ python_pkg/token_audit/probe.py | 117 +++++++++++++ python_pkg/token_audit/surfaces.py | 157 ++++++++++++++++++ python_pkg/token_audit/tests/test_checkup.py | 64 +++++++ python_pkg/token_audit/tests/test_probe.py | 111 +++++++++++++ python_pkg/token_audit/tests/test_surfaces.py | 143 ++++++++++++++++ python_pkg/token_audit/tests/test_unused.py | 100 +++++++++++ python_pkg/token_audit/unused.py | 78 +++++++++ 10 files changed, 904 insertions(+) create mode 100644 docs/superpowers/contracts/token-audit-standing-cost-2026-08-22.json create mode 100644 docs/superpowers/evidence/token-audit-standing-cost-2026-08-22.json create mode 100644 python_pkg/token_audit/checkup.py create mode 100644 python_pkg/token_audit/probe.py create mode 100644 python_pkg/token_audit/surfaces.py create mode 100644 python_pkg/token_audit/tests/test_checkup.py create mode 100644 python_pkg/token_audit/tests/test_probe.py create mode 100644 python_pkg/token_audit/tests/test_surfaces.py create mode 100644 python_pkg/token_audit/tests/test_unused.py create mode 100644 python_pkg/token_audit/unused.py diff --git a/docs/superpowers/contracts/token-audit-standing-cost-2026-08-22.json b/docs/superpowers/contracts/token-audit-standing-cost-2026-08-22.json new file mode 100644 index 00000000..6e9c4c70 --- /dev/null +++ b/docs/superpowers/contracts/token-audit-standing-cost-2026-08-22.json @@ -0,0 +1,16 @@ +{ + "title": "Land the token_audit standing-cost checkup: surfaces, probe, unused, checkup", + "objective": "Answer 'what is billed on every turn regardless of use' with measurement instead of recall. Earlier audits answered this from memory and got it wrong -- notably a seven-day dead list that included six surfaces used within thirty days -- so the module pairs a two-window liveness test with a weighted share estimate.", + "acceptance_criteria": [ + "python3 -m python_pkg.token_audit.checkup runs and prints the standing-cost report.", + "The whole token_audit package holds 100% branch coverage (116 tests).", + "Every file stays under the repo's 250-line cap.", + "A surface is only reported dead when unused in BOTH the short and long window, never on a single window.", + "pre-commit passes on all eight files with no suppression added." + ], + "out_of_scope": [ + "Acting on the report -- unregistering any MCP server, skill or rule file. This lands the instrument, not its conclusions.", + "The shell-coverage instrument work committed separately in 81191f40 and 29b8c5a7." + ], + "verifier": "timeout 120 python3 -m python_pkg.token_audit.checkup (expects a rendered report), plus python3 -m coverage run --branch --source=python_pkg/token_audit -m pytest python_pkg/token_audit/tests (expects 116 passed and 100% with zero partial branches), plus pre-commit run --files on the eight files." +} diff --git a/docs/superpowers/evidence/token-audit-standing-cost-2026-08-22.json b/docs/superpowers/evidence/token-audit-standing-cost-2026-08-22.json new file mode 100644 index 00000000..bc51327f --- /dev/null +++ b/docs/superpowers/evidence/token-audit-standing-cost-2026-08-22.json @@ -0,0 +1,47 @@ +{ + "intent": "Land the token_audit standing-cost checkup so the question 'what am I paying for on every turn' is answered by measurement rather than recall.", + "scope": [ + "python_pkg/token_audit/surfaces.py - enumerates always-on surfaces and estimates their token cost", + "python_pkg/token_audit/unused.py - two-window liveness test for registered surfaces", + "python_pkg/token_audit/probe.py - weighted share arithmetic", + "python_pkg/token_audit/checkup.py - renders the report; entry point", + "Four matching test modules", + "Non-goal: acting on the report's conclusions" + ], + "changes": [ + "A surface qualifies as dead only when unused in the short window AND the long one. A single seven-day window is not enough evidence: one earlier audit's seven-day dead list included six servers used within thirty days.", + "Shares are weighted, with cache-read weight 0.1, so a cached surface is not billed as if it were re-read in full each turn.", + "checkup.py composes surfaces + unused + probe rather than re-deriving any of them, so the report and the spend report cannot disagree." + ], + "verification": [ + { + "command": "timeout 120 python3 -m python_pkg.token_audit.checkup", + "result": "pass", + "evidence": "Rendered the standing-cost report over 44,311 turns: rules/ 4,907 tokens (1.77%), memories/ 3,758 (1.35%), MEMORY.md 4,312 (1.55%), CLAUDE.md 2,057 (0.74%), skill descriptions 3,279 (1.18%), plus per-MCP 7d/30d call counts with keep/park verdicts." + }, + { + "command": "python3 -m coverage run --branch --source=python_pkg/token_audit -m pytest python_pkg/token_audit/tests; coverage report -m", + "result": "pass", + "evidence": "116 passed. All 11 modules at 100%: TOTAL 574 stmts, 130 branches, 0 missed, 0 partial." + }, + { + "command": "wc -l python_pkg/token_audit/*.py python_pkg/token_audit/tests/*.py", + "result": "pass", + "evidence": "Largest files are report.py at 224 and parse.py at 223; every file is under the 250-line cap, so no split was needed." + }, + { + "command": "pre-commit run --files (the eight untracked files)", + "result": "pass", + "evidence": "All hooks pass, no suppressions added." + } + ], + "risks": [ + "Token counts are estimates from file sizes, not billed figures, so shares are directional rather than exact.", + "The two-window test reduces but does not eliminate false 'dead' verdicts: a surface used less often than the long window still reads as dead. The report says park, not remove, for this reason.", + "The files were found untracked in the working tree, ~2h old, from an earlier session; they are landed as-is after verifying they run and are fully covered." + ], + "rollback": [ + "git revert this commit; the package is self-contained and nothing else imports it.", + "After rollback, confirm python_pkg still collects cleanly with python3 -m pytest python_pkg -q." + ] +} diff --git a/python_pkg/token_audit/checkup.py b/python_pkg/token_audit/checkup.py new file mode 100644 index 00000000..1aa04a2c --- /dev/null +++ b/python_pkg/token_audit/checkup.py @@ -0,0 +1,71 @@ +"""The weekly standing-cost checkup. + +Complements the spend report. That report answers "where did the tokens go"; +this answers "what is still being paid for every turn regardless of use", which +is the question earlier audits kept answering by memory and kept getting wrong. + +Run as ``python3 -m python_pkg.token_audit.checkup``. +""" + +from __future__ import annotations + +import sys + +from python_pkg.token_audit import surfaces, unused +from python_pkg.token_audit.probe import weighted_pct + + +def _fmt_usage(rows: list[unused.Usage]) -> list[str]: + """Render call counts, marking only rows dead in BOTH windows.""" + header = ( + f"| name | {unused.SHORT_WINDOW_DAYS}d | {unused.LONG_WINDOW_DAYS}d | verdict |" + ) + lines = [header, "|---|---|---|---|"] + for row in rows: + verdict = "**park**" if row.dead else "keep" + lines.append( + f"| {row.name} | {row.short_calls} | {row.long_calls} | {verdict} |" + ) + return lines + + +def build(turns: int, weighted_total: int) -> str: + """Render the checkup as markdown.""" + out = ["# Standing-cost checkup", ""] + out.append( + "Every row below is billed on each of " + f"{turns:,} turns. Shares are weighted (cache-read weight 0.1)." + ) + out.append("") + out.append("## Surfaces") + out.append("| surface | est tokens | share | detail |") + out.append("|---|---|---|---|") + for surface in surfaces.collect(): + share = weighted_pct(surface.est_tokens, turns, weighted_total) + est = f"{surface.est_tokens:,}" if surface.est_tokens else "A/B only" + out.append(f"| {surface.name} | {est} | {share:.2f}% | {surface.detail} |") + out += ["", "## MCP servers", ""] + out += _fmt_usage(unused.mcp_usage()) + out += ["", "## Skills", ""] + out += _fmt_usage(unused.skill_usage()) + out += [ + "", + "_A surface marked `A/B only` cannot be sized from disk; measure it", + "with `probe.measure()`, which runs non-strict so project `.mcp.json`", + "files are included._", + "", + ] + return "\n".join(out) + + +def main(argv: list[str] | None = None) -> int: + """Print the checkup, taking turns and weighted total from the last report.""" + argv = sys.argv[1:] if argv is None else argv + turns = int(argv[0]) if argv else 44311 + total = int(argv[1]) if len(argv) > 1 else 1230346533 + sys.stdout.write(build(turns, total)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python_pkg/token_audit/probe.py b/python_pkg/token_audit/probe.py new file mode 100644 index 00000000..3c0576bd --- /dev/null +++ b/python_pkg/token_audit/probe.py @@ -0,0 +1,117 @@ +"""Measure the real per-turn prefix by starting Claude Code and reading usage. + +Disk sizes do not predict billed cost: deferred MCP schemas are absent while +their tool *names* are not, and skill bodies are summarised. The only honest +number comes from a real process start. + +Two lessons are encoded as defaults here, both learned by getting them wrong: + +* ``--strict-mcp-config`` suppresses project-scoped ``.mcp.json`` files, so an + A/B run with it understates the baseline (~2,000 tokens/turn in one audit). + Probes therefore run **non-strict** unless explicitly asked otherwise. +* A saving predicted from a synthetic config must be re-measured after the + change actually lands. :func:`check_prediction` exists to fail loudly when the + two disagree, because one audit predicted 6,122 and delivered 4,465. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +from pathlib import Path +import subprocess + +CLAUDE_BIN = Path.home() / ".local" / "bin" / "claude" +PROBE_PROMPT = "say ok" +PROBE_TIMEOUT = 300 +# Fraction by which a predicted saving may miss the measured one before the +# prediction is treated as unreliable rather than merely imprecise. +PREDICTION_TOLERANCE = 0.2 +PREFIX_KINDS = ( + "input_tokens", + "cache_creation_input_tokens", + "cache_read_input_tokens", +) + + +@dataclass(frozen=True) +class Probe: + """One measured process start.""" + + label: str + prefix_tokens: int + + +def _prefix_from_usage(usage: dict) -> int: + """Sum the token classes that make up the request prefix. + + Cache creation and cache read are alternatives — a cold start reports the + prefix as creation, a warm one as read — so summing both is correct and + stable across repeat runs. + """ + return sum(usage.get(kind) or 0 for kind in PREFIX_KINDS) + + +def measure( + label: str, + *, + cwd: Path | None = None, + mcp_config: Path | None = None, + strict: bool = False, +) -> Probe: + """Start Claude Code once and report the prefix it billed. + + ``strict`` defaults to False on purpose; see the module docstring. + """ + cmd = [str(CLAUDE_BIN), "-p", PROBE_PROMPT, "--output-format", "json"] + if mcp_config is not None: + cmd += ["--mcp-config", str(mcp_config)] + if strict: + cmd.append("--strict-mcp-config") + result = subprocess.run( + cmd, + cwd=str(cwd or Path.home()), + capture_output=True, + text=True, + timeout=PROBE_TIMEOUT, + check=True, + ) + usage = json.loads(result.stdout).get("usage") or {} + return Probe(label, _prefix_from_usage(usage)) + + +def check_prediction( + predicted_saving: int, measured_saving: int, tolerance: float = PREDICTION_TOLERANCE +) -> str | None: + """Return a warning when a predicted saving missed the measured one. + + Returns ``None`` when the prediction held, so callers can treat a truthy + result as a gate failure. + """ + if predicted_saving <= 0: + return None + drift = abs(measured_saving - predicted_saving) / predicted_saving + if drift <= tolerance: + return None + return ( + f"prediction off by {drift:.0%}: predicted {predicted_saving:,} " + f"tokens/turn, measured {measured_saving:,}" + ) + + +def weighted_pct( + tokens_per_turn: int, + turns: int, + weighted_total: int, + cache_read_weight: float = 0.1, +) -> float: + """Convert a per-turn prefix saving into a share of weighted spend. + + Every lever must be quoted through this function. Prefix tokens are billed + as cache reads, which carry a 0.1 weight, so a raw per-turn figure compared + against a weighted headline overstates the saving by roughly 10x — a + mistake made, and caught, in an earlier audit. + """ + if weighted_total <= 0: + return 0.0 + return 100.0 * tokens_per_turn * turns * cache_read_weight / weighted_total diff --git a/python_pkg/token_audit/surfaces.py b/python_pkg/token_audit/surfaces.py new file mode 100644 index 00000000..2f026396 --- /dev/null +++ b/python_pkg/token_audit/surfaces.py @@ -0,0 +1,157 @@ +"""Enumerate every context surface that is billed on every turn. + +The weekly audit kept missing surfaces — plugins and ``autoMode.environment`` +were both found only because the user asked "nothing else?". The fix is that +the list of surfaces lives in code: a surface that is not measured shows up as +an explicit row with ``measured=False`` instead of silently not existing. + +Sizes here are *estimates from disk*. Anything that can be A/B'd against a real +process start should be, via :mod:`python_pkg.token_audit.probe`; disk size and +measured prefix cost disagree (skill descriptions are summarised, MCP servers +contribute only tool names), and the measured number is the one to trust. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +from pathlib import Path + +CLAUDE_DIR = Path.home() / ".claude" +GLOBAL_CONFIG = Path.home() / ".claude.json" +SETTINGS = CLAUDE_DIR / "settings.json" +# Anthropic's rough byte-to-token ratio for English prose. +BYTES_PER_TOKEN = 4 + + +@dataclass(frozen=True) +class Surface: + """One always-loaded context component.""" + + name: str + detail: str + bytes_on_disk: int + measurable: bool = False + + @property + def est_tokens(self) -> int: + """Disk size converted to an approximate token count.""" + return self.bytes_on_disk // BYTES_PER_TOKEN + + +def _tree_bytes(root: Path, pattern: str = "*.md") -> int: + """Total size of every file under *root* matching *pattern*.""" + if not root.is_dir(): + return 0 + return sum(p.stat().st_size for p in root.rglob(pattern) if p.is_file()) + + +def _json_field_bytes(path: Path, field: str) -> int: + """Serialized size of one top-level field, or 0 when absent.""" + if not path.exists(): + return 0 + try: + data = json.loads(path.read_text()) + except (json.JSONDecodeError, OSError): + return 0 + if field not in data: + return 0 + return len(json.dumps(data[field])) + + +def _skill_description_bytes() -> int: + """Approximate the frontmatter that every skill contributes.""" + total = 0 + for skill in (CLAUDE_DIR / "skills").glob("*/SKILL.md"): + head = skill.read_text(errors="replace")[:4000] + _, _, rest = head.partition("description:") + total += len(rest.split("\n---", 1)[0]) if rest else 0 + return total + + +def _mcp_names(path: Path, field: str = "mcpServers") -> list[str]: + """Server names registered in a config file.""" + if not path.exists(): + return [] + try: + return sorted(json.loads(path.read_text()).get(field, {})) + except (json.JSONDecodeError, OSError): + return [] + + +def _enabled_plugins() -> list[str]: + """Plugins switched on in settings.json.""" + if not SETTINGS.exists(): + return [] + try: + plugins = json.loads(SETTINGS.read_text()).get("enabledPlugins", {}) + except (json.JSONDecodeError, OSError): + return [] + return sorted(name for name, on in plugins.items() if on) + + +def _project_mcp_files() -> list[Path]: + """Project-scoped .mcp.json files under the home directory. + + These are invisible to ``--strict-mcp-config``, which is exactly how an + earlier audit under-measured the real baseline by ~2,000 tokens/turn. + """ + return sorted(p for p in Path.home().glob("*/.mcp.json") if p.is_file()) + + +def collect() -> list[Surface]: + """Enumerate every known billed surface. + + Adding a surface here is what makes it appear in the report; the weekly + check walks this list rather than a human's memory of it. + """ + memory_dir = CLAUDE_DIR / "projects" / "-home-kuhy" / "memory" + plugins = _enabled_plugins() + project_mcp = _project_mcp_files() + global_mcp = _mcp_names(GLOBAL_CONFIG) + parked = _mcp_names(CLAUDE_DIR / "mcp-parked.json") + return [ + Surface( + "rules/", + "always-on, loaded recursively", + _tree_bytes(CLAUDE_DIR / "rules"), + ), + Surface("memories/", "always-on", _tree_bytes(CLAUDE_DIR / "memories")), + Surface( + "CLAUDE.md", + "global instructions", + (CLAUDE_DIR / "CLAUDE.md").stat().st_size + if (CLAUDE_DIR / "CLAUDE.md").exists() + else 0, + ), + Surface( + "MEMORY.md", + "auto-memory index", + (memory_dir / "MEMORY.md").stat().st_size + if (memory_dir / "MEMORY.md").exists() + else 0, + ), + Surface( + "skill descriptions", + f"{len(list((CLAUDE_DIR / 'skills').glob('*/SKILL.md')))} personal skills", + _skill_description_bytes(), + ), + Surface("plugins", ", ".join(plugins) or "none", 0, measurable=True), + Surface( + "global MCP", + f"{len(global_mcp)} active, {len(parked)} parked", + 0, + measurable=True, + ), + Surface( + "project .mcp.json", + f"{len(project_mcp)} repos (per-project only)", + 0, + measurable=True, + ), + Surface( + "autoMode.environment", + "injected in auto mode", + _json_field_bytes(SETTINGS, "autoMode"), + ), + ] diff --git a/python_pkg/token_audit/tests/test_checkup.py b/python_pkg/token_audit/tests/test_checkup.py new file mode 100644 index 00000000..7be31e04 --- /dev/null +++ b/python_pkg/token_audit/tests/test_checkup.py @@ -0,0 +1,64 @@ +"""Tests for the standing-cost checkup report.""" + +from __future__ import annotations + +import pytest + +from python_pkg.token_audit import checkup, surfaces, unused + + +@pytest.fixture(autouse=True) +def stub_sources(monkeypatch: pytest.MonkeyPatch) -> None: + """Replace disk scans with fixed rows so the render is deterministic.""" + monkeypatch.setattr( + surfaces, + "collect", + lambda: [ + surfaces.Surface("rules/", "always-on", 4000), + surfaces.Surface("global MCP", "13 active", 0, measurable=True), + ], + ) + monkeypatch.setattr( + unused, + "mcp_usage", + lambda: [unused.Usage("reaper", 0, 0), unused.Usage("ck3", 0, 9)], + ) + monkeypatch.setattr(unused, "skill_usage", lambda: [unused.Usage("grilling", 0, 0)]) + + +def test_report_lists_every_surface() -> None: + """A surface present in the enumeration reaches the report.""" + text = checkup.build(44311, 1230346533) + assert "rules/" in text + assert "global MCP" in text + + +def test_disk_sized_surface_shows_weighted_share() -> None: + """Shares are weighted, not raw — the unit error an audit made.""" + text = checkup.build(44311, 1230346533) + assert "| 1,000 | 0.36% |" in text + + +def test_unmeasurable_surface_is_labelled() -> None: + """A surface that needs an A/B says so rather than reporting zero bytes.""" + assert "A/B only" in checkup.build(1000, 1000000) + + +def test_only_both_window_zero_is_parked() -> None: + """ck3 (unused 7d, used 30d) must not be recommended for parking.""" + text = checkup.build(44311, 1230346533) + assert "| reaper | 0 | 0 | **park** |" in text + assert "| ck3 | 0 | 9 | keep |" in text + + +def test_skills_section_present() -> None: + """Skills are audited alongside MCP servers.""" + assert "## Skills" in checkup.build(1000, 1000000) + + +def test_main_defaults_and_args(capsys: pytest.CaptureFixture[str]) -> None: + """The CLI accepts turns and total, and defaults when omitted.""" + assert checkup.main([]) == 0 + assert "Standing-cost checkup" in capsys.readouterr().out + assert checkup.main(["100", "1000000"]) == 0 + assert "100 turns" in capsys.readouterr().out diff --git a/python_pkg/token_audit/tests/test_probe.py b/python_pkg/token_audit/tests/test_probe.py new file mode 100644 index 00000000..4d29e25c --- /dev/null +++ b/python_pkg/token_audit/tests/test_probe.py @@ -0,0 +1,111 @@ +"""Tests for the prefix probe and its prediction gate.""" + +from __future__ import annotations + +import json +import subprocess +from typing import TYPE_CHECKING + +import pytest + +from python_pkg.token_audit import probe + +if TYPE_CHECKING: + from pathlib import Path + + +def test_prefix_sums_cold_start() -> None: + """A cold start reports the prefix as cache creation.""" + usage = { + "input_tokens": 2, + "cache_creation_input_tokens": 46434, + "cache_read_input_tokens": 0, + } + assert probe._prefix_from_usage(usage) == 46436 + + +def test_prefix_sums_warm_start() -> None: + """A warm start reports the same prefix as cache read.""" + usage = {"input_tokens": 2, "cache_read_input_tokens": 46434} + assert probe._prefix_from_usage(usage) == 46436 + + +def test_prefix_handles_missing_and_null() -> None: + """Absent or null token classes count as zero.""" + assert probe._prefix_from_usage({}) == 0 + assert probe._prefix_from_usage({"input_tokens": None}) == 0 + + +def test_prediction_within_tolerance_passes() -> None: + """A prediction close to the measurement is not flagged.""" + assert probe.check_prediction(1000, 900) is None + + +def test_prediction_beyond_tolerance_is_flagged() -> None: + """The 6,122-vs-4,465 miss from a real audit must fail the gate.""" + warning = probe.check_prediction(6122, 4465) + assert warning is not None + assert "27%" in warning + + +def test_prediction_ignores_nonpositive() -> None: + """No prediction means nothing to check.""" + assert probe.check_prediction(0, 500) is None + + +def test_weighted_pct_applies_cache_read_weight() -> None: + """Prefix savings are billed as cache reads, at weight 0.1.""" + assert probe.weighted_pct(4465, 44311, 1230346533) == pytest.approx(1.608, abs=0.01) + + +def test_weighted_pct_guards_zero_total() -> None: + """An empty week cannot divide by zero.""" + assert probe.weighted_pct(100, 10, 0) == 0.0 + + +def test_measure_runs_non_strict_by_default( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Strict mode hides project .mcp.json, so it must be opt-in.""" + seen: dict[str, list[str]] = {} + + def fake_run(cmd: list[str], **_: object) -> subprocess.CompletedProcess[str]: + seen["cmd"] = cmd + return subprocess.CompletedProcess( + cmd, 0, json.dumps({"usage": {"input_tokens": 5}}), "" + ) + + monkeypatch.setattr(subprocess, "run", fake_run) + result = probe.measure("base", mcp_config=tmp_path / "x.json") + assert "--strict-mcp-config" not in seen["cmd"] + assert result.prefix_tokens == 5 + + +def test_measure_can_opt_into_strict( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Strict mode is still reachable when deliberately requested.""" + seen: dict[str, list[str]] = {} + + def fake_run(cmd: list[str], **_: object) -> subprocess.CompletedProcess[str]: + seen["cmd"] = cmd + return subprocess.CompletedProcess(cmd, 0, json.dumps({"usage": {}}), "") + + monkeypatch.setattr(subprocess, "run", fake_run) + probe.measure("strict", mcp_config=tmp_path / "x.json", strict=True) + assert "--strict-mcp-config" in seen["cmd"] + + +def test_measure_without_mcp_config_omits_flags( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The plain probe passes no MCP flags at all.""" + seen: dict[str, list[str]] = {} + + def fake_run(cmd: list[str], **_: object) -> subprocess.CompletedProcess[str]: + seen["cmd"] = cmd + return subprocess.CompletedProcess(cmd, 0, json.dumps({"usage": {}}), "") + + monkeypatch.setattr(subprocess, "run", fake_run) + probe.measure("plain") + assert "--mcp-config" not in seen["cmd"] diff --git a/python_pkg/token_audit/tests/test_surfaces.py b/python_pkg/token_audit/tests/test_surfaces.py new file mode 100644 index 00000000..26300454 --- /dev/null +++ b/python_pkg/token_audit/tests/test_surfaces.py @@ -0,0 +1,143 @@ +"""Tests for the billed-surface enumerator.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from python_pkg.token_audit import surfaces + + +@pytest.fixture +def fake_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Build a minimal ~/.claude tree and point the module at it.""" + claude = tmp_path / ".claude" + (claude / "rules").mkdir(parents=True) + (claude / "rules" / "a.instructions.md").write_text("x" * 400) + (claude / "memories").mkdir() + (claude / "memories" / "m.md").write_text("y" * 200) + (claude / "CLAUDE.md").write_text("z" * 100) + memory = claude / "projects" / "-home-kuhy" / "memory" + memory.mkdir(parents=True) + (memory / "MEMORY.md").write_text("w" * 80) + skill = claude / "skills" / "demo" + skill.mkdir(parents=True) + (skill / "SKILL.md").write_text("---\nname: demo\ndescription: hello there\n---\n") + (claude / "settings.json").write_text( + json.dumps( + { + "enabledPlugins": {"on@x": True, "off@y": False}, + "autoMode": {"environment": ["a", "b"]}, + } + ) + ) + (tmp_path / ".claude.json").write_text(json.dumps({"mcpServers": {"aseprite": {}}})) + (claude / "mcp-parked.json").write_text(json.dumps({"mcpServers": {"reaper": {}}})) + repo = tmp_path / "repo" + repo.mkdir() + (repo / ".mcp.json").write_text(json.dumps({"mcpServers": {"local": {}}})) + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + for name, value in [ + ("CLAUDE_DIR", claude), + ("GLOBAL_CONFIG", tmp_path / ".claude.json"), + ("SETTINGS", claude / "settings.json"), + ]: + monkeypatch.setattr(surfaces, name, value) + return tmp_path + + +def test_every_surface_is_reported(fake_home: Path) -> None: + """The enumeration is the contract: a missing row means a missed surface.""" + names = [s.name for s in surfaces.collect()] + assert names == [ + "rules/", + "memories/", + "CLAUDE.md", + "MEMORY.md", + "skill descriptions", + "plugins", + "global MCP", + "project .mcp.json", + "autoMode.environment", + ] + + +def test_sizes_come_from_disk(fake_home: Path) -> None: + """Byte counts reflect real files.""" + found = {s.name: s for s in surfaces.collect()} + assert found["rules/"].bytes_on_disk == 400 + assert found["rules/"].est_tokens == 100 + assert found["CLAUDE.md"].bytes_on_disk == 100 + assert found["MEMORY.md"].bytes_on_disk == 80 + + +def test_enabled_plugins_only(fake_home: Path) -> None: + """Disabled plugins cost nothing and must not be listed.""" + found = {s.name: s for s in surfaces.collect()} + assert found["plugins"].detail == "on@x" + assert found["plugins"].measurable + + +def test_mcp_split_between_active_and_parked(fake_home: Path) -> None: + """Parked servers are reported separately from active ones.""" + found = {s.name: s for s in surfaces.collect()} + assert found["global MCP"].detail == "1 active, 1 parked" + + +def test_project_mcp_files_counted(fake_home: Path) -> None: + """Project-scoped configs are surfaced, not hidden.""" + found = {s.name: s for s in surfaces.collect()} + assert "1 repos" in found["project .mcp.json"].detail + + +def test_skill_description_measured(fake_home: Path) -> None: + """Skill frontmatter contributes bytes.""" + found = {s.name: s for s in surfaces.collect()} + assert found["skill descriptions"].bytes_on_disk > 0 + assert "1 personal skills" in found["skill descriptions"].detail + + +def test_missing_files_are_zero( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An absent tree or config contributes zero rather than raising.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + for name in ("CLAUDE_DIR", "GLOBAL_CONFIG", "SETTINGS"): + monkeypatch.setattr(surfaces, name, tmp_path / "nope") + result = surfaces.collect() + assert all(s.bytes_on_disk == 0 for s in result) + assert surfaces._mcp_names(tmp_path / "nope.json") == [] + + +def test_corrupt_json_is_tolerated( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Malformed config files degrade to zero, never to a crash.""" + bad = tmp_path / "bad.json" + bad.write_text("{oops") + assert surfaces._json_field_bytes(bad, "autoMode") == 0 + assert surfaces._mcp_names(bad) == [] + monkeypatch.setattr(surfaces, "SETTINGS", bad) + assert surfaces._enabled_plugins() == [] + + +def test_absent_field_is_zero(tmp_path: Path) -> None: + """A config without the field contributes nothing.""" + cfg = tmp_path / "s.json" + cfg.write_text(json.dumps({"other": 1})) + assert surfaces._json_field_bytes(cfg, "autoMode") == 0 + + +def test_enabled_plugins_without_settings( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """No settings file means no plugins.""" + monkeypatch.setattr(surfaces, "SETTINGS", tmp_path / "gone.json") + assert surfaces._enabled_plugins() == [] + + +def test_tree_bytes_ignores_missing(tmp_path: Path) -> None: + """A non-directory root totals zero.""" + assert surfaces._tree_bytes(tmp_path / "absent") == 0 diff --git a/python_pkg/token_audit/tests/test_unused.py b/python_pkg/token_audit/tests/test_unused.py new file mode 100644 index 00000000..ea66824a --- /dev/null +++ b/python_pkg/token_audit/tests/test_unused.py @@ -0,0 +1,100 @@ +"""Tests for the dual-window unused-surface detector.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from python_pkg.token_audit import unused + + +@pytest.fixture +def transcripts(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Redirect the transcript scan at a temporary tree.""" + root = tmp_path / "projects" + (root / "proj").mkdir(parents=True) + monkeypatch.setattr(unused, "TRANSCRIPT_ROOT", root) + return root / "proj" + + +def test_dead_requires_both_windows_empty() -> None: + """Used at 30 days but not 7 is not dead — the rule that prevents overreach.""" + assert unused.Usage("reaper", 0, 0).dead + assert not unused.Usage("ck3", 0, 9).dead + assert not unused.Usage("aseprite", 232, 748).dead + + +def test_usage_counts_across_windows(transcripts: Path) -> None: + """A recent call lands in both windows; an old one only in the long window.""" + now = 1_000_000_000.0 + recent = transcripts / "recent.jsonl" + recent.write_text('{"name":"mcp__aseprite__draw_line"}\n') + import os + + os.utime(recent, (now - 3 * 86400, now - 3 * 86400)) + old = transcripts / "old.jsonl" + old.write_text('{"name":"mcp__ck3__ck3_search"}\n') + os.utime(old, (now - 20 * 86400, now - 20 * 86400)) + + result = { + u.name: u + for u in unused.usage_for( + ["aseprite", "ck3", "reaper"], unused._MCP_CALL, now=now + ) + } + assert result["aseprite"].short_calls == 1 + assert result["aseprite"].long_calls == 1 + assert result["ck3"].short_calls == 0 + assert result["ck3"].long_calls == 1 + assert result["reaper"].dead + + +def test_scan_skips_unreadable( + transcripts: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """An unreadable transcript is skipped rather than aborting the audit.""" + bad = transcripts / "bad.jsonl" + bad.write_text("{}") + + def boom(*_: object, **__: object) -> str: + msg = "nope" + raise OSError(msg) + + monkeypatch.setattr(Path, "read_text", boom) + assert unused._scan(unused._MCP_CALL, 0) == {} + + +def test_mcp_usage_reads_config(tmp_path: Path, transcripts: Path) -> None: + """Server names come from the global config.""" + cfg = tmp_path / "claude.json" + cfg.write_text(json.dumps({"mcpServers": {"reaper": {}, "aseprite": {}}})) + names = {u.name for u in unused.mcp_usage(cfg)} + assert names == {"reaper", "aseprite"} + + +def test_mcp_usage_survives_bad_config(tmp_path: Path) -> None: + """A corrupt config yields no rows instead of raising.""" + cfg = tmp_path / "broken.json" + cfg.write_text("{not json") + assert unused.mcp_usage(cfg) == [] + assert unused.mcp_usage(tmp_path / "missing.json") == [] + + +def test_skill_usage_lists_installed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, transcripts: Path +) -> None: + """Skill names come from the skills directory.""" + home = tmp_path / "home" + skill = home / ".claude" / "skills" / "grilling" + skill.mkdir(parents=True) + (skill / "SKILL.md").write_text("---\nname: grilling\n---\n") + monkeypatch.setattr(Path, "home", staticmethod(lambda: home)) + result = unused.skill_usage() + assert [u.name for u in result] == ["grilling"] + + +def test_skill_pattern_matches_invocation() -> None: + """The skill regex matches the transcript form.""" + assert unused._SKILL_CALL.findall('{"skill":"phone-deploy"}') == ["phone-deploy"] diff --git a/python_pkg/token_audit/unused.py b/python_pkg/token_audit/unused.py new file mode 100644 index 00000000..0594a647 --- /dev/null +++ b/python_pkg/token_audit/unused.py @@ -0,0 +1,78 @@ +"""Find registered surfaces that nothing has called, using two time windows. + +A single seven-day window is not enough evidence to unregister anything: one +audit's seven-day "dead" list included six servers that had been used within +thirty days. A surface therefore qualifies as dead only when it is unused in +the short window *and* the long one. +""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import dataclass +import json +from pathlib import Path +import re +import time + +TRANSCRIPT_ROOT = Path.home() / ".claude" / "projects" +SECONDS_PER_DAY = 86400 +SHORT_WINDOW_DAYS = 7 +LONG_WINDOW_DAYS = 30 +_MCP_CALL = re.compile(r'"name":"mcp__([a-zA-Z0-9_-]+)__') +_SKILL_CALL = re.compile(r'"skill":"([^"]+)"') + + +@dataclass(frozen=True) +class Usage: + """Call counts for one surface across both windows.""" + + name: str + short_calls: int + long_calls: int + + @property + def dead(self) -> bool: + """True when unused in both windows, so it is safe to park.""" + return self.short_calls == 0 and self.long_calls == 0 + + +def _scan(pattern: re.Pattern[str], since: float) -> Counter[str]: + """Count regex matches across transcripts modified after *since*.""" + found: Counter[str] = Counter() + for path in TRANSCRIPT_ROOT.glob("*/*.jsonl"): + try: + if path.stat().st_mtime < since: + continue + text = path.read_text(errors="replace") + except OSError: + continue + found.update(pattern.findall(text)) + return found + + +def usage_for( + names: list[str], pattern: re.Pattern[str], now: float | None = None +) -> list[Usage]: + """Report short- and long-window call counts for each of *names*.""" + now = time.time() if now is None else now + short = _scan(pattern, now - SHORT_WINDOW_DAYS * SECONDS_PER_DAY) + long_ = _scan(pattern, now - LONG_WINDOW_DAYS * SECONDS_PER_DAY) + return [Usage(n, short.get(n, 0), long_.get(n, 0)) for n in sorted(names)] + + +def mcp_usage(config: Path | None = None) -> list[Usage]: + """Call counts for every globally registered MCP server.""" + config = config or Path.home() / ".claude.json" + try: + names = list(json.loads(config.read_text()).get("mcpServers", {})) + except (json.JSONDecodeError, OSError): + return [] + return usage_for(names, _MCP_CALL) + + +def skill_usage() -> list[Usage]: + """Invocation counts for every personal skill.""" + root = Path.home() / ".claude" / "skills" + names = [p.parent.name for p in root.glob("*/SKILL.md")] + return usage_for(names, _SKILL_CALL)