diff --git a/docs/superpowers/contracts/token-audit-usd-per-note-2026-11-01.json b/docs/superpowers/contracts/token-audit-usd-per-note-2026-11-01.json new file mode 100644 index 00000000..b6d2c375 --- /dev/null +++ b/docs/superpowers/contracts/token-audit-usd-per-note-2026-11-01.json @@ -0,0 +1,16 @@ +{ + "title": "token_audit: USD per completed todo note", + "objective": "The weekly token audit reports how many list-price dollars each finished todo note cost, by linking sessions to notes (note_claim.sh, todo-images prompts, fuzzy first prompt, handoff chains) and dating completions from the Firebase dumps.", + "acceptance_criteria": [ + "Join rules a-d, the done-date recovery and the metric math are covered by python_pkg/token_audit/tests/test_notes_*.py at 100% branch coverage", + "The audit snapshot metrics carry usd_per_note (absent when no linked done note), note_linked_usd_share and notes_done; WEEKLY.md gains a 'Cost per todo note' section", + "notes.json in the report dir lists every note with linked spend (id, title, status, priority, done_at, sessions, usd, tokens)", + "A 30-day audit including the notes stage finishes under 60 s", + "token_dashboard stores the build-class A/B (agent-ab-build.jsonl) as meta ab_build, separate from the judgment meta ab (tests/test_ab_build.py)" + ], + "out_of_scope": [ + "python_pkg/app_icons changes in the working tree", + "Editing ~/.claude/CLAUDE.md (left to kuhy)" + ], + "verifier": "python3 -m pytest -q -o addopts='' python_pkg/token_audit/tests --cov=python_pkg.token_audit --cov-branch; capped.sh env PYTHONPATH=$HOME/src/testsAndMisc python3 -m python_pkg.token_audit --since $(date -d '30 days ago' +%s) --no-write" +} diff --git a/docs/superpowers/evidence/token-audit-usd-per-note-2026-11-01.json b/docs/superpowers/evidence/token-audit-usd-per-note-2026-11-01.json new file mode 100644 index 00000000..d9ac33c8 --- /dev/null +++ b/docs/superpowers/evidence/token-audit-usd-per-note-2026-11-01.json @@ -0,0 +1,38 @@ +{ + "intent": "Add a USD-per-completed-todo-note efficiency metric to the Claude Code token audit, plus notes.json for the control-panel dashboard. Also summarise the new build-class A/B log (agent-ab-build.jsonl) into its own meta row for the dashboard.", + "scope": [ + "python_pkg/token_audit/notes*.py (new), __main__.py and review.py (wiring), tests/test_notes_*.py + tests/notes_fixtures.py; python_pkg/token_dashboard/ab.py + tests/test_ab_build.py + tests/conftest.py", + "Non-goal: changing existing metrics or pricing; app_icons working-tree changes belong to another session" + ], + "changes": [ + "notes_backlog: BACKLOG.md parse plus done dates from the status-field HLC in the first Firebase dump showing the terminal status, BACKLOG updated as fallback", + "notes_scan/notes_link: per-session facts from typed prompts and tool_use blocks only; rules a (note_claim.sh), b (todo-images// in a prompt, <=2 notes), c (first prompt, max of body Dice and first-line/title Dice >= 0.5), d (handoff readers inherit the earliest creator's notes)", + "notes_cost/notes_collect/notes: lifetime spend of linked sessions (main + subagents, reusing audit-loaded files), usd_per_note, note_linked_usd_share, notes_done, notes_done_linked, WEEKLY section and notes.json; standalone python -m python_pkg.token_audit.notes", + "token_dashboard.ab: rebuild() also joins ~/.claude/logs/agent-ab-build.jsonl and stores meta 'ab_build' with a 'A/B build agents' banner; ab_rows stays judgment-only" + ], + "verification": [ + { + "command": "python3 -m pytest -q -o addopts='' python_pkg/token_audit/tests --cov=python_pkg.token_audit --cov-branch", + "result": "pass", + "evidence": "253 passed; token_audit TOTAL 100% line and branch" + }, + { + "command": "capped.sh env PYTHONPATH=$HOME/src/testsAndMisc python3 -m python_pkg.token_audit --since <30 days ago> --no-write", + "result": "pass", + "evidence": "48.2 s total; notes: 135 sessions linked (b=14 c=102 d=19); usd_per_note $55.77, note_linked_usd_share 64.6%, notes_done 98 (44 linked)" + }, + { + "command": "python3 -m pytest -q -o addopts='' python_pkg/token_dashboard/tests", + "result": "pass", + "evidence": "147 passed; ab.py 100% branch; live ingest stored ab_build and /api/tokens/panel/ab returns build banner 'A/B build agents: collecting (n=0)'" + } + ], + "risks": [ + "Fuzzy rule c can mislink a prompt that paraphrases another note; threshold calibrated by hand-labelling the 0.25-0.8 band (10/10 correct at >= 0.5, 15/16 image ground truths matched)", + "Done dates before the first dump (2026-09-26) are upper bounds" + ], + "rollback": [ + "git revert the commit; the audit falls back to its previous metrics", + "Run python3 -m python_pkg.token_audit --no-write and check it exits 0" + ] +} diff --git a/python_pkg/token_audit/__main__.py b/python_pkg/token_audit/__main__.py index 39bb447f..6f8d8b0f 100644 --- a/python_pkg/token_audit/__main__.py +++ b/python_pkg/token_audit/__main__.py @@ -22,6 +22,7 @@ from python_pkg.token_audit import ( discover, history, imagecost, + notes, parse, pricing, report, @@ -135,12 +136,16 @@ def main(argv: list[str] | None = None) -> int: drift = attribute.reconcile(totals, sessions) window = report.Window(since=since, until=until) snap = report.snapshot(totals, axes, window, len(sessions)) - priced = breakdown.build(sessions, pricing.load()) + book = pricing.load() + priced = breakdown.build(sessions, book) snap["v"] = review.SNAPSHOT_VERSION snap["effective_window"] = {"since": priced.earliest, "until": priced.latest} snap["metrics"] = breakdown.metrics(priced) | review.extra_metrics( totals, axes, priced.days ) + noted = notes.section(args.root, book, sessions, (since, until)) + if noted is not None: + snap["metrics"] |= noted[0] previous = history.last_snapshot(args.out) found = review.candidates(snap, previous) if args.recommend else None if found is not None: @@ -154,6 +159,7 @@ def main(argv: list[str] | None = None) -> int: *breakdown.render(priced), "", *review.render(snap, previous, levers=args.recommend, found=found), + *(noted[1] if noted is not None else []), "", ], ) @@ -161,6 +167,8 @@ def main(argv: list[str] | None = None) -> int: if not args.no_write: written = report.write(markdown, snap, args.out) history.save_snapshot(snap, args.out) + if noted is not None: + notes.write_json(noted[2], args.out) sys.stderr.write(f"Wrote {written}\n") sys.stdout.write(json.dumps(snap, indent=2) if args.json else markdown) diff --git a/python_pkg/token_audit/notes.py b/python_pkg/token_audit/notes.py new file mode 100644 index 00000000..fb29c608 --- /dev/null +++ b/python_pkg/token_audit/notes.py @@ -0,0 +1,198 @@ +"""USD per completed todo note: per-note rows, window metrics and outputs. + +``usd_per_note`` = lifetime linked USD of notes done or abandoned in the window +over the notes done in the window that have a linked session (lower is +better). ``note_linked_usd_share`` is the share of window spend linked to any +note: how far to trust the first number. +""" + +from __future__ import annotations + +import argparse +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +import json +from pathlib import Path +import sys +import time +from typing import TYPE_CHECKING + +from python_pkg.token_audit import history, notes_collect, pricing +from python_pkg.token_audit.notes_backlog import TERMINAL + +if TYPE_CHECKING: + from collections.abc import Mapping, Sequence + + from python_pkg.token_audit.model import Session + from python_pkg.token_audit.notes_backlog import Note + from python_pkg.token_audit.notes_cost import Spend + from python_pkg.token_audit.notes_link import Link + +JSON_NAME = "notes.json" +TOP = 15 +TITLE_CHARS = 60 + + +@dataclass(frozen=True) +class NoteRow: + """One note with the spend of every session linked to it.""" + + id: str + title: str + status: str + priority: str + done_at: float | None + sessions: list[str] + usd: float + tokens: int + + +def backlog_path_for(home: Path) -> Path: + """BACKLOG.md under a given home directory.""" + return home / "src" / "todo" / "BACKLOG.md" + + +def note_rows( + notes: list[Note], links: Mapping[str, Link], spend: Mapping[str, Spend] +) -> list[NoteRow]: + """Notes with >=1 linked session; a session's spend splits equally.""" + usd: dict[str, float] = {} + toks: dict[str, float] = {} + sessions: dict[str, list[str]] = {} + for sid in sorted(links): + share = len(links[sid].notes) + cost = spend.get(sid) + for note_id in links[sid].notes: + sessions.setdefault(note_id, []).append(sid) + usd[note_id] = usd.get(note_id, 0.0) + (cost.usd / share if cost else 0.0) + toks[note_id] = toks.get(note_id, 0) + (cost.tokens / share if cost else 0) + rows = [ + NoteRow( + n.id, + n.title, + n.status, + n.priority, + n.done_at, + sessions[n.id], + round(usd[n.id], 4), + round(toks[n.id]), + ) + for n in notes + if n.id in sessions + ] + return sorted(rows, key=lambda r: -r.usd) + + +def _in(stamp: float | None, window: tuple[float, float]) -> bool: + return stamp is not None and window[0] <= stamp <= window[1] + + +def metrics( + notes: list[Note], + rows: list[NoteRow], + window: tuple[float, float], + window_usd: Mapping[str, float], + linked: set[str], +) -> dict[str, float]: + """The snapshot metrics; ``usd_per_note`` is absent when unmeasurable.""" + done = {n.id for n in notes if n.status == "done" and _in(n.done_at, window)} + closed = {n.id for n in notes if n.status == "abandoned" and _in(n.done_at, window)} + linked_done = [r for r in rows if r.id in done] + spent = sum(r.usd for r in rows if r.id in done | closed) + total = sum(window_usd.values()) + linked_usd = sum(v for k, v in window_usd.items() if k in linked) + out: dict[str, float] = { + "note_linked_usd_share": round(linked_usd / total, 4) if total else 0.0, + "notes_done": len(done), + "notes_done_linked": len(linked_done), + } + if linked_done: + out = {"usd_per_note": round(spent / len(linked_done), 2)} | out + return out + + +def _day(stamp: float | None) -> str: + if not stamp: + return "-" + return datetime.fromtimestamp(stamp, tz=UTC).astimezone().strftime("%Y-%m-%d") + + +def _title(text: str) -> str: + text = text.replace("|", "\\|") + return text if len(text) <= TITLE_CHARS else text[: TITLE_CHARS - 1] + "…" + + +def render( + rows: list[NoteRow], found: Mapping[str, float], rules: Mapping[str, int] +) -> list[str]: + """The WEEKLY.md section: metrics, link sources, top closed notes by USD.""" + per_note = found.get("usd_per_note") + lines = [ + "## Cost per todo note", + "", + f"- usd_per_note: {'n/a' if per_note is None else f'${per_note:,.2f}'}" + " (lower is better)", + f"- note_linked_usd_share: {found['note_linked_usd_share'] * 100:.1f}%", + f"- notes_done: {found['notes_done']} " + f"({found.get('notes_done_linked', 0)} with a linked session)", + "- links by rule: " + + " ".join(f"{k}={rules.get(k, 0)}" for k in ("a", "b", "c", "d")), + "", + "| note | status | done | sessions | USD | tokens |", + "|---|---|---|---|---|---|", + ] + lines += [ + f"| {_title(r.title)} | {r.status} | {_day(r.done_at)} | " + f"{len(r.sessions)} | ${r.usd:,.2f} | {r.tokens:,} |" + for r in rows[:TOP] + ] + return lines + + +def write_json(rows: list[NoteRow], out: Path) -> Path: + """Write every linked note for the dashboard.""" + out.mkdir(parents=True, exist_ok=True) + path = out / JSON_NAME + data = {"generated": time.time(), "notes": [asdict(r) for r in rows]} + path.write_text(json.dumps(data, indent=1), encoding="utf-8") + return path + + +def section( + root: Path, + book: pricing.PriceBook, + sessions: Sequence[Session], + window: tuple[float, float], +) -> tuple[dict[str, float], list[str], list[NoteRow]] | None: + """Metrics, WEEKLY.md lines and rows for one audit window.""" + started = time.monotonic() + found = notes_collect.collect(root, book, sessions, window[0]) + if found is None: + return None + rows = note_rows(found.notes, found.links, found.spend) + spent = notes_collect.window_usd(sessions, book) + got = metrics(found.notes, rows, window, spent, set(found.links)) + rules = " ".join(f"{k}={v}" for k, v in sorted(found.rules.items())) + sys.stderr.write( + f"notes: {len(found.links)} sessions linked ({rules}) " + f"in {time.monotonic() - started:.2f}s\n" + ) + closed = [r for r in rows if r.status in TERMINAL and _in(r.done_at, window)] + return got, ["", *render(closed, got, found.rules)], rows + + +def main(argv: list[str] | None = None) -> int: + """Rewrite notes.json from all transcripts, independent of any window.""" + parser = argparse.ArgumentParser(prog="token_audit.notes") + parser.add_argument("--root", type=Path, default=notes_collect.TRANSCRIPTS) + parser.add_argument("--out", type=Path, default=history.REPORT_DIR) + args = parser.parse_args(argv) + found = notes_collect.collect(args.root, pricing.load(), [], float("inf")) + if found is not None: + written = write_json(note_rows(found.notes, found.links, found.spend), args.out) + sys.stderr.write(f"Wrote {written}\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python_pkg/token_audit/notes_backlog.py b/python_pkg/token_audit/notes_backlog.py new file mode 100644 index 00000000..7dd15fd9 --- /dev/null +++ b/python_pkg/token_audit/notes_backlog.py @@ -0,0 +1,145 @@ +"""Todo notes plus their done date: the status HLC in the first dump showing it. + +The app keeps no done timestamp; the BACKLOG parse copies todo_mcp/_backlog.py. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from datetime import UTC, datetime +import gzip +import json +from pathlib import Path +import re +from typing import Any + +STATUSES = ("todo", "inProgress", "done", "abandoned") +TERMINAL = frozenset({"done", "abandoned"}) +_MARKER = re.compile(r"^[ \t]*$", re.MULTILINE) +_ATTR = re.compile(r'(\w+)="([^"]*)"') +_HLC = re.compile(r"^(\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(?:\.\d+)?)Z") +_DUMP_GLOB = "kuhy-syncs-*.json.gz" +_FIELD_LEN = 2 + + +@dataclass(frozen=True) +class Note: + """One backlog note; ``done_at`` is set only for done/abandoned notes.""" + + id: str + title: str + body: str + priority: str + status: str + updated: float | None + done_at: float | None = None + + +def backlog_path() -> Path: + """Where the app's Settings -> Export notes writes BACKLOG.md.""" + return Path.home() / "src" / "todo" / "BACKLOG.md" + + +def dumps_dir() -> Path: + """Where the daily Firebase RTDB backups land.""" + return Path.home() / "data" / "cloud" / "firebase_backups" + + +def _local_time(raw: str | None) -> float | None: + if not raw: + return None + try: + return datetime.fromisoformat(raw).timestamp() + except ValueError: + return None + + +def parse_backlog(text: str) -> list[Note]: + """Every ```` block in an exported BACKLOG.md.""" + markers = list(_MARKER.finditer(text)) + out: list[Note] = [] + for index, marker in enumerate(markers): + attrs = dict(_ATTR.findall(marker.group(1))) + end = markers[index + 1].start() if index + 1 < len(markers) else len(text) + body = text[marker.end() : end].strip() + status = attrs.get("status", "todo") + out.append( + Note( + id=attrs.get("id", ""), + title=body.splitlines()[0].lstrip("#").strip() if body else "", + body=body, + priority=attrs.get("priority", "medium"), + status=status if status in STATUSES else "todo", + updated=_local_time(attrs.get("updated")), + ) + ) + return out + + +def hlc_time(hlc: object) -> float | None: + """The wall-clock part of an HLC string (``Z--``).""" + match = _HLC.match(hlc) if isinstance(hlc, str) else None + if match is None: + return None + return datetime.fromisoformat(match.group(1)).replace(tzinfo=UTC).timestamp() + + +def _status_field(note: object) -> tuple[int, str] | None: + if not isinstance(note, dict) or note.get("deleted"): + return None + field = (note.get("fields") or {}).get("status") + if not isinstance(field, list) or len(field) != _FIELD_LEN: + return None + value, hlc = field + return (value, hlc) if isinstance(value, int) and isinstance(hlc, str) else None + + +def merge_devices(devices: dict[str, Any]) -> dict[str, tuple[int, str]]: + """Last-writer-wins status per note across the per-device documents.""" + merged: dict[str, tuple[int, str]] = {} + for raw in devices.values(): + try: + doc = json.loads(raw) if isinstance(raw, str) else raw + except ValueError: + continue + for note_id, note in (doc if isinstance(doc, dict) else {}).items(): + found = _status_field(note) + if found is not None and ( + note_id not in merged or found[1] > merged[note_id][1] + ): + merged[note_id] = found + return merged + + +def _dump_statuses(path: Path) -> dict[str, tuple[int, str]]: + try: + data = json.loads(gzip.decompress(path.read_bytes())) + except OSError, ValueError, EOFError: + return {} + devices = ((data or {}).get("todo-sync") or {}).get("notes") or {} + return merge_devices(devices) if isinstance(devices, dict) else {} + + +def first_seen(directory: Path) -> dict[str, dict[str, float]]: + """Per note and terminal status: its status HLC in the first dump showing it.""" + seen: dict[str, dict[str, float]] = {} + for path in sorted(directory.glob(_DUMP_GLOB)) if directory.is_dir() else []: + for note_id, (value, hlc) in _dump_statuses(path).items(): + status = STATUSES[value] if 0 <= value < len(STATUSES) else "todo" + stamp = hlc_time(hlc) + if status in TERMINAL and stamp is not None: + seen.setdefault(note_id, {}).setdefault(status, stamp) + return seen + + +def _dated(note: Note, seen: dict[str, dict[str, float]]) -> Note: + if note.status not in TERMINAL: + return note + return replace(note, done_at=seen.get(note.id, {}).get(note.status, note.updated)) + + +def load_notes(backlog: Path, dumps: Path) -> list[Note]: + """BACKLOG notes with ``done_at`` filled for done/abandoned ones.""" + seen = first_seen(dumps) + text = backlog.read_text(encoding="utf-8") + return [_dated(note, seen) for note in parse_backlog(text)] diff --git a/python_pkg/token_audit/notes_collect.py b/python_pkg/token_audit/notes_collect.py new file mode 100644 index 00000000..fd289d4a --- /dev/null +++ b/python_pkg/token_audit/notes_collect.py @@ -0,0 +1,76 @@ +"""Gather notes, session links and linked spend for the notes report.""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import dataclass +from pathlib import Path +import sys +from typing import TYPE_CHECKING + +from python_pkg.token_audit import notes_backlog, notes_cost, notes_link, notes_scan + +if TYPE_CHECKING: + from collections.abc import Sequence + + from python_pkg.token_audit.model import Session + from python_pkg.token_audit.notes_backlog import Note + from python_pkg.token_audit.notes_cost import Spend + from python_pkg.token_audit.notes_link import Link + from python_pkg.token_audit.pricing import PriceBook + +TRANSCRIPTS = Path.home() / ".claude" / "projects" + + +@dataclass(frozen=True) +class Collected: + """Everything the notes metrics are computed from.""" + + notes: list[Note] + links: dict[str, Link] + spend: dict[str, Spend] + + @property + def rules(self) -> Counter[str]: + """How many sessions each rule linked.""" + return Counter(link.rule for link in self.links.values()) + + +def _reusable( + sessions: Sequence[Session], + facts: dict[str, notes_scan.Facts], + since: float, + book: PriceBook, +) -> dict[str, Spend]: + out: dict[str, Spend] = {} + for session in sessions: + fact = facts.get(session.parent_id or session.session_id) + if fact is not None and fact.start >= since: + out[session.path] = notes_cost.of_session(session, book) + return out + + +def collect( + root: Path, book: PriceBook, sessions: Sequence[Session], since: float +) -> Collected | None: + """Notes, links and lifetime spend; ``None`` when BACKLOG.md is missing.""" + backlog = notes_backlog.backlog_path() + if not backlog.exists(): + sys.stderr.write(f"notes: skipped, no {backlog}\n") + return None + notes = notes_backlog.load_notes(backlog, notes_backlog.dumps_dir()) + facts = notes_scan.scan_root(root) + links = notes_link.link_all(facts, notes) + mains = {sid: fact.path for sid, fact in facts.items()} + reuse = _reusable(sessions, facts, since, book) + spend = notes_cost.lifetime(mains, set(links), book, reuse) + return Collected(notes, links, spend) + + +def window_usd(sessions: Sequence[Session], book: PriceBook) -> dict[str, float]: + """In-window USD per main session id, subagents folded into their parent.""" + out: dict[str, float] = {} + for session in sessions: + key = session.parent_id or session.session_id + out[key] = out.get(key, 0.0) + notes_cost.of_session(session, book).usd + return out diff --git a/python_pkg/token_audit/notes_cost.py b/python_pkg/token_audit/notes_cost.py new file mode 100644 index 00000000..36c472f5 --- /dev/null +++ b/python_pkg/token_audit/notes_cost.py @@ -0,0 +1,63 @@ +"""Lifetime USD and tokens of the sessions linked to notes (main + subagents). + +Window-free on purpose: a note finished this week was often worked last week. +Files already loaded whole by the audit are reused instead of re-parsed. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from python_pkg.token_audit import parse, pricing +from python_pkg.token_audit.model import TOKEN_KINDS + +if TYPE_CHECKING: + from pathlib import Path + + from python_pkg.token_audit.model import Session + from python_pkg.token_audit.pricing import PriceBook + + +@dataclass(frozen=True) +class Spend: + """Dollars at list price and raw tokens (all four classes).""" + + usd: float = 0.0 + tokens: int = 0 + + def __add__(self, other: Spend) -> Spend: + """Sum two spends.""" + return Spend(self.usd + other.usd, self.tokens + other.tokens) + + +def of_session(session: Session, book: PriceBook) -> Spend: + """Spend of every turn a loaded session holds.""" + return Spend( + sum(pricing.usd(turn, book) for turn in session.turns), + sum(turn.usage.get(kind, 0) for turn in session.turns for kind in TOKEN_KINDS), + ) + + +def files_of(main: Path) -> list[Path]: + """A main transcript plus every subagent transcript filed under it.""" + return [main, *sorted(main.with_suffix("").glob("subagents/agent-*.jsonl"))] + + +def lifetime( + mains: dict[str, Path], + wanted: set[str], + book: PriceBook, + reuse: dict[str, Spend], +) -> dict[str, Spend]: + """Whole-file spend per wanted session id, reusing ``reuse[path]`` hits.""" + out: dict[str, Spend] = {} + for sid in sorted(wanted & mains.keys()): + total = Spend() + for path in files_of(mains[sid]): + hit = reuse.get(str(path)) + total += ( + hit if hit is not None else of_session(parse.load_session(path), book) + ) + out[sid] = total + return out diff --git a/python_pkg/token_audit/notes_link.py b/python_pkg/token_audit/notes_link.py new file mode 100644 index 00000000..88a484b9 --- /dev/null +++ b/python_pkg/token_audit/notes_link.py @@ -0,0 +1,157 @@ +"""Link sessions to todo notes; the first rule that yields a note decides. + +a: a Bash call ran ``note_claim.sh ``; b: a typed prompt names +``todo-images//``; c: the first prompt fuzzy-matches one note; +d: an ``Implement|Continue `` session inherits its writer's notes. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import functools +import re +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterable + + from python_pkg.token_audit.notes_backlog import Note + from python_pkg.token_audit.notes_scan import Facts + +FUZZY_THRESHOLD = 0.5 +MIN_TOKENS = 4 +MAX_IMAGE_NOTES = 2 +_TOKEN = re.compile(r"[^\W_]{3,}") +_STOP = frozenset( + [ + "the", + "and", + "for", + "that", + "this", + "with", + "you", + "are", + "not", + "but", + "have", + "from", + "what", + "when", + "where", + "should", + "would", + "can", + "will", + "just", + "all", + "any", + "its", + "into", + "also", + "then", + "than", + "them", + "they", + "their", + "there", + "was", + "were", + "been", + "being", + "lets", + "let", + "make", + "made", + "does", + "dont", + ] +) + + +@dataclass(frozen=True) +class Link: + """The notes one session worked on and the rule that found them.""" + + notes: tuple[str, ...] + rule: str + + +@functools.cache +def tokens(text: str) -> frozenset[str]: + """Normalised content words: lower-case, 3+ letters, no stopwords/digits.""" + return frozenset( + t for t in _TOKEN.findall(text.lower()) if t not in _STOP and not t.isdigit() + ) + + +def _dice(left: frozenset[str], right: frozenset[str]) -> float: + total = len(left) + len(right) + return 2 * len(left & right) / total if total else 0.0 + + +def score(prompt: str, note: Note) -> float: + """Body overlap, or the prompt's first line against the note title.""" + body = _dice(tokens(prompt), tokens(note.body)) + title = tokens(note.title) + if len(title) < MIN_TOKENS: + return body + first_line = prompt.strip().splitlines()[0] if prompt.strip() else "" + return max(body, _dice(tokens(first_line), title)) + + +def resolve(token: str, ids: Iterable[str]) -> str | None: + """The one note id ``token`` names (full id or unique prefix).""" + token = token.lower() + found = [i for i in ids if i.lower().startswith(token)] + return found[0] if len(found) == 1 else None + + +def _explicit(facts: Facts, ids: list[str]) -> tuple[str, ...]: + found = (resolve(token, ids) for token in facts.claims) + return tuple(dict.fromkeys(i for i in found if i is not None)) + + +def _images(facts: Facts, ids: set[str]) -> tuple[str, ...]: + known = tuple(i for i in facts.images if i in ids) + return known if len(known) <= MAX_IMAGE_NOTES else () + + +def _fuzzy(facts: Facts, notes: list[Note]) -> tuple[str, ...]: + if facts.handoff or len(tokens(facts.first_prompt)) < MIN_TOKENS: + return () + scored = [(score(facts.first_prompt, n), n.updated or 0.0, n.id) for n in notes] + best = max(scored, default=(0.0, 0.0, "")) + return (best[2],) if best[0] >= FUZZY_THRESHOLD else () + + +def _writers(facts: dict[str, Facts]) -> dict[str, tuple[float, str]]: + first: dict[str, tuple[float, str]] = {} + for sid, fact in facts.items(): + for rel, stamp in fact.creates.items(): + first[rel] = min(first.get(rel, (stamp, sid)), (stamp, sid)) + return first + + +def link_all(facts: dict[str, Facts], notes: list[Note]) -> dict[str, Link]: + """Links per session id, resolved in start order so writers come first.""" + ids = [n.id for n in notes if n.id] + writers = _writers(facts) + links: dict[str, Link] = {} + for sid in sorted(facts, key=lambda s: (facts[s].start, s)): + fact = facts[sid] + for rule, found in ( + ("a", _explicit(fact, ids)), + ("b", _images(fact, set(ids))), + ("c", _fuzzy(fact, notes)), + ): + if found: + links[sid] = Link(found, rule) + break + else: + writer = writers.get(fact.handoff or "") + if writer and writer[1] != sid and writer[0] <= fact.start: + inherited = links.get(writer[1]) + if inherited is not None: + links[sid] = Link(inherited.notes, "d") + return links diff --git a/python_pkg/token_audit/notes_scan.py b/python_pkg/token_audit/notes_scan.py new file mode 100644 index 00000000..5bdd2a65 --- /dev/null +++ b/python_pkg/token_audit/notes_scan.py @@ -0,0 +1,144 @@ +"""Per-session facts the note join needs, read from every main transcript. + +Only typed user prompts and assistant tool_use blocks are read; tool results +are never evidence (a session that reads BACKLOG.md names dozens of ids). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import re +from typing import TYPE_CHECKING, Any + +from python_pkg.token_audit import parse, records + +if TYPE_CHECKING: + from pathlib import Path + +CLAIM = re.compile(r"note_claim\.sh\s+[\"']?([0-9A-Fa-f][0-9A-Fa-f-]{7,35})") +IMAGE = re.compile(r"todo-images/([0-9a-f]{8}-[0-9a-f-]{27})/") +HANDOFF_PROMPT = re.compile(r"^\s*(?:implement|continue)\s+(\S+)", re.IGNORECASE) +HANDOFF_DIR = ".claude/handoff/" +_WRAPPER = re.compile( + r"<(system-reminder|command-[\w-]+|local-command[\w-]*)[^>]*>.*?", re.DOTALL +) +_PASTE_TAG = re.compile(r"]*>") +_BASH_CREATE = re.compile(r"(?:(?&\d])>(?!>)|\btee\s+(?!-a))\s*[\"']?(\S+)") +_SKIP_PREFIX = ("Caveat:",) + + +@dataclass +class Facts: + """What one main session contributes to the join.""" + + session_id: str + path: Path + start: float = 0.0 + first_prompt: str = "" + handoff: str | None = None + claims: list[str] = field(default_factory=list) + images: list[str] = field(default_factory=list) + creates: dict[str, float] = field(default_factory=dict) + + +def handoff_rel(raw: str) -> str | None: + """The path below ``~/.claude/handoff/``, or ``None`` if not under it.""" + _, sep, rel = raw.strip("\"'`").partition(HANDOFF_DIR) + return rel.rstrip(".,;)") or None if sep else None + + +def prompt_text(record: dict[str, Any]) -> str | None: + """Typed prompt text, wrappers removed; ``None`` for anything else.""" + if record.get("type") != "user" or record.get("isSidechain"): + return None + if record.get("isMeta"): + return None + content = (record.get("message") or {}).get("content") + if isinstance(content, list): + if any(b.get("type") == "tool_result" for b in content if isinstance(b, dict)): + return None + content = "\n".join( + str(b.get("text") or "") + for b in content + if isinstance(b, dict) and b.get("type") == "text" + ) + if not isinstance(content, str): + return None + text = _PASTE_TAG.sub("", _WRAPPER.sub("", content)).strip() + return "" if text.startswith(_SKIP_PREFIX) else text + + +def _created_paths(name: str, tool_input: dict[str, Any]) -> list[str]: + if name == "Write": + path = tool_input.get("file_path") + return [path] if isinstance(path, str) else [] + command = tool_input.get("command") + if name != "Bash" or not isinstance(command, str) or HANDOFF_DIR not in command: + return [] + return _BASH_CREATE.findall(command) + + +def _tool_uses(record: dict[str, Any]) -> list[dict[str, Any]]: + if record.get("type") != "assistant": + return [] + content = (record.get("message") or {}).get("content") + blocks = content if isinstance(content, list) else [] + return [b for b in blocks if isinstance(b, dict) and b.get("type") == "tool_use"] + + +def _use(facts: Facts, record: dict[str, Any], *, main: bool) -> None: + stamp = records.timestamp(record) or 0.0 + for block in _tool_uses(record): + tool_input = block.get("input") if isinstance(block.get("input"), dict) else {} + name = str(block.get("name")) + if main and name == "Bash": + facts.claims += CLAIM.findall(str(tool_input.get("command") or "")) + for raw in _created_paths(name, tool_input): + rel = handoff_rel(raw) + if rel is not None and rel not in facts.creates: + facts.creates[rel] = stamp + + +def _prompt(facts: Facts, record: dict[str, Any]) -> None: + text = prompt_text(record) + if not text: + return + facts.images += [i for i in IMAGE.findall(text) if i not in facts.images] + if not facts.first_prompt: + facts.first_prompt = text + match = HANDOFF_PROMPT.match(text) + facts.handoff = handoff_rel(match.group(1)) if match else None + + +def _wanted(line: str) -> bool: + if '"tool_use"' in line: + return "note_claim" in line or HANDOFF_DIR in line + return '"type":"user"' in line and '"tool_result"' not in line + + +def scan(path: Path, *, main: bool = True) -> Facts: + """Read one transcript's prompts, claims and handoff-file creations.""" + facts = Facts(session_id=path.stem, path=path) + with path.open(encoding="utf-8", errors="replace") as handle: + for line in handle: + if not facts.start and '"timestamp"' in line: + facts.start = records.timestamp(parse.load_line(line) or {}) or 0.0 + if not _wanted(line): + continue + record = parse.load_line(line) or {} + _use(facts, record, main=main) + if main: + _prompt(facts, record) + return facts + + +def scan_root(root: Path) -> dict[str, Facts]: + """Facts per main session; subagent creations fold into their parent.""" + found = {p.stem: scan(p) for p in sorted(root.glob("*/*.jsonl"))} + for sub in sorted(root.glob("*/*/subagents/agent-*.jsonl")): + parent = found.get(sub.parent.parent.name) + if parent is None: + continue + for rel, stamp in scan(sub, main=False).creates.items(): + parent.creates[rel] = min(stamp, parent.creates.get(rel, stamp)) + return found diff --git a/python_pkg/token_audit/review.py b/python_pkg/token_audit/review.py index a3ceb536..8c96fd01 100644 --- a/python_pkg/token_audit/review.py +++ b/python_pkg/token_audit/review.py @@ -20,6 +20,7 @@ SNAPSHOT_VERSION = 2 # windows differ in length (the "up 122%" of 2026-10-04 compared 37 days to 7). COMPARED: tuple[str, ...] = ( "usd_per_commit", + "usd_per_note", "usd_per_day", "usd_per_call", "calls_per_day", diff --git a/python_pkg/token_audit/tests/test_notes_backlog.py b/python_pkg/token_audit/tests/test_notes_backlog.py new file mode 100644 index 00000000..8bdb7916 --- /dev/null +++ b/python_pkg/token_audit/tests/test_notes_backlog.py @@ -0,0 +1,94 @@ +"""BACKLOG parsing and done dates recovered from the Firebase dumps.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import TYPE_CHECKING + +from python_pkg.token_audit import notes_backlog +from python_pkg.token_audit.tests import test_notes_fixtures as fx + +if TYPE_CHECKING: + from pathlib import Path + + +def _ts(iso: str) -> float: + return datetime.fromisoformat(iso).replace(tzinfo=UTC).timestamp() + + +def test_parse_reads_ids_status_title(tmp_path: Path) -> None: + text = fx.backlog(tmp_path / "BACKLOG.md").read_text() + notes = notes_backlog.parse_backlog(text) + assert [n.id for n in notes] == [fx.NOTE_A, fx.NOTE_B, fx.NOTE_C, fx.NOTE_D] + assert notes[0].status == "done" + assert notes[0].title == "diet guard meal times picker" + assert notes[1].status == "abandoned" + assert notes[0].updated is not None + + +def test_unknown_status_falls_back_to_todo() -> None: + text = '\nbody' + assert notes_backlog.parse_backlog(text)[0].status == "todo" + + +def test_hlc_time_parses_prefix_and_rejects_junk() -> None: + assert notes_backlog.hlc_time("2026-10-01T10:00:00.000Z-0000-dev") == _ts( + "2026-10-01T10:00:00" + ) + assert notes_backlog.hlc_time("nonsense") is None + assert notes_backlog.hlc_time(None) is None + + +def test_done_at_is_status_hlc_of_first_dump_showing_done(tmp_path: Path) -> None: + dumps = tmp_path / "dumps" + fx.dump(dumps, "2026-10-01T000000Z", {fx.NOTE_A: (1, "2026-09-30T08:00:00.000Z")}) + fx.dump(dumps, "2026-10-02T000000Z", {fx.NOTE_A: (2, "2026-10-01T12:00:00.000Z")}) + # Re-stamped later by an unrelated upsert: must not move the done date. + fx.dump(dumps, "2026-10-03T000000Z", {fx.NOTE_A: (2, "2026-10-02T20:00:00.000Z")}) + notes = notes_backlog.load_notes(fx.backlog(tmp_path / "B.md"), dumps) + done = {n.id: n.done_at for n in notes} + assert done[fx.NOTE_A] == _ts("2026-10-01T12:00:00") + + +def test_device_with_newest_hlc_wins(tmp_path: Path) -> None: + dumps = tmp_path / "dumps" + path = fx.dump(dumps, "2026-10-02T000000Z", {}) + first = notes_backlog.merge_devices( + { + "d1": '{"n": {"fields": {"status": [2, "2026-10-01T00:00:00.000Z-0-d1"]}}}', + "d2": '{"n": {"fields": {"status": [0, "2026-09-01T00:00:00.000Z-0-d2"]}}}', + "d3": '{"gone": {"deleted": true, "fields": {"status": [2, "x"]}}}', + "d4": "not json", + } + ) + assert first == {"n": (2, "2026-10-01T00:00:00.000Z-0-d1")} + assert path.exists() + + +def test_abandoned_dated_from_dump_and_todo_has_no_date(tmp_path: Path) -> None: + dumps = tmp_path / "dumps" + fx.dump(dumps, "2026-10-05T000000Z", {fx.NOTE_B: (3, "2026-10-04T09:00:00.000Z")}) + notes = { + n.id: n for n in notes_backlog.load_notes(fx.backlog(tmp_path / "B.md"), dumps) + } + assert notes[fx.NOTE_B].done_at == _ts("2026-10-04T09:00:00") + assert notes[fx.NOTE_C].done_at is None + + +def test_fallback_to_backlog_updated_when_no_dump_has_it(tmp_path: Path) -> None: + notes = { + n.id: n + for n in notes_backlog.load_notes( + fx.backlog(tmp_path / "B.md"), tmp_path / "none" + ) + } + assert notes[fx.NOTE_D].done_at == notes[fx.NOTE_D].updated + assert notes[fx.NOTE_D].done_at is not None + + +def test_unreadable_dump_is_skipped(tmp_path: Path) -> None: + dumps = tmp_path / "dumps" + dumps.mkdir() + (dumps / "kuhy-syncs-2026-10-01T000000Z.json.gz").write_bytes(b"not gzip") + notes = notes_backlog.load_notes(fx.backlog(tmp_path / "B.md"), dumps) + assert len(notes) == 4 diff --git a/python_pkg/token_audit/tests/test_notes_cli.py b/python_pkg/token_audit/tests/test_notes_cli.py new file mode 100644 index 00000000..0a665d5f --- /dev/null +++ b/python_pkg/token_audit/tests/test_notes_cli.py @@ -0,0 +1,80 @@ +"""Notes metrics wired into the audit CLI, and lifetime session pricing.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from python_pkg.token_audit import __main__ as cli +from python_pkg.token_audit import notes, notes_cost, pricing +from python_pkg.token_audit.tests import test_notes_fixtures as fx + + +def _session(root: Path) -> None: + fx.transcript( + root, + "s1", + [ + fx.prompt(fx.BODY_A), + fx.tool_use("Bash", {"command": "ls"}, msg="m1"), + ], + ) + fx.subagent(root, "s1", "a1", [fx.tool_use("Read", {"file_path": "/x"}, msg="m2")]) + + +def test_lifetime_spend_sums_main_and_subagents(tmp_path: Path) -> None: + root = tmp_path / "projects" + _session(root) + book = pricing.load() + got = notes_cost.lifetime({"s1": root / "proj" / "s1.jsonl"}, {"s1"}, book, {}) + assert got["s1"].tokens == 2000 + assert got["s1"].usd > 0 + reused = notes_cost.lifetime( + {"s1": root / "proj" / "s1.jsonl"}, + {"s1"}, + book, + {str(root / "proj" / "s1.jsonl"): notes_cost.Spend(1.0, 7)}, + ) + assert reused["s1"].tokens == 1007 + + +def test_audit_writes_notes_section_and_json( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + root = tmp_path / "projects" + _session(root) + fx.backlog(notes.backlog_path_for(Path.home())) + out = tmp_path / "out" + code = cli.main(["--root", str(root), "--out", str(out), "--days", "36500"]) + assert code == 0 + assert "## Cost per todo note" in (out / "WEEKLY.md").read_text() + snap = json.loads((out / "weekly.json").read_text()) + assert snap["metrics"]["notes_done"] == 2 # A and D (2026-09-01) + assert snap["metrics"]["note_linked_usd_share"] == 1.0 + assert snap["metrics"]["usd_per_note"] > 0 + data = json.loads((out / "notes.json").read_text()) + assert data["notes"][0]["id"] == fx.NOTE_A + assert "notes:" in capsys.readouterr().err + + +def test_audit_skips_notes_without_backlog( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + root = tmp_path / "projects" + _session(root) + out = tmp_path / "out" + assert cli.main(["--root", str(root), "--out", str(out), "--days", "36500"]) == 0 + assert "notes: skipped" in capsys.readouterr().err + assert not (out / "notes.json").exists() + + +def test_standalone_entry_writes_notes_json(tmp_path: Path) -> None: + root = tmp_path / "projects" + _session(root) + fx.backlog(notes.backlog_path_for(Path.home())) + out = tmp_path / "out" + assert notes.main(["--root", str(root), "--out", str(out)]) == 0 + assert (out / "notes.json").exists() + assert notes.main(["--root", str(tmp_path / "missing"), "--out", str(out)]) == 0 diff --git a/python_pkg/token_audit/tests/test_notes_edges.py b/python_pkg/token_audit/tests/test_notes_edges.py new file mode 100644 index 00000000..17d53fcc --- /dev/null +++ b/python_pkg/token_audit/tests/test_notes_edges.py @@ -0,0 +1,82 @@ +"""Edge cases of the note join that the rule tests do not reach.""" + +from __future__ import annotations + +from pathlib import Path + +from python_pkg.token_audit import ( + notes, + notes_backlog, + notes_collect, + notes_link, + notes_scan, + pricing, +) +from python_pkg.token_audit.model import Session +from python_pkg.token_audit.tests import test_notes_fixtures as fx + +HANDOFF = "/home/kuhy/.claude/handoff/e.md" + + +def test_prompt_text_rejects_tool_results_and_non_text() -> None: + result = fx.tool_result("x") + assert notes_scan.prompt_text(result) is None + empty = fx.prompt("x") + empty["message"]["content"] = None + assert notes_scan.prompt_text(empty) is None + + +def test_bad_backlog_stamp_and_malformed_status_field() -> None: + text = '\nbody' + assert notes_backlog.parse_backlog(text)[0].updated is None + odd = '{"n": {"fields": {"status": [2]}}}' + assert notes_backlog.merge_devices({"d": odd}) == {} + + +def test_non_handoff_writes_and_repeat_writes_are_ignored(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript( + root, + "w1", + [ + fx.tool_use("Write", {"file_path": "/home/kuhy/notes/x.md"}, msg="m1"), + fx.tool_use("Write", {"file_path": HANDOFF}, "2026-10-20T09:00:00Z", "m2"), + fx.tool_use("Write", {"file_path": HANDOFF}, "2026-10-20T10:00:00Z", "m3"), + ], + ) + fx.subagent(root, "orphan", "z", [fx.tool_use("Write", {"file_path": HANDOFF})]) + facts = notes_scan.scan_root(root) + assert set(facts) == {"w1"} + assert list(facts["w1"].creates) == ["e.md"] + + +def test_unlinked_writer_passes_nothing_on(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript( + root, + "w1", + [ + fx.prompt("nothing matching at all here", "2026-10-20T08:00:00Z"), + fx.tool_use("Write", {"file_path": HANDOFF}, "2026-10-20T08:10:00Z"), + ], + ) + fx.transcript( + root, "r1", [fx.prompt(f"Implement {HANDOFF}", "2026-10-20T09:00:00Z")] + ) + notes_list = notes_backlog.parse_backlog(fx.backlog(tmp_path / "B.md").read_text()) + assert notes_link.link_all(notes_scan.scan_root(root), notes_list) == {} + + +def test_sessions_started_before_window_are_not_reused(tmp_path: Path) -> None: + root = tmp_path / "projects" + path = fx.transcript(root, "s1", [fx.prompt("x", "2026-10-01T00:00:00Z")]) + facts = notes_scan.scan_root(root) + sessions = [Session("s1", str(path)), Session("ghost", "/nowhere")] + reuse = notes_collect._reusable(sessions, facts, 2e9, pricing.load()) + assert reuse == {} + + +def test_standalone_entry_without_backlog_writes_nothing(tmp_path: Path) -> None: + out = tmp_path / "out" + assert notes.main(["--root", str(tmp_path), "--out", str(out)]) == 0 + assert not out.exists() diff --git a/python_pkg/token_audit/tests/test_notes_fixtures.py b/python_pkg/token_audit/tests/test_notes_fixtures.py new file mode 100644 index 00000000..71361ebe --- /dev/null +++ b/python_pkg/token_audit/tests/test_notes_fixtures.py @@ -0,0 +1,150 @@ +"""Builders for fixture transcripts, BACKLOG files and Firebase dumps.""" + +from __future__ import annotations + +import gzip +import json +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from pathlib import Path + +NOTE_A = "aaaaaaaa-1111-4111-8111-000000000001" +NOTE_B = "bbbbbbbb-2222-4222-8222-000000000002" +NOTE_C = "cccccccc-3333-4333-8333-000000000003" +NOTE_D = "dddddddd-4444-4444-8444-000000000004" + +BODY_A = ( + "diet guard meal times picker\n\n" + "allow user to pick meal hours in intervals of fifteen minutes " + "instead of typing them, remember the last chosen catering option" +) +BODY_B = ( + "spellcode audio from first frame\n\n" + "play hub music and ambience immediately when the godot game opens, " + "no silent splash screen, crossfade between rooms" +) +BODY_C = "book guard volume buttons\n\nvolume keys turn reader pages forward backward" +BODY_D = "horror game" + + +def backlog(path: Path, notes: list[tuple[str, str, str, str]] | None = None) -> Path: + """Write a BACKLOG.md with ``(id, status, updated, body)`` notes.""" + notes = notes or [ + (NOTE_A, "done", "2026-10-20T10:00:00.000", BODY_A), + (NOTE_B, "abandoned", "2026-10-21T10:00:00.000", BODY_B), + (NOTE_C, "todo", "2026-10-22T10:00:00.000", BODY_C), + (NOTE_D, "done", "2026-09-01T10:00:00.000", BODY_D), + ] + path.parent.mkdir(parents=True, exist_ok=True) + parts = ["", ""] + for note_id, status, updated, body in notes: + parts += [ + f'', + body, + "", + ] + path.write_text("\n".join(parts), encoding="utf-8") + return path + + +def dump(directory: Path, stamp: str, statuses: dict[str, tuple[int, str]]) -> Path: + """Write one ``kuhy-syncs-.json.gz`` with ``id -> (status, hlc)``.""" + directory.mkdir(parents=True, exist_ok=True) + device = { + note_id: { + "id": note_id, + "fields": {"status": [status, f"{hlc}-0000-dev1"]}, + "deleted": False, + "deleted_hlc": None, + } + for note_id, (status, hlc) in statuses.items() + } + data = {"todo-sync": {"notes": {"dev1~2Ejson": json.dumps(device)}}} + path = directory / f"kuhy-syncs-{stamp}.json.gz" + path.write_bytes(gzip.compress(json.dumps(data).encode())) + return path + + +def prompt(text: str, stamp: str = "2026-10-20T09:00:00Z") -> dict[str, Any]: + """A typed user prompt record.""" + return { + "type": "user", + "timestamp": stamp, + "cwd": "/home/kuhy/demo", + "message": {"role": "user", "content": text}, + } + + +def tool_use( + name: str, + tool_input: dict[str, Any], + stamp: str = "2026-10-20T09:01:00Z", + msg: str = "m1", + usd_tokens: int = 1000, +) -> dict[str, Any]: + """An assistant record carrying one tool_use and a usage block.""" + return { + "type": "assistant", + "timestamp": stamp, + "message": { + "id": msg, + "model": "claude-opus-5-5", + "usage": {"input_tokens": 0, "output_tokens": usd_tokens}, + "content": [ + { + "type": "tool_use", + "id": f"t-{msg}", + "name": name, + "input": tool_input, + } + ], + }, + } + + +def tool_result(text: str, msg: str = "m1") -> dict[str, Any]: + """A tool_result record (not a prompt).""" + return { + "type": "user", + "timestamp": "2026-10-20T09:02:00Z", + "message": { + "content": [ + {"type": "tool_result", "tool_use_id": f"t-{msg}", "content": text} + ] + }, + } + + +def _write(path: Path, records: list[dict[str, Any]]) -> None: + lines = (json.dumps(r, separators=(",", ":")) for r in records) + path.write_text("\n".join(lines), encoding="utf-8") + + +def transcript( + root: Path, session: str, records: list[dict[str, Any]], slug: str = "proj" +) -> Path: + """Write ``//.jsonl``.""" + project = root / slug + project.mkdir(parents=True, exist_ok=True) + path = project / f"{session}.jsonl" + path.write_text( + "\n".join(json.dumps(r, separators=(",", ":")) for r in records), + encoding="utf-8", + ) + return path + + +def subagent( + root: Path, session: str, agent: str, records: list[dict[str, Any]] +) -> Path: + """Write ``/proj//subagents/agent-.jsonl``.""" + directory = root / "proj" / session / "subagents" + directory.mkdir(parents=True, exist_ok=True) + path = directory / f"agent-{agent}.jsonl" + path.write_text( + "\n".join(json.dumps(r, separators=(",", ":")) for r in records), + encoding="utf-8", + ) + return path diff --git a/python_pkg/token_audit/tests/test_notes_link_explicit.py b/python_pkg/token_audit/tests/test_notes_link_explicit.py new file mode 100644 index 00000000..49981596 --- /dev/null +++ b/python_pkg/token_audit/tests/test_notes_link_explicit.py @@ -0,0 +1,92 @@ +"""Rule (a): a Bash tool_use running note_claim.sh links the session.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from python_pkg.token_audit import notes_backlog, notes_link, notes_scan +from python_pkg.token_audit.tests import test_notes_fixtures as fx + +if TYPE_CHECKING: + from pathlib import Path + + +def _notes(tmp_path: Path) -> list[notes_backlog.Note]: + return notes_backlog.parse_backlog(fx.backlog(tmp_path / "B.md").read_text()) + + +def test_claim_with_prefix_links_note(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript( + root, + "s1", + [ + fx.prompt("do the thing"), + fx.tool_use( + "Bash", {"command": "~/.claude/scripts/note_claim.sh bbbbbbbb"} + ), + ], + ) + links = notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + assert links["s1"] == notes_link.Link((fx.NOTE_B,), "a") + + +def test_claim_beats_image_and_fuzzy(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript( + root, + "s1", + [ + fx.prompt(f"{fx.BODY_A}\n![x](todo-images/{fx.NOTE_C}/a.png)"), + fx.tool_use("Bash", {"command": f"note_claim.sh {fx.NOTE_B}"}), + ], + ) + links = notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + assert links["s1"].rule == "a" + assert links["s1"].notes == (fx.NOTE_B,) + + +def test_unknown_or_ambiguous_claim_is_ignored(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript( + root, + "s1", + [ + fx.prompt("hello there general"), + fx.tool_use("Bash", {"command": "note_claim.sh ffffffff"}), + ], + ) + notes = [ + *_notes(tmp_path), + notes_backlog.Note("aaaaaaaa-9999", "dup", "dup", "low", "todo", None), + ] + facts = notes_scan.scan_root(root) + assert "s1" not in notes_link.link_all(facts, notes) + assert notes_link.resolve("aaaaaaaa", [n.id for n in notes]) is None + + +def test_claim_in_tool_result_is_not_a_link(tmp_path: Path) -> None: + root = tmp_path / "projects" + ids = f"{fx.NOTE_A} {fx.NOTE_B} {fx.NOTE_C}" + fx.transcript( + root, + "s1", + [ + fx.prompt("read my backlog please now"), + fx.tool_use("Read", {"file_path": "/home/kuhy/src/todo/BACKLOG.md"}), + fx.tool_result(f"note_claim.sh {fx.NOTE_A} {ids}"), + ], + ) + assert notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) == {} + + +def test_claim_inside_subagent_does_not_count(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript(root, "s1", [fx.prompt("hello there general")]) + fx.subagent( + root, + "s1", + "x", + [fx.tool_use("Bash", {"command": f"note_claim.sh {fx.NOTE_A}"})], + ) + assert notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) == {} diff --git a/python_pkg/token_audit/tests/test_notes_link_fuzzy.py b/python_pkg/token_audit/tests/test_notes_link_fuzzy.py new file mode 100644 index 00000000..0a6988a6 --- /dev/null +++ b/python_pkg/token_audit/tests/test_notes_link_fuzzy.py @@ -0,0 +1,68 @@ +"""Rule (c): the first real prompt fuzzy-matched to one note's text.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from python_pkg.token_audit import notes_backlog, notes_link, notes_scan +from python_pkg.token_audit.tests import test_notes_fixtures as fx + +if TYPE_CHECKING: + from pathlib import Path + + +def _notes(tmp_path: Path) -> list[notes_backlog.Note]: + return notes_backlog.parse_backlog(fx.backlog(tmp_path / "B.md").read_text()) + + +def _link(tmp_path: Path, *texts: str) -> notes_link.Link | None: + root = tmp_path / "projects" + fx.transcript(root, "s1", [fx.prompt(t) for t in texts]) + return notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)).get("s1") + + +def test_pasted_note_body_links(tmp_path: Path) -> None: + pasted = f'\n{fx.BODY_B}\n\nplease' + assert _link(tmp_path, pasted) == notes_link.Link((fx.NOTE_B,), "c") + + +def test_title_only_prompt_links_by_first_line(tmp_path: Path) -> None: + link = _link(tmp_path, "diet guard meal times picker") + assert link == notes_link.Link((fx.NOTE_A,), "c") + + +def test_unrelated_prompt_does_not_link(tmp_path: Path) -> None: + assert _link(tmp_path, "restart the steam backlog enforcer while game runs") is None + + +def test_only_first_prompt_is_matched(tmp_path: Path) -> None: + assert _link(tmp_path, "unrelated opening words here now", fx.BODY_A) is None + + +def test_short_prompt_is_never_matched(tmp_path: Path) -> None: + assert _link(tmp_path, "horror game") is None + + +def test_wrappers_and_caveats_are_skipped(tmp_path: Path) -> None: + link = _link( + tmp_path, + "/clear", + "Caveat: the messages below were generated by the user", + f"noise{fx.BODY_A}", + ) + assert link == notes_link.Link((fx.NOTE_A,), "c") + + +def test_list_content_prompt_is_read(tmp_path: Path) -> None: + root = tmp_path / "projects" + record = fx.prompt("") + record["message"]["content"] = [{"type": "text", "text": fx.BODY_B}] + fx.transcript(root, "s1", [record]) + links = notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + assert links["s1"].notes == (fx.NOTE_B,) + + +def test_score_is_symmetric_dice_or_title() -> None: + note = notes_backlog.Note("n", "alpha beta gamma delta", "x", "low", "todo", None) + assert notes_link.score("alpha beta gamma delta", note) == 1.0 + assert notes_link.score("zeta eta theta iota", note) == 0.0 diff --git a/python_pkg/token_audit/tests/test_notes_link_handoff.py b/python_pkg/token_audit/tests/test_notes_link_handoff.py new file mode 100644 index 00000000..7a1bf352 --- /dev/null +++ b/python_pkg/token_audit/tests/test_notes_link_handoff.py @@ -0,0 +1,122 @@ +"""Rule (d): an Implement/Continue handoff session inherits its writer's notes.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from python_pkg.token_audit import notes_backlog, notes_link, notes_scan +from python_pkg.token_audit.tests import test_notes_fixtures as fx + +if TYPE_CHECKING: + from pathlib import Path + +HANDOFF = "/home/kuhy/.claude/handoff/sub/20261020-x.md" +HANDOFF2 = "/home/kuhy/.claude/handoff/20261021-y.md" + + +def _notes(tmp_path: Path) -> list[notes_backlog.Note]: + return notes_backlog.parse_backlog(fx.backlog(tmp_path / "B.md").read_text()) + + +def _writer(root: Path) -> None: + fx.transcript( + root, + "w1", + [ + fx.prompt(fx.BODY_A, "2026-10-20T08:00:00Z"), + fx.tool_use( + "Write", + {"file_path": HANDOFF, "content": "x"}, + "2026-10-20T08:30:00Z", + ), + ], + ) + + +def test_continuation_inherits_writer_notes(tmp_path: Path) -> None: + root = tmp_path / "projects" + _writer(root) + fx.transcript( + root, "r1", [fx.prompt(f"Implement {HANDOFF}", "2026-10-20T09:00:00Z")] + ) + links = notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + assert links["w1"].rule == "c" + assert links["r1"] == notes_link.Link((fx.NOTE_A,), "d") + + +def test_chain_follows_through_bash_heredoc_writer(tmp_path: Path) -> None: + root = tmp_path / "projects" + _writer(root) + fx.transcript( + root, + "r1", + [ + fx.prompt(f"Implement {HANDOFF}", "2026-10-20T09:00:00Z"), + fx.tool_use( + "Bash", + {"command": "cat > ~/.claude/handoff/20261021-y.md <<'EOF'\nx\nEOF"}, + "2026-10-20T10:00:00Z", + ), + ], + ) + fx.transcript( + root, "r2", [fx.prompt(f"continue {HANDOFF2}", "2026-10-20T11:00:00Z")] + ) + links = notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + assert links["r2"] == notes_link.Link((fx.NOTE_A,), "d") + + +def test_subagent_writer_maps_to_parent(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript(root, "w1", [fx.prompt(fx.BODY_B, "2026-10-20T08:00:00Z")]) + fx.subagent( + root, + "w1", + "a1", + [fx.tool_use("Write", {"file_path": HANDOFF}, "2026-10-20T08:10:00Z")], + ) + fx.transcript( + root, "r1", [fx.prompt(f"Implement {HANDOFF}", "2026-10-20T09:00:00Z")] + ) + links = notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + assert links["r1"] == notes_link.Link((fx.NOTE_B,), "d") + + +def test_appender_is_not_the_writer_and_unknown_handoff_unlinked( + tmp_path: Path, +) -> None: + root = tmp_path / "projects" + fx.transcript( + root, + "x1", + [ + fx.prompt(fx.BODY_B, "2026-10-20T07:00:00Z"), + fx.tool_use( + "Bash", {"command": f"cat >> {HANDOFF} <<'EOF'"}, "2026-10-20T07:10:00Z" + ), + ], + ) + fx.transcript( + root, "r1", [fx.prompt(f"Implement {HANDOFF}", "2026-10-20T09:00:00Z")] + ) + assert "r1" not in notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + + +def test_write_after_reader_started_is_ignored(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript( + root, "r1", [fx.prompt(f"Implement {HANDOFF}", "2026-10-20T07:00:00Z")] + ) + _writer(root) + assert "r1" not in notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + + +def test_handoff_prompt_is_not_fuzzy_matched(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript( + root, + "r1", + [fx.prompt(f"Implement /home/kuhy/.claude/handoff/{fx.BODY_A.split()[0]}.md")], + ) + assert notes_scan.scan_root(root)["r1"].handoff == f"{fx.BODY_A.split()[0]}.md" + assert "r1" not in notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) diff --git a/python_pkg/token_audit/tests/test_notes_link_images.py b/python_pkg/token_audit/tests/test_notes_link_images.py new file mode 100644 index 00000000..ea8f3ac0 --- /dev/null +++ b/python_pkg/token_audit/tests/test_notes_link_images.py @@ -0,0 +1,75 @@ +"""Rule (b): a typed user prompt naming todo-images// links the note.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from python_pkg.token_audit import notes_backlog, notes_link, notes_scan +from python_pkg.token_audit.tests import test_notes_fixtures as fx + +if TYPE_CHECKING: + from pathlib import Path + + +def _notes(tmp_path: Path) -> list[notes_backlog.Note]: + return notes_backlog.parse_backlog(fx.backlog(tmp_path / "B.md").read_text()) + + +def test_image_path_in_later_prompt_links(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript( + root, + "s1", + [ + fx.prompt("something unrelated entirely here"), + fx.prompt(f"see ~/data/todo-images/{fx.NOTE_C}/shot.png"), + ], + ) + links = notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + assert links["s1"] == notes_link.Link((fx.NOTE_C,), "b") + + +def test_image_only_in_tool_result_does_not_link(tmp_path: Path) -> None: + root = tmp_path / "projects" + fx.transcript( + root, + "s1", + [ + fx.prompt("something unrelated entirely here"), + fx.tool_use("Bash", {"command": "ls"}), + fx.tool_result(f"todo-images/{fx.NOTE_C}/shot.png"), + ], + ) + assert notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) == {} + + +def test_three_distinct_images_look_like_an_export(tmp_path: Path) -> None: + root = tmp_path / "projects" + refs = " ".join(f"todo-images/{n}/x.png" for n in (fx.NOTE_A, fx.NOTE_B, fx.NOTE_C)) + fx.transcript(root, "s1", [fx.prompt(f"zzz qqq www {refs}")]) + assert notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) == {} + + +def test_two_images_split_and_unknown_uuid_dropped(tmp_path: Path) -> None: + root = tmp_path / "projects" + unknown = "eeeeeeee-0000-4000-8000-000000000000" + fx.transcript( + root, + "s1", + [ + fx.prompt( + f"zzz todo-images/{fx.NOTE_A}/x.png todo-images/{fx.NOTE_C}/y.png " + f"todo-images/{unknown}/z.png" + ) + ], + ) + links = notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) + assert links["s1"] == notes_link.Link((fx.NOTE_A, fx.NOTE_C), "b") + + +def test_meta_and_sidechain_prompts_are_ignored(tmp_path: Path) -> None: + root = tmp_path / "projects" + meta = fx.prompt(f"todo-images/{fx.NOTE_C}/x.png") | {"isMeta": True} + side = fx.prompt(f"todo-images/{fx.NOTE_C}/x.png") | {"isSidechain": True} + fx.transcript(root, "s1", [meta, side, fx.prompt("plain words only here")]) + assert notes_link.link_all(notes_scan.scan_root(root), _notes(tmp_path)) == {} diff --git a/python_pkg/token_audit/tests/test_notes_metrics.py b/python_pkg/token_audit/tests/test_notes_metrics.py new file mode 100644 index 00000000..d3d5b97e --- /dev/null +++ b/python_pkg/token_audit/tests/test_notes_metrics.py @@ -0,0 +1,112 @@ +"""USD per completed note: the split, the window rules and the outputs.""" + +from __future__ import annotations + +from datetime import UTC, datetime +import json +from typing import TYPE_CHECKING + +from python_pkg.token_audit import notes, notes_backlog, notes_cost +from python_pkg.token_audit.notes_link import Link +from python_pkg.token_audit.tests import test_notes_fixtures as fx + +if TYPE_CHECKING: + from pathlib import Path + +NOTE_E = "eeeeeeee-5555-4555-8555-000000000005" + + +def _ts(iso: str) -> float: + return datetime.fromisoformat(iso).replace(tzinfo=UTC).timestamp() + + +def _note(note_id: str, status: str, done: str | None) -> notes_backlog.Note: + stamp = _ts(done) if done else None + title = f"title of {note_id[:4]} " + "x" * 80 + return notes_backlog.Note(note_id, title, title, "high", status, stamp, stamp) + + +NOTES = [ + _note(fx.NOTE_A, "done", "2026-10-20T00:00:00"), + _note(fx.NOTE_B, "abandoned", "2026-10-21T00:00:00"), + _note(fx.NOTE_C, "todo", None), + _note(fx.NOTE_D, "done", "2026-09-01T00:00:00"), + _note(NOTE_E, "done", "2026-10-22T00:00:00"), +] +LINKS = { + "s1": Link((fx.NOTE_A,), "a"), + "s2": Link((fx.NOTE_A, fx.NOTE_B), "b"), + "s3": Link((fx.NOTE_C,), "c"), +} +SPEND = { + "s1": notes_cost.Spend(10.0, 1000), + "s2": notes_cost.Spend(4.0, 400), + "s3": notes_cost.Spend(5.0, 500), +} +WINDOW = (_ts("2026-10-15T00:00:00"), _ts("2026-10-25T00:00:00")) + + +def test_rows_split_session_spend_equally() -> None: + rows = {r.id: r for r in notes.note_rows(NOTES, LINKS, SPEND)} + assert rows[fx.NOTE_A].usd == 12.0 + assert rows[fx.NOTE_A].tokens == 1200 + assert rows[fx.NOTE_A].sessions == ["s1", "s2"] + assert rows[fx.NOTE_B].usd == 2.0 + assert NOTE_E not in rows + + +def test_metrics_window_math() -> None: + rows = notes.note_rows(NOTES, LINKS, SPEND) + window_usd = {"s1": 10.0, "s2": 4.0, "s3": 5.0, "s4": 1.0} + got = notes.metrics(NOTES, rows, WINDOW, window_usd, set(LINKS)) + # (A 12 + abandoned B 2) / done-and-linked {A}; E is done but unlinked. + assert got == { + "usd_per_note": 14.0, + "note_linked_usd_share": 0.95, + "notes_done": 2, + "notes_done_linked": 1, + } + + +def test_no_linked_done_note_leaves_usd_per_note_unmeasured() -> None: + got = notes.metrics(NOTES[2:], [], WINDOW, {}, set()) + assert "usd_per_note" not in got + assert got["note_linked_usd_share"] == 0.0 + assert got["notes_done"] == 1 + + +def test_render_has_metrics_and_truncated_table() -> None: + rows = notes.note_rows(NOTES, LINKS, SPEND) + got = notes.metrics(NOTES, rows, WINDOW, {"s1": 1.0}, set(LINKS)) + text = "\n".join(notes.render(rows, got, {"a": 1, "c": 2})) + assert "## Cost per todo note" in text + assert "usd_per_note" in text + assert "a=1" in text + assert f"title of {fx.NOTE_A[:4]} " + "x" * 45 + "…" in text + assert "| abandoned |" in text + + +def test_render_without_measurement_says_so() -> None: + text = "\n".join( + notes.render([], {"notes_done": 0, "note_linked_usd_share": 0.0}, {}) + ) + assert "usd_per_note: n/a" in text + + +def test_write_json_lists_every_linked_note(tmp_path: Path) -> None: + rows = notes.note_rows(NOTES, LINKS, SPEND) + path = notes.write_json(rows, tmp_path) + data = json.loads(path.read_text()) + assert {r["id"] for r in data["notes"]} == {fx.NOTE_A, fx.NOTE_B, fx.NOTE_C} + first = data["notes"][0] + assert set(first) == { + "id", + "title", + "status", + "priority", + "done_at", + "sessions", + "usd", + "tokens", + } + assert first["usd"] == 12.0 diff --git a/python_pkg/token_dashboard/ab.py b/python_pkg/token_dashboard/ab.py index 14588f60..c009a689 100644 --- a/python_pkg/token_dashboard/ab.py +++ b/python_pkg/token_dashboard/ab.py @@ -1,22 +1,12 @@ -"""A/B test of judgment subagents: Sonnet arm vs Opus arm, 0 tokens. +"""A/B test of subagents: Sonnet arm vs Opus arm, one verdict per log. -Assignments come from ``~/.claude/logs/agent-ab.jsonl`` (one line per -dispatched review/debug/design/plan agent: ``tool_use_id``, ``arm``, ...). -Each is joined on ``tool_use_id`` to the ``Agent`` call the ingester stored, -and judged on outcomes visible in the transcript after the agent returned: - -* **redo** -- the parent dispatched another Agent in the same session whose - description shares >= :data:`REDO_JACCARD` of its word set; -* **follow-up** -- the parent sent that agent a ``SendMessage``; -* **error** -- the call errored or the agent was aborted; -* **correction** (noisy) -- the next user prompt opens with a correction - phrase (:data:`CORRECTION`); -* the parent session's end tag. - -Rates get 95% Wilson intervals and per-arm differences Newcombe's hybrid -interval. The banner turns RED only when, with >= :data:`MIN_N` per arm, the -Sonnet arm's redo or correction rate is higher and the two Wilson intervals do -not overlap. +Logs: ``agent-ab.jsonl`` (judgment, closed), ``agent-ab-build.jsonl`` (build). +Assignments join on ``tool_use_id`` to the stored ``Agent`` call: **redo** (a +later Agent shares >= :data:`REDO_JACCARD` of its words), **follow-up** (a +``SendMessage`` to it), **error**, **correction** (noisy: next prompt matches +:data:`CORRECTION`), end tag. Wilson / Newcombe 95% intervals; the banner is +RED only when, with >= :data:`MIN_N` per arm, Sonnet's redo or correction rate +is higher and the two Wilson intervals do not overlap. """ from __future__ import annotations @@ -34,6 +24,7 @@ if TYPE_CHECKING: import sqlite3 LOG = Path.home() / ".claude" / "logs" / "agent-ab.jsonl" +BUILD_LOG = LOG.with_name("agent-ab-build.jsonl") # build-class arm, own verdict Z = 1.96 MIN_N = 20 REDO_JACCARD = 0.6 @@ -45,6 +36,7 @@ CORRECTION = re.compile( ) WORD = re.compile(r"[a-z0-9]+") RATES = ("redo", "followup", "error", "correction") +_UPSERT = "INSERT OR REPLACE INTO meta VALUES (?, ?)" def wilson(k: int, n: int, z: float = Z) -> tuple[float, float]: @@ -90,11 +82,13 @@ class ArmCounts: correction: int -def banner(sonnet: ArmCounts, opus: ArmCounts) -> dict[str, str]: +def banner( + sonnet: ArmCounts, opus: ArmCounts, label: str = "judgment" +) -> dict[str, str]: """RED / GREEN / GREY verdict for the top of every /tokens view.""" n = min(sonnet.n, opus.n) if n < MIN_N: - return {"level": "grey", "text": f"A/B judgment agents: collecting (n={n})"} + return {"level": "grey", "text": f"A/B {label} agents: collecting (n={n})"} for name in ("redo", "correction"): ks, ko = getattr(sonnet, name), getattr(opus, name) low_s, _ = wilson(ks, sonnet.n) @@ -107,7 +101,7 @@ def banner(sonnet: ArmCounts, opus: ArmCounts) -> dict[str, str]: } return { "level": "green", - "text": f"A/B judgment agents: no detectable drop (n={n})", + "text": f"A/B {label} agents: no detectable drop (n={n})", } @@ -198,19 +192,20 @@ def _arm(rows: list[dict[str, Any]]) -> dict[str, Any]: return out -def rebuild(conn: sqlite3.Connection, log: Path | None = None) -> dict[str, Any]: - """Join every assignment, summarise both arms, store in ``meta['ab']``.""" +def _summary( + conn: sqlite3.Connection, log: Path, label: str +) -> tuple[dict[str, Any], list[tuple[str, str, dict[str, Any]]]]: + """Join one log's assignments; the per-arm payload and the joined rows.""" joined: dict[str, list[dict[str, Any]]] = {"sonnet": [], "opus": []} kept: list[tuple[str, str, dict[str, Any]]] = [] unjoined = 0 - for row in _assignments(log or LOG): + for row in _assignments(log): found = _outcome(conn, str(row["tool_use_id"])) if found is None or row["arm"] not in joined: unjoined += 1 continue joined[row["arm"]].append(found) kept.append((row["arm"], str(row["tool_use_id"]), found)) - ab_rows.store(conn, kept) arms = {name: _arm(rows) for name, rows in joined.items()} diffs = { name: dict( @@ -237,7 +232,18 @@ def rebuild(conn: sqlite3.Connection, log: Path | None = None) -> dict[str, Any] "arms": arms, "diffs": diffs, "unjoined": unjoined, - "banner": banner(counts["sonnet"], counts["opus"]), + "banner": banner(counts["sonnet"], counts["opus"], label), } - conn.execute("INSERT OR REPLACE INTO meta VALUES ('ab', ?)", (json.dumps(payload),)) + return payload, kept + + +def rebuild( + conn: sqlite3.Connection, log: Path | None = None, build_log: Path | None = None +) -> dict[str, Any]: + """Store the judgment A/B in ``meta['ab']``, the build A/B in ``ab_build``.""" + payload, kept = _summary(conn, log or LOG, "judgment") + ab_rows.store(conn, kept) + build, _ = _summary(conn, build_log or BUILD_LOG, "build") + for key, value in (("ab", payload), ("ab_build", build)): + conn.execute(_UPSERT, (key, json.dumps(value))) return payload diff --git a/python_pkg/token_dashboard/tests/conftest.py b/python_pkg/token_dashboard/tests/conftest.py index 398e6232..9106fb53 100644 --- a/python_pkg/token_dashboard/tests/conftest.py +++ b/python_pkg/token_dashboard/tests/conftest.py @@ -30,6 +30,7 @@ def _isolated_sources(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(aux, "PROMPT_HISTORY", claude / "history.jsonl") monkeypatch.setattr(aux, "TIME_CAP", claude / "time-cap.jsonl") monkeypatch.setattr(ab, "LOG", claude / "agent-ab.jsonl") + monkeypatch.setattr(ab, "BUILD_LOG", claude / "agent-ab-build.jsonl") monkeypatch.setattr(context_files, "CLAUDE_DIR", claude) monkeypatch.setattr(context_files, "SRC_DIR", claude / "src") diff --git a/python_pkg/token_dashboard/tests/test_ab_build.py b/python_pkg/token_dashboard/tests/test_ab_build.py new file mode 100644 index 00000000..cf8bec94 --- /dev/null +++ b/python_pkg/token_dashboard/tests/test_ab_build.py @@ -0,0 +1,69 @@ +"""The build-class A/B (agent-ab-build.jsonl) is summarised on its own.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +from python_pkg.token_dashboard import ab, ingest, schema +from python_pkg.token_dashboard.tests.conftest import assistant, prompt, result, use +from python_pkg.token_dashboard.tests.conftest import write as write_jsonl + +if TYPE_CHECKING: + from pathlib import Path + + +def _store(tmp_path: Path) -> Path: + root = tmp_path / "projects" + write_jsonl( + root / "p" / "s1.jsonl", + [ + prompt("go", 0), + assistant("m1", 1, blocks=[use("t1", "Agent", description="Build x")]), + result("t1", 50, "done", toolUseResult={"agentId": "abc"}), + assistant("m2", 60, blocks=[use("t2", "Agent", description="Review y")]), + result("t2", 70, "done", toolUseResult={"agentId": "def"}), + ], + ) + db = tmp_path / "t.sqlite" + ingest.run(root, db, tmp_path / "report", 30.0) + return db + + +def _log(path: Path, rows: list[dict[str, str]]) -> Path: + path.write_text("\n".join(json.dumps(r) for r in rows), encoding="utf-8") + return path + + +def test_build_log_lands_in_its_own_meta_row(tmp_path: Path) -> None: + conn = schema.connect(_store(tmp_path)) + judgment = _log( + tmp_path / "agent-ab.jsonl", + [{"tool_use_id": "t2", "arm": "opus", "class": "review"}], + ) + build = _log( + tmp_path / "agent-ab-build.jsonl", + [{"tool_use_id": "t1", "arm": "sonnet", "class": "build"}], + ) + ab.rebuild(conn, judgment, build) + stored = { + k: json.loads(v) + for k, v in conn.execute("SELECT key, value FROM meta WHERE key LIKE 'ab%'") + } + rows = conn.execute("SELECT tool_use_id FROM ab_rows").fetchall() + conn.close() + assert stored["ab"]["arms"]["opus"]["n"] == 1 + assert stored["ab"]["arms"]["sonnet"]["n"] == 0 + assert stored["ab_build"]["arms"]["sonnet"]["n"] == 1 + assert stored["ab_build"]["arms"]["opus"]["n"] == 0 + assert stored["ab_build"]["banner"]["text"].startswith("A/B build agents") + assert stored["ab"]["banner"]["text"].startswith("A/B judgment agents") + assert rows == [("t2",)] + + +def test_missing_build_log_stores_an_empty_summary(tmp_path: Path) -> None: + conn = schema.connect(_store(tmp_path)) + ab.rebuild(conn, tmp_path / "none.jsonl", tmp_path / "none-build.jsonl") + raw = conn.execute("SELECT value FROM meta WHERE key = 'ab_build'").fetchone() + conn.close() + assert json.loads(raw[0])["arms"]["sonnet"]["n"] == 0