token_audit v2: dedup, subagents, per-model USD, snapshots+ledger; repo contract; dep bumps (fixes red freshness CI)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-10-04 16:39:42 +02:00
parent 5a2a4bcbda
commit 32ea8ba949
42 changed files with 2486 additions and 315 deletions

25
.github/workflows/repo-contract.yml vendored Normal file
View File

@ -0,0 +1,25 @@
name: repo-contract
# The local pre-commit hook can be skipped (a fresh clone has no hooks), so the
# contract is enforced on push too. Mirrors md-naming.yml.
on:
push:
branches: [main]
pull_request:
workflow_dispatch:
jobs:
repo-contract:
name: repo contract
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Check out the shared gate
uses: actions/checkout@v4
with:
repository: kuhyx/utils
path: .utils
- name: Enforce the repo contract
run: bash .utils/scripts/check_repo_contract.sh

View File

@ -348,13 +348,19 @@ repos:
# on unrelated commits. Skips itself where ~/.claude is absent (CI). # on unrelated commits. Skips itself where ~/.claude is absent (CI).
files: ^(linux_configuration/tests/test_session_autopsy_hooks\.py|meta/scripts/mutate_session_autopsy_hooks\.py)$ files: ^(linux_configuration/tests/test_session_autopsy_hooks\.py|meta/scripts/mutate_session_autopsy_hooks\.py)$
- id: ci-mirror - id: ci-mirror
name: CI mirror (clean-venv install + --all-files + changed-pkg pytest) name: CI mirror (background; a failed verdict blocks the next push)
entry: bash meta/scripts/ci_mirror.sh entry: bash meta/scripts/ci_mirror.sh
language: system language: system
pass_filenames: false pass_filenames: false
always_run: true always_run: true
require_serial: true require_serial: true
stages: [pre-push] stages: [pre-push]
- id: repo-contract
name: repo contract (CLAUDE.md Commands + test_changed)
entry: scripts/check_repo_contract.sh
language: system
pass_filenames: false
files: ^(CLAUDE\.md|AGENTS\.md|scripts/test_changed\.sh)$
# =========================================================================== # ===========================================================================
# VULTURE - Dead code detection (disabled - doesn't work well with pre-commit) # VULTURE - Dead code detection (disabled - doesn't work well with pre-commit)

View File

@ -81,7 +81,14 @@ hooks travel with the repo. `.git/hooks/` is unused — **never run
then `pre-commit run --hook-stage pre-commit`. then `pre-commit run --hook-stage pre-commit`.
- **pre-push**: `pre-commit run --hook-stage pre-push` — `prettier` and - **pre-push**: `pre-commit run --hook-stage pre-push` — `prettier` and
`ci-mirror`. `ci-mirror` runs a clean-venv install, `pre-commit run` over all `ci-mirror`. `ci-mirror` runs a clean-venv install, `pre-commit run` over all
files, and pytest for changed packages. files, and pytest for changed packages — but **not on the push's critical
path**. The push spawns a detached, niced worker and returns in well under a
second; the worker records a verdict for the tree it checked. A recorded
failure refuses the *next* push until the tree is verified in the foreground
with `CI_MIRROR_SYNC=1 git push`, so a broken tree reaches GitHub at most
once. A tree that already passed costs nothing to re-push.
`prettier` caches its verdict per file content + prettier version + config,
so unchanged files never start Node.
Bootstrap a clone or new machine: Bootstrap a clone or new machine:
@ -161,3 +168,12 @@ number of error codes, so it is weaker than the `strict = true` in the config.
- `docs/superpowers/evidence/template.json` — evidence template - `docs/superpowers/evidence/template.json` — evidence template
- `.github/workflows/` — `pre-commit`, `python-tests`, `shell-tests`, and one - `.github/workflows/` — `pre-commit`, `python-tests`, `shell-tests`, and one
CI workflow per app CI workflow per app
## Commands
- run: n/a: monorepo of independent scripts and apps, no single entry point (see Repository Layout)
- test: `python3 -m pytest -q -o addopts="--strict-markers --strict-config"`
- test-changed: `scripts/test_changed.sh`
- lint: `pre-commit run --hook-stage pre-commit`
- coverage: `python3 -m pytest -q --cov=python_pkg --cov-branch --cov-report=xml -o addopts="--strict-markers --strict-config"`
- coverage-gaps: `coverage-gaps coverage.xml`

View File

@ -0,0 +1,15 @@
{
"title": "token_audit v2 + repo contract in testsAndMisc",
"objective": "token_audit measures deduplicated, subagent-inclusive, per-model USD spend over the window since the last snapshot and scores ledger levers; testsAndMisc satisfies the shared repo contract.",
"acceptance_criteria": [
"python_pkg/token_audit at 100% branch coverage, no suppressions, every file <= 250 lines",
"Tests never write to ~/.claude/token-report (autouse conftest redirect)",
"python3 -m python_pkg.token_audit --no-write runs and reports the window since the previous snapshot",
"check_repo_contract.sh passes; dependency freshness --all passes"
],
"out_of_scope": [
"~/.claude hooks and the /token-optimization skill (separate repo)",
"Other packages' coverage"
],
"verifier": "pytest --cov-branch, pre-commit on changed files, CI on push"
}

View File

@ -0,0 +1,48 @@
{
"intent": "Give the new signal-bot phone app (~/signal-bot/app) a launcher icon from the shared family, so it stops shipping the stock Flutter beach icon.",
"scope": [
"python_pkg/app_icons/_glyph_art_more.py - the new SVG fragment",
"python_pkg/app_icons/glyphs.py - register the glyph in GLYPHS",
"python_pkg/app_icons/apps.py - register the app so --app signalbot_remote passes argument validation",
"Non-goal: no change to the shared accent #B8862E or the charcoal field; a glyph is distinguished by silhouette only",
"Non-goal: no change to any existing app's icon",
"Non-goal: the app itself, its Flutter code and its Cloudflare transport all live in ~/signal-bot and are committed there"
],
"changes": [
"Added _REMOTE_WAVE: a filled handset (x=412..612, y=440..792, r=56) with two punched-out buttons (r=40 at 512,540 and 512,680) under a detached stroked arc (r=132 about 512,440, drawn 135deg to 45deg). Total ink y=272..792, x=383..641, inside the 560 safe box.",
"The arc is detached from the body on purpose: an arc that meets the body is a padlock shackle, and padlock-closed is already in this family.",
"The body is filled rather than stroked, for the same reason as _QUILL_NIB and _ANVIL: a 200-wide stroked body leaves a 56px interior at STROKE_WIDTH=72 and fills in solid at 48dp.",
"Registered Glyph('remote-wave', ...) in GLYPHS and AppIcon(key='signalbot_remote', glyph='remote-wave', linux=False) in APPS, pointing at ~/signal-bot/app."
],
"verification": [
{
"command": "PYTHONPATH=~/testsAndMisc python3 -m python_pkg.app_icons list",
"result": "pass",
"evidence": "remote-wave appears under Glyphs; 'signalbot_remote #B8862E remote-wave /home/kuhy/signal-bot/app' appears under Apps."
},
{
"command": "PYTHONPATH=~/testsAndMisc python3 -m python_pkg.app_icons generate --app signalbot_remote --android",
"result": "pass",
"evidence": "wrote 6 asset files to /home/kuhy/signal-bot/app/assets/icon and regenerated the Android mipmaps. The generator's safe-box and negative-space checks did not fire. A first attempt failed because flutter_launcher_icons was not yet in the app's pubspec; the config block was added and it then succeeded."
},
{
"command": "visual review of the rendered preview sheet (full size, circle-masked, 48dp)",
"result": "pass",
"evidence": "Reads as a handset with a signal arc at all three sizes. The two buttons stay separate at 48dp (60px edge gap against MIN_NEGATIVE_SPACE=36) and the arc keeps it from reading as a domino."
},
{
"command": "python3 -m coverage run --branch --source=python_pkg/app_icons -m pytest python_pkg/app_icons/tests",
"result": "pass",
"evidence": "44 passed; python_pkg/app_icons at 100% line and branch coverage. The registry invariants (every app references a real glyph, accents are shared, no glyph paints the background) iterate over APPS and GLYPHS, so the new entries are exercised by the existing tests."
}
],
"risks": [
"The glyph is new artwork, so only a visual check can confirm it reads correctly at 48dp. Mitigated by rendering and reviewing the actual preview sheet rather than trusting the SVG source.",
"Not yet verified on the phone: kuhy's device is unavailable until 2026-09-10, so the launcher has not rendered this icon on real hardware.",
"_glyph_art_more.py is now 160 lines. Together with _glyph_art.py it exists solely to stay under the 250-line cap, so a third file will be needed before long."
],
"rollback": [
"git revert the commit. The three edits are additive - no existing glyph, app or shared token is touched - so a revert cannot change any other app's icon.",
"After rollback, re-run 'python3 -m python_pkg.app_icons list' and confirm the remaining 16 glyphs still resolve. The already-generated PNGs in ~/signal-bot/app would then have no source in this repo, so revert that repo's icon commit too or regenerate from a restored glyph."
]
}

View File

@ -0,0 +1,51 @@
{
"intent": "Make `git push` return in well under a second instead of 22s-to-minutes, by taking the CI mirror off the push's critical path without giving up the guarantee it exists for.",
"scope": [
"meta/scripts/ci_mirror_async.sh - new: verdict bookkeeping, the detached worker, and the green-tree short-circuit moved here",
"meta/scripts/ci_mirror.sh - main() gains three modes (worker / sync / async); the gates themselves are unchanged",
".pre-commit-config.yaml - hook name now describes the background behaviour",
"AGENTS.md (CLAUDE.md symlinks to it) - documents the new pre-push contract",
"meta/requirements.txt - matplotlib 3.11.2, which went red on main after release",
"Non-goal: changing which gates run or what they check. A verified tree is verified by exactly the same commands as before."
],
"changes": [
"`git push` spawns a detached, niced (nice -n 19, ionice -c 3) worker for HEAD's tree and returns. Measured at 0.028s.",
"The worker records a verdict of ok or fail for the tree it checked, plus the full log, under .ci-mirror-venv/verdicts/.",
"A recorded failure refuses the NEXT push, printing the last 40 lines of the failing log and the foreground escape hatch, so a broken tree can reach GitHub at most once - where CI sees the same commit within minutes.",
"CI_MIRROR_SYNC=1 restores the old blocking behaviour for a push that must be verified before it lands.",
"Workers serialise on a flock rather than skipping when one is already in flight: skipping would have left a newly pushed tree permanently unverified, which was a hole in the first draft of this patch.",
"The green-tree short-circuit is unchanged in behaviour - a tree that already passed still costs nothing to re-push."
],
"verification": [
{
"command": "PRE_COMMIT_FROM_REF=42e41319~1 PRE_COMMIT_TO_REF=42e41319 bash meta/scripts/ci_mirror.sh # async push path",
"result": "pass",
"evidence": "0.028s total, printing 'verification running in the background (tree 68346b01e46a)'; `pgrep` confirmed the detached worker was running after the command returned."
},
{
"command": "block_on_previous_failure with a synthetic 'fail' verdict in an isolated VENV_DIR",
"result": "pass",
"evidence": "exit 1, the failing log's tail echoed to stderr, and the CI_MIRROR_SYNC=1 hint printed. With the verdict flipped to 'ok' the same call passes straight through."
},
{
"command": "timing baseline before the change: bash meta/scripts/ci_mirror.sh with the green marker cleared",
"result": "pass",
"evidence": "22.6s for a 6-file scoped range with a warm venv; 70.8s when requirements.txt had changed and the clean venv was rebuilt. The clean worktree checkout itself is only 0.048s, so the cost is the verification work, not overhead."
},
{
"command": "shellcheck -x -S warning meta/scripts/ci_mirror.sh meta/scripts/ci_mirror_async.sh",
"result": "pass",
"evidence": "no findings; shfmt applied. ci_mirror.sh is 244 lines, under the repo's 250-line cap."
}
],
"risks": [
"A commit that fails the mirror can reach GitHub once, before the verdict lands. This is the tradeoff kuhy chose explicitly over a 22s-to-minutes blocking push; GitHub CI and the next push both catch it.",
"Several pushes in quick succession queue workers that serialise on the lock, so a verdict can lag the push by the length of the runs ahead of it. Each worker still checks its own tree.",
"The worker is detached with setsid, so it survives the terminal that started it. It is niced and ionice idle, so it yields to interactive work, but it will still be running after the push prompt returns.",
"Paths in ci_mirror_async.sh are computed in functions because the file is sourced before ci_mirror.sh defines VENV_DIR; defining them as readonly constants there would silently resolve to /verdicts."
],
"rollback": [
"git revert the commit. ci_mirror.sh's main() returns to calling ensure_venv / run_gates_in_clean_worktree / record_tree_green inline, and pushes block again.",
"After rollback, delete .ci-mirror-venv/verdicts/, which is then unused, and confirm `git push` still runs the gate in the foreground."
]
}

View File

@ -1,61 +0,0 @@
{
"intent": "Flatten linux_configuration by dissolving the scripts/ and single_use/ levels, cutting directory-depth violations nearly in half and making the depth-2 cap reachable.",
"scope": [
"linux_configuration/scripts/{features,fixes,utils,misc,fresh-install,feedback-appimage,lib,boot_recovery,diagnostics,gaming,meta,periodic_background} -> linux_configuration/<same>",
"single_use's loose installer scripts -> linux_configuration/installers/",
"single_use/lib/joplin_server.sh -> installers/, freeing lib/ for the shared library",
"~90 files rewritten for the new paths; 16 $HOME symlinks repointed",
"Non-goal: periodic_background/digital_wellbeing and hosts, deferred pending the guard-lib repoint"
],
"changes": [
"git mv'd every category out of scripts/single_use/ and scripts/ up to linux_configuration/, then removed the two empty parents. scripts/tests/ held only an untracked __pycache__ and was deleted.",
"Rewrote absolute references (linux_configuration/scripts/... -> linux_configuration/...) across 38 files, and relative ../scripts/X forms across the docs and harnesses that used them.",
"Recomputed every relative `source` of lib/common.sh and lib/android.sh per file depth: 40 files were ../../ and 12 were ../../../, and the correct prefix differs by where the file now sits.",
"Fixed REPO_ROOT in 7 fixes/lib/tests harnesses plus setup_syncyomi_guard.sh, which walked up four levels from a directory that is now two levels shallower.",
"Repointed 16 symlinks in $HOME: the whole i3blocks bar, keepass-open, and two diagnostics tools.",
"Fixed linux_configuration/.gitignore's two secret paths and meta/shell-coverage-allowlist.txt's joplin_server entry.",
"Fixed linux_configuration/tests/conftest.py, whose sys.path entries for fast_count and the transcribe tools still named the old tree.",
"Documented the new layout in AGENTS.md and removed the last stale billsplit references."
],
"verification": [
{
"command": "check_directory_depth.sh --all",
"result": "pass (improvement)",
"evidence": "248 -> 139 violations. Roughly 40 of the remainder are inside digital_wellbeing and hosts, which are deliberately deferred."
},
{
"command": "resolve every '$SCRIPT_DIR/../lib/...' source against the filesystem",
"result": "fail then pass",
"evidence": "First sweep left 5 files in misc/testsAndMisc-bash pointing at ../../../lib/common.sh (they had a different prefix than the 40 already fixed). After the second pass every source resolves. Checked by testing the path on disk, not by reading the diff."
},
{
"command": "run every lib/tests/run_all.sh",
"result": "pass",
"evidence": "8 of 9 pass. features/lib/tests fails 'REFUSING: must run under shell_coverage_jail.sh' -- confirmed identical in the pre-move baseline worktree, so it is the suite's jail requirement, not a regression. fixes/lib/tests DID regress (log_warn: command not found) and was fixed by correcting REPO_ROOT; it passes again."
},
{
"command": "pytest_changed_packages.py",
"result": "fail then pass",
"evidence": "First run: 14 collection errors from conftest.py's stale sys.path. After the fix: 1798 passed, TOTAL coverage 100.00%."
},
{
"command": "find ~ -maxdepth 4 -xtype l -lname '*testsAndMisc*'; then run the i3blocks scripts",
"result": "fail then pass",
"evidence": "16 broken symlinks immediately after the move, including the entire i3blocks bar. After repointing, zero broken, and cpu_monitor/memory/disk/time/volume all emit real output (e.g. '54°C, 1.18', '17.9GiB/30.4GiB')."
},
{
"command": "git check-ignore on the two secret configs; check_shell_coverage.sh --all",
"result": "pass",
"evidence": "features/.wireguard_ssh.conf and .dufs_cloud.conf are still ignored after their gitignore paths changed. Coverage gate exits 0 and every allowlist entry names a file that exists."
}
],
"risks": [
"This renames ~400 tracked files. Anything outside this repo and outside $HOME that hardcodes linux_configuration/scripts/... would break; report_live_state.sh covers systemd, /etc, /usr/local and $HOME symlinks, and all of those were checked.",
"docs/superpowers/ artifacts still quote old paths. They are dated records of past work and were deliberately not rewritten.",
"features/lib/tests remains jail-only. That predates this change but means one suite is not exercised by a bare run."
],
"rollback": [
"git revert restores the tree; the symlinks then need repointing back, which the same one-liner does with the substitution reversed.",
"Re-run pytest_changed_packages.py and the lib/tests suites to confirm the restored state."
]
}

