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) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-08-22 14:33:03 +02:00
parent 29b8c5a788
commit 11aea86920
10 changed files with 904 additions and 0 deletions

View File

@ -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."
}

View File

@ -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."
]
}

View File

@ -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())

View File

@ -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

View File

@ -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"),
),
]

View File

@ -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

View File

@ -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"]

View File

@ -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

View File

@ -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"]

View File

@ -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)