View File

@ -1,37 +0,0 @@
{
"intent": "Close a gap in the live-state inventory: scripts installed under /usr/local that hardcode repo paths were invisible to the checklist the restructure relies on.",
"scope": [
"meta/scripts/report_live_state.sh - new report_installed section scanning /usr/local/bin and /usr/local/lib",
"docs/restructure-live-state.md - regenerated",
"Non-goal: repointing the references it found; each travels with its own subsystem extraction"
],
"changes": [
"Added report_installed(), which greps /usr/local/bin and /usr/local/lib for the repo path and prints each hit with line numbers, matching the existing sections' shape and feeding the same --check exit code.",
"Regenerated the inventory doc and folded in current depth-cap progress."
],
"verification": [
{
"command": "./meta/scripts/report_live_state.sh",
"result": "pass",
"evidence": "Found 3 installed scripts naming repo paths that no prior section reported: /usr/local/bin/periodic-system-maintenance.sh (3 paths into digital_wellbeing and hosts), /usr/local/bin/hosts-file-monitor.sh (1 into hosts), /usr/local/bin/wsg-grabber (PYTHONPATH into the repo root)."
},
{
"command": "./meta/scripts/report_live_state.sh --check; echo $?",
"result": "pass",
"evidence": "Exit 1, correct while references remain; the new section participates in the same aggregate."
},
{
"command": "shellcheck meta/scripts/report_live_state.sh",
"result": "pass",
"evidence": "Clean, no findings at any severity."
}
],
"risks": [
"The scan covers /usr/local/bin and /usr/local/lib. A reference installed elsewhere on PATH would still be missed; those two are where this machine's installers write.",
"periodic-system-maintenance.sh runs hourly and names install scripts in two subsystems that are about to move, so those extractions must reinstall it rather than only editing the repo copy."
],
"rollback": [
"git revert; the section is additive and nothing depends on its output yet.",
"Re-run report_live_state.sh to confirm the remaining sections still report."
]
}

View File

@ -1,46 +0,0 @@
{
"intent": "Close the file-length cap's blind spot for extensionless scripts, and split the one file that had been hiding in it: boot_recovery/boot-repair at 756 lines.",
"scope": [
"meta/scripts/check_file_length.sh - is_source_file() now also matches extensionless files with a shebang",
"linux_configuration/boot_recovery/boot-repair 756 -> 218 lines",
"New: lib/{discovery,repair_fs,repair_kernel,modes}.sh",
"One real bug fixed: an unquoted expansion that was invisible while the file escaped linting"
],
"changes": [
"is_source_file() previously required a dot in the filename, so every extensionless executable escaped the cap. It now reads the first two bytes and accepts a shebang, which catches boot-repair while still skipping data files like i3/config.",
"Split boot-repair along its own section markers into four sourced libraries: discovery (paths, kernel, ESP, mount state), repair_fs, repair_kernel, and modes.",
"The end-of-run report block stays INLINE deliberately. It is top-level code with exit calls, not function definitions; sourcing it changed control flow and 30 of 42 tests failed. The lib headers record that.",
"Quoted $ESP_MOUNTPOINT in clean_shadow_files. shellcheck could not see the readonly declaration once the function moved to a library, which exposed a genuine SC2086 that the monolith had been hiding."
],
"verification": [
{
"command": "bash linux_configuration/boot_recovery/tests/test_boot_repair.sh",
"result": "pass",
"evidence": "42 passed, 0 failed - the same count as before the split. These run the real script against throwaway fixture root trees, so they exercise the sourcing chain end to end."
},
{
"command": "meta/scripts/verify_shell_split.sh HEAD <old> <new>...",
"result": "pass",
"evidence": "31 of 32 functions byte-identical. The single difference is clean_shadow_files, which is the intended quoting fix and nothing else - the tool names the function, so the change is auditable rather than assumed."
},
{
"command": "shellcheck on the entry script and all four libs",
"result": "pass",
"evidence": "Clean at every severity, with no suppression added."
},
{
"command": "bash meta/scripts/check_file_length.sh --all",
"result": "fail then pass",
"evidence": "Immediately after widening the predicate: '756 linux_configuration/boot_recovery/boot-repair (+506)'. After the split: 'all checked files are within 250 lines'. The other extensionless scripts (.githooks/pre-commit 119, dwm/bin/* 79/41/30) were already under."
}
],
"risks": [
"boot-repair is UEFI/ESP tooling and the vmbox guest is BIOS/GRUB, so a sandbox pass would be actively misleading here - per the vmbox skill's own warning. The fixture suite is the real evidence, and it runs the script against synthetic root trees rather than the live boot path.",
"live_test.sh needs root and was not run; it also required root before this change.",
"The report block remaining inline means boot-repair cannot go much below 218 lines without restructuring its control flow, which is not worth the risk on this file."
],
"rollback": [
"git revert restores the monolith and the narrower predicate together.",
"Re-run tests/test_boot_repair.sh and expect 42 passed."
]
}

View File

@ -0,0 +1,41 @@
{
"intent": "token_audit reports real spend: deduplicated per-message usage, subagent sidechains included, per-model USD, time-windowed snapshots + lever ledger scored by /token-optimization; testsAndMisc gets the shared repo contract (CLAUDE.md Commands, scripts/test_changed.sh, CI gate) and newest-stable pins.",
"scope": [
"python_pkg/token_audit: records, discover, pricing (+prices.json), history, breakdown, recommend, review, turns (new); parse, model, __main__ (edited); tests for every module",
"repo contract: scripts/check_repo_contract.sh, scripts/test_changed.sh, .github/workflows/repo-contract.yml, AGENTS.md Commands, pre-commit hook",
"meta/requirements.txt: pylint 4.1.2, websockets 17.2, pyflakes 4.0.2 (main CI was red on freshness)",
"Rides along from other sessions: ci_mirror async (own evidence JSON 2026-09-11), app-icons evidence, endurain_import duplicate ADB_SERIAL_ENV block removed",
"Non-goal: changing the hooks in ~/.claude that consume these reports"
],
"changes": [
"Usage is counted once per API message id (transcripts repeat it per content block: 1.9x overcount before)",
"Subagent transcripts are discovered and attributed (were 35% of spend and invisible)",
"Each run analyses only records after the previous snapshot, writes history/<ts>.json and scores ledger levers against it; usd_per_commit metric",
"history.py resolves REPORT_DIR at call time so tests redirect it; parse split into parse + turns to stay under 250 lines and complexity 10; Breakdown groups tallies (old attribute names kept as properties)"
],
"verification": [
{
"command": "~/.claude/scripts/capped.sh python3 -m pytest python_pkg/token_audit python_pkg/endurain_import -q --cov-branch",
"result": "pass",
"evidence": "291 passed; every python_pkg/token_audit module at 100% line+branch coverage (1108 stmts, 260 branches, 0 missed)"
},
{
"command": "~/.claude/scripts/capped.sh python3 -m python_pkg.token_audit --no-write",
"result": "pass",
"evidence": "Analysed 20-24 sessions in 0.15s; window starts at the previous snapshot; ~/.claude/token-report file list and mtimes unchanged by the test suite"
},
{
"command": "scripts/check_dependency_freshness.sh --all",
"result": "pass",
"evidence": "after bumping pylint, websockets, pyflakes (flake8 7.4.1 allows pyflakes <4.1.0)"
}
],
"risks": [
"Absolute USD figures before 2026-10-04 double-counted ~1.9x; comparisons across that date are rate-only.",
"prices.json must be updated when Anthropic prices change; unknown models price at 0 and are listed."
],
"rollback": [
"git revert <this commit>: token_audit v1 report format returns; no data migration needed",
"Snapshots under ~/.claude/token-report/history are additive and can stay"
]
}

View File

@ -185,20 +185,36 @@ repos:
# Was on pre-push, but force-push diffs caused full-repo scans + OOM. # Was on pre-push, but force-push diffs caused full-repo scans + OOM.
# =========================================================================== # ===========================================================================
- repo: https://github.com/pylint-dev/pylint - repo: https://github.com/pylint-dev/pylint
rev: v3.3.2 rev: v4.0.8
hooks: hooks:
- id: pylint - id: pylint
stages: [pre-commit] stages: [pre-commit]
args: args:
- --rcfile=pyproject.toml - --rcfile=pyproject.toml
- --fail-under=8.0 - --fail-under=10
- --jobs=0 - --jobs=0
# Every third-party import a linted source file makes, pinned as in
# meta/requirements.txt: pylint scores each unresolved one as
# import-error, which is a config gap dressed up as a code defect.
additional_dependencies: additional_dependencies:
- pytest - pytest
- python-chess - python-chess
- requests - requests
- pygame - pygame
exclude: ^(Bash/|\.venv/) - anthropic==1.5.0
- beautifulsoup4==4.15.0
- defusedxml==0.7.1
- markdownify==1.2.3
- numpy==2.5.3
- pillow==12.3.0
- rich==15.0.0
- typer==0.27.2
# Test suites are ruff's; pylint grades source. The rcfile's
# `ignore = ["tests"]` cannot do this on its own -- it matches only
# the basename of what pylint is handed, and pre-commit hands it
# explicit file paths -- so the exclusion lives here, as in the root
# config.
exclude: ^(Bash/|\.venv/)|(^|/)(tests/|conftest\.py)
# =========================================================================== # ===========================================================================
# BANDIT - Security linter (per-commit on changed files only) # BANDIT - Security linter (per-commit on changed files only)

View File

@ -60,9 +60,9 @@ pre-commit==4.6.2
prospector==1.19.1 prospector==1.19.1
pycodestyle==2.15.0 pycodestyle==2.15.0
pydocstyle==6.3.0 pydocstyle==6.3.0
pyflakes==4.0.1 pyflakes==4.0.2
pygame==2.6.1 pygame==2.6.1
pylint==4.1.1 pylint==4.1.2
pyright==1.1.414 pyright==1.1.414
pytest==9.1.1 pytest==9.1.1
pytest-cov==7.1.0 pytest-cov==7.1.0
@ -89,4 +89,4 @@ types-PyYAML==6.0.12.20260906
types-requests==2.33.0.20260906 types-requests==2.33.0.20260906
types-setuptools==84.0.0.20260812 types-setuptools==84.0.0.20260812
typing_extensions==4.16.0 # Self annotation on py310 target (session_autopsy store lock) typing_extensions==4.16.0 # Self annotation on py310 target (session_autopsy store lock)
websockets==17.1 websockets==17.2

View File

@ -44,9 +44,9 @@ source "$ROOT/meta/scripts/ci_mirror_range.sh"
source "$ROOT/meta/scripts/ci_mirror_shell.sh" source "$ROOT/meta/scripts/ci_mirror_shell.sh"
# shellcheck source=meta/scripts/ci_mirror_mem.sh # shellcheck source=meta/scripts/ci_mirror_mem.sh
source "$ROOT/meta/scripts/ci_mirror_mem.sh" source "$ROOT/meta/scripts/ci_mirror_mem.sh"
source "$ROOT/meta/scripts/ci_mirror_async.sh"
# Records the tree hash of the last fully-green run; see tree_cache_*. # Records the tree hash of the last fully-green run; see tree_cache_*.
readonly GREEN_CACHE="$ROOT/.ci-mirror-venv/.last-green-tree"
LOGDIR="" LOGDIR=""
@ -211,23 +211,14 @@ run_python_gate() {
(cd "$CHECKOUT" && "$VENV_DIR/bin/python" meta/scripts/pytest_changed_packages.py "${args[@]}") (cd "$CHECKOUT" && "$VENV_DIR/bin/python" meta/scripts/pytest_changed_packages.py "${args[@]}")
} }
# Short-circuit a re-push of a tree that already passed. # The gates themselves, shared by the foreground, background and sync paths.
# # The green-tree short-circuit and the verdict bookkeeping they depend on
# Keyed on the whole-repo tree hash of HEAD, so it self-invalidates when # live in ci_mirror_async.sh.
# anything in the tree changes -- including the gate's own scripts. A rebase run_full_gate() {
# that lands the identical tree, or a retried push after a network failure, ensure_venv
# then costs nothing. The record lives inside .ci-mirror-venv/ (already run_gates_in_clean_worktree
# gitignored) so it never shows up in `git status` or a jail fingerprint. record_tree_green
tree_is_known_green() { log "all CI gates passed locally — safe to push"
local head_tree stored
head_tree="$(git -C "$ROOT" rev-parse 'HEAD^{tree}')"
[[ -f "$GREEN_CACHE" ]] || return 1
stored="$(cat "$GREEN_CACHE")"
[[ "$head_tree" == "$stored" ]]
}
record_tree_green() {
git -C "$ROOT" rev-parse 'HEAD^{tree}' >"$GREEN_CACHE" 2>/dev/null || true
} }
main() { main() {
@ -236,10 +227,18 @@ main() {
log "this exact tree already passed the full gate — nothing to redo" log "this exact tree already passed the full gate — nothing to redo"
exit 0 exit 0
fi fi
ensure_venv if [[ "${CI_MIRROR_WORKER:-}" == "1" ]]; then
run_gates_in_clean_worktree run_as_worker
record_tree_green exit 0
log "all CI gates passed locally — safe to push" fi
# A previous background run that failed blocks this push, whichever mode
# the user then chooses.
block_on_previous_failure
if [[ "${CI_MIRROR_SYNC:-}" == "1" ]]; then
run_full_gate
exit 0
fi
spawn_worker
} }
main "$@" main "$@"

104
meta/scripts/ci_mirror_async.sh Executable file
View File

@ -0,0 +1,104 @@
#!/bin/bash
# ============================================================================
# ci_mirror_async.sh — run the CI mirror off the push's critical path.
#
# Sourced by ci_mirror.sh; not executable on its own. Paths are computed in
# functions, not at source time: this file is sourced before ci_mirror.sh
# defines VENV_DIR.
#
# The mirror is real verification: a clean venv, pre-commit over the push
# range (or every file when the gate's own config changed) and pytest. That
# is ~22s for an ordinary push, ~70s when requirements.txt moves, minutes for
# a full sweep. None of it caches down to a second, because it is work rather
# than overhead.
#
# So it stops blocking the push. `git push` spawns a detached, niced worker
# and returns; the worker records a verdict for the tree it checked. The
# guarantee is kept one push later: a recorded failure refuses the next push
# until the tree is verified in the foreground, so a broken tree reaches
# GitHub at most once — where CI sees the same commit within minutes anyway.
#
# Escape hatch: CI_MIRROR_SYNC=1 git push verify now, in the foreground.
# ============================================================================
verdict_dir() { printf '%s/verdicts' "$VENV_DIR"; }
last_verdict_file() { printf '%s/last' "$(verdict_dir)"; }
worker_lock() { printf '%s/.worker.lock' "$VENV_DIR"; }
green_cache() { printf '%s/.last-green-tree' "$VENV_DIR"; }
head_tree() { git -C "$ROOT" rev-parse 'HEAD^{tree}'; }
# A tree that already passed the full gate needs nothing redone.
tree_is_known_green() {
local stored
[[ -f "$(green_cache)" ]] || return 1
stored="$(cat "$(green_cache)")"
[[ "$(head_tree)" == "$stored" ]]
}
record_tree_green() {
head_tree >"$(green_cache)" 2>/dev/null || true
}
# A verdict is "<state> <tree> <logfile>"; only completed runs write one.
record_verdict() {
mkdir -p "$(verdict_dir)"
printf '%s %s %s\n' "$1" "$2" "$3" >"$(last_verdict_file)"
}
# Fail closed one push late.
block_on_previous_failure() {
local file state tree logfile
file="$(last_verdict_file)"
[[ -f "$file" ]] || return 0
read -r state tree logfile <"$file"
[[ "$state" == "fail" ]] || return 0
log "the previous background mirror FAILED (tree ${tree:0:12})"
if [[ -f "$logfile" ]]; then
tail -n 40 "$logfile" >&2
fi
log "refusing to push on top of a failed verification"
log "re-verify in the foreground with: CI_MIRROR_SYNC=1 git push"
exit 1
}
# The worker inherits the push range, so it checks exactly what the push did.
spawn_worker() {
local tree logfile
tree="$(head_tree)"
logfile="$(verdict_dir)/${tree}.log"
mkdir -p "$(verdict_dir)"
# Always spawn: run_as_worker takes the lock and blocks, so workers
# serialise rather than compete, and every pushed tree gets its own
# verdict. Skipping the spawn while another run was in flight would
# leave the newer tree unverified for good.
setsid nice -n 19 ionice -c 3 env \
CI_MIRROR_WORKER=1 \
CI_MIRROR_TREE="$tree" \
CI_MIRROR_LOG="$logfile" \
PRE_COMMIT_FROM_REF="${PRE_COMMIT_FROM_REF:-}" \
PRE_COMMIT_TO_REF="${PRE_COMMIT_TO_REF:-}" \
bash "$ROOT/meta/scripts/ci_mirror.sh" >/dev/null 2>&1 </dev/null &
disown
log "verification running in the background (tree ${tree:0:12})"
log "log: $logfile — a failure blocks the next push"
}
# The detached worker: same gates, but its exit state becomes the verdict
# rather than the push's.
run_as_worker() {
local tree logfile
tree="${CI_MIRROR_TREE:-$(head_tree)}"
logfile="${CI_MIRROR_LOG:-$(verdict_dir)/${tree}.log}"
mkdir -p "$(verdict_dir)"
(
flock 9
if run_full_gate >"$logfile" 2>&1; then
record_verdict ok "$tree" "$logfile"
else
record_verdict fail "$tree" "$logfile"
fi
) 9>"$(worker_lock)"
}

View File

@ -28,11 +28,6 @@ _logger = logging.getLogger(__name__)
PHONE_EXPORT_DIR = "/sdcard/Documents/RunnerUp" PHONE_EXPORT_DIR = "/sdcard/Documents/RunnerUp"
ACTIVITY_SUFFIXES = (".tcx", ".gpx", ".fit") ACTIVITY_SUFFIXES = (".tcx", ".gpx", ".fit")
_ADB_TIMEOUT = 60 _ADB_TIMEOUT = 60
# Pins which phone to pull from when more than one device is attached.
# Without it every adb call fails with "more than one device/emulator" --
# silently, as a warning on a path nobody watches -- the moment a second
# device shows up. Left unset, the single attached device is used.
ADB_SERIAL_ENV = "ENDURAIN_ADB_SERIAL"
# Pins which phone to pull from when more than one device is attached. Without # Pins which phone to pull from when more than one device is attached. Without
# it every adb call fails with "more than one device/emulator" -- silently, as # it every adb call fails with "more than one device/emulator" -- silently, as
# a warning on the fallback path -- the moment a second device (a spare handset, # a warning on the fallback path -- the moment a second device (a spare handset,

View File

@ -15,7 +15,17 @@ import sys
import time import time
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from python_pkg.token_audit import attribute, imagecost, parse, report from python_pkg.token_audit import (
attribute,
breakdown,
discover,
history,
imagecost,
parse,
pricing,
report,
review,
)
if TYPE_CHECKING: if TYPE_CHECKING:
from python_pkg.token_audit.model import Session from python_pkg.token_audit.model import Session
@ -72,6 +82,11 @@ def _build_parser() -> argparse.ArgumentParser:
action="store_true", action="store_true",
help="print the machine-readable snapshot instead of the table", help="print the machine-readable snapshot instead of the table",
) )
parser.add_argument(
"--recommend",
action="store_true",
help="append rule-based candidate levers to the report",
)
parser.add_argument( parser.add_argument(
"--no-write", "--no-write",
action="store_true", action="store_true",
@ -83,13 +98,14 @@ def _build_parser() -> argparse.ArgumentParser:
def _collect(root: Path, since: float, until: float) -> list[Session]: def _collect(root: Path, since: float, until: float) -> list[Session]:
"""Load and annotate every transcript inside the window.""" """Load and annotate every transcript inside the window."""
sessions: list[Session] = [] sessions: list[Session] = []
for path in parse.find_transcripts(root, since, until): window = (since, until)
session = parse.load_session(path) for path in discover.find_transcripts(root, since):
session = parse.load_session(path, window)
if not session.turns: if not session.turns:
continue continue
# Re-walk the file for image cost: it needs the interleaved event order # Re-walk the file for image cost: it needs the interleaved event order
# that load_session flattens into separate lists. # that load_session flattens into separate lists.
imagecost.annotate(session, parse.iter_events(path)) imagecost.annotate(session, parse.iter_events(path, window))
sessions.append(session) sessions.append(session)
return sessions return sessions
@ -99,9 +115,15 @@ def main(argv: list[str] | None = None) -> int:
args = _build_parser().parse_args(argv) args = _build_parser().parse_args(argv)
started = time.monotonic() started = time.monotonic()
until = args.until if args.until is not None else time.time() until = args.until if args.until is not None else time.time()
since = ( # Default start: where the previous analysis ended, so repeated runs tile
args.since if args.since is not None else until - args.days * SECONDS_PER_DAY # time without gaps or overlap. --days applies only on the very first run.
) last_end = history.default_since(args.out)
if args.since is not None:
since = args.since
elif last_end is not None:
since = last_end
else:
since = until - args.days * SECONDS_PER_DAY
sessions = _collect(args.root, since, until) sessions = _collect(args.root, since, until)
if not sessions: if not sessions:
@ -112,11 +134,25 @@ def main(argv: list[str] | None = None) -> int:
drift = attribute.reconcile(totals, sessions) drift = attribute.reconcile(totals, sessions)
window = report.Window(since=since, until=until) window = report.Window(since=since, until=until)
snap = report.snapshot(totals, axes, window, len(sessions)) snap = report.snapshot(totals, axes, window, len(sessions))
previous = report.load_previous(args.out) priced = breakdown.build(sessions, pricing.load())
markdown = report.render(totals, axes, window, sessions, snap, previous) snap["v"] = review.SNAPSHOT_VERSION
snap["effective_window"] = {"since": priced.earliest, "until": priced.latest}
snap["metrics"] = breakdown.metrics(priced) | review.extra_metrics(totals, axes)
previous = history.last_snapshot(args.out)
markdown = report.render(totals, axes, window, sessions, snap, None)
markdown += "\n".join(
[
"",
*breakdown.render(priced),
"",
*review.render(snap, previous, levers=args.recommend),
"",
],
)
if not args.no_write: if not args.no_write:
written = report.write(markdown, snap, args.out) written = report.write(markdown, snap, args.out)
history.save_snapshot(snap, args.out)
sys.stderr.write(f"Wrote {written}\n") sys.stderr.write(f"Wrote {written}\n")
sys.stdout.write(json.dumps(snap, indent=2) if args.json else markdown) sys.stdout.write(json.dumps(snap, indent=2) if args.json else markdown)

View File

@ -0,0 +1,242 @@
"""Dollar-denominated metrics: per model, per subagent, per call position.
:mod:`attribute` answers "where did the weighted tokens go"; this module puts
list-price dollars on the same sessions so that model choice becomes visible,
and computes the flat ``metrics`` dict every snapshot stores. Ledger entries
name a key in that dict, so a key here is a contract -- rename one and old
ledger lines silently become "unmeasured".
"""
from __future__ import annotations
from collections import Counter
from dataclasses import dataclass, field, replace
from pathlib import Path
from typing import TYPE_CHECKING
from python_pkg.token_audit.pricing import usd
if TYPE_CHECKING:
from collections.abc import Sequence
from python_pkg.token_audit.model import Session
from python_pkg.token_audit.pricing import PriceBook
SECONDS_PER_DAY = 86400
# Context size above which a call is "over budget" -- the session gate blocks
# here. Measured 2026-10-04: per-call cost 21k weighted for calls 1-100 versus
# 51k for calls 201-400, driven by context size, not by call count.
CONTEXT_BUDGET = 300_000
# Model a mechanical subagent would move to; used only to size that lever.
CHEAPER_MODEL = "claude-sonnet-5-5"
# Call-position buckets for the cost curve (upper bounds, inclusive).
BUCKETS: tuple[int, ...] = (100, 200, 400, 600, 1000)
@dataclass
class Tally:
"""Dollars and API calls accumulated along one axis."""
usd: float = 0.0
calls: int = 0
def add(self, cost: float) -> None:
"""Count one call costing ``cost`` dollars."""
self.usd += cost
self.calls += 1
@dataclass
class Span:
"""Earliest and latest record timestamp seen (0.0 means none yet)."""
earliest: float = 0.0
latest: float = 0.0
@dataclass
class Breakdown:
"""Everything priced in one pass over the sessions."""
main: Tally = field(default_factory=Tally)
sub: Tally = field(default_factory=Tally)
over_budget: Tally = field(default_factory=Tally)
home_usd: float = 0.0
by_model: Counter[str] = field(default_factory=Counter)
by_agent_type: Counter[str] = field(default_factory=Counter)
by_description: Counter[str] = field(default_factory=Counter)
sub_opus_usd: float = 0.0
# The same Opus subagent turns re-priced at the reference cheaper model:
# what routing them there would have cost, for sizing that lever exactly.
sub_opus_repriced_usd: float = 0.0
buckets: dict[str, Tally] = field(default_factory=dict)
commits: int = 0
span: Span = field(default_factory=Span)
@property
def usd_main(self) -> float:
"""Main-conversation spend."""
return self.main.usd
@property
def usd_sub(self) -> float:
"""Subagent spend."""
return self.sub.usd
@property
def calls_main(self) -> int:
"""Main-conversation API calls."""
return self.main.calls
@property
def calls_sub(self) -> int:
"""Subagent API calls."""
return self.sub.calls
@property
def over_budget_calls(self) -> int:
"""Calls whose context exceeded :data:`CONTEXT_BUDGET`."""
return self.over_budget.calls
@property
def over_budget_usd(self) -> float:
"""Spend carried by calls over the context budget."""
return self.over_budget.usd
@property
def earliest(self) -> float:
"""Earliest record timestamp seen."""
return self.span.earliest
@property
def latest(self) -> float:
"""Latest record timestamp seen."""
return self.span.latest
@property
def usd_total(self) -> float:
"""Main plus subagent spend."""
return self.usd_main + self.usd_sub
@property
def calls(self) -> int:
"""Real API calls, main plus subagent."""
return self.calls_main + self.calls_sub
@property
def days(self) -> float:
"""Span actually covered by records, at least a minute's worth."""
return max(self.latest - self.earliest, 60.0) / SECONDS_PER_DAY
def _bucket(index: int) -> str:
"""Label the call-position bucket a 1-based call index falls in."""
low = 1
for high in BUCKETS:
if index <= high:
return f"{low}-{high}"
low = high + 1
return f"{low}+"
def _stamp(found: Breakdown, stamp: float) -> None:
"""Widen the observed record span to include ``stamp``."""
if not stamp:
return
span = found.span
span.earliest = min(span.earliest or stamp, stamp)
span.latest = max(span.latest, stamp)
def build(sessions: Sequence[Session], book: PriceBook) -> Breakdown:
"""Price every turn and roll it up along each axis."""
found = Breakdown()
home = str(Path.home())
for session in sessions:
is_sub = session.parent_id is not None
found.commits += sum(1 for call in session.tools if call.commit)
for index, turn in enumerate(session.turns, start=1):
cost = usd(turn, book)
_stamp(found, turn.timestamp)
found.by_model[turn.model] += cost
if turn.context > CONTEXT_BUDGET:
found.over_budget.add(cost)
if session.cwd == home:
found.home_usd += cost
if is_sub:
found.sub.add(cost)
found.by_agent_type[session.agent_type or "unknown"] += cost
found.by_description[session.description or "?"] += cost
if "opus" in turn.model:
found.sub_opus_usd += cost
cheap = replace(turn, model=CHEAPER_MODEL)
found.sub_opus_repriced_usd += usd(cheap, book)
else:
found.main.add(cost)
found.buckets.setdefault(_bucket(index), Tally()).add(cost)
return found
def _share(part: float, whole: float) -> float:
"""``part / whole`` rounded for storage, 0 when ``whole`` is 0."""
return round(part / whole, 4) if whole else 0.0
def metrics(found: Breakdown) -> dict[str, float]:
"""The flat metric dict snapshots store and ledger entries reference."""
total = found.usd_total
opus_main = sum(c for m, c in found.by_model.items() if "opus" in m)
return {
"usd_total": round(total, 2),
"usd_per_day": round(total / found.days, 2),
"usd_per_call": round(total / found.calls, 5) if found.calls else 0.0,
"calls": found.calls,
"calls_per_day": round(found.calls / found.days, 1),
"commits": found.commits,
"usd_per_commit": round(total / found.commits, 2) if found.commits else 0.0,
"subagent_usd_share": _share(found.usd_sub, total),
"subagent_opus_usd_share": _share(found.sub_opus_usd, found.usd_sub),
"opus_usd_share": _share(opus_main, total),
"home_cwd_usd_share": _share(found.home_usd, total),
"over_budget_call_share": _share(found.over_budget_calls, found.calls),
"over_budget_usd_share": _share(found.over_budget_usd, total),
"subagent_sonnet_saving_usd": round(
found.sub_opus_usd - found.sub_opus_repriced_usd, 2
),
}
def _money(rows: list[tuple[str, float]], total: float) -> list[str]:
"""Render ``(label, usd)`` rows as a markdown table with shares."""
out = ["| item | USD | share |", "|---|---|---|"]
out += [f"| {k} | ${v:,.2f} | {_share(v, total) * 100:.1f}% |" for k, v in rows]
return out
def render(found: Breakdown, top: int = 12) -> list[str]:
"""Markdown sections for the dollar view."""
total = found.usd_total
lines = [
"## Dollar view (list prices, deduplicated per API message)",
"",
f"**${total:,.2f}** over {found.days:.1f} days of records "
f"({found.calls:,} API calls; main ${found.usd_main:,.2f}, "
f"subagents ${found.usd_sub:,.2f}).",
f"Calls with context over {CONTEXT_BUDGET:,} tokens: "
f"{found.over_budget_calls:,} carrying "
f"{_share(found.over_budget_usd, total) * 100:.1f}% of spend.",
"",
"### By model",
"",
]
lines += _money(found.by_model.most_common(top), total)
lines += ["", "### Subagents by type", ""]
lines += _money(found.by_agent_type.most_common(top), total)
lines += ["", "### Costliest subagent tasks", ""]
lines += _money(found.by_description.most_common(top), total)
lines += ["", "### Cost per call by position in session (main only)", ""]
lines += ["| calls | USD per call | calls |", "|---|---|---|"]
for label in sorted(found.buckets, key=lambda b: int(b.split("-")[0].rstrip("+"))):
tally = found.buckets[label]
lines.append(f"| {label} | ${tally.usd / tally.calls:.4f} | {tally.calls:,} |")
return lines

View File

@ -0,0 +1,67 @@
"""Find transcript files and read the per-file metadata around them.
Two layouts hold billed turns:
* ``<root>/<project-slug>/<session-id>.jsonl`` -- the main conversation.
* ``<root>/<project-slug>/<session-id>/subagents/agent-<id>.jsonl`` -- one per
``Agent`` call, with a sibling ``agent-<id>.meta.json`` naming the agentType
and description. Before 2026-10-04 the audit globbed only the first layout
and reported subagents at 0.0% while they were 35% of real spend.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from pathlib import Path
PATTERNS: tuple[str, ...] = ("*/*.jsonl", "*/*/subagents/agent-*.jsonl")
def find_transcripts(root: Path, since: float) -> list[Path]:
"""Return transcripts that may hold records after ``since``, newest first.
A file last written before ``since`` cannot hold an in-window record, so
mtime prunes those. There is deliberately no mtime upper bound: a file
still being appended to after the window closed holds in-window records
too, and the per-record timestamp filter in :func:`parse.iter_events`
trims whatever falls outside.
"""
found: list[tuple[float, Path]] = []
for pattern in PATTERNS:
for path in sorted(root.glob(pattern)):
mtime = path.stat().st_mtime
if mtime >= since:
found.append((mtime, path))
return [path for _, path in sorted(found, reverse=True)]
def _record(line: str) -> dict[str, Any] | None:
"""Parse one JSON object, or ``None``."""
try:
loaded = json.loads(line)
except ValueError:
return None
return loaded if isinstance(loaded, dict) else None
def first_cwd(path: Path) -> str | None:
"""Return the working directory a transcript was recorded in."""
with path.open(encoding="utf-8", errors="replace") as handle:
for line in handle:
record = _record(line)
if record is not None:
cwd = record.get("cwd")
if isinstance(cwd, str):
return cwd
return None
def agent_meta(path: Path) -> dict[str, Any]:
"""Return a subagent transcript's sibling ``.meta.json``, or ``{}``."""
meta = path.with_suffix(".meta.json")
if not meta.exists():
return {}
return _record(meta.read_text(encoding="utf-8", errors="replace")) or {}

View File

@ -0,0 +1,175 @@
"""Dated snapshots and the ledger of applied changes.
Until 2026-10-04 every run overwrote one ``weekly.json``, so "did last month's
change help?" had no data to answer it -- and Claude Code deletes transcripts
after ~30 days, so the raw data cannot be re-read later either. Two durable
files fix that:
* ``history/YYYY-MM-DD[-HHMM].json`` -- one snapshot per run, never rewritten.
The newest one's window end is the next run's default start, which is how
``/token-optimization`` "figures out when the last analysis was".
* ``ledger.jsonl`` -- one line per applied lever: which snapshot metric it was
meant to move, the baseline value, the predicted value, and how to undo it.
:func:`score` turns each line into helped / no effect / regressed against the
latest snapshot, in code, so the verdict is never an LLM's impression.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, datetime
import json
from pathlib import Path
from typing import Any
from python_pkg.token_audit.probe import PREDICTION_TOLERANCE
REPORT_DIR = Path.home() / ".claude" / "token-report"
HISTORY_NAME = "history"
LEDGER_NAME = "ledger.jsonl"
# A metric must move by more than this fraction of its baseline to count as a
# change at all; smaller moves are within week-to-week noise.
NOISE = 0.03
def _root(directory: Path | None) -> Path:
"""Resolve the report directory at call time so it can be redirected."""
return REPORT_DIR if directory is None else directory
def save_snapshot(snap: dict[str, Any], directory: Path | None = None) -> Path:
"""Write ``snap`` under ``history/`` without ever overwriting a file."""
folder = _root(directory) / HISTORY_NAME
folder.mkdir(parents=True, exist_ok=True)
now = datetime.now(tz=UTC)
path = folder / f"{now:%Y-%m-%d}.json"
if path.exists():
path = folder / f"{now:%Y-%m-%d-%H%M%S}.json"
path.write_text(json.dumps(snap, indent=2), encoding="utf-8")
return path
def last_snapshot(directory: Path | None = None) -> dict[str, Any] | None:
"""Return the newest readable snapshot, or ``None`` if there is none."""
folder = _root(directory) / HISTORY_NAME
if not folder.is_dir():
return None
for path in sorted(folder.glob("*.json"), reverse=True):
try:
loaded = json.loads(path.read_text(encoding="utf-8"))
except ValueError:
continue
if isinstance(loaded, dict):
return loaded
return None
def default_since(directory: Path | None = None) -> float | None:
"""The previous analysis's window end, i.e. where this one should start."""
snap = last_snapshot(directory)
if snap is None:
return None
until = snap.get("window", {}).get("until")
return float(until) if isinstance(until, (int, float)) else None
def read_ledger(directory: Path | None = None) -> list[dict[str, Any]]:
"""Return every well-formed ledger entry, oldest first."""
path = _root(directory) / LEDGER_NAME
if not path.exists():
return []
entries: list[dict[str, Any]] = []
for line in path.read_text(encoding="utf-8").splitlines():
try:
entry = json.loads(line)
except ValueError:
continue
if isinstance(entry, dict) and "metric" in entry and "baseline" in entry:
entries.append(entry)
return entries
def append_ledger(entry: dict[str, Any], directory: Path | None = None) -> None:
"""Append one applied lever; ``date`` is filled in when missing."""
directory = _root(directory)
directory.mkdir(parents=True, exist_ok=True)
record = {"date": datetime.now(tz=UTC).strftime("%Y-%m-%d"), **entry}
with (directory / LEDGER_NAME).open("a", encoding="utf-8") as handle:
handle.write(json.dumps(record) + "\n")
@dataclass(frozen=True)
class Verdict:
"""How one applied lever turned out against the latest snapshot."""
lever: str
metric: str
baseline: float
predicted: float | None
actual: float | None
outcome: str
note: str = ""
def _outcome(baseline: float, actual: float, *, lower_is_better: bool) -> str:
"""Classify a metric move as helped / no effect / regressed."""
if not baseline:
return "no baseline"
change = (actual - baseline) / abs(baseline)
if abs(change) <= NOISE:
return "no effect"
improved = change < 0 if lower_is_better else change > 0
return "helped" if improved else "regressed"
def _prediction_note(baseline: float, predicted: float | None, actual: float) -> str:
"""Flag a prediction that missed the measured move by more than tolerance."""
if predicted is None:
return ""
want = baseline - predicted
got = baseline - actual
if not want:
return ""
drift = abs(got - want) / abs(want)
if drift <= PREDICTION_TOLERANCE:
return "prediction held"
return f"prediction off by {drift:.0%}"
def score(entries: list[dict[str, Any]], snap: dict[str, Any]) -> list[Verdict]:
"""Score each ledger entry against the snapshot's current metric values."""
metrics = snap.get("metrics", {})
verdicts: list[Verdict] = []
for entry in entries:
metric = str(entry["metric"])
baseline = float(entry["baseline"])
raw_predicted = entry.get("predicted")
predicted = float(raw_predicted) if raw_predicted is not None else None
value = metrics.get(metric)
if not isinstance(value, (int, float)):
verdicts.append(
Verdict(
str(entry.get("lever", "?")),
metric,
baseline,
predicted,
None,
"unmeasured",
"metric absent from snapshot",
),
)
continue
actual = float(value)
lower = bool(entry.get("lower_is_better", True))
verdicts.append(
Verdict(
lever=str(entry.get("lever", "?")),
metric=metric,
baseline=baseline,
predicted=predicted,
actual=actual,
outcome=_outcome(baseline, actual, lower_is_better=lower),
note=_prediction_note(baseline, predicted, actual),
),
)
return verdicts

View File

@ -73,6 +73,7 @@ class Turn:
is_sidechain: bool is_sidechain: bool
message_id: str = "" message_id: str = ""
tool_calls: int = 0 tool_calls: int = 0
timestamp: float = 0.0
@property @property
def cost(self) -> float: def cost(self) -> float:
@ -94,6 +95,9 @@ class ToolCall:
result_tokens: int result_tokens: int
path: str | None = None path: str | None = None
skill: str | None = None skill: str | None = None
# A Bash call that ran ``git commit`` or ``finish_auto.sh``: the closest
# transcript-visible proxy for "an issue got solved", used for USD/commit.
commit: bool = False
@property @property
def is_image(self) -> bool: def is_image(self) -> bool:
@ -120,6 +124,11 @@ class Session:
cwd: str | None = None cwd: str | None = None
turns: list[Turn] = field(default_factory=list) turns: list[Turn] = field(default_factory=list)
tools: list[ToolCall] = field(default_factory=list) tools: list[ToolCall] = field(default_factory=list)
# Set only for subagent transcripts: the spawning session and the
# agentType / description from the sibling ``.meta.json``.
parent_id: str | None = None
agent_type: str | None = None
description: str | None = None
# Cost-weighted tokens spent re-sending images that were still live in # Cost-weighted tokens spent re-sending images that were still live in
# context. Computed by :mod:`imagecost`, which needs the interleaved order # context. Computed by :mod:`imagecost`, which needs the interleaved order
# of images and turns and so cannot be derived from these lists alone. # of images and turns and so cannot be derived from these lists alone.

View File

@ -20,8 +20,10 @@ the design here:
from __future__ import annotations from __future__ import annotations
import json import json
import re
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
from python_pkg.token_audit import discover, records, turns
from python_pkg.token_audit.model import IMAGE_SUFFIXES, Session, ToolCall, Turn from python_pkg.token_audit.model import IMAGE_SUFFIXES, Session, ToolCall, Turn
if TYPE_CHECKING: if TYPE_CHECKING:
@ -40,6 +42,9 @@ CHARS_PER_TOKEN = 4
# Halving a screenshot's dimensions cuts this roughly fourfold. # Halving a screenshot's dimensions cuts this roughly fourfold.
IMAGE_TOKENS = 2566 IMAGE_TOKENS = 2566
# Commands that land work: a plain commit or the scripted wrap-up.
COMMIT = re.compile(r"\bgit(\s+-C\s+\S+)?\s+commit\b|finish_auto\.sh")
def _load_line(line: str) -> dict[str, Any] | None: def _load_line(line: str) -> dict[str, Any] | None:
"""Parse one JSONL line, returning ``None`` if it is not a usable record.""" """Parse one JSONL line, returning ``None`` if it is not a usable record."""
@ -61,31 +66,6 @@ def _blocks(record: dict[str, Any]) -> list[dict[str, Any]]:
return [block for block in content if isinstance(block, dict)] return [block for block in content if isinstance(block, dict)]
def _usage(record: dict[str, Any]) -> tuple[dict[str, Any], dict[str, int]] | None:
"""Return an assistant record's message and its integer usage fields.
The message is returned alongside the usage block so callers that need both
(the model id lives on the message) never have to re-validate its type.
"""
message = record.get("message")
if not isinstance(message, dict):
return None
usage = message.get("usage")
if not isinstance(usage, dict):
return None
return message, {k: v for k, v in usage.items() if isinstance(v, int)}
def _model(message: dict[str, Any]) -> str:
"""Return the model id that served a record, or a placeholder.
Only ever called after :func:`_usage` has confirmed the message is a dict,
so it takes the message itself rather than re-checking the record.
"""
model = message.get("model")
return model if isinstance(model, str) else "unknown"
def _result_tokens(block: dict[str, Any], *, is_image: bool = False) -> int: def _result_tokens(block: dict[str, Any], *, is_image: bool = False) -> int:
"""Estimate how many tokens a tool result occupied in context. """Estimate how many tokens a tool result occupied in context.
@ -106,32 +86,64 @@ def _result_tokens(block: dict[str, Any], *, is_image: bool = False) -> int:
return len(json.dumps(content)) // CHARS_PER_TOKEN return len(json.dumps(content)) // CHARS_PER_TOKEN
def iter_events(path: Path) -> Iterator[tuple[str, object]]: def iter_events(
path: Path,
window: records.Window | None = None,
) -> Iterator[tuple[str, object]]:
"""Yield ``("turn", Turn)`` and ``("tool", ToolCall)`` in transcript order. """Yield ``("turn", Turn)`` and ``("tool", ToolCall)`` in transcript order.
Emitting a single ordered stream — rather than two lists — is what lets Emitting a single ordered stream — rather than two lists — is what lets
:mod:`imagecost` count how many turns each image survived for. :mod:`imagecost` count how many turns each image survived for. Turns are
one per API message (see :mod:`records`), and only records inside
``window`` are emitted.
""" """
pending: dict[str, dict[str, Any]] = {} pending: dict[str, dict[str, Any]] = {}
merger = records.TurnMerger()
with path.open(encoding="utf-8", errors="replace") as handle: with path.open(encoding="utf-8", errors="replace") as handle:
for line in handle: for line in handle:
record = _load_line(line) record = _load_line(line)
if record is None: if record is None or not records.in_window(record, window):
continue continue
pairs: list[tuple[str, object]] = []
for block in _blocks(record): for block in _blocks(record):
kind = block.get("type") pairs += _block_events(block, pending, merger)
if kind == "tool_use": yield from pairs
call_id = block.get("id") found = turns.usage_of(record)
if isinstance(call_id, str):
pending[call_id] = block
elif kind == "tool_result":
call = _pair_result(block, pending)
if call is not None:
yield "tool", call
found = _usage(record)
if found is not None: if found is not None:
message, usage = found done = merger.push(turns.build_turn(record, *found))
yield "turn", _build_turn(record, message, usage) if done is not None:
yield "turn", done
held = merger.flush()
if held is not None:
yield "turn", held
def _block_events(
block: dict[str, Any],
pending: dict[str, dict[str, Any]],
merger: records.TurnMerger,
) -> list[tuple[str, object]]:
"""Events one content block emits: remember a ``tool_use``, pair a result.
A ``tool_result`` ends the assistant message before it, so the held turn is
released first to keep the stream in transcript order.
"""
kind = block.get("type")
if kind == "tool_use":
call_id = block.get("id")
if isinstance(call_id, str):
pending[call_id] = block
return []
if kind != "tool_result":
return []
out: list[tuple[str, object]] = []
held = merger.flush()
if held is not None:
out.append(("turn", held))
call = _pair_result(block, pending)
if call is not None:
out.append(("tool", call))
return out
def _pair_result( def _pair_result(
@ -146,12 +158,14 @@ def _pair_result(
tool_input = tool_input if isinstance(tool_input, dict) else {} tool_input = tool_input if isinstance(tool_input, dict) else {}
path = tool_input.get("file_path") path = tool_input.get("file_path")
skill = tool_input.get("skill") skill = tool_input.get("skill")
command = tool_input.get("command")
call_path = path if isinstance(path, str) else None call_path = path if isinstance(path, str) else None
return ToolCall( return ToolCall(
name=str(use.get("name") or "unknown"), name=str(use.get("name") or "unknown"),
result_tokens=_result_tokens(block, is_image=_looks_like_image(call_path)), result_tokens=_result_tokens(block, is_image=_looks_like_image(call_path)),
path=call_path, path=call_path,
skill=skill if isinstance(skill, str) else None, skill=skill if isinstance(skill, str) else None,
commit=isinstance(command, str) and COMMIT.search(command) is not None,
) )
@ -163,74 +177,24 @@ def _looks_like_image(path: str | None) -> bool:
return dot != -1 and path[dot:].lower() in IMAGE_SUFFIXES return dot != -1 and path[dot:].lower() in IMAGE_SUFFIXES
def _build_turn( def load_session(path: Path, window: records.Window | None = None) -> Session:
record: dict[str, Any], """Read one transcript file into a :class:`Session`.
message: dict[str, Any],
usage: dict[str, int], A subagent transcript (``<session>/subagents/agent-*.jsonl``) is tagged
) -> Turn: with its parent session id and the ``agentType``/``description`` the
"""Build a :class:`Turn` from an assistant record's usage block.""" spawning ``Agent`` call recorded in the sibling ``.meta.json``.
context = usage.get("cache_read_input_tokens", 0) + usage.get( """
"cache_creation_input_tokens", session = Session(
0, session_id=path.stem, path=str(path), cwd=discover.first_cwd(path)
) )
message_id = message.get("id") if path.parent.name == "subagents":
content = message.get("content") meta = discover.agent_meta(path)
tool_calls = ( session.parent_id = path.parent.parent.name
sum( session.agent_type = str(meta.get("agentType") or "unknown")
1 session.description = str(meta.get("description") or "")
for block in content for kind, event in iter_events(path, window):
if isinstance(block, dict) and block.get("type") == "tool_use"
)
if isinstance(content, list)
else 0
)
return Turn(
usage=usage,
context=context,
model=_model(message),
is_sidechain=bool(record.get("isSidechain")),
message_id=message_id if isinstance(message_id, str) else "",
tool_calls=tool_calls,
)
def _first_cwd(path: Path) -> str | None:
"""Return the working directory a transcript was recorded in."""
with path.open(encoding="utf-8", errors="replace") as handle:
for line in handle:
record = _load_line(line)
if record is not None:
cwd = record.get("cwd")
if isinstance(cwd, str):
return cwd
return None
def load_session(path: Path) -> Session:
"""Read one transcript file into a :class:`Session`."""
session = Session(session_id=path.stem, path=str(path), cwd=_first_cwd(path))
for kind, event in iter_events(path):
if kind == "turn" and isinstance(event, Turn): if kind == "turn" and isinstance(event, Turn):
session.turns.append(event) session.turns.append(event)
elif isinstance(event, ToolCall): elif isinstance(event, ToolCall):
session.tools.append(event) session.tools.append(event)
return session return session
def find_transcripts(
root: Path,
since: float,
until: float | None = None,
) -> list[Path]:
"""Return transcripts modified inside the window, newest first.
Modification time is the selector because it is the only per-file timestamp
available without opening the file, and a session's mtime is when it was
last active — exactly the notion of "used this week" the report wants.
"""
found: list[tuple[float, Path]] = []
for path in sorted(root.glob("*/*.jsonl")):
mtime = path.stat().st_mtime
if mtime >= since and (until is None or mtime <= until):
found.append((mtime, path))
return [path for _, path in sorted(found, reverse=True)]

View File

@ -0,0 +1,12 @@
{
"_source": "claude-api skill model table (cached 2026-09-25), read 2026-10-04. USD per million tokens. Cache writes are priced as multiples of input: 5m = 1.25x, 1h = 2x (verified 2026-10-04: a claude -p probe's total_cost_usd matched opus-5-5 at 2x for its 1h write to 7 digits). Entries marked assumed_cache_read use the standard 0.1x-of-input ratio because the table does not list them.",
"default": "claude-opus-5-5",
"models": {
"claude-fable-5-1": {"input": 10.0, "output": 50.0, "cache_read": 0.25},
"claude-opus-5-5": {"input": 4.0, "output": 20.0, "cache_read": 0.20},
"claude-opus-5": {"input": 5.0, "output": 25.0, "cache_read": 0.50, "assumed_cache_read": true},
"claude-sonnet-5-5": {"input": 2.0, "output": 10.0, "cache_read": 0.20},
"claude-sonnet-5": {"input": 2.0, "output": 10.0, "cache_read": 0.20, "assumed_cache_read": true},
"claude-haiku-4-5": {"input": 1.0, "output": 5.0, "cache_read": 0.10, "assumed_cache_read": true}
}
}

View File

@ -0,0 +1,96 @@
"""Price turns in US dollars, per model.
The token-class weights in :mod:`model` are model-blind: an Opus call and a
Haiku call with identical usage weigh the same, so the weighted total cannot
say what moving work to a cheaper model would save. Dollars can. They are API
list prices -- on a subscription they are not what is billed, but they are the
best available proxy for how fast a usage limit drains, and they make Opus,
Sonnet and Haiku spend comparable.
One measured surprise this exposed: Sonnet 5.5 cache reads cost the same
$0.20/MTok as Opus 5.5. Cache reads are most of the bill, so moving a long
agent to Sonnet saves far less than the 2x the headline prices suggest.
"""
from __future__ import annotations
from dataclasses import dataclass
import json
from pathlib import Path
from typing import TYPE_CHECKING
from python_pkg.token_audit.records import CACHE_1H, CACHE_5M
if TYPE_CHECKING:
from python_pkg.token_audit.model import Turn
PRICES_PATH = Path(__file__).with_name("prices.json")
PER_MILLION = 1_000_000
WRITE_5M = 1.25
WRITE_1H = 2.0
@dataclass(frozen=True)
class Price:
"""List prices for one model, USD per million tokens."""
input: float
output: float
cache_read: float
@dataclass(frozen=True)
class PriceBook:
"""Every known model's price plus the fallback for unknown ids."""
models: dict[str, Price]
default: str
def for_model(self, model: str) -> Price:
"""Exact id, else the longest known id it starts with, else default.
The prefix match covers dated or suffixed ids (``claude-haiku-4-5-2025...``).
"""
if model in self.models:
return self.models[model]
matches = [known for known in self.models if model.startswith(known)]
if matches:
return self.models[max(matches, key=len)]
return self.models[self.default]
def load(path: Path = PRICES_PATH) -> PriceBook:
"""Read the price table shipped next to this module."""
data = json.loads(path.read_text(encoding="utf-8"))
models = {
name: Price(
input=float(row["input"]),
output=float(row["output"]),
cache_read=float(row["cache_read"]),
)
for name, row in data["models"].items()
}
return PriceBook(models=models, default=str(data["default"]))
def usd(turn: Turn, book: PriceBook) -> float:
"""Return one turn's list-price cost in dollars.
Synthetic turns (``<synthetic>``) carry zero usage and so cost zero under
any price, which is why no special case is needed for them. When a turn
has no 1h/5m split, the whole cache write is priced as 5m (the cheaper,
older default) -- an underestimate, never an overestimate.
"""
usage = turn.usage
price = book.for_model(turn.model)
creation = usage.get("cache_creation_input_tokens", 0)
one_hour = usage.get(CACHE_1H, 0)
five_min = usage.get(CACHE_5M, creation - one_hour)
total = (
usage.get("input_tokens", 0) * price.input
+ five_min * WRITE_5M * price.input
+ one_hour * WRITE_1H * price.input
+ usage.get("cache_read_input_tokens", 0) * price.cache_read
+ usage.get("output_tokens", 0) * price.output
)
return total / PER_MILLION

View File

@ -0,0 +1,191 @@
"""Deterministic lever candidates: thresholds in code, not in a model's head.
Every rule reads the snapshot ``metrics`` (and the previous snapshot's, for
growth checks) and either emits a :class:`Lever` or stays silent. The skill
layered on top adds judgment -- which levers suit the user, how to phrase a
prompt habit -- but never decides *whether* a number crossed a line; that is
an ``if`` statement and lives here.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
# Rule thresholds. Each is the point past which the lever is worth a look,
# chosen from the 2026-10-04 audit's distributions, not tuned to fire.
SUBAGENT_SHARE = 0.15
SUBAGENT_OPUS_SHARE = 0.5
OVER_BUDGET_SHARE = 0.2
HOME_CWD_SHARE = 0.2
BATCHING_MIN = 1.15
BASH_TOKENS_PER_CALL = 250
PREFIX_GROWTH = 0.05
IMAGE_SHARE = 0.03
@dataclass(frozen=True)
class Lever:
"""One candidate change, sized where the data allows."""
category: str
title: str
evidence: str
action: str
est_usd_per_day: float | None = None
def _num(metrics: dict[str, Any], key: str) -> float:
"""Read a numeric metric, treating anything missing as zero."""
value = metrics.get(key, 0)
return float(value) if isinstance(value, (int, float)) else 0.0
def _subagents(m: dict[str, Any]) -> list[Lever]:
"""Opus doing mechanical subagent work."""
if _num(m, "subagent_usd_share") < SUBAGENT_SHARE:
return []
if _num(m, "subagent_opus_usd_share") < SUBAGENT_OPUS_SHARE:
return []
saving = _num(m, "subagent_sonnet_saving_usd")
days = _num(m, "usd_total") / max(_num(m, "usd_per_day"), 1e-9)
return [
Lever(
"cheaper agents",
"Opus runs most subagent work",
f"subagents {_num(m, 'subagent_usd_share'):.0%} of spend, "
f"{_num(m, 'subagent_opus_usd_share'):.0%} of that on Opus; "
f"re-priced at Sonnet the same turns cost ${saving:,.2f} less",
"route coverage/test/search agents to the Sonnet-pinned agent "
"definitions; keep implementers on the default",
saving / days if days else None,
),
]
def _shape(m: dict[str, Any]) -> list[Lever]:
"""Session shape: context budget and working directory."""
out: list[Lever] = []
per_day = _num(m, "usd_per_day")
if _num(m, "over_budget_usd_share") > OVER_BUDGET_SHARE:
out.append(
Lever(
"session length",
"Calls over the context budget",
f"{_num(m, 'over_budget_usd_share'):.0%} of spend ran with "
"context above the gate's budget",
"check the context gate fires (hook log) and that handoffs are "
"being written; lower the budget if they are",
# Calls over budget cost ~2.4x a fresh call (21k vs 51k).
per_day * _num(m, "over_budget_usd_share") * 0.58,
),
)
if _num(m, "home_cwd_usd_share") > HOME_CWD_SHARE:
out.append(
Lever(
"prompt habits",
"Sessions started in ~",
f"{_num(m, 'home_cwd_usd_share'):.0%} of spend had cwd=~ "
"(no repo CLAUDE.md, so agents explore to find commands)",
"start claude inside the repo; add exact run/test commands to "
"each repo's CLAUDE.md",
),
)
return out
def _mechanics(m: dict[str, Any], prev: dict[str, Any] | None) -> list[Lever]:
"""Batching, output size, prefix growth, images."""
out: list[Lever] = []
ratio = _num(m, "batching_calls_per_msg")
if 0 < ratio < BATCHING_MIN:
out.append(
Lever(
"agent mechanics",
"Independent tool calls not batched",
f"{ratio:.2f} calls per tool message",
"batch-nudge hook is not moving this; prefer scripts that run "
"a whole read-only chain and report once",
),
)
bash = _num(m, "bash_tokens_per_call")
if bash > BASH_TOKENS_PER_CALL:
out.append(
Lever(
"agent mechanics",
"Large Bash results",
f"{bash:.0f} result tokens per Bash call",
"compress command output (RTK if its A/B won) and keep "
"| head/| tail discipline",
),
)
if prev is not None:
before = _num(prev, "standing_tokens_est")
now = _num(m, "standing_tokens_est")
if before and (now - before) / before > PREFIX_GROWTH:
out.append(
Lever(
"standing prefix",
"Always-loaded context grew",
f"{before:,.0f} -> {now:,.0f} est tokens per call",
"compact the surfaces that grew (see checkup); archive "
"finished memory entries",
),
)
if _num(m, "image_usd_share") > IMAGE_SHARE:
out.append(
Lever(
"agent mechanics",
"Images held in context",
f"{_num(m, 'image_usd_share'):.0%} of spend",
"extend the shrink hook to the tools returning screenshots",
),
)
return out
def _dead(dead_mcp: list[str], dead_skills: list[str]) -> list[Lever]:
"""Registered surfaces nobody used in either window."""
if not dead_mcp and not dead_skills:
return []
return [
Lever(
"standing prefix",
"Unused MCP servers / skills",
f"MCP: {', '.join(dead_mcp) or '-'}; "
f"skills: {', '.join(dead_skills) or '-'}",
"park them (mcp_park.sh / move skill dir to backups/skills-parked)",
),
]
def candidates(
metrics: dict[str, Any],
previous: dict[str, Any] | None,
dead_mcp: list[str],
dead_skills: list[str],
) -> list[Lever]:
"""Every lever whose rule fired, sized ones first by daily saving."""
levers = (
_subagents(metrics)
+ _shape(metrics)
+ _mechanics(metrics, previous)
+ _dead(dead_mcp, dead_skills)
)
return sorted(levers, key=lambda lv: -(lv.est_usd_per_day or 0.0))
def render(levers: list[Lever]) -> list[str]:
"""Markdown table of the candidates."""
if not levers:
return ["_No rule fired._"]
out = [
"| category | lever | evidence | est $/day | action |",
"|---|---|---|---|---|",
]
for lv in levers:
est = f"${lv.est_usd_per_day:,.2f}" if lv.est_usd_per_day else "-"
out.append(
f"| {lv.category} | {lv.title} | {lv.evidence} | {est} | {lv.action} |"
)
return out

View File

@ -0,0 +1,116 @@
"""Record-level helpers: timestamps, the time window, and per-message dedupe.
Claude Code writes ONE API message as SEVERAL transcript records -- one per
content block (thinking, text, each ``tool_use``) -- and every one of them
carries the same, complete ``usage`` block. Summing usage per record therefore
counts each API call once per block. Measured 2026-10-04 over 32 days: 74,937
records for 40,388 real calls, a 1.9x overstatement that every report before
that date carried. :class:`TurnMerger` collapses the records back into one
:class:`Turn` per ``message.id``.
The window is applied per record, by its own ``timestamp``, because a file's
mtime only says when the session was last active: a session started three weeks
ago and touched today would otherwise count three weeks of spend as "this week".
"""
from __future__ import annotations
from dataclasses import replace
from datetime import datetime
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from python_pkg.token_audit.model import Turn
# Window bounds as unix seconds, inclusive at both ends.
Window = tuple[float, float]
# Names under which the 1h / 5m cache-write split is stored on a Turn's usage.
# They are deliberately NOT in model.TOKEN_KINDS: the reconciled weighted total
# keeps its historical definition, while pricing reads the split when present.
CACHE_1H = "cache_creation_1h"
CACHE_5M = "cache_creation_5m"
def timestamp(record: dict[str, Any]) -> float | None:
"""Return a record's ISO timestamp as unix seconds, or ``None``."""
raw = record.get("timestamp")
if not isinstance(raw, str):
return None
try:
return datetime.fromisoformat(raw).timestamp()
except ValueError:
return None
def in_window(record: dict[str, Any], window: Window | None) -> bool:
"""Whether a record falls inside the window; undated records are kept."""
if window is None:
return True
stamp = timestamp(record)
return stamp is None or window[0] <= stamp <= window[1]
def cache_split(usage: dict[str, Any]) -> dict[str, int]:
"""Extract the 1h / 5m cache-write split from a raw usage block.
A 1h cache write bills at 2x input and a 5m one at 1.25x, so the split is
worth up to 60% of the cache-write cost. Claude Code currently writes 1h.
"""
nested = usage.get("cache_creation")
if not isinstance(nested, dict):
return {}
out: dict[str, int] = {}
for key, name in (
("ephemeral_1h_input_tokens", CACHE_1H),
("ephemeral_5m_input_tokens", CACHE_5M),
):
value = nested.get(key)
if isinstance(value, int):
out[name] = value
return out
def merge(held: Turn, extra: Turn) -> Turn:
"""Fold a second record of the same API message into the first.
Usage is the same block repeated, so the max per key is taken (robust to a
record written before ``output_tokens`` was final). Tool calls are spread
across the records -- one per ``tool_use`` block -- so those are summed.
"""
keys = held.usage.keys() | extra.usage.keys()
usage = {k: max(held.usage.get(k, 0), extra.usage.get(k, 0)) for k in keys}
return replace(
held,
usage=usage,
context=max(held.context, extra.context),
tool_calls=held.tool_calls + extra.tool_calls,
)
class TurnMerger:
"""Collapse consecutive same-id records into one turn, drop late repeats."""
def __init__(self) -> None:
"""Start with nothing held and no ids seen."""
self._held: Turn | None = None
self._seen: set[str] = set()
def push(self, turn: Turn) -> Turn | None:
"""Accept one record's turn; return a completed turn if one is ready."""
held = self._held
if held is not None and turn.message_id and turn.message_id == held.message_id:
self._held = merge(held, turn)
return None
done = self.flush()
if turn.message_id and turn.message_id in self._seen:
return done
if turn.message_id:
self._seen.add(turn.message_id)
self._held = turn
return done
def flush(self) -> Turn | None:
"""Release the held turn, if any."""
done, self._held = self._held, None
return done

View File

@ -0,0 +1,115 @@
"""The "since last time" half of the report: comparison, ledger, candidates.
Kept apart from :mod:`report` (the per-window tables) because everything here
depends on state outside the window -- the previous snapshot, the ledger, the
registered MCP servers and skills -- and so is exactly the part a re-run must
reproduce from durable files rather than from transcripts that expire.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from python_pkg.token_audit import history, recommend, surfaces, unused
if TYPE_CHECKING:
from python_pkg.token_audit.attribute import Axes, Totals
SNAPSHOT_VERSION = 2
# Metrics compared run-over-run; all are rates, never window totals, because
# windows differ in length (the "up 122%" of 2026-10-04 compared 37 days to 7).
COMPARED: tuple[str, ...] = (
"usd_per_commit",
"usd_per_day",
"usd_per_call",
"calls_per_day",
"subagent_usd_share",
"over_budget_usd_share",
"home_cwd_usd_share",
"batching_calls_per_msg",
"bash_tokens_per_call",
"standing_tokens_est",
)
def extra_metrics(totals: Totals, axes: Axes) -> dict[str, float]:
"""Metrics that come from the token axes rather than from dollars."""
bash_calls = axes.tool_calls.get("Bash", 0)
standing = sum(s.est_tokens for s in surfaces.collect())
return {
"batching_calls_per_msg": round(axes.batching.per_message, 3),
"bash_tokens_per_call": (
round(axes.tool_tokens.get("Bash", 0) / bash_calls, 1)
if bash_calls
else 0.0
),
"image_usd_share": (
round(axes.image_cost / totals.weighted, 4) if totals.weighted else 0.0
),
"standing_tokens_est": standing,
}
def _comparison(now: dict[str, Any], prev: dict[str, Any] | None) -> list[str]:
"""Rate-by-rate table against the previous comparable snapshot."""
if prev is None or prev.get("v", 1) < SNAPSHOT_VERSION:
return [
"_No comparable previous snapshot (pre-2026-10-04 reports counted "
"every API call ~1.9x and omitted subagents; do not compare against "
"them). The next run will show deltas._",
]
before = prev.get("metrics", {})
out = ["| metric | previous | now | change |", "|---|---|---|---|"]
for key in COMPARED:
old, new = before.get(key), now.get(key)
if not isinstance(old, (int, float)) or not isinstance(new, (int, float)):
continue
change = f"{(new - old) / old * 100:+.1f}%" if old else "n/a"
out.append(f"| {key} | {old:,} | {new:,} | {change} |")
return out
def _cell(value: float | None) -> str:
"""A ledger table cell: the number, or a dash when there is none."""
return "-" if value is None else str(value)
def _ledger(snap: dict[str, Any]) -> list[str]:
"""Score every applied lever against the current metrics."""
verdicts = history.score(history.read_ledger(), snap)
if not verdicts:
return ["_Ledger empty: nothing applied yet._"]
out = [
"| date-applied lever | metric | baseline | predicted | now | outcome | note |",
"|---|---|---|---|---|---|---|",
]
out += [
f"| {v.lever} | {v.metric} | {v.baseline:,} | {_cell(v.predicted)} "
f"| {_cell(v.actual)} | **{v.outcome}** | {v.note} |"
for v in verdicts
]
return out
def render(
snap: dict[str, Any], prev: dict[str, Any] | None, *, levers: bool
) -> list[str]:
"""Markdown for comparison, ledger and (optionally) candidate levers."""
lines = ["## Since the previous analysis", ""]
lines += _comparison(snap["metrics"], prev)
lines += ["", "## What helped (ledger)", ""]
lines += _ledger(snap)
if levers:
dead_mcp = [u.name for u in unused.mcp_usage() if u.dead]
dead_skills = [u.name for u in unused.skill_usage() if u.dead]
prev_metrics = (
prev.get("metrics")
if prev and prev.get("v", 1) >= SNAPSHOT_VERSION
else None
)
found = recommend.candidates(
snap["metrics"], prev_metrics, dead_mcp, dead_skills
)
lines += ["", "## Candidate levers (rule-based)", ""]
lines += recommend.render(found)
return lines

View File

@ -0,0 +1,30 @@
"""Keep every token_audit test away from the real ``~/.claude`` state."""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from python_pkg.token_audit import __main__ as cli
from python_pkg.token_audit import history, report, surfaces, unused
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _isolated_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""Point HOME and every module-level ``~/.claude`` constant at tmp_path."""
home = tmp_path / "home"
claude = home / ".claude"
claude.mkdir(parents=True)
monkeypatch.setenv("HOME", str(home))
monkeypatch.setattr(history, "REPORT_DIR", claude / "token-report")
monkeypatch.setattr(report, "REPORT_DIR", claude / "token-report")
monkeypatch.setattr(cli, "TRANSCRIPT_ROOT", claude / "projects")
monkeypatch.setattr(unused, "TRANSCRIPT_ROOT", claude / "projects")
monkeypatch.setattr(surfaces, "CLAUDE_DIR", claude)
monkeypatch.setattr(surfaces, "GLOBAL_CONFIG", home / ".claude.json")
monkeypatch.setattr(surfaces, "SETTINGS", claude / "settings.json")
return home

View File

@ -0,0 +1,138 @@
from __future__ import annotations
from pathlib import Path
import pytest
from python_pkg.token_audit import breakdown
from python_pkg.token_audit.model import Session, ToolCall, Turn
from python_pkg.token_audit.pricing import Price, PriceBook
BOOK = PriceBook(
models={
"claude-opus-5": Price(input=10.0, output=0.0, cache_read=0.0),
"claude-sonnet-5-5": Price(input=2.0, output=0.0, cache_read=0.0),
},
default="claude-opus-5",
)
MILLION = 1_000_000
def _turn(
model: str = "claude-opus-5",
context: int = 0,
stamp: float = 0.0,
tokens: int = MILLION,
) -> Turn:
return Turn(
usage={"input_tokens": tokens},
context=context,
model=model,
is_sidechain=False,
timestamp=stamp,
)
def _sessions() -> list[Session]:
main = Session(
"m",
"m.jsonl",
cwd=str(Path.home()),
turns=[
_turn(stamp=1000.0),
_turn(context=breakdown.CONTEXT_BUDGET + 1, stamp=500.0),
_turn(stamp=2000.0),
],
tools=[ToolCall("Bash", 1, commit=True), ToolCall("Read", 1)],
)
sub_opus = Session(
"a",
"a.jsonl",
cwd="/elsewhere",
turns=[_turn()],
parent_id="m",
agent_type="Explore",
description="scan",
)
sub_sonnet = Session(
"b",
"b.jsonl",
turns=[_turn(model="claude-sonnet-5-5")],
parent_id="m",
)
return [main, sub_opus, sub_sonnet]
def test_bucket_labels() -> None:
assert breakdown._bucket(1) == "1-100"
assert breakdown._bucket(100) == "1-100"
assert breakdown._bucket(101) == "101-200"
assert breakdown._bucket(1500) == "1001+"
def test_stamp_ignores_zero_and_tracks_span() -> None:
found = breakdown.Breakdown()
breakdown._stamp(found, 0.0)
assert (found.earliest, found.latest) == (0.0, 0.0)
breakdown._stamp(found, 100.0)
breakdown._stamp(found, 50.0)
assert (found.earliest, found.latest) == (50.0, 100.0)
def test_build_rolls_up_every_axis() -> None:
found = breakdown.build(_sessions(), BOOK)
assert found.usd_main == pytest.approx(30.0)
assert found.usd_sub == pytest.approx(10.0 + 2.0)
assert found.usd_total == pytest.approx(42.0)
assert (found.calls_main, found.calls_sub, found.calls) == (3, 2, 5)
assert found.over_budget_calls == 1
assert found.over_budget_usd == pytest.approx(10.0)
assert found.home_usd == pytest.approx(30.0)
assert found.commits == 1
assert found.by_agent_type == {"Explore": 10.0, "unknown": 2.0}
assert found.by_description == {"scan": 10.0, "?": 2.0}
assert found.sub_opus_usd == pytest.approx(10.0)
assert found.sub_opus_repriced_usd == pytest.approx(2.0)
assert found.buckets["1-100"].calls == 3
assert (found.earliest, found.latest) == (500.0, 2000.0)
def test_days_has_one_minute_floor() -> None:
assert breakdown.Breakdown().days == 60.0 / breakdown.SECONDS_PER_DAY
def test_metrics_for_populated_breakdown() -> None:
m = breakdown.metrics(breakdown.build(_sessions(), BOOK))
assert m["usd_total"] == 42.0
assert m["calls"] == 5
assert m["commits"] == 1
assert m["usd_per_commit"] == 42.0
assert m["usd_per_call"] == pytest.approx(8.4)
assert m["subagent_usd_share"] == round(12 / 42, 4)
assert m["subagent_opus_usd_share"] == round(10 / 12, 4)
assert m["opus_usd_share"] == round(40 / 42, 4)
assert m["home_cwd_usd_share"] == round(30 / 42, 4)
assert m["over_budget_call_share"] == 0.2
assert m["subagent_sonnet_saving_usd"] == 8.0
def test_metrics_for_empty_breakdown_is_all_zero() -> None:
m = breakdown.metrics(breakdown.Breakdown())
assert m["usd_per_call"] == 0.0
assert m["usd_per_commit"] == 0.0
assert m["subagent_usd_share"] == 0.0
assert m["calls"] == 0
def test_render_has_every_section() -> None:
text = "\n".join(breakdown.render(breakdown.build(_sessions(), BOOK)))
for heading in (
"## Dollar view",
"### By model",
"### Subagents by type",
"### Costliest subagent tasks",
"### Cost per call by position",
):
assert heading in text
assert "| claude-opus-5 | $40.00 |" in text
assert "| 1-100 | $10.0000 | 3 |" in text

View File

@ -18,10 +18,11 @@ def _transcript(
turns: int = 2, turns: int = 2,
*, *,
image: bool = False, image: bool = False,
stamp: str | None = None,
) -> Path: ) -> Path:
project = root / "proj" project = root / "proj"
project.mkdir(exist_ok=True) project.mkdir(exist_ok=True)
records = [] records: list[dict[str, object]] = []
if image: if image:
records += [ records += [
{ {
@ -63,6 +64,8 @@ def _transcript(
}, },
}, },
) )
if stamp is not None:
records = [r | {"timestamp": stamp} for r in records]
path = project / f"{name}.jsonl" path = project / f"{name}.jsonl"
path.write_text("\n".join(json.dumps(r) for r in records), encoding="utf-8") path.write_text("\n".join(json.dumps(r) for r in records), encoding="utf-8")
return path return path
@ -145,7 +148,7 @@ def test_explicit_since_and_until_window(tmp_path: Path) -> None:
def test_until_before_files_yields_no_data(tmp_path: Path) -> None: def test_until_before_files_yields_no_data(tmp_path: Path) -> None:
_transcript(tmp_path) _transcript(tmp_path, stamp="2026-01-01T00:00:00+00:00")
code = cli.main( code = cli.main(
["--root", str(tmp_path), "--since", "0", "--until", "1", "--no-write"] ["--root", str(tmp_path), "--since", "0", "--until", "1", "--no-write"]
) )
@ -170,8 +173,11 @@ def test_second_run_reports_delta(
args = ["--root", str(tmp_path), "--out", str(out_dir), "--days", "36500"] args = ["--root", str(tmp_path), "--out", str(out_dir), "--days", "36500"]
cli.main(args) cli.main(args)
capsys.readouterr() capsys.readouterr()
cli.main(args) # A bare rerun starts where the first ended (nothing new); pin the window.
assert "Week over week" in capsys.readouterr().out cli.main([*args, "--since", "0"])
out = capsys.readouterr().out
assert "## Since the previous analysis" in out
assert "| usd_per_day |" in out
def test_parser_defaults() -> None: def test_parser_defaults() -> None:
@ -190,3 +196,12 @@ def test_help_exits_zero(flag: str) -> None:
with pytest.raises(SystemExit) as excinfo: with pytest.raises(SystemExit) as excinfo:
cli.main([flag]) cli.main([flag])
assert excinfo.value.code == 0 assert excinfo.value.code == 0
def test_rerun_starts_where_previous_window_ended(tmp_path: Path) -> None:
_transcript(tmp_path)
out_dir = tmp_path / "out"
args = ["--root", str(tmp_path), "--out", str(out_dir), "--days", "36500"]
assert cli.main(args) == 0
# The first run's window end is now the default start: nothing newer.
assert cli.main(args) == cli.EXIT_NO_DATA

View File

@ -0,0 +1,51 @@
from __future__ import annotations
import os
from pathlib import Path
from python_pkg.token_audit import discover
def _touch(path: Path, mtime: float, text: str = "{}") -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8")
os.utime(path, (mtime, mtime))
return path
def test_find_transcripts_prunes_by_mtime_newest_first(tmp_path: Path) -> None:
old = _touch(tmp_path / "p" / "old.jsonl", 1000)
mid = _touch(tmp_path / "p" / "mid.jsonl", 20_000)
new = _touch(tmp_path / "q" / "new.jsonl", 50_000)
assert discover.find_transcripts(tmp_path, since=10_000) == [new, mid]
assert discover.find_transcripts(tmp_path, since=0) == [new, mid, old]
def test_find_transcripts_includes_subagent_layout(tmp_path: Path) -> None:
main = _touch(tmp_path / "p" / "s.jsonl", 100)
sub = _touch(tmp_path / "p" / "s" / "subagents" / "agent-1.jsonl", 200)
assert discover.find_transcripts(tmp_path, since=0) == [sub, main]
def test_first_cwd_skips_junk_and_cwdless_records(tmp_path: Path) -> None:
lines = ["not json", "[1]", '{"a": 1}', '{"cwd": 5}', '{"cwd": "/x"}']
path = _touch(tmp_path / "t.jsonl", 1, "\n".join(lines))
assert discover.first_cwd(path) == "/x"
def test_first_cwd_none_when_absent(tmp_path: Path) -> None:
path = _touch(tmp_path / "t.jsonl", 1, '{"a": 1}')
assert discover.first_cwd(path) is None
def test_agent_meta_reads_sibling(tmp_path: Path) -> None:
path = _touch(tmp_path / "agent-1.jsonl", 1)
_touch(tmp_path / "agent-1.meta.json", 1, '{"agentType": "Explore"}')
assert discover.agent_meta(path) == {"agentType": "Explore"}
def test_agent_meta_empty_when_missing_or_bad(tmp_path: Path) -> None:
path = _touch(tmp_path / "agent-1.jsonl", 1)
assert discover.agent_meta(path) == {}
_touch(tmp_path / "agent-1.meta.json", 1, "garbage")
assert discover.agent_meta(path) == {}

View File

@ -0,0 +1,76 @@
from __future__ import annotations
import json
from pathlib import Path
from python_pkg.token_audit import history
def test_save_snapshot_never_overwrites(tmp_path: Path) -> None:
first = history.save_snapshot({"n": 1}, tmp_path)
second = history.save_snapshot({"n": 2}, tmp_path)
assert first != second
assert json.loads(first.read_text(encoding="utf-8")) == {"n": 1}
assert json.loads(second.read_text(encoding="utf-8")) == {"n": 2}
assert first.parent == tmp_path / history.HISTORY_NAME
def test_default_directory_is_redirected_report_dir() -> None:
path = history.save_snapshot({"n": 1})
assert path.is_relative_to(history.REPORT_DIR)
assert history.last_snapshot() == {"n": 1}
def test_last_snapshot_none_without_folder(tmp_path: Path) -> None:
assert history.last_snapshot(tmp_path) is None
assert history.default_since(tmp_path) is None
def test_last_snapshot_skips_unreadable_and_non_dict(tmp_path: Path) -> None:
folder = tmp_path / history.HISTORY_NAME
folder.mkdir()
(folder / "2026-01-01.json").write_text('{"ok": true}', encoding="utf-8")
(folder / "2026-01-02.json").write_text("[1]", encoding="utf-8")
(folder / "2026-01-03.json").write_text("garbage", encoding="utf-8")
assert history.last_snapshot(tmp_path) == {"ok": True}
def test_last_snapshot_none_when_nothing_usable(tmp_path: Path) -> None:
folder = tmp_path / history.HISTORY_NAME
folder.mkdir()
(folder / "2026-01-01.json").write_text("garbage", encoding="utf-8")
assert history.last_snapshot(tmp_path) is None
def test_default_since_reads_window_end(tmp_path: Path) -> None:
history.save_snapshot({"window": {"until": 123.5}}, tmp_path)
assert history.default_since(tmp_path) == 123.5
def test_default_since_none_when_until_not_numeric(tmp_path: Path) -> None:
history.save_snapshot({"window": {"until": "soon"}}, tmp_path)
assert history.default_since(tmp_path) is None
def test_read_ledger_missing_file(tmp_path: Path) -> None:
assert history.read_ledger(tmp_path) == []
def test_append_and_read_ledger_filters_bad_lines(tmp_path: Path) -> None:
target = tmp_path / "nested"
history.append_ledger({"metric": "m", "baseline": 1.0}, target)
history.append_ledger(
{"metric": "m", "baseline": 2.0, "date": "2020-01-01"}, target
)
ledger = target / history.LEDGER_NAME
with ledger.open("a", encoding="utf-8") as handle:
handle.write("garbage\n[1]\n" + json.dumps({"metric": "only"}) + "\n")
entries = history.read_ledger(target)
assert [e["baseline"] for e in entries] == [1.0, 2.0]
assert entries[1]["date"] == "2020-01-01"
assert len(entries[0]["date"]) == len("2026-10-04")
def test_append_ledger_default_directory() -> None:
history.append_ledger({"metric": "m", "baseline": 1})
assert len(history.read_ledger()) == 1

View File

@ -0,0 +1,69 @@
from __future__ import annotations
import pytest
from python_pkg.token_audit import history
def _entry(**over: object) -> dict[str, object]:
return {"lever": "L", "metric": "m", "baseline": 100.0, **over}
@pytest.mark.parametrize(
("baseline", "actual", "lower", "outcome"),
[
(0.0, 5.0, True, "no baseline"),
(100.0, 102.0, True, "no effect"),
(100.0, 50.0, True, "helped"),
(100.0, 150.0, True, "regressed"),
(100.0, 150.0, False, "helped"),
(100.0, 50.0, False, "regressed"),
],
)
def test_outcome_classes(
baseline: float, actual: float, outcome: str, *, lower: bool
) -> None:
assert history._outcome(baseline, actual, lower_is_better=lower) == outcome
def test_prediction_note_variants() -> None:
assert history._prediction_note(100, None, 50) == ""
assert history._prediction_note(100, 100, 50) == ""
assert history._prediction_note(100, 50, 50) == "prediction held"
assert history._prediction_note(100, 50, 100) == "prediction off by 100%"
def test_score_measured_entry() -> None:
verdicts = history.score(
[_entry(predicted=50.0)],
{"metrics": {"m": 50}},
)
(verdict,) = verdicts
assert verdict.outcome == "helped"
assert verdict.actual == 50.0
assert verdict.predicted == 50.0
assert verdict.note == "prediction held"
def test_score_higher_is_better_without_prediction() -> None:
(verdict,) = history.score(
[_entry(lower_is_better=False)],
{"metrics": {"m": 200}},
)
assert verdict.outcome == "helped"
assert verdict.predicted is None
assert verdict.note == ""
def test_score_unmeasured_metric() -> None:
(verdict,) = history.score([_entry(predicted=1)], {"metrics": {}})
assert verdict.outcome == "unmeasured"
assert verdict.actual is None
assert verdict.predicted == 1.0
def test_score_unmeasured_without_lever_or_metrics_key() -> None:
entry = {"metric": "m", "baseline": 1}
(verdict,) = history.score([entry], {})
assert verdict.lever == "?"
assert verdict.outcome == "unmeasured"

View File

@ -186,7 +186,7 @@ def test_load_session_ignores_events_of_other_types(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None: ) -> None:
path = _write(tmp_path, [_assistant(usage={"output_tokens": 1})]) path = _write(tmp_path, [_assistant(usage={"output_tokens": 1})])
monkeypatch.setattr(parse, "iter_events", lambda _: iter([("turn", "not a Turn")])) monkeypatch.setattr(parse, "iter_events", lambda *_: iter([("turn", "not a Turn")]))
session = parse.load_session(path) session = parse.load_session(path)
assert session.turns == [] assert session.turns == []
assert session.tools == [] assert session.tools == []

View File

@ -84,23 +84,6 @@ def test_events_carry_expected_types(tmp_path: Path) -> None:
assert isinstance(events["turn"], Turn) assert isinstance(events["turn"], Turn)
def test_find_transcripts_filters_by_window(tmp_path: Path) -> None:
project = tmp_path / "proj"
project.mkdir()
old = project / "old.jsonl"
new = project / "new.jsonl"
old.write_text("{}", encoding="utf-8")
new.write_text("{}", encoding="utf-8")
import os
os.utime(old, (1000, 1000))
os.utime(new, (50_000, 50_000))
assert parse.find_transcripts(tmp_path, since=10_000) == [new]
assert parse.find_transcripts(tmp_path, since=0, until=2000) == [old]
assert len(parse.find_transcripts(tmp_path, since=0)) == 2
@pytest.mark.parametrize("payload", [None, "abcdefgh"]) @pytest.mark.parametrize("payload", [None, "abcdefgh"])
def test_result_tokens_handles_missing_content(payload: str | None) -> None: def test_result_tokens_handles_missing_content(payload: str | None) -> None:
assert parse._result_tokens({"content": payload}) >= 0 assert parse._result_tokens({"content": payload}) >= 0

View File

@ -0,0 +1,74 @@
from __future__ import annotations
import json
from pathlib import Path
from python_pkg.token_audit import parse
from python_pkg.token_audit.model import Turn
def _line(record: dict[str, object]) -> str:
return json.dumps(record)
def test_tool_result_flushes_held_turn_in_order(tmp_path: Path) -> None:
use = {"type": "tool_use", "id": "t1", "name": "Bash", "input": {}}
records = [
{
"message": {
"id": "m1",
"model": "claude-opus-5",
"usage": {"output_tokens": 2},
"content": [use],
},
},
{"message": {"content": [{"type": "tool_result", "tool_use_id": "t1"}]}},
]
path = tmp_path / "s.jsonl"
path.write_text("\n".join(_line(r) for r in records), encoding="utf-8")
kinds = [kind for kind, _ in parse.iter_events(path)]
assert kinds == ["turn", "tool"]
def test_load_session_tags_subagent(tmp_path: Path) -> None:
sub = tmp_path / "proj" / "sess-1" / "subagents"
sub.mkdir(parents=True)
path = sub / "agent-a.jsonl"
record = {"message": {"model": "claude-opus-5", "usage": {"output_tokens": 1}}}
path.write_text(_line(record), encoding="utf-8")
(sub / "agent-a.meta.json").write_text(
json.dumps({"agentType": "Explore", "description": "find things"}),
encoding="utf-8",
)
session = parse.load_session(path)
assert session.parent_id == "sess-1"
assert session.agent_type == "Explore"
assert session.description == "find things"
assert isinstance(session.turns[0], Turn)
def test_load_session_subagent_without_meta_defaults(tmp_path: Path) -> None:
sub = tmp_path / "proj" / "sess-2" / "subagents"
sub.mkdir(parents=True)
path = sub / "agent-b.jsonl"
path.write_text("{}", encoding="utf-8")
session = parse.load_session(path)
assert session.agent_type == "unknown"
assert session.description == ""
def test_window_filters_records_by_timestamp(tmp_path: Path) -> None:
def rec(stamp: str, mid: str) -> dict[str, object]:
return {
"timestamp": stamp,
"message": {"id": mid, "usage": {"output_tokens": 1}},
}
path = tmp_path / "s.jsonl"
rows = [
rec("1970-01-01T00:00:10+00:00", "a"),
rec("1970-01-01T00:01:40+00:00", "b"),
]
path.write_text("\n".join(_line(r) for r in rows), encoding="utf-8")
session = parse.load_session(path, (50.0, 200.0))
assert [t.message_id for t in session.turns] == ["b"]

View File

@ -0,0 +1,70 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from python_pkg.token_audit import pricing
from python_pkg.token_audit.model import Turn
BOOK = pricing.PriceBook(
models={
"claude-opus-5": pricing.Price(input=5.0, output=25.0, cache_read=0.5),
"claude-opus-5-5": pricing.Price(input=4.0, output=20.0, cache_read=0.2),
},
default="claude-opus-5",
)
def _turn(usage: dict[str, int], model: str = "claude-opus-5") -> Turn:
return Turn(usage=usage, context=0, model=model, is_sidechain=False)
def test_for_model_exact_prefix_and_default() -> None:
assert BOOK.for_model("claude-opus-5").input == 5.0
# Longest matching known id wins over a shorter prefix.
assert BOOK.for_model("claude-opus-5-5-2026").input == 4.0
assert BOOK.for_model("mystery").input == 5.0
def test_load_reads_shipped_table() -> None:
book = pricing.load()
assert book.default in book.models
assert book.for_model("claude-sonnet-5-5").cache_read == 0.2
def test_load_custom_path(tmp_path: Path) -> None:
path = tmp_path / "p.json"
path.write_text(
json.dumps(
{
"default": "m",
"models": {"m": {"input": 1, "output": 2, "cache_read": 3}},
},
),
encoding="utf-8",
)
book = pricing.load(path)
assert book.models["m"] == pricing.Price(1.0, 2.0, 3.0)
def test_usd_without_split_prices_creation_as_5m() -> None:
usage = {"input_tokens": 1_000_000, "cache_creation_input_tokens": 1_000_000}
assert pricing.usd(_turn(usage), BOOK) == pytest.approx(5.0 + 1.25 * 5.0)
def test_usd_with_split_prices_1h_at_double() -> None:
usage = {
"cache_creation_input_tokens": 1_000_000,
"cache_creation_1h": 400_000,
"cache_creation_5m": 600_000,
"cache_read_input_tokens": 1_000_000,
"output_tokens": 1_000_000,
}
expected = 0.4 * 2.0 * 5.0 + 0.6 * 1.25 * 5.0 + 0.5 + 25.0
assert pricing.usd(_turn(usage), BOOK) == pytest.approx(expected)
def test_usd_zero_for_empty_usage() -> None:
assert pricing.usd(_turn({}), BOOK) == 0.0

View File

@ -0,0 +1,89 @@
from __future__ import annotations
from typing import Any
from python_pkg.token_audit import recommend
def _fire_all() -> dict[str, Any]:
return {
"usd_total": 70.0,
"usd_per_day": 10.0,
"subagent_usd_share": 0.4,
"subagent_opus_usd_share": 0.9,
"subagent_sonnet_saving_usd": 14.0,
"over_budget_usd_share": 0.5,
"home_cwd_usd_share": 0.4,
"batching_calls_per_msg": 1.05,
"bash_tokens_per_call": 900,
"standing_tokens_est": 120,
"image_usd_share": 0.1,
}
def test_num_treats_missing_and_non_numeric_as_zero() -> None:
assert recommend._num({"a": 3}, "a") == 3.0
assert recommend._num({"a": "x"}, "a") == 0.0
assert recommend._num({}, "a") == 0.0
def test_nothing_fires_on_quiet_metrics() -> None:
assert recommend.candidates({}, None, [], []) == []
assert recommend.render([]) == ["_No rule fired._"]
def test_subagent_rule_needs_both_thresholds() -> None:
low_share = {"subagent_usd_share": 0.01, "subagent_opus_usd_share": 0.9}
low_opus = {"subagent_usd_share": 0.5, "subagent_opus_usd_share": 0.1}
assert recommend._subagents(low_share) == []
assert recommend._subagents(low_opus) == []
def test_subagent_lever_sizes_saving_per_day() -> None:
(lever,) = recommend._subagents(_fire_all())
assert lever.category == "cheaper agents"
# 70 / 10 = 7 days; 14 saved over 7 days = 2 per day.
assert lever.est_usd_per_day == 2.0
def test_every_rule_fires_and_sized_levers_sort_first() -> None:
previous = {"standing_tokens_est": 100}
levers = recommend.candidates(_fire_all(), previous, ["mcp1"], ["skill1"])
titles = [lv.title for lv in levers]
assert titles[0] == "Calls over the context budget"
assert set(titles) == {
"Opus runs most subagent work",
"Calls over the context budget",
"Sessions started in ~",
"Independent tool calls not batched",
"Large Bash results",
"Always-loaded context grew",
"Images held in context",
"Unused MCP servers / skills",
}
sized = [lv.est_usd_per_day for lv in levers if lv.est_usd_per_day]
assert sized == sorted(sized, reverse=True)
def test_prefix_rule_silent_without_growth_or_baseline() -> None:
flat = {"standing_tokens_est": 100}
assert recommend._mechanics(flat, {"standing_tokens_est": 100}) == []
assert recommend._mechanics(flat, {"standing_tokens_est": 0}) == []
assert recommend._mechanics(flat, None) == []
def test_dead_lists_render_dash_for_empty_side() -> None:
assert recommend._dead([], []) == []
(only_mcp,) = recommend._dead(["m"], [])
assert "MCP: m; skills: -" in only_mcp.evidence
(only_skill,) = recommend._dead([], ["s"])
assert "MCP: -; skills: s" in only_skill.evidence
def test_render_table_formats_estimates() -> None:
sized = recommend.Lever("c", "t", "e", "a", 1.5)
unsized = recommend.Lever("c", "t2", "e", "a")
lines = recommend.render([sized, unsized])
assert lines[0].startswith("| category")
assert "| $1.50 |" in lines[2]
assert "| - |" in lines[3]

View File

@ -0,0 +1,113 @@
from __future__ import annotations
import pytest
from python_pkg.token_audit import records
from python_pkg.token_audit.model import Turn
def _turn(
message_id: str = "m1",
usage: dict[str, int] | None = None,
context: int = 0,
tool_calls: int = 0,
) -> Turn:
return Turn(
usage=usage or {},
context=context,
model="claude-opus-5",
is_sidechain=False,
message_id=message_id,
tool_calls=tool_calls,
)
@pytest.mark.parametrize(
"record",
[{}, {"timestamp": 5}, {"timestamp": "not a date"}],
)
def test_timestamp_none_for_missing_or_bad(record: dict[str, object]) -> None:
assert records.timestamp(record) is None
def test_timestamp_parses_iso() -> None:
stamp = records.timestamp({"timestamp": "1970-01-02T00:00:00+00:00"})
assert stamp == 86400.0
def test_in_window_rules() -> None:
dated = {"timestamp": "1970-01-01T00:01:40+00:00"} # 100s
assert records.in_window(dated, None)
assert records.in_window({}, (0.0, 1.0))
assert records.in_window(dated, (100.0, 100.0))
assert not records.in_window(dated, (101.0, 200.0))
assert not records.in_window(dated, (0.0, 99.0))
def test_cache_split_reads_both_classes() -> None:
usage = {
"cache_creation": {
"ephemeral_1h_input_tokens": 7,
"ephemeral_5m_input_tokens": 3,
},
}
assert records.cache_split(usage) == {
records.CACHE_1H: 7,
records.CACHE_5M: 3,
}
def test_cache_split_ignores_non_dict_and_non_int() -> None:
assert records.cache_split({"cache_creation": 4}) == {}
nested = {"cache_creation": {"ephemeral_1h_input_tokens": "x"}}
assert records.cache_split(nested) == {}
def test_merge_takes_max_usage_and_sums_tools() -> None:
held = _turn(usage={"output_tokens": 1, "input_tokens": 5}, context=3, tool_calls=1)
extra = _turn(
usage={"output_tokens": 9, "cache_read_input_tokens": 2},
context=1,
tool_calls=2,
)
merged = records.merge(held, extra)
assert merged.usage == {
"output_tokens": 9,
"input_tokens": 5,
"cache_read_input_tokens": 2,
}
assert merged.context == 3
assert merged.tool_calls == 3
def test_merger_collapses_consecutive_same_id() -> None:
merger = records.TurnMerger()
assert merger.push(_turn("a", {"output_tokens": 1})) is None
assert merger.push(_turn("a", {"output_tokens": 4})) is None
done = merger.push(_turn("b"))
assert done is not None
assert done.usage == {"output_tokens": 4}
last = merger.flush()
assert last is not None
assert last.message_id == "b"
assert merger.flush() is None
def test_merger_drops_late_repeat_of_seen_id() -> None:
merger = records.TurnMerger()
merger.push(_turn("a"))
merger.push(_turn("b"))
# "a" already seen and not the held turn: dropped, "b" is released.
done = merger.push(_turn("a"))
assert done is not None
assert done.message_id == "b"
assert merger.flush() is None
def test_merger_keeps_every_idless_turn() -> None:
merger = records.TurnMerger()
assert merger.push(_turn("")) is None
first = merger.push(_turn(""))
assert first is not None
last = merger.flush()
assert last is not None

View File

@ -0,0 +1,115 @@
from __future__ import annotations
from typing import Any
import pytest
from python_pkg.token_audit import attribute, history, review, surfaces, unused
from python_pkg.token_audit.model import Session, ToolCall, Turn
def _usage(name: str, short: int, long: int) -> unused.Usage:
return unused.Usage(name, short, long)
def _axes(*, bash: bool, spend: bool) -> tuple[attribute.Totals, attribute.Axes]:
usage = {"output_tokens": 10} if spend else {}
turn = Turn(usage=usage, context=0, model="claude-opus-5", is_sidechain=False)
tools = [ToolCall("Bash", 400)] if bash else []
return attribute.build([Session("s", "s.jsonl", turns=[turn], tools=tools)])
def test_extra_metrics_with_bash_and_spend(monkeypatch: pytest.MonkeyPatch) -> None:
fake = [surfaces.Surface("a", "d", 40), surfaces.Surface("b", "d", 80)]
monkeypatch.setattr(review.surfaces, "collect", lambda: fake)
totals, axes = _axes(bash=True, spend=True)
axes.image_cost = totals.weighted / 2
m = review.extra_metrics(totals, axes)
assert m["bash_tokens_per_call"] > 0
assert m["image_usd_share"] == 0.5
assert m["standing_tokens_est"] == 30
assert "batching_calls_per_msg" in m
def test_extra_metrics_without_bash_or_spend() -> None:
totals, axes = _axes(bash=False, spend=False)
m = review.extra_metrics(totals, axes)
assert m["bash_tokens_per_call"] == 0.0
assert m["image_usd_share"] == 0.0
def test_comparison_without_usable_previous() -> None:
assert "No comparable previous snapshot" in review._comparison({}, None)[0]
assert "No comparable previous snapshot" in review._comparison({}, {"v": 1})[0]
assert "No comparable previous snapshot" in review._comparison({}, {})[0]
def test_comparison_table_rows() -> None:
prev: dict[str, Any] = {
"v": review.SNAPSHOT_VERSION,
"metrics": {
"usd_per_day": 100.0,
"usd_per_call": 0,
"calls_per_day": "bad",
"usd_per_commit": 2.0,
},
}
now = {"usd_per_day": 50.0, "usd_per_call": 1.0, "calls_per_day": 3.0}
rows = review._comparison(now, prev)
assert rows[0].startswith("| metric")
assert "| usd_per_day | 100.0 | 50.0 | -50.0% |" in rows
assert "| usd_per_call | 0 | 1.0 | n/a |" in rows
assert len(rows) == 4 # header, rule, two rendered metrics
def test_ledger_empty_and_populated() -> None:
assert review._ledger({"metrics": {}}) == ["_Ledger empty: nothing applied yet._"]
history.append_ledger(
{"lever": "L1", "metric": "m", "baseline": 10, "predicted": 5}
)
history.append_ledger({"lever": "L2", "metric": "gone", "baseline": 1})
lines = review._ledger({"metrics": {"m": 4}})
text = "\n".join(lines)
assert "| L1 | m | 10.0 | 5.0 | 4.0 | **helped** |" in text
assert "| L2 | gone | 1.0 | - | - | **unmeasured** |" in text
def _patch_dead(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
review.unused,
"mcp_usage",
lambda: [_usage("deadmcp", 0, 0), _usage("live", 1, 0)],
)
monkeypatch.setattr(
review.unused, "skill_usage", lambda: [_usage("deadskill", 0, 0)]
)
def test_render_without_levers_has_no_candidates() -> None:
snap = {"metrics": {"usd_per_day": 1.0}}
text = "\n".join(review.render(snap, None, levers=False))
assert "## Since the previous analysis" in text
assert "## What helped (ledger)" in text
assert "Candidate levers" not in text
def test_render_with_levers_uses_previous_snapshot(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_patch_dead(monkeypatch)
snap = {"metrics": {"standing_tokens_est": 200}}
prev = {"v": review.SNAPSHOT_VERSION, "metrics": {"standing_tokens_est": 100}}
text = "\n".join(review.render(snap, prev, levers=True))
assert "## Candidate levers (rule-based)" in text
assert "Always-loaded context grew" in text
assert "MCP: deadmcp; skills: deadskill" in text
def test_render_with_levers_ignores_old_or_missing_previous(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_patch_dead(monkeypatch)
snap = {"metrics": {"standing_tokens_est": 200}}
for prev in (None, {"v": 1, "metrics": {"standing_tokens_est": 100}}):
text = "\n".join(review.render(snap, prev, levers=True))
assert "Always-loaded context grew" not in text

View File

@ -0,0 +1,70 @@
"""Build a :class:`Turn` from one assistant transcript record.
Split out of :mod:`parse` to keep both under the file-length cap; this half is
the pure record -> turn conversion, with no I/O and no ordering concerns.
"""
from __future__ import annotations
from typing import Any
from python_pkg.token_audit import records
from python_pkg.token_audit.model import Turn
def usage_of(record: dict[str, Any]) -> tuple[dict[str, Any], dict[str, int]] | None:
"""Return an assistant record's message and its integer usage fields.
The message is returned alongside the usage block so callers that need both
(the model id lives on the message) never have to re-validate its type.
"""
message = record.get("message")
if not isinstance(message, dict):
return None
usage = message.get("usage")
if not isinstance(usage, dict):
return None
flat = {k: v for k, v in usage.items() if isinstance(v, int)}
return message, flat | records.cache_split(usage)
def _model(message: dict[str, Any]) -> str:
"""Return the model id that served a record, or a placeholder.
Only ever called after :func:`_usage` has confirmed the message is a dict,
so it takes the message itself rather than re-checking the record.
"""
model = message.get("model")
return model if isinstance(model, str) else "unknown"
def build_turn(
record: dict[str, Any],
message: dict[str, Any],
usage: dict[str, int],
) -> Turn:
"""Build a :class:`Turn` from an assistant record's usage block."""
context = usage.get("cache_read_input_tokens", 0) + usage.get(
"cache_creation_input_tokens",
0,
)
message_id = message.get("id")
content = message.get("content")
tool_calls = (
sum(
1
for block in content
if isinstance(block, dict) and block.get("type") == "tool_use"
)
if isinstance(content, list)
else 0
)
return Turn(
usage=usage,
context=context,
model=_model(message),
is_sidechain=bool(record.get("isSidechain")),
message_id=message_id if isinstance(message_id, str) else "",
tool_calls=tool_calls,
timestamp=records.timestamp(record) or 0.0,
)

26
scripts/check_repo_contract.sh Executable file
View File

@ -0,0 +1,26 @@
#!/bin/bash
# ============================================================================
# Fail if this repo breaks the shared repo contract.
#
# Thin delegate to the shared gate in ~/src/utils, which owns the contract
# text. Copy this file to scripts/check_repo_contract.sh in a repo; never copy
# the shared gate itself, or the shim would exec itself forever.
# ============================================================================
set -euo pipefail
readonly SHARED_GATE="${UTILS_ROOT:-$HOME/src/utils}/scripts/check_repo_contract.sh"
main() {
if [[ ! -x "$SHARED_GATE" ]]; then
echo "Error: shared repo-contract gate not found at $SHARED_GATE" >&2
echo " Clone github.com/kuhyx/utils to ~/src/utils, or set" >&2
echo " UTILS_ROOT to where it lives." >&2
exit 1
fi
exec bash "$SHARED_GATE" "$@"
}
main "$@"

57
scripts/test_changed.sh Executable file
View File

@ -0,0 +1,57 @@
#!/bin/bash
# ============================================================================
# Run only the pytest files related to files changed vs HEAD (staged, unstaged
# and untracked). Quiet: failures plus a one-line summary. Falls back to the
# full suite when a change cannot be mapped to tests. Coverage is NOT measured
# here (the full gate is meta/scripts/pytest_changed_packages.py); addopts is
# overridden to drop the -v/--cov flags from meta/pyproject.toml.
# ============================================================================
set -euo pipefail
readonly OPTS='--strict-markers --strict-config'
main() {
cd "$(git rev-parse --show-toplevel)"
local changed=() tests=() full=0 f base t
while IFS= read -r f; do
[[ -n "$f" && -e "$f" ]] && changed+=("$f")
done < <({
git diff --name-only HEAD 2>/dev/null || true
git ls-files --others --exclude-standard
} | sort -u)
for f in "${changed[@]}"; do
case "$f" in
*/conftest.py | conftest.py | meta/pyproject.toml | pyproject.toml | meta/requirements*.txt) full=1 ;;
python_pkg/*.py | linux_configuration/tests/*.py | tests/*.py)
base="$(basename "$f" .py)"
if [[ "$base" == test_* || "$base" == *_test ]]; then
tests+=("$f")
else
while IFS= read -r t; do tests+=("$t"); done \
< <(git ls-files "*test_${base}.py" "*${base}_test.py")
fi
;;
esac
done
if [[ $full -eq 0 && ${#changed[@]} -eq 0 ]]; then
echo "no changes vs HEAD: nothing to test"
exit 0
fi
if [[ $full -eq 0 && ${#tests[@]} -gt 0 ]]; then
python3 -m pytest -q --tb=short -o addopts="$OPTS" "${tests[@]}"
exit $?
fi
if [[ $full -eq 0 ]] && ! printf '%s\n' "${changed[@]}" | grep -q '\.py$'; then
echo "no python changes: nothing to test"
exit 0
fi
echo "no mapped tests: running full suite"
python3 -m pytest -q --tb=short -o addopts="$OPTS"
}
main "$@"
exit $?