From a83b5abc27fc295013b2def7dc6b39fb33c234c2 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 25 Jul 2026 16:56:43 +0200 Subject: [PATCH] python_pkg: track session_autopsy, finishing its interrupted refactor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The transcript analyzer behind /compile-candidate lived only in the working tree. Untracked meant unlinted — pre-commit only inspects files handed to it — so nothing had ever checked it, and it had been abandoned mid-refactor with every parse raising AttributeError: '_Accumulator' object has no attribute 'first_timestamp'. The accumulator had been moved onto the grouped records dataclasses, but three increment sites and parse_session still read flat attributes that no longer existed. parse_session now hands over acc.meta and acc.counts directly; only subagent_count comes from elsewhere, since no transcript line carries it. Two modules broke the 500-line limit, so they were split along seams that were already there. signatures.py takes the text-signature helpers — traces.py was importing them from parse.py, making parse an accidental hub, and now neither imports the other. stats.py takes the pure number and time helpers. parse.py 521 -> 458, detectors.py 559 -> 467. The structural lint findings were refactored, not configured away. _counter_candidates took eight arguments; the five naming which counter field to mine and at what thresholds became a frozen _CounterSpec, so both call sites are one line. _extract_skill carried in_span and a line budget as loop variables across thirteen branches; a _SpanBuffer owns that state now, leaving the walk at four. record() took twelve flat kwargs and now takes one override per record block, with session_meta() so a partial meta override keeps the standard timestamp — a bare SessionMeta(slug=...) would have nulled started_at and quietly changed the per-week arithmetic. All 36 call sites rewritten by an AST pass. tests/factories.py moved into conftest.py, where every other package here keeps shared builders and where name-tests-test does not object. 965 tests pass at 100% statement and branch coverage, every hook is green, and no suppression was added anywhere. `candidates` and `traces skill-finish` both run against the real corpus. Co-Authored-By: Claude Opus 5 (1M context) --- .../session-autopsy-package-2026-07-25.json | 18 + .../session-autopsy-package-2026-07-25.json | 49 ++ python_pkg/session_autopsy/__init__.py | 7 + python_pkg/session_autopsy/__main__.py | 8 + python_pkg/session_autopsy/cli.py | 386 +++++++++++++++ python_pkg/session_autopsy/config.py | 30 ++ python_pkg/session_autopsy/detectors.py | 467 ++++++++++++++++++ python_pkg/session_autopsy/parse.py | 458 +++++++++++++++++ python_pkg/session_autopsy/records.py | 328 ++++++++++++ python_pkg/session_autopsy/report.py | 377 ++++++++++++++ python_pkg/session_autopsy/signatures.py | 67 +++ python_pkg/session_autopsy/stats.py | 120 +++++ python_pkg/session_autopsy/store.py | 136 +++++ python_pkg/session_autopsy/tests/__init__.py | 1 + python_pkg/session_autopsy/tests/conftest.py | 218 ++++++++ python_pkg/session_autopsy/tests/test_cli.py | 194 ++++++++ .../session_autopsy/tests/test_detectors.py | 240 +++++++++ .../session_autopsy/tests/test_parse.py | 307 ++++++++++++ .../session_autopsy/tests/test_records.py | 150 ++++++ .../session_autopsy/tests/test_report.py | 197 ++++++++ .../session_autopsy/tests/test_signatures.py | 28 ++ .../session_autopsy/tests/test_store.py | 65 +++ .../session_autopsy/tests/test_traces.py | 266 ++++++++++ python_pkg/session_autopsy/traces.py | 392 +++++++++++++++ 24 files changed, 4509 insertions(+) create mode 100644 docs/superpowers/contracts/session-autopsy-package-2026-07-25.json create mode 100644 docs/superpowers/evidence/session-autopsy-package-2026-07-25.json create mode 100644 python_pkg/session_autopsy/__init__.py create mode 100644 python_pkg/session_autopsy/__main__.py create mode 100644 python_pkg/session_autopsy/cli.py create mode 100644 python_pkg/session_autopsy/config.py create mode 100644 python_pkg/session_autopsy/detectors.py create mode 100644 python_pkg/session_autopsy/parse.py create mode 100644 python_pkg/session_autopsy/records.py create mode 100644 python_pkg/session_autopsy/report.py create mode 100644 python_pkg/session_autopsy/signatures.py create mode 100644 python_pkg/session_autopsy/stats.py create mode 100644 python_pkg/session_autopsy/store.py create mode 100644 python_pkg/session_autopsy/tests/__init__.py create mode 100644 python_pkg/session_autopsy/tests/conftest.py create mode 100644 python_pkg/session_autopsy/tests/test_cli.py create mode 100644 python_pkg/session_autopsy/tests/test_detectors.py create mode 100644 python_pkg/session_autopsy/tests/test_parse.py create mode 100644 python_pkg/session_autopsy/tests/test_records.py create mode 100644 python_pkg/session_autopsy/tests/test_report.py create mode 100644 python_pkg/session_autopsy/tests/test_signatures.py create mode 100644 python_pkg/session_autopsy/tests/test_store.py create mode 100644 python_pkg/session_autopsy/tests/test_traces.py create mode 100644 python_pkg/session_autopsy/traces.py diff --git a/docs/superpowers/contracts/session-autopsy-package-2026-07-25.json b/docs/superpowers/contracts/session-autopsy-package-2026-07-25.json new file mode 100644 index 00000000..aa4b920c --- /dev/null +++ b/docs/superpowers/contracts/session-autopsy-package-2026-07-25.json @@ -0,0 +1,18 @@ +{ + "title": "Track the session-autopsy package", + "objective": "Bring python_pkg/session_autopsy — the transcript analyzer that ranks automation candidates and backs /compile-candidate — under version control. It was authored in an earlier session and left untracked mid-refactor, so it existed only in the working tree, was never linted (hooks only see files handed to them), and its tests were red. Success means the package is committed in a state that passes every repo gate on the first try.", + "acceptance_criteria": [ + "The whole package (11 modules + 11 test modules) is tracked; no store, report or __pycache__ artifacts come with it — its data lives under ~/.claude/autopsy.", + "The interrupted accumulator refactor is finished: parse_session hands over acc.meta and acc.counts instead of rebuilding SessionMeta/ActivityCounts from flat attributes that no longer exist.", + "No module exceeds the repo's 500-line limit.", + "ruff, ruff-format, mypy --strict, pylint, bandit and codespell all pass on every file, with no new per-file-ignores, noqa or type: ignore anywhere.", + "pytest passes with 100% statement and branch coverage on the package.", + "The CLI runs for real: `candidates` lists ranked candidates and `traces skill-finish` renders spans from actual transcripts." + ], + "out_of_scope": [ + "Any behaviour change to the analyzer's heuristics, thresholds or scoring — the detectors' judgment is the author's, and this commit does not touch it.", + "The two files over 500 lines were split, not rewritten: no detector logic moved semantics.", + "The candidate corpus itself (~/.claude/autopsy/REPORT.md and the store) stays outside the repo." + ], + "verifier": "bash ~/.claude/scripts/finish.sh gate python_pkg/session_autopsy/*.py python_pkg/session_autopsy/tests/*.py (exit 0, risk low); python -m pytest python_pkg -q (100% coverage gate); and a real run of `python -m python_pkg.session_autopsy candidates` plus `traces skill-finish`." +} diff --git a/docs/superpowers/evidence/session-autopsy-package-2026-07-25.json b/docs/superpowers/evidence/session-autopsy-package-2026-07-25.json new file mode 100644 index 00000000..8be6c471 --- /dev/null +++ b/docs/superpowers/evidence/session-autopsy-package-2026-07-25.json @@ -0,0 +1,49 @@ +{ + "intent": "Commit python_pkg/session_autopsy, the transcript analyzer behind /compile-candidate, which had been left untracked and mid-refactor: red tests, two files over the 500-line limit, and ten lint findings that no hook had ever seen because pre-commit only inspects files it is given.", + "scope": [ + "python_pkg/session_autopsy/ - 11 modules and 11 test modules, newly tracked", + "The analyzer's heuristics, thresholds and scoring are NOT touched: the detectors' judgment belongs to the session that wrote them", + "The candidate store and REPORT.md stay outside the repo, under ~/.claude/autopsy" + ], + "changes": [ + "Finished the interrupted accumulator refactor. _Accumulator had been moved onto the grouped records dataclasses (self.meta, self.counts), but three increment sites and parse_session still used flat attributes that no longer existed, so every parse raised AttributeError: '_Accumulator' object has no attribute 'first_timestamp'. parse_session now hands over acc.meta and acc.counts directly; only subagent_count is set from the file listing, since no transcript line carries it.", + "Split the two files over the 500-line limit along seams that already existed. signatures.py takes normalize_signature/command_signature — traces.py was already importing them from parse.py, so parse was acting as an accidental hub and now neither module imports the other. stats.py takes the pure number/time helpers (parse_timestamp, per_turn_avg, per_week, mean, mean_pairwise_similarity, slope). parse.py 521 -> 458 lines, detectors.py 559 -> 467.", + "Refactored the two structural lint findings rather than configuring them away. _counter_candidates took 8 arguments; the five describing which counter field to mine at what thresholds became a frozen _CounterSpec, so both call sites are now one line against _ERROR_SPEC/_PROMPT_SPEC. _extract_skill had 13 branches carrying in_span and span_lines as loop variables; a _SpanBuffer now owns the span state and the per-span budget, with _capture_results/_capture_block beside it, leaving the walk at four branches.", + "The record() test builder took 12 flat keyword arguments (PLR0913). Regrouped onto the record's own blocks — file/meta/counts/tokens/obs — with a session_meta() helper so a partial meta override keeps the standard timestamp; a bare SessionMeta(slug=...) would otherwise have nulled started_at and quietly changed the per-week arithmetic. All 36 call sites rewritten by an AST pass so none was missed.", + "tests/factories.py matched neither test_*.py nor the name-tests-test hook's exemptions; merged into conftest.py, which is where every other package here keeps shared builders (see code_tutor/tests/conftest.py), and rewired six test modules.", + "Remaining lint fixed at the source: six over-length lines in report.py, one each in cli.py and traces.py, cli.py's module docstring (it repeated PYTHONPATH=... on every usage line), RUF001/RUF002 ambiguous multiplication signs in two output strings and one docstring, codespell's 'resetted' in a fake prompt (the retry detector keys on 'limit' plus 'reset'/'continue', so the corrected wording exercises the same branch), and FBT003 boolean positionals in test_records.py (the bool is now bound to a named object, keeping the point of the assertion).", + "Added tests for the branch the _SpanBuffer refactor exposed: an end-to-end case for a content block that is neither tool_use nor text, plus direct unit tests for _capture_block's unrecognised-block handling, the in-span invariant, and the per-span budget reset." + ], + "verification": [ + { + "command": "bash ~/.claude/scripts/finish.sh gate /home/kuhy/testsAndMisc python_pkg/session_autopsy/*.py python_pkg/session_autopsy/tests/*.py", + "result": "pass", + "evidence": "exit 0, risk: low. 965 passed; 'Required test coverage of 100.0% reached. Total coverage: 100.00%'. Earlier runs of the same gate returned exit 20 (11 failing tests) and then exit 10 (pylint/ruff/codespell/coverage) — this is the same command going green, not a narrowed one." + }, + { + "command": "python -m pytest python_pkg -q (repo-wide, coverage gate on)", + "result": "pass", + "evidence": "880 passed, TOTAL 100% across 3899 statements and 1126 branches. session_autopsy/traces.py specifically: 180 statements, 100 branches, 100% with no partials." + }, + { + "command": "pre-commit run ruff / ruff-format / mypy / pylint / bandit / codespell / name-tests-test --files ", + "result": "pass", + "evidence": "All pass. Zero suppressions added: no noqa, no type: ignore, no new per-file-ignores in meta/pyproject.toml." + }, + { + "command": "PYTHONPATH=~/testsAndMisc python -m python_pkg.session_autopsy candidates && ... traces skill-finish", + "result": "pass", + "evidence": "candidates lists the real ranked corpus (skill-explain-diff 35x ~2.3M/wk, skill-finish 63x ~2.1M/wk, err-549ef470 528x ~437k/wk ...). traces skill-finish renders real spans from a 2026-07-23 transcript: 'TOOL Skill | finish', 'TOOL Bash | {\"command\": \"git diff --stat\"...}', 'RESULT | ...' — the refactored span walk working on actual data, not only under test." + } + ], + "risks": [ + "This package was authored in another session and may still be edited there. Committing it does not conflict with that work — later edits simply stack on tracked files — but the refactors above changed module boundaries (signatures.py, stats.py) and the record() builder's signature, which a parallel edit would have to rebase onto.", + "The interrupted refactor's intent was inferred from the half-finished state (grouped dataclasses in __init__ and _note_metadata, flat attributes at the call sites). The direction is unambiguous, but the author may have wanted a different destination.", + "record() call sites were rewritten mechanically via AST. The 100% coverage gate and 880 passing tests are the check on that; nothing else asserts on the builder's shape.", + "Two output strings changed cosmetically: the CLI and report tables now print '12x / 3 sessions' rather than '12× / 3 sessions', because RUF001 flags the multiplication sign as ambiguous." + ], + "rollback": [ + "git revert the commit; the package returns to untracked-and-red, so re-check with python -m pytest python_pkg/session_autopsy (expect the AttributeError on parse_session) to confirm the revert restored the prior state rather than a third one.", + "Nothing outside python_pkg/session_autopsy/ and the two docs artifacts is touched, so no other package needs revalidation." + ] +} diff --git a/python_pkg/session_autopsy/__init__.py b/python_pkg/session_autopsy/__init__.py new file mode 100644 index 00000000..b6fc2d0b --- /dev/null +++ b/python_pkg/session_autopsy/__init__.py @@ -0,0 +1,7 @@ +"""Deterministic Claude Code transcript analyzer (zero LLM tokens). + +Parses ``~/.claude/projects/**/*.jsonl`` session transcripts into compact +per-session records, then mines them for automation candidates: skills worth +"compiling" into deterministic scripts, repeated command sequences, repeated +errors worth a fix-once script, and repeated typed prompts. +""" diff --git a/python_pkg/session_autopsy/__main__.py b/python_pkg/session_autopsy/__main__.py new file mode 100644 index 00000000..79b51a22 --- /dev/null +++ b/python_pkg/session_autopsy/__main__.py @@ -0,0 +1,8 @@ +"""Module entry point: ``python -m python_pkg.session_autopsy``.""" + +from __future__ import annotations + +from python_pkg.session_autopsy.cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/python_pkg/session_autopsy/cli.py b/python_pkg/session_autopsy/cli.py new file mode 100644 index 00000000..d7880b98 --- /dev/null +++ b/python_pkg/session_autopsy/cli.py @@ -0,0 +1,386 @@ +"""Command-line entry point for the session autopsy analyzer. + +Usage, with ``PYTHONPATH=~/testsAndMisc`` and ``python3 -m +python_pkg.session_autopsy``: + + ingest + scan --jobs 8 + report [--mark-reviewed] + traces skill-finish +""" + +from __future__ import annotations + +import argparse +import concurrent.futures +from datetime import datetime, timezone +import json +from pathlib import Path +import sys +import time +from typing import TYPE_CHECKING + +from python_pkg.session_autopsy import config, detectors, report, store, traces +from python_pkg.session_autopsy.parse import parse_session + +if TYPE_CHECKING: + from python_pkg.session_autopsy.records import SessionRecord + +DEFAULT_SCAN_JOBS = 4 +DEFAULT_TRACE_INVOCATIONS = 5 +EXIT_NO_FILE = 2 +EXIT_UNKNOWN_ID = 3 + + +def _emit(message: str) -> None: + """Write a line to stdout. + + Args: + message: Line to write, without a trailing newline. + """ + sys.stdout.write(f"{message}\n") + + +def _emit_err(message: str) -> None: + """Write a line to stderr. + + Args: + message: Line to write, without a trailing newline. + """ + sys.stderr.write(f"{message}\n") + + +def _now() -> datetime: + """Current UTC time (single seam for tests). + + Returns: + Timezone-aware now. + """ + return datetime.now(tz=timezone.utc) + + +def build_parser() -> argparse.ArgumentParser: + """Construct the argument parser. + + Returns: + Parser with all subcommands. + """ + parser = argparse.ArgumentParser( + prog="python -m python_pkg.session_autopsy", + description="Deterministic Claude Code transcript analyzer.", + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + ingest = subparsers.add_parser("ingest", help="analyze one session transcript") + ingest.add_argument( + "transcript", type=Path, help="path to the .jsonl transcript" + ) + ingest.add_argument( + "--quiet", action="store_true", help="suppress the summary line" + ) + ingest.add_argument( + "--no-report", action="store_true", help="skip REPORT.md regeneration" + ) + + scan = subparsers.add_parser( + "scan", help="analyze every transcript under the projects dir" + ) + scan.add_argument( + "--all", + action="store_true", + help="accepted for symmetry; scan always walks everything", + ) + scan.add_argument( + "--jobs", type=int, default=DEFAULT_SCAN_JOBS, help="parallel worker processes" + ) + scan.add_argument( + "--force", action="store_true", help="reanalyze transcripts even if unchanged" + ) + scan.add_argument( + "--no-report", action="store_true", help="skip REPORT.md regeneration" + ) + + report_cmd = subparsers.add_parser( + "report", help="regenerate REPORT.md from stored records" + ) + report_cmd.add_argument( + "--mark-reviewed", + action="store_true", + help="mark all current candidates reviewed", + ) + + candidates = subparsers.add_parser("candidates", help="list ranked candidates") + candidates.add_argument( + "--json", action="store_true", help="machine-readable output" + ) + + trace_cmd = subparsers.add_parser( + "traces", help="export compact excerpts for a candidate" + ) + trace_cmd.add_argument( + "candidate_id", help="candidate id from REPORT.md (e.g. skill-finish)" + ) + trace_cmd.add_argument( + "--max-invocations", type=int, default=DEFAULT_TRACE_INVOCATIONS + ) + trace_cmd.add_argument( + "--out", type=Path, default=None, help="write to a file instead of stdout" + ) + + measure = subparsers.add_parser( + "measure", help="before/after tokens for compiled skills" + ) + measure.add_argument( + "skill", nargs="?", default=None, help="restrict to one skill name" + ) + return parser + + +def _regenerate_report(records: list[SessionRecord] | None = None) -> int: + """Re-run detectors and rewrite REPORT.md + state.json. + + Args: + records: Preloaded records, or None to load from the store. + + Returns: + The unreviewed-candidate count. + """ + home = config.autopsy_home() + if records is None: + records = store.load_records(home) + result = detectors.analyze(records, _now()) + return report.write_report(home, records, result, _now()) + + +def _summary_line(record: SessionRecord) -> str: + """Render the one-line ingest summary for a record. + + Args: + record: The freshly parsed record. + + Returns: + A compact human-readable summary. + """ + return ( + f"{record.session_id}: {record.counts.assistant_msgs} assistant msgs, " + f"{record.tokens.output} out / {record.tokens.cache_write} cache-write tokens, " + f"{len(record.obs.skill_invocations)} skill spans, " + f"{record.counts.subagent_count} subagents, " + f"{record.counts.malformed_lines} malformed lines" + ) + + +def cmd_ingest(args: argparse.Namespace) -> int: + """Analyze one transcript, upsert its record, refresh the report. + + Args: + args: Parsed CLI arguments. + + Returns: + Process exit code (0 on success, 2 on missing file). + """ + transcript: Path = args.transcript.expanduser() + if not transcript.is_file(): + _emit_err(f"ingest: no such transcript: {transcript}") + return EXIT_NO_FILE + record = parse_session(transcript) + total = store.upsert_records(config.autopsy_home(), [record]) + if not args.no_report: + _regenerate_report() + if not args.quiet: + _emit(_summary_line(record)) + _emit(f"store: {total} sessions") + return 0 + + +def _find_transcripts(root: Path) -> list[Path]: + """List every top-level session transcript under the projects dir. + + Subagent transcripts (``**/subagents/agent-*.jsonl``) are excluded; + they are folded into their parent session by the parser. + + Args: + root: The projects directory. + + Returns: + Sorted transcript paths. + """ + return sorted(path for path in root.glob("*/*.jsonl") if path.is_file()) + + +def _needs_analysis(path: Path, index: dict[str, tuple[int, float]]) -> bool: + """Decide whether a transcript changed since its stored record. + + Args: + path: Transcript path. + index: Stored ``{path: (size, mtime)}`` map. + + Returns: + True when the transcript is new or its size/mtime moved. + """ + stored = index.get(str(path)) + if stored is None: + return True + stat = path.stat() + return (stat.st_size, stat.st_mtime) != stored + + +def _parse_one(path: Path) -> SessionRecord | None: + """Worker-side parse wrapper that never raises across the pool. + + Args: + path: Transcript path to parse. + + Returns: + The record, or None when the file vanished or was unreadable. + """ + try: + return parse_session(path) + except OSError as error: + _emit_err(f"scan: skipping {path}: {error}") + return None + + +def cmd_scan(args: argparse.Namespace) -> int: + """Analyze every changed transcript in parallel and upsert results. + + Args: + args: Parsed CLI arguments. + + Returns: + Process exit code (always 0; unreadable files are reported and skipped). + """ + started = time.monotonic() + home = config.autopsy_home() + root = config.projects_dir() + transcripts = _find_transcripts(root) + index = {} if args.force else store.load_file_index(home) + pending = [path for path in transcripts if _needs_analysis(path, index)] + records: list[SessionRecord] = [] + if pending: + with concurrent.futures.ProcessPoolExecutor( + max_workers=max(1, args.jobs) + ) as pool: + records = [ + record for record in pool.map(_parse_one, pending) if record is not None + ] + store.upsert_records(home, records) + if not args.no_report: + _regenerate_report() + elapsed = time.monotonic() - started + _emit( + f"scan: {len(transcripts)} transcripts, {len(pending)} analyzed, " + f"{len(transcripts) - len(pending)} unchanged, {elapsed:.1f}s", + ) + return 0 + + +def cmd_report(args: argparse.Namespace) -> int: + """Regenerate REPORT.md (optionally marking candidates reviewed). + + Args: + args: Parsed CLI arguments. + + Returns: + Process exit code (always 0). + """ + home = config.autopsy_home() + records = store.load_records(home) + result = detectors.analyze(records, _now()) + report.write_report(home, records, result, _now()) + if args.mark_reviewed: + report.write_state( + home, [cand.id for cand in result.candidates], _now(), mark_reviewed=True + ) + _emit("report: all candidates marked reviewed") + _emit(f"report: {len(result.candidates)} candidates → {home / report.REPORT_FILE}") + return 0 + + +def cmd_candidates(args: argparse.Namespace) -> int: + """List ranked candidates. + + Args: + args: Parsed CLI arguments. + + Returns: + Process exit code (always 0). + """ + records = store.load_records(config.autopsy_home()) + result = detectors.analyze(records, _now()) + if args.json: + _emit(json.dumps([cand.to_dict() for cand in result.candidates], indent=2)) + return 0 + for cand in result.candidates: + savings = report.fmt_tokens(cand.est_weekly_savings) + _emit( + f"{cand.id} {cand.kind:6} {cand.occurrences:4}x " + f"~{savings}/wk {cand.action}" + ) + return 0 + + +def cmd_traces(args: argparse.Namespace) -> int: + """Export compact excerpts for one candidate. + + Args: + args: Parsed CLI arguments. + + Returns: + Process exit code (0, or 3 for an unknown candidate id). + """ + records = store.load_records(config.autopsy_home()) + result = detectors.analyze(records, _now()) + match = next( + (cand for cand in result.candidates if cand.id == args.candidate_id), None + ) + if match is None: + _emit_err(f"traces: unknown candidate id: {args.candidate_id}") + return EXIT_UNKNOWN_ID + records_by_id = {record.session_id: record for record in records} + text = traces.render_traces( + match, records_by_id, max_invocations=args.max_invocations + ) + if args.out is not None: + args.out.write_text(text, encoding="utf-8") + _emit(f"traces: wrote {args.out}") + return 0 + _emit(text) + return 0 + + +def cmd_measure(args: argparse.Namespace) -> int: + """Report before/after tokens per invocation for compiled skills. + + Args: + args: Parsed CLI arguments. + + Returns: + Process exit code (always 0). + """ + home = config.autopsy_home() + records = store.load_records(home) + for line in report.measure_lines(records, report.load_compiled(home), args.skill): + _emit(line) + return 0 + + +def main(argv: list[str] | None = None) -> int: + """Run the CLI. + + Args: + argv: Argument vector (None means ``sys.argv[1:]``). + + Returns: + Process exit code. + """ + args = build_parser().parse_args(argv) + handlers = { + "ingest": cmd_ingest, + "scan": cmd_scan, + "report": cmd_report, + "candidates": cmd_candidates, + "traces": cmd_traces, + "measure": cmd_measure, + } + return handlers[args.command](args) diff --git a/python_pkg/session_autopsy/config.py b/python_pkg/session_autopsy/config.py new file mode 100644 index 00000000..33cc3f5f --- /dev/null +++ b/python_pkg/session_autopsy/config.py @@ -0,0 +1,30 @@ +"""Filesystem locations for the session-autopsy store. + +Both locations are overridable via environment variables so tests (and any +future relocation) never touch the real ``~/.claude`` tree. +""" + +from __future__ import annotations + +import os +from pathlib import Path + + +def autopsy_home() -> Path: + """Return the directory holding sessions.jsonl, REPORT.md and state.json. + + Returns: + The ``CLAUDE_AUTOPSY_HOME`` override or ``~/.claude/autopsy``. + """ + return Path(os.environ.get("CLAUDE_AUTOPSY_HOME", "~/.claude/autopsy")).expanduser() + + +def projects_dir() -> Path: + """Return the directory containing Claude Code session transcripts. + + Returns: + The ``CLAUDE_PROJECTS_DIR`` override or ``~/.claude/projects``. + """ + return Path( + os.environ.get("CLAUDE_PROJECTS_DIR", "~/.claude/projects") + ).expanduser() diff --git a/python_pkg/session_autopsy/detectors.py b/python_pkg/session_autopsy/detectors.py new file mode 100644 index 00000000..bfb6f7e6 --- /dev/null +++ b/python_pkg/session_autopsy/detectors.py @@ -0,0 +1,467 @@ +"""Pure heuristics turning session records into ranked automation candidates. + +No LLM anywhere: the compile-worthiness score comes from turn composition and +cross-invocation command similarity, savings from measured token usage. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import hashlib +from typing import TYPE_CHECKING + +from python_pkg.session_autopsy import stats + +if TYPE_CHECKING: + from datetime import datetime + + from python_pkg.session_autopsy.records import SessionRecord, SkillInvocation + +MIN_SKILL_INVOCATIONS = 3 +NGRAM_SIZES = range(3, 7) +NGRAM_MIN_OCCURRENCES = 8 +NGRAM_MIN_SESSIONS = 4 +NGRAM_MIN_DISTINCT_RATIO = 0.5 +ERROR_MIN_OCCURRENCES = 10 +ERROR_MIN_SESSIONS = 3 +PROMPT_MIN_OCCURRENCES = 4 +PROMPT_MIN_WORDS = 3 +MAX_EVIDENCE_SESSIONS = 10 +TREND_TOP_SESSIONS = 3 + + +@dataclass(frozen=True) +class _CounterSpec: + """Which signature-counter field to mine, and the thresholds to mine it at.""" + + field_name: str + """Attribute on ``record.obs`` holding the ``sig -> count`` map.""" + kind: str + """Candidate kind tag.""" + min_occurrences: int + """Total-count threshold.""" + min_sessions: int + """Distinct-session threshold (ORed with the occurrence threshold).""" + action: str + """Suggested action text.""" + + +_ERROR_SPEC = _CounterSpec( + field_name="error_signatures", + kind="err", + min_occurrences=ERROR_MIN_OCCURRENCES, + min_sessions=ERROR_MIN_SESSIONS, + action="fix once: eliminate the failure's root cause", +) +_PROMPT_SPEC = _CounterSpec( + field_name="typed_prompt_signatures", + kind="prompt", + min_occurrences=PROMPT_MIN_OCCURRENCES, + min_sessions=1, + action="promote to slash command or script", +) + + +@dataclass +class Candidate: + """One ranked automation opportunity.""" + + id: str + kind: str + title: str + sig: str + sessions: int + occurrences: int + avg_tokens: int + per_week: float + est_weekly_savings: int + score: float + action: str + session_ids: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, object]: + """Serialize for ``candidates --json``. + + Returns: + A JSON-compatible dict of all fields. + """ + return { + "id": self.id, + "kind": self.kind, + "title": self.title, + "sig": self.sig, + "sessions": self.sessions, + "occurrences": self.occurrences, + "avg_tokens": self.avg_tokens, + "per_week": round(self.per_week, 2), + "est_weekly_savings": self.est_weekly_savings, + "score": round(self.score, 3), + "action": self.action, + "session_ids": self.session_ids, + } + + +@dataclass +class TrendInfo: + """Weekly cache-read-per-turn averages and their direction.""" + + weekly: list[tuple[str, int]] = field(default_factory=list) + slope_per_week: float = 0.0 + top_sessions: list[tuple[str, int]] = field(default_factory=list) + + +@dataclass +class AnalysisResult: + """Everything the report renders: ranked candidates and the trend.""" + + candidates: list[Candidate] = field(default_factory=list) + trend: TrendInfo = field(default_factory=TrendInfo) + + +def analyze(records: list[SessionRecord], now: datetime) -> AnalysisResult: + """Run every detector and rank the merged candidates. + + Args: + records: All stored session records. + now: Current UTC time (injected for determinism in tests). + + Returns: + Ranked candidates and the context trend. + """ + per_turn_avg = stats.per_turn_avg(records) + candidates = [ + *_skill_candidates(records, now), + *_ngram_candidates(records, now, per_turn_avg), + *_error_candidates(records, now, per_turn_avg), + *_prompt_candidates(records, now, per_turn_avg), + ] + candidates.sort(key=lambda cand: (-cand.est_weekly_savings, cand.id)) + return AnalysisResult(candidates=candidates, trend=_trend(records)) + + +def _hex_id(text: str) -> str: + """Derive the stable 8-hex candidate id suffix for a signature. + + Args: + text: The normalized signature. + + Returns: + First 8 hex chars of its SHA-256. + """ + return hashlib.sha256(text.encode("utf-8")).hexdigest()[:8] + + +def _skill_candidates(records: list[SessionRecord], now: datetime) -> list[Candidate]: + """Rank every skill with enough invocations by measured spend. + + No compile-vs-keep verdict is attempted here: real-corpus probing showed no + cheap scalar separates mechanical skills from judgment-heavy ones (a giant + Write turn hides among mechanical read turns). The score is published as + evidence; the verdict belongs to /compile-candidate reading real traces. + + Args: + records: All session records. + now: Current UTC time. + + Returns: + One candidate per skill with at least MIN_SKILL_INVOCATIONS uses. + """ + by_skill: dict[str, list[tuple[SessionRecord, SkillInvocation]]] = {} + for record in records: + for invocation in record.obs.skill_invocations: + by_skill.setdefault(invocation.name, []).append((record, invocation)) + return [ + _score_skill(name, pairs, now) + for name, pairs in sorted(by_skill.items()) + if len(pairs) >= MIN_SKILL_INVOCATIONS + ] + + +def _score_skill( + name: str, + pairs: list[tuple[SessionRecord, SkillInvocation]], + now: datetime, +) -> Candidate: + """Build the candidate for one skill from all its invocations. + + Args: + name: Skill name. + pairs: (session record, invocation) for every invocation. + now: Current UTC time. + + Returns: + The scored candidate (score = determinism * similarity). + """ + determinism = stats.mean( + [ + invocation.tool_only_turns / total + for _, invocation in pairs + if (total := invocation.tool_only_turns + invocation.text_turns) > 0 + ], + ) + similarity = stats.mean_pairwise_similarity( + [invocation.bash_sig for _, invocation in pairs] + ) + score = determinism * similarity + avg_tokens = int( + stats.mean( + [float(inv.tokens_output + inv.tokens_cache_write) for _, inv in pairs] + ) + ) + per_week = stats.per_week( + [stats.parse_timestamp(record.meta.started_at) for record, _ in pairs], now + ) + session_ids = _recent_session_ids([record for record, _ in pairs]) + action = f"review for compilation: /compile-candidate skill-{name}" + return Candidate( + id=f"skill-{name}", + kind="skill", + title=f"Skill '{name}' ({len(pairs)} invocations)", + sig=name, + sessions=len({record.session_id for record, _ in pairs}), + occurrences=len(pairs), + avg_tokens=avg_tokens, + per_week=per_week, + est_weekly_savings=int(avg_tokens * per_week), + score=score, + action=action, + session_ids=session_ids, + ) + + +def _recent_session_ids(records: list[SessionRecord]) -> list[str]: + """Most recent contributing session ids for the traces subcommand. + + Args: + records: Contributing session records (may repeat). + + Returns: + Up to :data:`MAX_EVIDENCE_SESSIONS` unique ids, newest first. + """ + unique: dict[str, str] = {} + for record in records: + unique[record.session_id] = record.meta.started_at or "" + ranked = sorted(unique.items(), key=lambda item: item[1], reverse=True) + return [session_id for session_id, _ in ranked[:MAX_EVIDENCE_SESSIONS]] + + +def _counter_candidates( + records: list[SessionRecord], + now: datetime, + per_turn_avg: int, + spec: _CounterSpec, +) -> list[Candidate]: + """Shared aggregation for signature-counter fields (errors, prompts). + + Args: + records: All session records. + now: Current UTC time. + per_turn_avg: Corpus-wide tokens per assistant turn. + spec: Which counter field to mine, and at what thresholds. + + Returns: + Candidates for every signature crossing either threshold. + """ + totals: dict[str, int] = {} + contributors: dict[str, list[SessionRecord]] = {} + for record in records: + counter: dict[str, int] = getattr(record.obs, spec.field_name) + for sig, count in counter.items(): + totals[sig] = totals.get(sig, 0) + count + contributors.setdefault(sig, []).append(record) + candidates = [] + for sig, total in sorted(totals.items()): + sessions = contributors[sig] + if total < spec.min_occurrences and len(sessions) < spec.min_sessions: + continue + timestamps = [ + stats.parse_timestamp(record.meta.started_at) for record in sessions + ] + per_week = stats.per_week(timestamps, now) * (total / len(sessions)) + candidates.append( + Candidate( + id=f"{spec.kind}-{_hex_id(sig)}", + kind=spec.kind, + title=sig[:70], + sig=sig, + sessions=len({record.session_id for record in sessions}), + occurrences=total, + avg_tokens=per_turn_avg, + per_week=per_week, + est_weekly_savings=int(per_turn_avg * per_week), + score=1.0, + action=spec.action, + session_ids=_recent_session_ids(sessions), + ), + ) + return candidates + + +def _error_candidates( + records: list[SessionRecord], now: datetime, per_turn_avg: int +) -> list[Candidate]: + """Repeated error signatures worth a fix-once script. + + Args: + records: All session records. + now: Current UTC time. + per_turn_avg: Corpus-wide tokens per assistant turn (crude cost proxy). + + Returns: + Fix-once candidates. + """ + return _counter_candidates(records, now, per_turn_avg, _ERROR_SPEC) + + +def _prompt_candidates( + records: list[SessionRecord], now: datetime, per_turn_avg: int +) -> list[Candidate]: + """Repeated typed prompts worth promoting to a command or script. + + Args: + records: All session records. + now: Current UTC time. + per_turn_avg: Corpus-wide tokens per assistant turn. + + Returns: + Prompt-promotion candidates (short generic prompts filtered out). + """ + raw = _counter_candidates(records, now, per_turn_avg, _PROMPT_SPEC) + return [ + cand + for cand in raw + if len(cand.sig.split()) >= PROMPT_MIN_WORDS + and cand.occurrences >= PROMPT_MIN_OCCURRENCES + ] + + +def _ngram_candidates( + records: list[SessionRecord], now: datetime, per_turn_avg: int +) -> list[Candidate]: + """Repeated cross-session Bash command sequences outside skill spans. + + Args: + records: All session records. + now: Current UTC time. + per_turn_avg: Corpus-wide tokens per assistant turn. + + Returns: + Sequence candidates, longest surviving grams only. + """ + counts: dict[tuple[str, ...], int] = {} + contributors: dict[tuple[str, ...], list[SessionRecord]] = {} + for record in records: + seen_here: set[tuple[str, ...]] = set() + for sequence in record.obs.bash_sequences: + for size in NGRAM_SIZES: + for start in range(len(sequence) - size + 1): + gram = tuple(sequence[start : start + size]) + counts[gram] = counts.get(gram, 0) + 1 + if gram not in seen_here: + seen_here.add(gram) + contributors.setdefault(gram, []).append(record) + surviving = { + gram: count + for gram, count in counts.items() + if count >= NGRAM_MIN_OCCURRENCES + and len(contributors[gram]) >= NGRAM_MIN_SESSIONS + and len(set(gram)) / len(gram) >= NGRAM_MIN_DISTINCT_RATIO + } + top = _drop_subgrams(surviving) + candidates = [] + for gram in sorted(top): + sessions = contributors[gram] + joined = " → ".join(gram) + occurrences = counts[gram] + per_week = stats.per_week( + [stats.parse_timestamp(record.meta.started_at) for record in sessions], now + ) + candidates.append( + Candidate( + id=f"ngram-{_hex_id(joined)}", + kind="ngram", + title=joined[:70], + sig=joined, + sessions=len({record.session_id for record in sessions}), + occurrences=occurrences, + avg_tokens=per_turn_avg * len(gram), + per_week=per_week, + est_weekly_savings=int(per_turn_avg * len(gram) * per_week), + score=1.0, + action="script this sequence", + session_ids=_recent_session_ids(sessions), + ), + ) + return candidates + + +def _drop_subgrams(surviving: dict[tuple[str, ...], int]) -> set[tuple[str, ...]]: + """Remove grams contained in a longer surviving gram. + + Args: + surviving: Grams that crossed the frequency thresholds. + + Returns: + The maximal grams only. + """ + grams = sorted(surviving, key=len, reverse=True) + kept: set[tuple[str, ...]] = set() + for gram in grams: + if not any(_contains(longer, gram) for longer in kept): + kept.add(gram) + return kept + + +def _contains(longer: tuple[str, ...], shorter: tuple[str, ...]) -> bool: + """Whether ``shorter`` appears contiguously inside ``longer``. + + Args: + longer: The candidate container gram. + shorter: The gram to look for. + + Returns: + True when contained. + """ + if len(shorter) >= len(longer): + return False + return any( + longer[i : i + len(shorter)] == shorter + for i in range(len(longer) - len(shorter) + 1) + ) + + +def _trend(records: list[SessionRecord]) -> TrendInfo: + """Weekly cache-read-per-turn trend plus the heaviest sessions. + + Args: + records: All session records. + + Returns: + The populated trend info (empty when timestamps are missing). + """ + weekly_turns: dict[str, int] = {} + weekly_reads: dict[str, int] = {} + heaviest: list[tuple[str, int]] = [] + for record in records: + stamp = stats.parse_timestamp(record.meta.started_at) + if stamp is None or record.counts.assistant_msgs == 0: + continue + week = f"{stamp.isocalendar().year}-W{stamp.isocalendar().week:02d}" + weekly_turns[week] = weekly_turns.get(week, 0) + record.counts.assistant_msgs + weekly_reads[week] = weekly_reads.get(week, 0) + record.tokens.cache_read + heaviest.append( + ( + record.meta.slug or record.session_id, + record.tokens.cache_read // record.counts.assistant_msgs, + ) + ) + weekly = [ + (week, weekly_reads[week] // weekly_turns[week]) + for week in sorted(weekly_turns) + ] + heaviest.sort(key=lambda item: -item[1]) + return TrendInfo( + weekly=weekly, + slope_per_week=stats.slope([float(avg) for _, avg in weekly]), + top_sessions=heaviest[:TREND_TOP_SESSIONS], + ) diff --git a/python_pkg/session_autopsy/parse.py b/python_pkg/session_autopsy/parse.py new file mode 100644 index 00000000..8c8cac5a --- /dev/null +++ b/python_pkg/session_autopsy/parse.py @@ -0,0 +1,458 @@ +"""Streaming parser turning one transcript (plus subagents) into a record. + +Transcripts can reach 12 MB, so every file is read line-by-line; no line's +parse result outlives the accumulator fields it updates. +""" + +from __future__ import annotations + +from collections import Counter +from datetime import datetime, timezone +import json +import re +from typing import TYPE_CHECKING + +from python_pkg.session_autopsy.records import ( + ActivityCounts, + Observations, + SessionMeta, + SessionRecord, + SkillInvocation, + TokenTotals, +) +from python_pkg.session_autopsy.signatures import command_signature, normalize_signature + +if TYPE_CHECKING: + from pathlib import Path + +GEN_TURN_MIN_CHARS = 400 +MAX_BASH_RUN = 30 +MIN_BASH_RUN = 3 +MAX_ERROR_SIGS = 20 +MAX_PROMPT_SIGS = 40 +MAX_BASH_FIRST_LINES = 40 +MAX_SKILL_BASH_SIG = 60 +RESULT_SCAN_LINES = 10 +RESULT_TAIL_LINES = 5 +RETRY_PROMPT_MAX_LEN = 70 + +_ERROR_LINE_RE = re.compile(r"^(?:rm: |fatal: |Traceback)|\w*error\b", re.IGNORECASE) +_COMPACT_PREFIX = "This session is being continued from a previous conversation" +_SLASH_COMMAND_RE = re.compile(r"/?([\w:-]+)") +# Harness built-ins that appear as but are not skills. +_BUILTIN_COMMANDS = frozenset( + { + "clear", + "compact", + "config", + "context", + "cost", + "doctor", + "effort", + "exit", + "fast", + "help", + "hooks", + "ide", + "login", + "logout", + "loop", + "mcp", + "memory", + "model", + "permissions", + "plan", + "plugin", + "reload-plugins", + "remote-control", + "rename", + "resume", + "skills", + "status", + "usage", + }, +) +# Tool inputs that carry generated content (prose/code the model wrote). +_GENERATIVE_INPUT_KEYS = ("content", "new_string") +# Error lines too generic to act on; the informative line is elsewhere. +_GENERIC_ERROR_SIGS = frozenset( + { + "Error: Exit code ", + "Traceback (most recent call last):", + }, +) + + +class _Accumulator: + """Mutable per-session state fed one transcript line at a time.""" + + def __init__(self) -> None: + """Initialize empty counters and open-span state.""" + self.tokens = TokenTotals() + self.meta = SessionMeta() + self.counts = ActivityCounts() + self.tool_histogram: Counter[str] = Counter() + self.bash_first_lines: Counter[str] = Counter() + self.error_signatures: Counter[str] = Counter() + self.typed_prompt_signatures: Counter[str] = Counter() + self.bash_sequences: list[list[str]] = [] + self.skill_invocations: list[SkillInvocation] = [] + self._current_run: list[str] = [] + self._current_skill: SkillInvocation | None = None + + def feed_line(self, raw_line: str) -> None: + """Parse and account one transcript line. + + Args: + raw_line: One line of the JSONL transcript. + """ + stripped = raw_line.strip() + if not stripped: + return + try: + obj = json.loads(stripped) + except json.JSONDecodeError: + self.counts.malformed_lines += 1 + return + if not isinstance(obj, dict): + self.counts.malformed_lines += 1 + return + self._dispatch(obj) + + def finish(self) -> None: + """Flush any still-open Bash run and skill span.""" + self._flush_run() + self._flush_skill() + + def _dispatch(self, obj: dict[str, object]) -> None: + """Route a parsed line by its ``type``. + + Args: + obj: The parsed transcript line. + """ + line_type = obj.get("type") + if line_type == "assistant": + self._on_assistant(obj) + elif line_type == "user": + self._on_user(obj) + elif line_type == "ai-title": + self._on_title(obj) + elif line_type == "system" and obj.get("subtype") == "compact_boundary": + self.counts.compaction_count += 1 + + def _on_title(self, obj: dict[str, object]) -> None: + """Record the session's AI-generated title (last one wins). + + Args: + obj: An ``ai-title`` transcript line. + """ + for key in ("title", "aiTitle"): + value = obj.get(key) + if isinstance(value, str) and value: + self.meta.title = value + + def _note_metadata(self, obj: dict[str, object]) -> None: + """Capture session metadata from the first message line carrying it. + + Args: + obj: A message-bearing transcript line. + """ + timestamp = obj.get("timestamp") + if isinstance(timestamp, str): + if self.meta.started_at is None: + self.meta.started_at = timestamp + self.meta.ended_at = timestamp + if self.meta.cwd is None and isinstance(obj.get("cwd"), str): + self.meta.cwd = str(obj["cwd"]) + if self.meta.git_branch is None and isinstance(obj.get("gitBranch"), str): + self.meta.git_branch = str(obj["gitBranch"]) + if self.meta.slug is None and isinstance(obj.get("slug"), str): + self.meta.slug = str(obj["slug"]) + + def _on_assistant(self, obj: dict[str, object]) -> None: + """Account one assistant message: usage, tools, turn class. + + Args: + obj: An ``assistant`` transcript line. + """ + message = obj.get("message") + if not isinstance(message, dict): + return + self._note_metadata(obj) + self.counts.assistant_msgs += 1 + usage = message.get("usage") + if isinstance(usage, dict): + self.tokens.add_usage(usage) + text_chars, had_tool_use = self._scan_blocks(message) + if self._current_skill is not None: + skill = self._current_skill + skill.tokens_output += _usage_int(usage, "output_tokens") + skill.tokens_cache_write += _usage_int(usage, "cache_creation_input_tokens") + if had_tool_use and text_chars <= GEN_TURN_MIN_CHARS: + skill.tool_only_turns += 1 + else: + skill.text_turns += 1 + + def _scan_blocks(self, message: dict[str, object]) -> tuple[int, bool]: + """Walk an assistant message's content blocks. + + Args: + message: The ``message`` object of an assistant line. + + Returns: + Total text characters and whether any tool_use block was present. + """ + text_chars = 0 + had_tool_use = False + content = message.get("content") + if not isinstance(content, list): + return (0, False) + for block in content: + if not isinstance(block, dict): + continue + block_type = block.get("type") + if block_type == "text": + text = block.get("text") + if isinstance(text, str): + text_chars += len(text.strip()) + elif block_type == "tool_use": + had_tool_use = True + text_chars += _generated_chars(block) + self._on_tool_use(block) + return (text_chars, had_tool_use) + + def _on_tool_use(self, block: dict[str, object]) -> None: + """Account one tool_use block (histogram, Bash runs, skill spans). + + Args: + block: A ``tool_use`` content block. + """ + name = str(block.get("name") or "?") + self.tool_histogram[name] += 1 + tool_input = block.get("input") + if not isinstance(tool_input, dict): + tool_input = {} + if name == "Bash": + self._on_bash(tool_input) + return + self._flush_run() + if name == "Skill": + self._flush_skill() + skill_name = tool_input.get("skill") + self._current_skill = SkillInvocation(name=str(skill_name or "?")) + + def _on_bash(self, tool_input: dict[str, object]) -> None: + """Account one Bash tool_use: first-line counter, run, skill sig. + + Args: + tool_input: The Bash tool's ``input`` object. + """ + command = tool_input.get("command") + if not isinstance(command, str): + return + first_line = normalize_signature(command) + if first_line: + self.bash_first_lines[first_line] += 1 + sig = command_signature(command) + if not sig: + return + if self._current_skill is not None: + if len(self._current_skill.bash_sig) < MAX_SKILL_BASH_SIG: + self._current_skill.bash_sig.append(sig) + elif len(self._current_run) < MAX_BASH_RUN: + self._current_run.append(sig) + + def _on_user(self, obj: dict[str, object]) -> None: + """Account one user line: typed prompts, retries, tool errors. + + Args: + obj: A ``user`` transcript line. + """ + message = obj.get("message") + if not isinstance(message, dict): + return + self._note_metadata(obj) + self._scan_tool_result(obj.get("toolUseResult")) + content = message.get("content") + is_external = ( + obj.get("userType") == "external" and obj.get("isSidechain") is not True + ) + if not (is_external and isinstance(content, str)): + return + if content.startswith(_COMPACT_PREFIX): + self.counts.compaction_count += 1 + return + slash = _SLASH_COMMAND_RE.search(content) + if slash is not None: + self._flush_run() + self._flush_skill() + if slash.group(1) not in _BUILTIN_COMMANDS: + self._current_skill = SkillInvocation(name=slash.group(1)) + return + self.counts.external_prompts += 1 + self._flush_run() + self._flush_skill() + lowered = content.lower() + if ( + len(content) <= RETRY_PROMPT_MAX_LEN + and "limit" in lowered + and ("reset" in lowered or "continue" in lowered) + ): + self.counts.retry_prompt_count += 1 + if content.startswith(("<", "[")): + return + sig = normalize_signature(content) + if sig: + self.typed_prompt_signatures[sig] += 1 + + def _scan_tool_result(self, result: object) -> None: + """Mine a ``toolUseResult`` for repeated error lines. + + Args: + result: The top-level ``toolUseResult`` value (str, dict, or absent). + """ + texts: list[str] = [] + if isinstance(result, str): + texts.append(result) + elif isinstance(result, dict): + for key in ("stderr", "stdout"): + value = result.get(key) + if isinstance(value, str) and value: + texts.append(value) + for text in texts: + all_lines = text.splitlines() + scan = all_lines[:RESULT_SCAN_LINES] + if len(all_lines) > RESULT_SCAN_LINES: + scan += all_lines[-RESULT_TAIL_LINES:] + for line in scan: + if _ERROR_LINE_RE.search(line): + sig = normalize_signature(line) + if sig and sig not in _GENERIC_ERROR_SIGS: + self.error_signatures[sig] += 1 + + def _flush_run(self) -> None: + """Close the current Bash run, keeping it only if long enough.""" + if len(self._current_run) >= MIN_BASH_RUN: + self.bash_sequences.append(self._current_run) + self._current_run = [] + + def _flush_skill(self) -> None: + """Close the current skill span, dropping spans with no turns.""" + skill = self._current_skill + if skill is not None: + if skill.tool_only_turns + skill.text_turns > 0: + self.skill_invocations.append(skill) + self._current_skill = None + + +def _generated_chars(block: dict[str, object]) -> int: + """Characters of model-generated content hidden in a tool_use input. + + Write/Edit-style inputs carry the prose or code the model wrote; counting + them stops generation-heavy skills from masquerading as tool-only. + + Args: + block: A ``tool_use`` content block. + + Returns: + Character count of generative input fields. + """ + tool_input = block.get("input") + if not isinstance(tool_input, dict): + return 0 + return sum( + len(value) + for key in _GENERATIVE_INPUT_KEYS + if isinstance(value := tool_input.get(key), str) + ) + + +def _usage_int(usage: object, key: str) -> int: + """Read one int field from a possibly-absent usage dict. + + Args: + usage: The ``message.usage`` value. + key: Field name to read. + + Returns: + The int value, or 0. + """ + if isinstance(usage, dict): + value = usage.get(key) + if isinstance(value, int) and not isinstance(value, bool): + return value + return 0 + + +def _top(counter: Counter[str], limit: int) -> dict[str, int]: + """Keep only the most common counter entries, deterministically ordered. + + Args: + counter: Any string counter. + limit: Maximum number of entries to keep. + + Returns: + A dict of the top entries, ties broken alphabetically. + """ + ranked = sorted(counter.items(), key=lambda item: (-item[1], item[0])) + return dict(ranked[:limit]) + + +def subagent_files(transcript_path: Path) -> list[Path]: + """Locate the nested subagent transcripts of a session. + + Args: + transcript_path: Path to the main ``.jsonl`` transcript. + + Returns: + Sorted ``agent-*.jsonl`` paths under ``/subagents/``. + """ + nested = transcript_path.parent / transcript_path.stem / "subagents" + if not nested.is_dir(): + return [] + return sorted(nested.glob("agent-*.jsonl")) + + +def parse_session(transcript_path: Path) -> SessionRecord: + """Parse one session transcript (and its subagents) into a record. + + Subagent token usage and tool calls fold into the parent session's + totals; subagent prompts never count as external (they are sidechains). + + Args: + transcript_path: Path to the main ``.jsonl`` transcript. + + Returns: + The populated :class:`SessionRecord`. + """ + acc = _Accumulator() + subagents = subagent_files(transcript_path) + for path in [transcript_path, *subagents]: + with path.open(encoding="utf-8", errors="replace") as handle: + for raw_line in handle: + acc.feed_line(raw_line) + acc.finish() + stat = transcript_path.stat() + # The accumulator owns the metadata and counts records outright, so they are + # handed over rather than copied field by field. Only the subagent count is + # not something a transcript line can carry: it comes from the file listing. + acc.counts.subagent_count = len(subagents) + return SessionRecord( + session_id=transcript_path.stem, + project_slug=transcript_path.parent.name, + transcript_path=str(transcript_path), + file_size=stat.st_size, + file_mtime=stat.st_mtime, + analyzed_at=datetime.now(tz=timezone.utc).isoformat(timespec="seconds"), + meta=acc.meta, + counts=acc.counts, + tokens=acc.tokens, + obs=Observations( + tool_histogram=dict(acc.tool_histogram), + bash_first_lines=_top(acc.bash_first_lines, MAX_BASH_FIRST_LINES), + bash_sequences=acc.bash_sequences, + skill_invocations=acc.skill_invocations, + error_signatures=_top(acc.error_signatures, MAX_ERROR_SIGS), + typed_prompt_signatures=_top(acc.typed_prompt_signatures, MAX_PROMPT_SIGS), + ), + ) diff --git a/python_pkg/session_autopsy/records.py b/python_pkg/session_autopsy/records.py new file mode 100644 index 00000000..65222104 --- /dev/null +++ b/python_pkg/session_autopsy/records.py @@ -0,0 +1,328 @@ +"""Dataclass schema for per-session autopsy records (schema v2).""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, field + +SCHEMA_VERSION = 2 + + +@dataclass +class TokenTotals: + """Token usage summed over every assistant message in a session.""" + + output: int = 0 + input: int = 0 + cache_read: int = 0 + cache_write: int = 0 + + def add_usage(self, usage: dict[str, object]) -> None: + """Accumulate one assistant message's ``usage`` block. + + Args: + usage: The ``message.usage`` object from a transcript line. + """ + self.output += _as_int(usage.get("output_tokens")) + self.input += _as_int(usage.get("input_tokens")) + self.cache_read += _as_int(usage.get("cache_read_input_tokens")) + self.cache_write += _as_int(usage.get("cache_creation_input_tokens")) + + +@dataclass +class SessionMeta: + """Where and when the session happened.""" + + started_at: str | None = None + ended_at: str | None = None + cwd: str | None = None + git_branch: str | None = None + slug: str | None = None + title: str | None = None + + +@dataclass +class ActivityCounts: + """How much happened, by category.""" + + assistant_msgs: int = 0 + external_prompts: int = 0 + subagent_count: int = 0 + compaction_count: int = 0 + retry_prompt_count: int = 0 + malformed_lines: int = 0 + + +@dataclass +class SkillInvocation: + """One skill span (Skill tool_use or slash command) and its turns.""" + + name: str + tool_only_turns: int = 0 + text_turns: int = 0 + tokens_output: int = 0 + tokens_cache_write: int = 0 + bash_sig: list[str] = field(default_factory=list) + + +@dataclass +class Observations: + """Signals mined from the transcript for the detectors.""" + + tool_histogram: dict[str, int] = field(default_factory=dict) + bash_first_lines: dict[str, int] = field(default_factory=dict) + bash_sequences: list[list[str]] = field(default_factory=list) + skill_invocations: list[SkillInvocation] = field(default_factory=list) + error_signatures: dict[str, int] = field(default_factory=dict) + typed_prompt_signatures: dict[str, int] = field(default_factory=dict) + + +@dataclass +class SessionRecord: + """Compact, deterministic summary of one Claude Code session transcript.""" + + session_id: str + project_slug: str + transcript_path: str + file_size: int + file_mtime: float + analyzed_at: str + schema_version: int = SCHEMA_VERSION + meta: SessionMeta = field(default_factory=SessionMeta) + counts: ActivityCounts = field(default_factory=ActivityCounts) + tokens: TokenTotals = field(default_factory=TokenTotals) + obs: Observations = field(default_factory=Observations) + + def to_dict(self) -> dict[str, object]: + """Serialize to a JSON-compatible dict. + + Returns: + A plain dict with nested dataclasses expanded. + """ + return asdict(self) + + +def _meta_from_dict(data: object) -> SessionMeta: + """Rebuild the meta block from its serialized form. + + Args: + data: The ``meta`` value of a stored record. + + Returns: + The reconstructed meta (empty on junk). + """ + return SessionMeta( + started_at=_opt_str(_dict_get(data, "started_at")), + ended_at=_opt_str(_dict_get(data, "ended_at")), + cwd=_opt_str(_dict_get(data, "cwd")), + git_branch=_opt_str(_dict_get(data, "git_branch")), + slug=_opt_str(_dict_get(data, "slug")), + title=_opt_str(_dict_get(data, "title")), + ) + + +def _counts_from_dict(data: object) -> ActivityCounts: + """Rebuild the counts block from its serialized form. + + Args: + data: The ``counts`` value of a stored record. + + Returns: + The reconstructed counts (zeros on junk). + """ + return ActivityCounts( + assistant_msgs=_as_int(_dict_get(data, "assistant_msgs")), + external_prompts=_as_int(_dict_get(data, "external_prompts")), + subagent_count=_as_int(_dict_get(data, "subagent_count")), + compaction_count=_as_int(_dict_get(data, "compaction_count")), + retry_prompt_count=_as_int(_dict_get(data, "retry_prompt_count")), + malformed_lines=_as_int(_dict_get(data, "malformed_lines")), + ) + + +def _tokens_from_dict(data: object) -> TokenTotals: + """Rebuild token totals from their serialized form. + + Args: + data: The ``tokens`` value of a stored record. + + Returns: + The reconstructed totals (zeros on junk). + """ + return TokenTotals( + output=_as_int(_dict_get(data, "output")), + input=_as_int(_dict_get(data, "input")), + cache_read=_as_int(_dict_get(data, "cache_read")), + cache_write=_as_int(_dict_get(data, "cache_write")), + ) + + +def _obs_from_dict(data: object) -> Observations: + """Rebuild the observations block from its serialized form. + + Args: + data: The ``obs`` value of a stored record. + + Returns: + The reconstructed observations (empty on junk). + """ + invocations = [ + SkillInvocation( + name=str(_dict_get(raw, "name") or "?"), + tool_only_turns=_as_int(_dict_get(raw, "tool_only_turns")), + text_turns=_as_int(_dict_get(raw, "text_turns")), + tokens_output=_as_int(_dict_get(raw, "tokens_output")), + tokens_cache_write=_as_int(_dict_get(raw, "tokens_cache_write")), + bash_sig=_str_list(_dict_get(raw, "bash_sig")), + ) + for raw in _list_of_dicts(_dict_get(data, "skill_invocations")) + ] + return Observations( + tool_histogram=_str_int_map(_dict_get(data, "tool_histogram")), + bash_first_lines=_str_int_map(_dict_get(data, "bash_first_lines")), + bash_sequences=[ + _str_list(seq) for seq in _list_value(_dict_get(data, "bash_sequences")) + ], + skill_invocations=invocations, + error_signatures=_str_int_map(_dict_get(data, "error_signatures")), + typed_prompt_signatures=_str_int_map( + _dict_get(data, "typed_prompt_signatures") + ), + ) + + +def record_from_dict(data: dict[str, object]) -> SessionRecord: + """Rebuild a :class:`SessionRecord` from its serialized form. + + Args: + data: A dict previously produced by :meth:`SessionRecord.to_dict`. + + Returns: + The reconstructed record; unknown keys are ignored so old readers + survive future additive schema changes. + """ + return SessionRecord( + session_id=str(data.get("session_id") or ""), + project_slug=str(data.get("project_slug") or ""), + transcript_path=str(data.get("transcript_path") or ""), + file_size=_as_int(data.get("file_size")), + file_mtime=_as_float(data.get("file_mtime")), + analyzed_at=str(data.get("analyzed_at") or ""), + schema_version=_as_int(data.get("schema_version")), + meta=_meta_from_dict(data.get("meta")), + counts=_counts_from_dict(data.get("counts")), + tokens=_tokens_from_dict(data.get("tokens")), + obs=_obs_from_dict(data.get("obs")), + ) + + +def _as_int(value: object) -> int: + """Coerce a JSON value to int, treating bools and junk as 0. + + Args: + value: Any value read from a transcript or store line. + + Returns: + The int value, or 0 when absent/bool/non-numeric. + """ + if isinstance(value, bool): + return 0 + if isinstance(value, (int, float)): + return int(value) + return 0 + + +def _as_float(value: object) -> float: + """Coerce a JSON value to float, treating bools and junk as 0.0. + + Args: + value: Any value read from a transcript or store line. + + Returns: + The float value, or 0.0 when absent/bool/non-numeric. + """ + if isinstance(value, bool): + return 0.0 + if isinstance(value, (int, float)): + return float(value) + return 0.0 + + +def _opt_str(value: object) -> str | None: + """Return the value if it is a non-empty string, else None. + + Args: + value: Any JSON value. + + Returns: + The string, or None. + """ + if isinstance(value, str) and value: + return value + return None + + +def _dict_get(value: object, key: str) -> object: + """Safely index into a value that may not be a dict. + + Args: + value: Any JSON value. + key: Key to read when the value is a dict. + + Returns: + The mapped value, or None. + """ + if isinstance(value, dict): + return value.get(key) + return None + + +def _list_value(value: object) -> list[object]: + """Return the value if it is a list, else an empty list. + + Args: + value: Any JSON value. + + Returns: + A list (possibly empty). + """ + if isinstance(value, list): + return value + return [] + + +def _list_of_dicts(value: object) -> list[dict[str, object]]: + """Filter a JSON value down to its dict elements. + + Args: + value: Any JSON value. + + Returns: + The dict elements of the list, or an empty list. + """ + return [item for item in _list_value(value) if isinstance(item, dict)] + + +def _str_list(value: object) -> list[str]: + """Filter a JSON value down to its string elements. + + Args: + value: Any JSON value. + + Returns: + The string elements of the list, or an empty list. + """ + return [item for item in _list_value(value) if isinstance(item, str)] + + +def _str_int_map(value: object) -> dict[str, int]: + """Coerce a JSON value to a ``str -> int`` counter mapping. + + Args: + value: Any JSON value. + + Returns: + A dict of the string keys to int counts, or an empty dict. + """ + if not isinstance(value, dict): + return {} + return {str(key): _as_int(val) for key, val in value.items()} diff --git a/python_pkg/session_autopsy/report.py b/python_pkg/session_autopsy/report.py new file mode 100644 index 00000000..14ef27d0 --- /dev/null +++ b/python_pkg/session_autopsy/report.py @@ -0,0 +1,377 @@ +"""Render REPORT.md, maintain state.json, and measure compiled skills.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +from python_pkg.session_autopsy.stats import parse_timestamp + +if TYPE_CHECKING: + from datetime import datetime + from pathlib import Path + + from python_pkg.session_autopsy.detectors import AnalysisResult, Candidate + from python_pkg.session_autopsy.records import SessionRecord + +REPORT_FILE = "REPORT.md" +STATE_FILE = "state.json" +COMPILED_FILE = "compiled.json" +TRACES_HINT = "PYTHONPATH=~/testsAndMisc python3 -m python_pkg.session_autopsy traces" +THOUSAND = 1_000 +MILLION = 1_000_000 +TABLE_TOP = 40 +DETAIL_TOP = 25 + + +def fmt_tokens(count: int) -> str: + """Format a token count for humans. + + Args: + count: Token count. + + Returns: + ``"842"``, ``"74k"``, or ``"1.2M"``. + """ + if count >= MILLION: + return f"{count / MILLION:.1f}M" + if count >= THOUSAND: + return f"{count // THOUSAND}k" + return str(count) + + +def load_state(home: Path) -> dict[str, object]: + """Read state.json, tolerating absence and corruption. + + Args: + home: The autopsy home directory. + + Returns: + The stored state dict, or an empty dict. + """ + path = home / STATE_FILE + if not path.is_file(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return {} + if isinstance(data, dict): + return data + return {} + + +def load_compiled(home: Path) -> list[dict[str, object]]: + """Read compiled.json entries, tolerating absence and corruption. + + Args: + home: The autopsy home directory. + + Returns: + The list of ``{candidate_id, skill, script, compiled_at}`` entries. + """ + path = home / COMPILED_FILE + if not path.is_file(): + return [] + try: + data = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return [] + if isinstance(data, list): + return [entry for entry in data if isinstance(entry, dict)] + return [] + + +def write_state( + home: Path, candidate_ids: list[str], now: datetime, *, mark_reviewed: bool = False +) -> int: + """Update state.json and return the unreviewed-candidate count. + + Args: + home: The autopsy home directory. + candidate_ids: Every current candidate id, ranked. + now: Current UTC time. + mark_reviewed: When True, every current candidate becomes reviewed. + + Returns: + The number of candidates not yet marked reviewed. + """ + state = load_state(home) + stored = state.get("reviewed_ids") + reviewed = ( + {item for item in stored if isinstance(item, str)} + if isinstance(stored, list) + else set() + ) + if mark_reviewed: + reviewed = set(candidate_ids) + reviewed &= set(candidate_ids) + unreviewed = len([cid for cid in candidate_ids if cid not in reviewed]) + payload = { + "unreviewed_count": unreviewed, + "candidate_ids": candidate_ids, + "reviewed_ids": sorted(reviewed), + "last_generated": now.isoformat(timespec="seconds"), + } + home.mkdir(parents=True, exist_ok=True) + (home / STATE_FILE).write_text( + json.dumps(payload, indent=2) + "\n", encoding="utf-8" + ) + return unreviewed + + +def measure_lines( + records: list[SessionRecord], + compiled: list[dict[str, object]], + only_skill: str | None, +) -> list[str]: + """Compare tokens/invocation before vs after each compilation. + + Args: + records: All session records. + compiled: Entries from compiled.json. + only_skill: Restrict output to one skill name, when given. + + Returns: + One human-readable line per compiled skill. + """ + lines = [] + for entry in compiled: + skill = str(entry.get("skill") or "") + if ( + not skill + or not entry.get("script") + or (only_skill is not None and skill != only_skill) + ): + continue + compiled_at = parse_timestamp(str(entry.get("compiled_at") or "") or None) + before: list[int] = [] + after: list[int] = [] + for record in records: + stamp = parse_timestamp(record.meta.started_at) + for invocation in record.obs.skill_invocations: + if invocation.name != skill: + continue + spent = invocation.tokens_output + invocation.tokens_cache_write + if ( + compiled_at is not None + and stamp is not None + and stamp >= compiled_at + ): + after.append(spent) + else: + before.append(spent) + lines.append(_measure_line(skill, before, after)) + if not lines: + lines.append( + "nothing compiled yet — run /compile-candidate on a ranked candidate" + ) + return lines + + +def _measure_line(skill: str, before: list[int], after: list[int]) -> str: + """Render one scoreboard line. + + Args: + skill: Skill name. + before: Tokens/invocation samples before compilation. + after: Samples after compilation. + + Returns: + The formatted line, with a delta when both sides have data. + """ + avg_before = sum(before) // len(before) if before else 0 + if not after: + return ( + f"- {skill}: before {fmt_tokens(avg_before)}/inv " + f"(n={len(before)}) — no post-compile invocations yet" + ) + avg_after = sum(after) // len(after) + delta = "" + if avg_before > 0: + delta = f" — {100 * (avg_after - avg_before) // avg_before:+d}%" + return ( + f"- {skill}: before {fmt_tokens(avg_before)}/inv (n={len(before)}) " + f"→ after {fmt_tokens(avg_after)}/inv (n={len(after)}){delta}" + ) + + +def render_report( + records: list[SessionRecord], + result: AnalysisResult, + compiled: list[dict[str, object]], + now: datetime, +) -> str: + """Render the full REPORT.md content. + + Args: + records: All session records. + result: Detector output. + compiled: Entries from compiled.json. + now: Current UTC time. + + Returns: + The Markdown document. + """ + parts = [_header(records, now), _candidate_table(result.candidates)] + parts.extend( + _candidate_detail(candidate) for candidate in result.candidates[:DETAIL_TOP] + ) + parts.append(_reviewed_keep_llm(compiled)) + parts.append(_trend_section(result)) + parts.append( + "## Compiled scoreboard\n\n" + + "\n".join(measure_lines(records, compiled, None)) + + "\n" + ) + return "\n".join(parts) + + +def _header(records: list[SessionRecord], now: datetime) -> str: + """Render the totals header. + + Args: + records: All session records. + now: Current UTC time. + + Returns: + The header block. + """ + turns = sum(record.counts.assistant_msgs for record in records) + output = sum(record.tokens.output for record in records) + cache_write = sum(record.tokens.cache_write for record in records) + cache_read = sum(record.tokens.cache_read for record in records) + per_turn = cache_read // turns if turns else 0 + return ( + "# Session Autopsy Report\n\n" + f"Generated: {now.isoformat(timespec='seconds')} | Sessions: {len(records)} | " + f"Assistant msgs: {turns}\n" + f"Tokens: {fmt_tokens(output)} output | " + f"{fmt_tokens(cache_write)} cache-write | " + f"{fmt_tokens(cache_read)} cache-read (avg {fmt_tokens(per_turn)}/turn)\n" + ) + + +def _candidate_table(candidates: list[Candidate]) -> str: + """Render the ranked candidate table. + + Args: + candidates: Ranked candidates. + + Returns: + The Markdown table (or a placeholder when empty). + """ + if not candidates: + return "## Ranked automation candidates\n\nnone yet — need more sessions\n" + rows = [ + f"| {rank} | {cand.id} | {cand.kind} | " + f"{cand.occurrences}x / {cand.sessions} sessions " + f"| {fmt_tokens(cand.est_weekly_savings)}/wk | {cand.action} |" + for rank, cand in enumerate(candidates[:TABLE_TOP], start=1) + ] + overflow = "" + if len(candidates) > TABLE_TOP: + overflow = ( + f"\n… plus {len(candidates) - TABLE_TOP} below the fold — " + "see `candidates --json` for all.\n" + ) + return ( + "## Ranked automation candidates\n\n" + "| # | id | kind | evidence | est. weekly tokens | action |\n" + "|---|----|------|----------|--------------------|--------|\n" + + "\n".join(rows) + + "\n" + + overflow + ) + + +def _candidate_detail(candidate: Candidate) -> str: + """Render one candidate's detail block. + + Args: + candidate: The candidate. + + Returns: + The detail block with the exact traces command. + """ + return ( + f"### {candidate.id} — {candidate.title}\n\n" + f"- score {candidate.score:.2f}, avg " + f"{fmt_tokens(candidate.avg_tokens)} tokens/occurrence, " + f"{candidate.per_week:.1f}/week\n" + f"- signature: `{candidate.sig}`\n" + f"- traces: `{TRACES_HINT} {candidate.id}`\n" + ) + + +def _reviewed_keep_llm(compiled: list[dict[str, object]]) -> str: + """Render candidates already reviewed and ruled judgment-heavy. + + These come from /compile-candidate verdicts in compiled.json, not from a + heuristic: real-trace review decided the workflow needs the LLM. + + Args: + compiled: Entries from compiled.json. + + Returns: + The section text. + """ + kept = [entry for entry in compiled if entry.get("verdict") == "keep-llm"] + if not kept: + return "## Reviewed — keeping LLM\n\nnone reviewed yet\n" + lines = [ + f"- {entry.get('candidate_id')} — " + f"reviewed {entry.get('compiled_at')} — keep LLM" + for entry in kept + ] + return "## Reviewed — keeping LLM\n\n" + "\n".join(lines) + "\n" + + +def _trend_section(result: AnalysisResult) -> str: + """Render the context-bloat trend section. + + Args: + result: Detector output holding the trend. + + Returns: + The section text. + """ + trend = result.trend + if not trend.weekly: + return "## Context-bloat trend\n\nno dated sessions yet\n" + weeks = "\n".join( + f"- {week}: {fmt_tokens(avg)}/turn" for week, avg in trend.weekly[-8:] + ) + offenders = ", ".join( + f"{name} ({fmt_tokens(avg)}/turn)" for name, avg in trend.top_sessions + ) + return ( + "## Context-bloat trend\n\n" + f"slope: {fmt_tokens(int(trend.slope_per_week))}/turn per week\n" + f"heaviest sessions: {offenders}\n\n{weeks}\n" + ) + + +def write_report( + home: Path, + records: list[SessionRecord], + result: AnalysisResult, + now: datetime, +) -> int: + """Write REPORT.md and refresh state.json. + + Args: + home: The autopsy home directory. + records: All session records. + result: Detector output. + now: Current UTC time. + + Returns: + The unreviewed-candidate count (for the caller's summary line). + """ + home.mkdir(parents=True, exist_ok=True) + compiled = load_compiled(home) + (home / REPORT_FILE).write_text( + render_report(records, result, compiled, now), encoding="utf-8" + ) + return write_state(home, [cand.id for cand in result.candidates], now) diff --git a/python_pkg/session_autopsy/signatures.py b/python_pkg/session_autopsy/signatures.py new file mode 100644 index 00000000..cbf80203 --- /dev/null +++ b/python_pkg/session_autopsy/signatures.py @@ -0,0 +1,67 @@ +"""Stable text signatures for messages, commands and errors. + +Two modules need these: :mod:`python_pkg.session_autopsy.parse` while streaming +a transcript, and :mod:`python_pkg.session_autopsy.traces` when re-reading one to +pull evidence. They live here rather than in either of those so neither has to +import the other. +""" + +from __future__ import annotations + +import re + +MAX_SIG_LEN = 120 + +_PATH_RE = re.compile(r"(?:~|/)[\w.@+/-]{4,}") +_HEX_RE = re.compile(r"[0-9a-fA-F]{8,}") +_NUM_RE = re.compile(r"\b\d+\b") +_WS_RE = re.compile(r"\s+") +# Wrapper commands whose real command is the following word. +_WRAPPERS = frozenset({"cd", "sudo", "env", "nice", "command", "timeout"}) + + +def normalize_signature(text: str) -> str: + """Collapse a message/command/error into a stable cross-session signature. + + Takes the first line only and masks paths, long hex runs, and numbers so + that reoccurrences of the same event count as one signature. + + Args: + text: Raw text (may be multi-line). + + Returns: + The normalized first line, truncated to :data:`MAX_SIG_LEN` chars. + """ + stripped = text.strip() + if not stripped: + return "" + first = stripped.splitlines()[0] + first = _PATH_RE.sub("", first) + first = _HEX_RE.sub("", first) + first = _NUM_RE.sub("", first) + return _WS_RE.sub(" ", first).strip()[:MAX_SIG_LEN] + + +def command_signature(command: str) -> str: + """Reduce a Bash command to its leading command word(s). + + Wrapper commands (``cd``, ``sudo``, ``timeout`` …) keep their first + non-flag, non-numeric argument so sequences stay meaningful. + + Args: + command: The raw (possibly multi-line) Bash tool input. + + Returns: + A short signature such as ``"git"`` or ``"cd "``, or ``""``. + """ + normalized = normalize_signature(command) + words = normalized.split() + if not words: + return "" + head = words[0] + if head not in _WRAPPERS: + return head + for word in words[1:]: + if not word.startswith(("-", "")): + return f"{head} {word}" + return head diff --git a/python_pkg/session_autopsy/stats.py b/python_pkg/session_autopsy/stats.py new file mode 100644 index 00000000..b9cf3f83 --- /dev/null +++ b/python_pkg/session_autopsy/stats.py @@ -0,0 +1,120 @@ +"""Small numeric and time helpers shared by the detectors and the report. + +These are pure functions over plain numbers, timestamps and token totals — no +knowledge of what a candidate is. Keeping them here leaves +:mod:`python_pkg.session_autopsy.detectors` to the heuristics themselves. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from difflib import SequenceMatcher +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from python_pkg.session_autopsy.records import SessionRecord + +TRAILING_WEEKS = 8 +MAX_SIMILARITY_INVOCATIONS = 5 +MIN_SLOPE_SAMPLES = 2 + + +def parse_timestamp(value: str | None) -> datetime | None: + """Parse an ISO-8601 transcript timestamp (with trailing ``Z``). + + Args: + value: The timestamp string, or None. + + Returns: + A timezone-aware datetime, or None when absent/invalid. + """ + if not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + + +def per_turn_avg(records: list[SessionRecord]) -> int: + """Average output+cache-write tokens per assistant turn, corpus-wide. + + Args: + records: All session records. + + Returns: + The average, or 0 for an empty corpus. + """ + turns = sum(record.counts.assistant_msgs for record in records) + if turns == 0: + return 0 + spent = sum(record.tokens.output + record.tokens.cache_write for record in records) + return spent // turns + + +def per_week(timestamps: list[datetime | None], now: datetime) -> float: + """Occurrences per week over the trailing window. + + Args: + timestamps: One entry per occurrence (None when the session had no + parseable start time; those never count as recent). + now: Current UTC time. + + Returns: + Recent occurrences divided by :data:`TRAILING_WEEKS`. + """ + cutoff = now - timedelta(weeks=TRAILING_WEEKS) + recent = sum(1 for stamp in timestamps if stamp is not None and stamp >= cutoff) + return recent / TRAILING_WEEKS + + +def mean(values: list[float]) -> float: + """Arithmetic mean, 0.0 for an empty list. + + Args: + values: Sample values. + + Returns: + The mean. + """ + if not values: + return 0.0 + return sum(values) / len(values) + + +def mean_pairwise_similarity(sequences: list[list[str]]) -> float: + """Mean SequenceMatcher ratio over the first few invocation pairs. + + Args: + sequences: One Bash-signature sequence per invocation. + + Returns: + Mean similarity in [0, 1]; 0.0 when fewer than two non-empty + sequences exist (nothing to compare means no evidence of stability). + """ + sample = [seq for seq in sequences if seq][:MAX_SIMILARITY_INVOCATIONS] + ratios = [ + SequenceMatcher(None, sample[i], sample[j]).ratio() + for i in range(len(sample)) + for j in range(i + 1, len(sample)) + ] + return mean(ratios) + + +def slope(values: list[float]) -> float: + """Least-squares slope of evenly spaced samples. + + Args: + values: One sample per week, in order. + + Returns: + Tokens-per-turn change per week; 0.0 with fewer than two samples. + """ + count = len(values) + if count < MIN_SLOPE_SAMPLES: + return 0.0 + mean_x = (count - 1) / 2 + mean_y = sum(values) / count + numerator = sum((i - mean_x) * (val - mean_y) for i, val in enumerate(values)) + denominator = sum((i - mean_x) ** 2 for i in range(count)) + return numerator / denominator diff --git a/python_pkg/session_autopsy/store.py b/python_pkg/session_autopsy/store.py new file mode 100644 index 00000000..334dd334 --- /dev/null +++ b/python_pkg/session_autopsy/store.py @@ -0,0 +1,136 @@ +"""Flock-guarded persistence for sessions.jsonl. + +The store is small (~1 record per session, hundreds total), so every write +is a full read-modify-replace under an exclusive lock; this keeps concurrent +SessionEnd hook firings and a parallel ``scan`` safe without a database. +""" + +from __future__ import annotations + +import fcntl +import json +from typing import TYPE_CHECKING + +from typing_extensions import Self + +from python_pkg.session_autopsy.records import SessionRecord, record_from_dict + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +SESSIONS_FILE = "sessions.jsonl" +LOCK_FILE = ".lock" + + +class StoreLock: + """Context manager holding the store's exclusive flock.""" + + def __init__(self, home: Path) -> None: + """Remember the store directory. + + Args: + home: The autopsy home directory. + """ + self._lock_path = home / LOCK_FILE + self._handle = None + + def __enter__(self) -> Self: + """Create the home dir if needed and take the exclusive lock. + + Returns: + Self, with the lock held. + """ + self._lock_path.parent.mkdir(parents=True, exist_ok=True) + self._handle = self._lock_path.open("a") + fcntl.flock(self._handle, fcntl.LOCK_EX) + return self + + def __exit__(self, *exc_info: object) -> None: + """Release the lock and close the handle. + + Args: + exc_info: Ignored exception triple. + """ + if self._handle is not None: + fcntl.flock(self._handle, fcntl.LOCK_UN) + self._handle.close() + self._handle = None + + +def _iter_stored(home: Path) -> Iterator[SessionRecord]: + """Yield every record currently in sessions.jsonl. + + Args: + home: The autopsy home directory. + + Yields: + Stored records; malformed lines are skipped silently. + """ + sessions_path = home / SESSIONS_FILE + if not sessions_path.is_file(): + return + with sessions_path.open(encoding="utf-8") as handle: + for raw_line in handle: + stripped = raw_line.strip() + if not stripped: + continue + try: + data = json.loads(stripped) + except json.JSONDecodeError: + continue + if isinstance(data, dict): + yield record_from_dict(data) + + +def load_records(home: Path) -> list[SessionRecord]: + """Load all stored records under the lock. + + Args: + home: The autopsy home directory. + + Returns: + All stored records, in file order. + """ + with StoreLock(home): + return list(_iter_stored(home)) + + +def load_file_index(home: Path) -> dict[str, tuple[int, float]]: + """Map transcript paths to their (size, mtime) at last analysis. + + Used by ``scan`` to skip transcripts that have not changed. + + Args: + home: The autopsy home directory. + + Returns: + ``{transcript_path: (file_size, file_mtime)}`` for every record. + """ + with StoreLock(home): + return { + record.transcript_path: (record.file_size, record.file_mtime) + for record in _iter_stored(home) + } + + +def upsert_records(home: Path, new_records: list[SessionRecord]) -> int: + """Insert or replace records by session id, atomically. + + Args: + home: The autopsy home directory. + new_records: Records to insert (replacing same-session entries). + + Returns: + The total number of stored records after the write. + """ + with StoreLock(home): + merged = {record.session_id: record for record in _iter_stored(home)} + for record in new_records: + merged[record.session_id] = record + tmp_path = home / (SESSIONS_FILE + ".tmp") + with tmp_path.open("w", encoding="utf-8") as handle: + for record in merged.values(): + handle.write(json.dumps(record.to_dict(), separators=(",", ":")) + "\n") + tmp_path.replace(home / SESSIONS_FILE) + return len(merged) diff --git a/python_pkg/session_autopsy/tests/__init__.py b/python_pkg/session_autopsy/tests/__init__.py new file mode 100644 index 00000000..bc6548c7 --- /dev/null +++ b/python_pkg/session_autopsy/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for the session_autopsy package.""" diff --git a/python_pkg/session_autopsy/tests/conftest.py b/python_pkg/session_autopsy/tests/conftest.py new file mode 100644 index 00000000..b915c947 --- /dev/null +++ b/python_pkg/session_autopsy/tests/conftest.py @@ -0,0 +1,218 @@ +"""Shared fixtures and builders for the session-autopsy tests. + +Fixtures isolate the store and transcript root per test; the builders below make +synthetic transcript lines, transcript files, and records. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +from typing import TYPE_CHECKING + +import pytest + +from python_pkg.session_autopsy.records import ( + ActivityCounts, + Observations, + SessionMeta, + SessionRecord, + SkillInvocation, + TokenTotals, +) + +if TYPE_CHECKING: + from pathlib import Path + + +@dataclass +class AutopsyEnv: + """Paths of the per-test isolated store and transcript root.""" + + home: Path + projects: Path + + +@pytest.fixture +def autopsy_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> AutopsyEnv: + """Point the analyzer at temp dirs so tests never touch ~/.claude.""" + home = tmp_path / "autopsy" + projects = tmp_path / "projects" + projects.mkdir() + monkeypatch.setenv("CLAUDE_AUTOPSY_HOME", str(home)) + monkeypatch.setenv("CLAUDE_PROJECTS_DIR", str(projects)) + return AutopsyEnv(home=home, projects=projects) + + +Line = dict[str, object] + +STAMP = "2026-07-20T10:00:00.000Z" + + +def usage_block(output: int = 10, cache_write: int = 20) -> dict[str, int]: + """A minimal assistant usage block.""" + return { + "output_tokens": output, + "input_tokens": 1, + "cache_read_input_tokens": 100, + "cache_creation_input_tokens": cache_write, + } + + +def text_block(text: str) -> Line: + """A text content block.""" + return {"type": "text", "text": text} + + +def tool_block(name: str, **tool_input: object) -> Line: + """A tool_use content block.""" + return {"type": "tool_use", "name": name, "input": dict(tool_input)} + + +def bash_block(command: str) -> Line: + """A Bash tool_use block.""" + return tool_block("Bash", command=command) + + +def skill_block(skill: str) -> Line: + """A Skill tool_use block.""" + return tool_block("Skill", skill=skill) + + +def assistant_line( + *blocks: Line, timestamp: str = STAMP, with_usage: bool = True +) -> Line: + """An assistant transcript line with the given content blocks.""" + message: dict[str, object] = {"role": "assistant", "content": list(blocks)} + if with_usage: + message["usage"] = usage_block() + return {"type": "assistant", "timestamp": timestamp, "message": message} + + +def prompt_line( + content: str, + *, + user_type: str = "external", + sidechain: bool = False, + timestamp: str = STAMP, +) -> Line: + """A user transcript line with string content (a typed prompt).""" + return { + "type": "user", + "userType": user_type, + "isSidechain": sidechain, + "timestamp": timestamp, + "message": {"role": "user", "content": content}, + } + + +def tool_result_line(result: object) -> Line: + """A user transcript line carrying a toolUseResult.""" + return { + "type": "user", + "userType": "external", + "message": {"role": "user", "content": [{"type": "tool_result"}]}, + "toolUseResult": result, + } + + +def write_transcript( + projects: Path, + session_id: str, + lines: list[Line | str], + project: str = "-home-kuhy", +) -> Path: + """Write transcript lines (dicts, or raw strings verbatim) as JSONL.""" + project_dir = projects / project + project_dir.mkdir(parents=True, exist_ok=True) + path = project_dir / f"{session_id}.jsonl" + rendered = [line if isinstance(line, str) else json.dumps(line) for line in lines] + path.write_text("\n".join(rendered) + "\n", encoding="utf-8") + return path + + +def invocation( + name: str = "finish", + tool_only: int = 6, + text: int = 4, + output: int = 50_000, + cache_write: int = 100_000, + bash_sig: list[str] | None = None, +) -> SkillInvocation: + """A skill invocation span with sensible defaults.""" + return SkillInvocation( + name=name, + tool_only_turns=tool_only, + text_turns=text, + tokens_output=output, + tokens_cache_write=cache_write, + bash_sig=bash_sig if bash_sig is not None else ["pre-commit", "pytest", "git"], + ) + + +@dataclass(frozen=True) +class FileFacts: + """The transcript-file scalars a record carries, defaulted for tests.""" + + path: str | None = None + size: int = 100 + mtime: float = 1.0 + + +def session_meta( + started_at: str | None = STAMP, slug: str | None = None +) -> SessionMeta: + """SessionMeta carrying the standard test timestamp unless overridden. + + Args: + started_at: Session start stamp; defaults to :data:`STAMP`. + slug: Optional project slug. + + Returns: + The metadata block. + """ + return SessionMeta(started_at=started_at, slug=slug) + + +def record( + session_id: str = "s1", + *, + file: FileFacts | None = None, + meta: SessionMeta | None = None, + counts: ActivityCounts | None = None, + tokens: TokenTotals | None = None, + obs: Observations | None = None, +) -> SessionRecord: + """A SessionRecord with sensible defaults; override a whole block at a time. + + Grouping the overrides by the record's own blocks is what keeps this builder + to six parameters; the flat kwargs it used to take were one per field. + + Args: + session_id: The session id, also used for the default transcript path. + file: Transcript-file scalars. + meta: Session metadata; defaults to :func:`session_meta`. + counts: Activity counts; defaults to ten assistant messages. + tokens: Token totals; defaults to a typical spend. + obs: Observations; defaults to empty. + + Returns: + The record. + """ + facts = file or FileFacts() + return SessionRecord( + session_id=session_id, + project_slug="-home-kuhy", + transcript_path=facts.path or f"/nonexistent/{session_id}.jsonl", + file_size=facts.size, + file_mtime=facts.mtime, + analyzed_at="2026-07-20T10:00:00+00:00", + meta=meta if meta is not None else session_meta(), + counts=counts if counts is not None else ActivityCounts(assistant_msgs=10), + tokens=tokens + if tokens is not None + else TokenTotals( + output=10_000, input=100, cache_read=500_000, cache_write=20_000 + ), + obs=obs if obs is not None else Observations(), + ) diff --git a/python_pkg/session_autopsy/tests/test_cli.py b/python_pkg/session_autopsy/tests/test_cli.py new file mode 100644 index 00000000..706f5fcf --- /dev/null +++ b/python_pkg/session_autopsy/tests/test_cli.py @@ -0,0 +1,194 @@ +"""Tests for the CLI, config env overrides, and the module entry point.""" + +from __future__ import annotations + +import json +import runpy +import sys +from typing import TYPE_CHECKING + +import pytest + +from python_pkg.session_autopsy import cli, config, report, store +from python_pkg.session_autopsy.records import Observations +from python_pkg.session_autopsy.tests.conftest import ( + FileFacts, + assistant_line, + bash_block, + invocation, + prompt_line, + record, + skill_block, + text_block, + write_transcript, +) + +if TYPE_CHECKING: + from pathlib import Path + + from python_pkg.session_autopsy.tests.conftest import AutopsyEnv + + +def test_config_env_overrides(autopsy_env: AutopsyEnv) -> None: + """Both locations follow their environment variables.""" + assert config.autopsy_home() == autopsy_env.home + assert config.projects_dir() == autopsy_env.projects + + +def test_config_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + """Without overrides, both locations live under ~/.claude.""" + monkeypatch.delenv("CLAUDE_AUTOPSY_HOME", raising=False) + monkeypatch.delenv("CLAUDE_PROJECTS_DIR", raising=False) + assert config.autopsy_home().name == "autopsy" + assert config.projects_dir().name == "projects" + + +def _seed_transcript(env: AutopsyEnv, session_id: str = "sess-1") -> Path: + """Write a small but representative transcript into the projects dir.""" + return write_transcript( + env.projects, + session_id, + [ + assistant_line(skill_block("finish")), + assistant_line(bash_block("git status"), text_block("checking")), + prompt_line("wrap it up please"), + ], + ) + + +def test_ingest_missing_file( + autopsy_env: AutopsyEnv, capsys: pytest.CaptureFixture[str] +) -> None: + """A nonexistent transcript exits 2 with a stderr message.""" + assert ( + cli.main(["ingest", str(autopsy_env.projects / "nope.jsonl")]) + == cli.EXIT_NO_FILE + ) + assert "no such transcript" in capsys.readouterr().err + + +def test_ingest_and_report_regeneration( + autopsy_env: AutopsyEnv, capsys: pytest.CaptureFixture[str] +) -> None: + """ingest upserts, prints a summary, and regenerates REPORT.md.""" + transcript = _seed_transcript(autopsy_env) + assert cli.main(["ingest", str(transcript), "--no-report"]) == 0 + out = capsys.readouterr().out + assert "sess-1" in out + assert not (autopsy_env.home / report.REPORT_FILE).exists() + assert cli.main(["ingest", str(transcript), "--quiet"]) == 0 + assert capsys.readouterr().out == "" + assert (autopsy_env.home / report.REPORT_FILE).is_file() + + +def test_scan_skip_force_and_nonfile( + autopsy_env: AutopsyEnv, capsys: pytest.CaptureFixture[str] +) -> None: + """scan analyzes changed transcripts, skips unchanged, honors --force.""" + _seed_transcript(autopsy_env, "s1") + _seed_transcript(autopsy_env, "s2") + (autopsy_env.projects / "-home-kuhy" / "dir.jsonl").mkdir() + assert cli.main(["scan", "--all", "--jobs", "0", "--no-report"]) == 0 + assert "2 analyzed, 0 unchanged" in capsys.readouterr().out + assert cli.main(["scan", "--no-report"]) == 0 + assert "0 analyzed, 2 unchanged" in capsys.readouterr().out + assert cli.main(["scan", "--force"]) == 0 + assert "2 analyzed" in capsys.readouterr().out + + +def test_needs_analysis_on_size_change(autopsy_env: AutopsyEnv) -> None: + """A stored (size, mtime) mismatch marks the transcript changed.""" + transcript = _seed_transcript(autopsy_env) + index = {str(transcript): (transcript.stat().st_size, transcript.stat().st_mtime)} + assert not cli._needs_analysis(transcript, index) + index[str(transcript)] = (1, 1.0) + assert cli._needs_analysis(transcript, index) + + +def test_parse_one_swallows_oserror( + autopsy_env: AutopsyEnv, capsys: pytest.CaptureFixture[str] +) -> None: + """A vanished transcript logs to stderr and yields None.""" + assert cli._parse_one(autopsy_env.projects / "gone.jsonl") is None + assert "skipping" in capsys.readouterr().err + + +def test_regenerate_report_with_preloaded_records(autopsy_env: AutopsyEnv) -> None: + """The records argument skips the store read.""" + assert cli._regenerate_report([]) == 0 + + +def _seed_store(env: AutopsyEnv) -> None: + """Store records that produce one deterministic skill candidate.""" + transcript = _seed_transcript(env, "t1") + store.upsert_records( + env.home, + [ + record( + "t1", + file=FileFacts(path=str(transcript)), + obs=Observations(skill_invocations=[invocation()]), + ), + record("t2", obs=Observations(skill_invocations=[invocation()])), + record("t3", obs=Observations(skill_invocations=[invocation()])), + ], + ) + + +def test_report_and_mark_reviewed( + autopsy_env: AutopsyEnv, capsys: pytest.CaptureFixture[str] +) -> None: + """report writes the file; --mark-reviewed zeroes the badge count.""" + _seed_store(autopsy_env) + assert cli.main(["report"]) == 0 + assert "candidates" in capsys.readouterr().out + assert cli.main(["report", "--mark-reviewed"]) == 0 + assert "marked reviewed" in capsys.readouterr().out + state = json.loads( + (autopsy_env.home / report.STATE_FILE).read_text(encoding="utf-8") + ) + assert state["unreviewed_count"] == 0 + + +def test_candidates_json_and_plain( + autopsy_env: AutopsyEnv, capsys: pytest.CaptureFixture[str] +) -> None: + """candidates renders both machine and human output.""" + _seed_store(autopsy_env) + assert cli.main(["candidates", "--json"]) == 0 + payload = json.loads(capsys.readouterr().out) + assert payload[0]["id"] == "skill-finish" + assert cli.main(["candidates"]) == 0 + assert "skill-finish" in capsys.readouterr().out + + +def test_traces_unknown_and_output_modes( + autopsy_env: AutopsyEnv, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """traces exits 3 on unknown ids and writes --out files.""" + _seed_store(autopsy_env) + assert cli.main(["traces", "nope-123"]) == cli.EXIT_UNKNOWN_ID + assert "unknown candidate id" in capsys.readouterr().err + out_file = tmp_path / "traces.txt" + assert cli.main(["traces", "skill-finish", "--out", str(out_file)]) == 0 + assert "TOOL Skill | finish" in out_file.read_text(encoding="utf-8") + assert cli.main(["traces", "skill-finish", "--max-invocations", "1"]) == 0 + assert "traces skill-finish" in capsys.readouterr().out + + +def test_measure_command( + autopsy_env: AutopsyEnv, capsys: pytest.CaptureFixture[str] +) -> None: + """measure prints the placeholder when nothing is compiled.""" + assert cli.main(["measure"]) == 0 + assert "nothing compiled yet" in capsys.readouterr().out + + +def test_main_module_entry( + autopsy_env: AutopsyEnv, monkeypatch: pytest.MonkeyPatch +) -> None: + """python -m python_pkg.session_autopsy dispatches and exits.""" + monkeypatch.setattr(sys, "argv", ["session_autopsy", "measure"]) + with pytest.raises(SystemExit) as excinfo: + runpy.run_module("python_pkg.session_autopsy", run_name="__main__") + assert excinfo.value.code == 0 diff --git a/python_pkg/session_autopsy/tests/test_detectors.py b/python_pkg/session_autopsy/tests/test_detectors.py new file mode 100644 index 00000000..67127fe4 --- /dev/null +++ b/python_pkg/session_autopsy/tests/test_detectors.py @@ -0,0 +1,240 @@ +"""Tests for the pure candidate detectors.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from python_pkg.session_autopsy import detectors, stats +from python_pkg.session_autopsy.detectors import analyze +from python_pkg.session_autopsy.records import ( + ActivityCounts, + Observations, + TokenTotals, +) +from python_pkg.session_autopsy.stats import parse_timestamp +from python_pkg.session_autopsy.tests.conftest import ( + invocation, + record, + session_meta, +) + +NOW = datetime(2026, 7, 24, 12, 0, tzinfo=timezone.utc) +OLD = "2026-01-01T00:00:00.000Z" + + +def test_parse_timestamp() -> None: + """None, empty, junk, and Z-suffixed ISO forms.""" + assert parse_timestamp(None) is None + assert parse_timestamp("") is None + assert parse_timestamp("junk") is None + stamp = parse_timestamp("2026-07-20T10:00:00.000Z") + assert stamp is not None + assert stamp.tzinfo is not None + + +def test_analyze_empty_corpus() -> None: + """No records → no candidates, empty trend, zero per-turn average.""" + result = analyze([], NOW) + assert result.candidates == [] + assert result.trend.weekly == [] + assert result.trend.slope_per_week == 0.0 + + +def test_skill_candidates_threshold_and_score() -> None: + """Skills need >=3 invocations; zero-turn spans don't skew determinism.""" + records = [ + record( + "a", + obs=Observations( + skill_invocations=[invocation(), invocation(tool_only=0, text=0)] + ), + ), + record("b", obs=Observations(skill_invocations=[invocation()])), + record("c", obs=Observations(skill_invocations=[invocation(name="rare")])), + ] + result = analyze(records, NOW) + skills = [cand for cand in result.candidates if cand.kind == "skill"] + assert [cand.id for cand in skills] == ["skill-finish"] + finish = skills[0] + assert finish.occurrences == 3 + assert finish.sessions == 2 + assert finish.score == 0.6 + assert finish.avg_tokens == 150_000 + assert finish.per_week == 3 / 8 + assert finish.est_weekly_savings == int(150_000 * 3 / 8) + assert "compile-candidate skill-finish" in finish.action + + +def test_skill_similarity_zero_without_sequences() -> None: + """All-empty bash signatures give zero similarity evidence.""" + records = [ + record("a", obs=Observations(skill_invocations=[invocation(bash_sig=[])])), + record("b", obs=Observations(skill_invocations=[invocation(bash_sig=[])])), + record("c", obs=Observations(skill_invocations=[invocation(bash_sig=[])])), + ] + result = analyze(records, NOW) + assert result.candidates[0].score == 0.0 + + +def test_old_invocations_have_zero_weekly_rate() -> None: + """Occurrences outside the trailing window contribute no weekly rate.""" + records = [ + record( + "a", + meta=session_meta(started_at=OLD), + obs=Observations(skill_invocations=[invocation()]), + ), + record( + "b", + meta=session_meta(started_at=OLD), + obs=Observations(skill_invocations=[invocation()]), + ), + record( + "c", + meta=session_meta(started_at=None), + obs=Observations(skill_invocations=[invocation()]), + ), + ] + result = analyze(records, NOW) + assert result.candidates[0].per_week == 0.0 + assert result.candidates[0].est_weekly_savings == 0 + + +def test_error_candidates_thresholds() -> None: + """Errors qualify by total occurrences OR by session spread.""" + records = [ + record("a", obs=Observations(error_signatures={"boom: ": 10, "meh": 2})), + record("b", obs=Observations(error_signatures={"spread: x": 1})), + record("c", obs=Observations(error_signatures={"spread: x": 1})), + record("d", obs=Observations(error_signatures={"spread: x": 1})), + ] + result = analyze(records, NOW) + errs = {cand.sig: cand for cand in result.candidates if cand.kind == "err"} + assert set(errs) == {"boom: ", "spread: x"} + assert errs["boom: "].occurrences == 10 + assert errs["spread: x"].sessions == 3 + + +def test_prompt_candidates_filters() -> None: + """Prompts need >=4 occurrences and >=3 words.""" + records = [ + record( + "a", + obs=Observations( + typed_prompt_signatures={ + "do the big thing": 4, + "hi there friend": 3, + "short one": 9, + } + ), + ), + ] + result = analyze(records, NOW) + prompts = [cand.sig for cand in result.candidates if cand.kind == "prompt"] + assert prompts == ["do the big thing"] + + +def test_ngram_candidates_and_subgram_drop() -> None: + """Frequent distinct sequences survive; subgrams and degenerates drop.""" + seq = ["a", "b", "c", "d"] + degenerate = ["x", "x", "x", "x", "x", "x"] + records = [ + record(sid, obs=Observations(bash_sequences=[seq, seq, degenerate])) + for sid in ("s1", "s2", "s3", "s4") + ] + result = analyze(records, NOW) + ngrams = [cand for cand in result.candidates if cand.kind == "ngram"] + assert [cand.sig for cand in ngrams] == ["a → b → c → d"] + assert ngrams[0].occurrences == 8 + assert ngrams[0].sessions == 4 + + +def test_ngram_below_session_threshold() -> None: + """Many occurrences in too few sessions do not qualify.""" + records = [record("only", obs=Observations(bash_sequences=[["a", "b", "c"]] * 20))] + result = analyze(records, NOW) + assert [cand for cand in result.candidates if cand.kind == "ngram"] == [] + + +def test_contains_and_drop_subgrams() -> None: + """Containment handles equal-length and absent cases.""" + assert detectors._contains(("a", "b", "c"), ("b", "c")) + assert not detectors._contains(("a", "b"), ("a", "b")) + assert not detectors._contains(("a", "b", "c"), ("c", "d")) + kept = detectors._drop_subgrams({("a", "b", "c"): 9, ("a", "b"): 9, ("z", "q"): 9}) + assert kept == {("a", "b", "c"), ("z", "q")} + + +def test_trend_weekly_slope_and_offenders() -> None: + """Weekly averages, slope, and cache-heaviest sessions.""" + records = [ + record( + "w1", + tokens=TokenTotals(cache_read=1000000), + meta=session_meta(started_at="2026-07-06T10:00:00Z"), + counts=ActivityCounts(assistant_msgs=10), + ), + record( + "w2", + tokens=TokenTotals(cache_read=2000000), + meta=session_meta(started_at="2026-07-13T10:00:00Z"), + counts=ActivityCounts(assistant_msgs=10), + ), + record( + "w3", + tokens=TokenTotals(cache_read=3000000), + meta=session_meta(started_at="2026-07-20T10:00:00Z", slug="heavy-slug"), + counts=ActivityCounts(assistant_msgs=10), + ), + record( + "w4", + tokens=TokenTotals(cache_read=100), + meta=session_meta(started_at="2026-07-21T10:00:00Z"), + counts=ActivityCounts(assistant_msgs=10), + ), + record("undated", meta=session_meta(started_at=None)), + record("silent", counts=ActivityCounts(assistant_msgs=0)), + ] + result = analyze(records, NOW) + trend = result.trend + assert len(trend.weekly) == 3 + assert trend.slope_per_week > 0 + assert trend.top_sessions[0] == ("heavy-slug", 300_000) + assert len(trend.top_sessions) == 3 + + +def test_slope_needs_two_points() -> None: + """A single weekly sample yields slope 0.""" + assert stats.slope([5.0]) == 0.0 + + +def test_candidate_to_dict() -> None: + """Serialization exposes rounded score and per_week.""" + records = [record("a", obs=Observations(skill_invocations=[invocation()] * 3))] + cand = analyze(records, NOW).candidates[0] + data = cand.to_dict() + assert data["id"] == "skill-finish" + assert isinstance(data["per_week"], float) + assert isinstance(data["session_ids"], list) + + +def test_recent_session_ids_cap_and_order() -> None: + """Evidence session ids are newest-first, capped at 10, deduplicated.""" + many = [ + record( + f"s{index:02d}", + meta=session_meta(started_at=f"2026-07-{index + 1:02d}T00:00:00Z"), + obs=Observations(skill_invocations=[invocation()]), + ) + for index in range(12) + ] + many.append( + record( + "s00", + meta=session_meta(started_at="2026-07-01T00:00:00Z"), + obs=Observations(skill_invocations=[invocation()]), + ) + ) + cand = analyze(many, NOW).candidates[0] + assert len(cand.session_ids) == 10 + assert cand.session_ids[0] == "s11" diff --git a/python_pkg/session_autopsy/tests/test_parse.py b/python_pkg/session_autopsy/tests/test_parse.py new file mode 100644 index 00000000..9ebb40cb --- /dev/null +++ b/python_pkg/session_autopsy/tests/test_parse.py @@ -0,0 +1,307 @@ +"""Tests for the streaming transcript parser.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +from python_pkg.session_autopsy import parse +from python_pkg.session_autopsy.parse import ( + _Accumulator, + parse_session, + subagent_files, +) +from python_pkg.session_autopsy.tests.conftest import ( + assistant_line, + bash_block, + prompt_line, + skill_block, + text_block, + tool_block, + tool_result_line, + write_transcript, +) + +if TYPE_CHECKING: + from pathlib import Path + + +def _feed(acc: _Accumulator, *lines: dict[str, object]) -> None: + """Feed dict lines through the accumulator as JSON.""" + for line in lines: + acc.feed_line(json.dumps(line)) + + +def test_feed_line_malformed() -> None: + """Blank, invalid-JSON, and non-dict lines are tolerated and counted.""" + acc = _Accumulator() + acc.feed_line("") + acc.feed_line(" ") + acc.feed_line("{not json") + acc.feed_line("[1, 2]") + assert acc.counts.malformed_lines == 2 + + +def test_dispatch_types() -> None: + """Titles, compact boundaries, and unknown types route correctly.""" + acc = _Accumulator() + _feed( + acc, + {"type": "ai-title", "title": "My Session"}, + {"type": "ai-title", "aiTitle": "Renamed", "title": ""}, + {"type": "ai-title"}, + {"type": "system", "subtype": "compact_boundary"}, + {"type": "system", "subtype": "other"}, + {"type": "attachment"}, + ) + assert acc.meta.title == "Renamed" + assert acc.counts.compaction_count == 1 + + +def test_assistant_usage_and_metadata() -> None: + """Usage sums; first/last timestamps and first cwd/branch/slug stick.""" + acc = _Accumulator() + first = assistant_line(text_block("hello"), timestamp="2026-07-01T00:00:00Z") + first["cwd"] = "/home/kuhy" + first["gitBranch"] = "main" + first["slug"] = "slug-one" + second = assistant_line(timestamp="2026-07-02T00:00:00Z", with_usage=False) + second["cwd"] = "/elsewhere" + second["timestamp"] = "2026-07-02T00:00:00Z" + _feed(acc, first, second, {"type": "assistant", "message": "not-a-dict"}) + assert acc.counts.assistant_msgs == 2 + assert acc.tokens.output == 10 + assert acc.meta.started_at == "2026-07-01T00:00:00Z" + assert acc.meta.ended_at == "2026-07-02T00:00:00Z" + assert (acc.meta.cwd, acc.meta.git_branch, acc.meta.slug) == ( + "/home/kuhy", + "main", + "slug-one", + ) + + +def test_scan_blocks_edge_shapes() -> None: + """Non-list content, non-dict blocks, and non-str text are tolerated.""" + acc = _Accumulator() + _feed( + acc, + {"type": "assistant", "message": {"content": "not-a-list", "usage": None}}, + { + "type": "assistant", + "message": { + "content": ["junk", {"type": "text", "text": 5}, {"type": "other"}] + }, + }, + ) + assert acc.counts.assistant_msgs == 2 + assert acc.tokens.output == 0 + + +def test_skill_span_via_tool_and_slash() -> None: + """Skill spans open via Skill tool_use AND via slash-command prompts.""" + acc = _Accumulator() + _feed( + acc, + assistant_line(skill_block("finish")), + assistant_line(bash_block("git status")), + assistant_line(text_block("x" * 500)), + prompt_line("/phone-deploy args"), + assistant_line(bash_block("adb devices")), + prompt_line("/model"), + prompt_line("looks good, thanks"), + ) + acc.finish() + names = [inv.name for inv in acc.skill_invocations] + assert names == ["finish", "phone-deploy"] + finish_inv = acc.skill_invocations[0] + assert finish_inv.tool_only_turns == 2 + assert finish_inv.text_turns == 1 + assert finish_inv.bash_sig == ["git"] + assert acc.counts.external_prompts == 1 + assert acc.typed_prompt_signatures == {"looks good, thanks": 1} + + +def test_generated_chars_counts_write_inputs() -> None: + """Big Write/Edit inputs make a turn text-bearing despite tool_use.""" + acc = _Accumulator() + _feed( + acc, + assistant_line(skill_block("explain-diff")), + assistant_line(tool_block("Write", content="y" * 1000)), + assistant_line(tool_block("Edit", new_string="z" * 1000)), + assistant_line(tool_block("Write", junk=3)), + { + "type": "assistant", + "message": { + "content": [ + {"type": "tool_use", "name": "Write", "input": "not-a-dict"} + ] + }, + }, + ) + acc.finish() + span = acc.skill_invocations[0] + assert span.text_turns == 2 + assert span.tool_only_turns == 3 + + +def test_skill_bash_sig_cap() -> None: + """A skill span's bash signature list stops at the cap.""" + acc = _Accumulator() + _feed(acc, assistant_line(skill_block("finish"))) + _feed( + acc, + { + "type": "assistant", + "message": {"content": [], "usage": {"output_tokens": True}}, + }, + ) + for _ in range(parse.MAX_SKILL_BASH_SIG + 5): + _feed(acc, assistant_line(bash_block("git status"))) + acc.finish() + span = acc.skill_invocations[0] + assert len(span.bash_sig) == parse.MAX_SKILL_BASH_SIG + assert span.tokens_output == 10 * (parse.MAX_SKILL_BASH_SIG + 6) + + +def test_bash_runs_and_caps() -> None: + """Contiguous Bash runs are kept at >=3, broken by other tools, capped.""" + acc = _Accumulator() + _feed(acc, assistant_line(bash_block("git status"), bash_block("pytest -q"))) + _feed(acc, assistant_line(tool_block("Read", file_path="/x"))) + for _ in range(parse.MAX_BASH_RUN + 5): + _feed(acc, assistant_line(bash_block("ls"))) + acc.finish() + assert acc.bash_sequences == [["ls"] * parse.MAX_BASH_RUN] + assert acc.tool_histogram["Bash"] == 2 + parse.MAX_BASH_RUN + 5 + assert acc.tool_histogram["Read"] == 1 + + +def test_bash_edge_inputs() -> None: + """Non-string and empty commands are ignored; unnamed tools count as ?.""" + acc = _Accumulator() + _feed( + acc, + assistant_line(tool_block("Bash", command=5)), + assistant_line(bash_block(" ")), + assistant_line({"type": "tool_use", "name": None, "input": {}}), + ) + acc.finish() + assert not acc.bash_first_lines + assert acc.tool_histogram["?"] == 1 + + +def test_user_line_filters() -> None: + """Sidechain, non-external, and non-string content never count.""" + acc = _Accumulator() + _feed( + acc, + {"type": "user", "message": "not-a-dict"}, + prompt_line("real prompt here"), + prompt_line("agent prompt", user_type="agent"), + prompt_line("sidechain prompt", sidechain=True), + { + "type": "user", + "userType": "external", + "message": {"content": [{"type": "tool_result"}]}, + }, + ) + assert acc.counts.external_prompts == 1 + assert list(acc.typed_prompt_signatures) == ["real prompt here"] + + +def test_compact_retry_and_skipped_prompts() -> None: + """Compact continuations, retries, and pasted/wrapped prompts.""" + acc = _Accumulator() + _feed( + acc, + prompt_line("This session is being continued from a previous conversation."), + prompt_line("limit got reset, continue"), + prompt_line("continue where you left off, the limit has reset ok" + "x" * 40), + prompt_line("[Image: original 1080x2460]"), + prompt_line("stuff"), + prompt_line(" "), + ) + assert acc.counts.compaction_count == 1 + assert acc.counts.retry_prompt_count == 1 + assert acc.counts.external_prompts == 5 + assert "limit got reset, continue" in acc.typed_prompt_signatures + assert len(acc.typed_prompt_signatures) == 2 + + +def test_error_signatures() -> None: + """Error lines are mined from str and dict results; generics dropped.""" + acc = _Accumulator() + long_tail = ( + "\n".join(["noise"] * 20) + + "\nrm: cannot remove '/home/kuhy/.cache/x': Permission denied" + ) + _feed( + acc, + tool_result_line("Error: Exit code 5\nfatal: not a git repository"), + tool_result_line( + { + "stderr": "Traceback (most recent call last):\nValueError: boom", + "stdout": "ok", + "junk": 3, + } + ), + tool_result_line({"stdout": long_tail}), + tool_result_line(None), + ) + sigs = acc.error_signatures + assert sigs["fatal: not a git repository"] == 1 + assert sigs["ValueError: boom"] == 1 + assert sigs["rm: cannot remove '': Permission denied"] == 1 + assert "Error: Exit code " not in sigs + assert "Traceback (most recent call last):" not in sigs + + +def test_flush_skill_drops_empty_spans() -> None: + """A span with zero turns (e.g. a bare slash command) is dropped.""" + acc = _Accumulator() + _feed(acc, prompt_line("/finish")) + acc.finish() + acc.finish() + assert acc.skill_invocations == [] + + +def test_top_truncation_and_ordering() -> None: + """bash_first_lines keeps the most frequent entries, ties alphabetical.""" + acc = _Accumulator() + for command, count in [("ls", 3), ("git status", 3), ("pwd", 1)]: + for _ in range(count): + _feed(acc, assistant_line(bash_block(command))) + top = parse._top(acc.bash_first_lines, 2) + assert list(top) == ["git status", "ls"] + + +def test_parse_session_with_subagents(tmp_path: Path) -> None: + """Subagent transcripts fold into the parent record.""" + projects = tmp_path / "projects" + main = write_transcript( + projects, + "sess-1", + [assistant_line(text_block("hi")), "not json"], + ) + sub_dir = main.parent / "sess-1" / "subagents" + sub_dir.mkdir(parents=True) + (sub_dir / "agent-a.jsonl").write_text( + json.dumps(assistant_line(bash_block("git status"))) + "\n", + encoding="utf-8", + ) + rec = parse_session(main) + assert rec.session_id == "sess-1" + assert rec.counts.subagent_count == 1 + assert rec.counts.assistant_msgs == 2 + assert rec.counts.malformed_lines == 1 + assert rec.obs.tool_histogram == {"Bash": 1} + + +def test_parse_session_without_subagents(tmp_path: Path) -> None: + """A lone transcript parses with zero subagents.""" + projects = tmp_path / "projects" + main = write_transcript(projects, "sess-2", [assistant_line(text_block("hi"))]) + rec = parse_session(main) + assert rec.counts.subagent_count == 0 + assert subagent_files(main) == [] diff --git a/python_pkg/session_autopsy/tests/test_records.py b/python_pkg/session_autopsy/tests/test_records.py new file mode 100644 index 00000000..b4d3e995 --- /dev/null +++ b/python_pkg/session_autopsy/tests/test_records.py @@ -0,0 +1,150 @@ +"""Tests for the record schema and its defensive JSON coercions.""" + +from __future__ import annotations + +from python_pkg.session_autopsy.records import ( + ActivityCounts, + Observations, + SessionMeta, + SessionRecord, + SkillInvocation, + TokenTotals, + _as_float, + _as_int, + _dict_get, + _list_of_dicts, + _list_value, + _opt_str, + _str_int_map, + _str_list, + record_from_dict, +) + + +def test_as_int_coercions() -> None: + """bools and junk yield 0; numerics coerce.""" + flag: object = True + assert _as_int(flag) == 0 + assert _as_int(5) == 5 + assert _as_int(5.9) == 5 + assert _as_int("x") == 0 + assert _as_int(None) == 0 + + +def test_as_float_coercions() -> None: + """bools and junk yield 0.0; numerics coerce.""" + flag: object = True + assert _as_float(flag) == 0.0 + assert _as_float(2) == 2.0 + assert _as_float(2.5) == 2.5 + assert _as_float("x") == 0.0 + + +def test_opt_str() -> None: + """Only non-empty strings survive.""" + assert _opt_str("a") == "a" + assert _opt_str("") is None + assert _opt_str(3) is None + + +def test_dict_get() -> None: + """Non-dict values index to None.""" + assert _dict_get({"k": 1}, "k") == 1 + assert _dict_get([1], "k") is None + + +def test_list_helpers() -> None: + """List filters tolerate wrong types and mixed elements.""" + assert _list_value([1]) == [1] + assert _list_value("no") == [] + assert _list_of_dicts([{"a": 1}, 2, "x"]) == [{"a": 1}] + assert _str_list(["a", 1, "b"]) == ["a", "b"] + + +def test_str_int_map() -> None: + """Counter maps coerce keys to str and values to int.""" + assert _str_int_map({"a": 2, 3: "junk"}) == {"a": 2, "3": 0} + assert _str_int_map("no") == {} + + +def test_token_totals_add_usage() -> None: + """Usage fields accumulate; missing fields count as 0.""" + totals = TokenTotals() + totals.add_usage({"output_tokens": 5, "cache_read_input_tokens": 7}) + totals.add_usage( + {"output_tokens": 2, "input_tokens": 1, "cache_creation_input_tokens": 4} + ) + assert (totals.output, totals.input, totals.cache_read, totals.cache_write) == ( + 7, + 1, + 7, + 4, + ) + + +def test_round_trip() -> None: + """to_dict → record_from_dict preserves every field.""" + original = SessionRecord( + session_id="s", + project_slug="p", + transcript_path="/t", + file_size=9, + file_mtime=1.5, + analyzed_at="2026-01-01T00:00:00+00:00", + meta=SessionMeta( + started_at="a", + ended_at="b", + cwd="/c", + git_branch="main", + slug="sl", + title="ti", + ), + counts=ActivityCounts( + assistant_msgs=3, + external_prompts=2, + subagent_count=1, + compaction_count=1, + retry_prompt_count=2, + malformed_lines=3, + ), + tokens=TokenTotals(output=1, input=2, cache_read=3, cache_write=4), + obs=Observations( + tool_histogram={"Bash": 2}, + bash_first_lines={"git status": 1}, + bash_sequences=[["git", "pytest", "git"]], + skill_invocations=[ + SkillInvocation( + name="finish", + tool_only_turns=1, + text_turns=2, + tokens_output=3, + tokens_cache_write=4, + bash_sig=["git"], + ) + ], + error_signatures={"rm: x": 2}, + typed_prompt_signatures={"do the thing": 1}, + ), + ) + assert record_from_dict(original.to_dict()) == original + + +def test_from_dict_tolerates_junk() -> None: + """Missing keys, wrong types, and junk entries never raise.""" + rebuilt = record_from_dict( + { + "tokens": "junk", + "counts": "junk", + "obs": { + "skill_invocations": [{"name": None}, "junk"], + "bash_sequences": ["junk", ["a", 1]], + }, + "meta": {"started_at": 5}, + }, + ) + assert rebuilt.session_id == "" + assert rebuilt.tokens == TokenTotals() + assert rebuilt.counts == ActivityCounts() + assert rebuilt.obs.skill_invocations[0].name == "?" + assert rebuilt.obs.bash_sequences == [[], ["a"]] + assert rebuilt.meta.started_at is None diff --git a/python_pkg/session_autopsy/tests/test_report.py b/python_pkg/session_autopsy/tests/test_report.py new file mode 100644 index 00000000..79ac861f --- /dev/null +++ b/python_pkg/session_autopsy/tests/test_report.py @@ -0,0 +1,197 @@ +"""Tests for report rendering, state.json, and the measure scoreboard.""" + +from __future__ import annotations + +from datetime import datetime, timezone +import json +from typing import TYPE_CHECKING + +from python_pkg.session_autopsy import report +from python_pkg.session_autopsy.detectors import Candidate, analyze +from python_pkg.session_autopsy.records import ActivityCounts, Observations +from python_pkg.session_autopsy.tests.conftest import ( + invocation, + record, + session_meta, +) + +if TYPE_CHECKING: + from pathlib import Path + +NOW = datetime(2026, 7, 24, 12, 0, tzinfo=timezone.utc) + + +def test_fmt_tokens() -> None: + """Plain, k, and M ranges.""" + assert report.fmt_tokens(842) == "842" + assert report.fmt_tokens(74_500) == "74k" + assert report.fmt_tokens(1_234_567) == "1.2M" + + +def test_load_state_variants(tmp_path: Path) -> None: + """Missing, corrupt, and wrong-type state files read as empty.""" + assert report.load_state(tmp_path) == {} + state_path = tmp_path / report.STATE_FILE + state_path.write_text("{bad", encoding="utf-8") + assert report.load_state(tmp_path) == {} + state_path.write_text("[1]", encoding="utf-8") + assert report.load_state(tmp_path) == {} + state_path.write_text('{"unreviewed_count": 2}', encoding="utf-8") + assert report.load_state(tmp_path) == {"unreviewed_count": 2} + + +def test_load_compiled_variants(tmp_path: Path) -> None: + """Missing, corrupt, wrong-type, and mixed compiled files.""" + assert report.load_compiled(tmp_path) == [] + compiled_path = tmp_path / report.COMPILED_FILE + compiled_path.write_text("{bad", encoding="utf-8") + assert report.load_compiled(tmp_path) == [] + compiled_path.write_text('{"a": 1}', encoding="utf-8") + assert report.load_compiled(tmp_path) == [] + compiled_path.write_text('[{"skill": "finish"}, 3]', encoding="utf-8") + assert report.load_compiled(tmp_path) == [{"skill": "finish"}] + + +def test_write_state_lifecycle(tmp_path: Path) -> None: + """Unreviewed counting, mark-reviewed, stale-id pruning, junk tolerance.""" + assert report.write_state(tmp_path, ["a", "b"], NOW) == 2 + assert report.write_state(tmp_path, ["a", "b"], NOW, mark_reviewed=True) == 0 + assert report.write_state(tmp_path, ["a", "b", "c"], NOW) == 1 + assert report.write_state(tmp_path, ["c"], NOW) == 1 + state_path = tmp_path / report.STATE_FILE + state_path.write_text('{"reviewed_ids": "junk"}', encoding="utf-8") + assert report.write_state(tmp_path, ["z"], NOW) == 1 + + +def _compiled_entry(**over: object) -> dict[str, object]: + """A compiled.json entry with sensible defaults.""" + entry: dict[str, object] = { + "candidate_id": "skill-finish", + "skill": "finish", + "script": "/x/finish.sh", + "verdict": "compiled", + "compiled_at": "2026-07-10T00:00:00+00:00", + } + entry.update(over) + return entry + + +def test_measure_lines_variants() -> None: + """Placeholder, skips, before-only, delta, and after-only cases.""" + before = record( + "old", + meta=session_meta(started_at="2026-07-01T00:00:00Z"), + obs=Observations( + skill_invocations=[ + invocation(output=100000, cache_write=0), + invocation(name="other"), + ] + ), + ) + after = record( + "new", + meta=session_meta(started_at="2026-07-20T00:00:00Z"), + obs=Observations(skill_invocations=[invocation(output=10000, cache_write=0)]), + ) + assert report.measure_lines([], [], None) == [ + "nothing compiled yet — run /compile-candidate on a ranked candidate", + ] + skipped = [ + _compiled_entry(skill=""), + _compiled_entry(script=None), + _compiled_entry(skill="other"), + ] + assert len(report.measure_lines([before], skipped, "finish")) == 1 + only_before = report.measure_lines([before], [_compiled_entry()], None) + assert "no post-compile invocations yet" in only_before[0] + delta = report.measure_lines([before, after], [_compiled_entry()], None) + assert "→ after" in delta[0] + assert "-90%" in delta[0] + after_only = report.measure_lines([after], [_compiled_entry()], None) + assert "before 0/inv (n=0)" in after_only[0] + assert "%" not in after_only[0] + + +def _fake_candidate(index: int) -> Candidate: + """A minimal ranked candidate for table-rendering tests.""" + return Candidate( + id=f"err-{index:08d}", + kind="err", + title=f"error {index}", + sig=f"sig {index}", + sessions=1, + occurrences=2, + avg_tokens=10, + per_week=0.1, + est_weekly_savings=100 - index, + score=1.0, + action="fix once", + session_ids=["s1"], + ) + + +def test_candidate_table_empty_and_overflow() -> None: + """Empty corpus placeholder and the beyond-top-40 overflow note.""" + assert "none yet" in report._candidate_table([]) + table = report._candidate_table([_fake_candidate(index) for index in range(45)]) + assert "plus 5 below the fold" in table + + +def test_reviewed_keep_llm_section() -> None: + """Empty and populated reviewed-keep-LLM sections.""" + assert "none reviewed yet" in report._reviewed_keep_llm([]) + section = report._reviewed_keep_llm( + [_compiled_entry(verdict="keep-llm", candidate_id="skill-x")] + ) + assert "skill-x" in section + + +def test_render_and_write_report(tmp_path: Path) -> None: + """Full render includes all sections; write_report persists + counts.""" + records = [ + record( + "a", + obs=Observations( + skill_invocations=[invocation()] * 3, error_signatures={"boom: x": 12} + ), + ), + ] + result = analyze(records, NOW) + compiled = [ + _compiled_entry(), + _compiled_entry( + verdict="keep-llm", candidate_id="skill-explain-diff", script=None + ), + ] + text = report.render_report(records, result, compiled, NOW) + assert "# Session Autopsy Report" in text + assert "skill-finish" in text + assert "Reviewed — keeping LLM" in text + assert "Context-bloat trend" in text + assert "Compiled scoreboard" in text + unreviewed = report.write_report(tmp_path, records, result, NOW) + assert unreviewed == len(result.candidates) + assert (tmp_path / report.REPORT_FILE).is_file() + + +def test_trend_and_header_empty_states() -> None: + """Undated corpus → trend placeholder; zero turns → zero per-turn.""" + records = [ + record( + "a", + meta=session_meta(started_at=None), + counts=ActivityCounts(assistant_msgs=0), + ) + ] + result = analyze(records, NOW) + assert "no dated sessions yet" in report._trend_section(result) + header = report._header(records, NOW) + assert "avg 0/turn" in header + + +def test_state_file_is_valid_json(tmp_path: Path) -> None: + """state.json holds the ranked ids and the timestamp.""" + report.write_state(tmp_path, ["x"], NOW) + payload = json.loads((tmp_path / report.STATE_FILE).read_text(encoding="utf-8")) + assert payload["candidate_ids"] == ["x"] + assert payload["last_generated"].startswith("2026-07-24") diff --git a/python_pkg/session_autopsy/tests/test_signatures.py b/python_pkg/session_autopsy/tests/test_signatures.py new file mode 100644 index 00000000..eec254ab --- /dev/null +++ b/python_pkg/session_autopsy/tests/test_signatures.py @@ -0,0 +1,28 @@ +"""Tests for the cross-session text signatures.""" + +from __future__ import annotations + +from python_pkg.session_autopsy import signatures +from python_pkg.session_autopsy.signatures import command_signature, normalize_signature + + +def test_normalize_signature() -> None: + """Paths, hex runs, and numbers are masked; first line only.""" + assert normalize_signature("") == "" + assert normalize_signature(" \n\n") == "" + sig = normalize_signature("rm -rf /home/kuhy/.cache/yay/python-jaxlib\nsecond line") + assert sig == "rm -rf " + assert normalize_signature("deadbeefcafe1234") == "" + assert normalize_signature("retry 12 times") == "retry times" + assert len(normalize_signature("x" * 500)) == signatures.MAX_SIG_LEN + + +def test_command_signature() -> None: + """Wrappers keep their first non-flag, non-numeric argument.""" + assert command_signature("") == "" + assert command_signature("git status") == "git" + assert command_signature("cd /home/kuhy/testsAndMisc && ls") == "cd " + assert command_signature("timeout 120 python3 -m x") == "timeout python3" + assert command_signature("sudo -n true") == "sudo true" + assert command_signature("sudo -n") == "sudo" + assert command_signature("cd") == "cd" diff --git a/python_pkg/session_autopsy/tests/test_store.py b/python_pkg/session_autopsy/tests/test_store.py new file mode 100644 index 00000000..2e17267d --- /dev/null +++ b/python_pkg/session_autopsy/tests/test_store.py @@ -0,0 +1,65 @@ +"""Tests for the flock-guarded sessions.jsonl store.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +from python_pkg.session_autopsy.records import ActivityCounts +from python_pkg.session_autopsy.store import ( + SESSIONS_FILE, + StoreLock, + load_file_index, + load_records, + upsert_records, +) +from python_pkg.session_autopsy.tests.conftest import ( + FileFacts, + record, +) + +if TYPE_CHECKING: + from pathlib import Path + + +def test_lock_exit_without_enter(tmp_path: Path) -> None: + """__exit__ with no held handle is a safe no-op.""" + StoreLock(tmp_path).__exit__(None, None, None) + + +def test_load_records_missing_store(tmp_path: Path) -> None: + """An absent sessions.jsonl reads as empty.""" + assert load_records(tmp_path / "home") == [] + + +def test_iter_skips_bad_lines(tmp_path: Path) -> None: + """Blank, malformed, and non-dict store lines are skipped.""" + home = tmp_path + good = record("keep") + (home / SESSIONS_FILE).write_text( + "\n".join(["", "{bad json", json.dumps([1, 2]), json.dumps(good.to_dict())]) + + "\n", + encoding="utf-8", + ) + records = load_records(home) + assert [rec.session_id for rec in records] == ["keep"] + + +def test_upsert_inserts_and_replaces(tmp_path: Path) -> None: + """Upsert keys by session id: new ids append, same ids replace.""" + home = tmp_path / "home" + assert upsert_records(home, [record("a"), record("b")]) == 2 + replacement = record("a", counts=ActivityCounts(assistant_msgs=99)) + assert upsert_records(home, [replacement]) == 2 + by_id = {rec.session_id: rec for rec in load_records(home)} + assert by_id["a"].counts.assistant_msgs == 99 + assert by_id["b"].counts.assistant_msgs == 10 + + +def test_load_file_index(tmp_path: Path) -> None: + """The index maps transcript paths to (size, mtime).""" + home = tmp_path / "home" + upsert_records( + home, [record("a", file=FileFacts(path="/t/a.jsonl", size=5, mtime=2.5))] + ) + assert load_file_index(home) == {"/t/a.jsonl": (5, 2.5)} diff --git a/python_pkg/session_autopsy/tests/test_traces.py b/python_pkg/session_autopsy/tests/test_traces.py new file mode 100644 index 00000000..a3b3ac95 --- /dev/null +++ b/python_pkg/session_autopsy/tests/test_traces.py @@ -0,0 +1,266 @@ +"""Tests for compact trace extraction.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from python_pkg.session_autopsy import traces +from python_pkg.session_autopsy.detectors import Candidate +from python_pkg.session_autopsy.tests.conftest import ( + FileFacts, + assistant_line, + bash_block, + prompt_line, + record, + skill_block, + text_block, + tool_block, + tool_result_line, + write_transcript, +) + +if TYPE_CHECKING: + from pathlib import Path + + import pytest + + from python_pkg.session_autopsy.records import SessionRecord + + +DENIED = "rm: cannot remove '/home/kuhy/.cache/yay/pkg': Permission denied" + + +def _candidate(kind: str, sig: str, session_ids: list[str]) -> Candidate: + """A candidate pointing at the given evidence sessions.""" + return Candidate( + id=f"{kind}-x", + kind=kind, + title="t", + sig=sig, + sessions=len(session_ids), + occurrences=1, + avg_tokens=1, + per_week=0.0, + est_weekly_savings=0, + score=1.0, + action="a", + session_ids=session_ids, + ) + + +def _session( + tmp_path: Path, session_id: str, lines: list[dict[str, object] | str] +) -> SessionRecord: + """Write a transcript and a record whose path points at it.""" + path = write_transcript(tmp_path / "projects", session_id, lines) + return record(session_id, file=FileFacts(path=str(path))) + + +def test_skill_traces_full_span(tmp_path: Path) -> None: + """Skill spans render TOOL/TEXT/RESULT lines and close on prompts.""" + lines: list[dict[str, object] | str] = [ + {"type": "attachment"}, + {"type": "assistant", "message": "junk"}, + {"type": "assistant", "message": {"content": "not-a-list"}}, + prompt_line("start", user_type="agent"), + prompt_line("side", sidechain=True), + {"type": "user", "message": "junk"}, + assistant_line(skill_block("finish")), + assistant_line( + bash_block("git status"), text_block("Running checks now on this repo.") + ), + assistant_line(text_block(" ")), + tool_result_line("line one\n\nline three\nline four"), + tool_result_line(None), + prompt_line("done, thanks"), + assistant_line(skill_block("other")), + assistant_line(bash_block("should not appear")), + ] + rec = _session(tmp_path, "sess", lines) + text = traces.render_traces( + _candidate("skill", "finish", ["sess"]), {"sess": rec}, 5 + ) + assert "TOOL Skill | finish" in text + assert "TOOL Bash" in text + assert "TEXT | Running checks now" in text + assert "RESULT | line one" in text + assert "should not appear" not in text + + +def test_skill_traces_ignores_other_block_types(tmp_path: Path) -> None: + """A block that is neither tool_use nor text contributes nothing.""" + lines: list[dict[str, object] | str] = [ + assistant_line(skill_block("finish")), + assistant_line({"type": "thinking", "thinking": "quietly"}), + assistant_line(bash_block("git status")), + ] + rec = _session(tmp_path, "sess", lines) + text = traces.render_traces( + _candidate("skill", "finish", ["sess"]), {"sess": rec}, 5 + ) + assert "TOOL Skill | finish" in text + assert "TOOL Bash" in text + assert "quietly" not in text + + +def test_capture_block_ignores_unrecognised_blocks() -> None: + """Only tool_use and text blocks can contribute a line.""" + span = traces._SpanBuffer() + span.start(matched=True, header="TOOL Skill | finish") + traces._capture_block(span, {"type": "thinking", "thinking": "quietly"}, "finish") + traces._capture_block(span, {"type": "tool_result"}, "finish") + traces._capture_block(span, {}, "finish") + # A text block still needs actual text: neither a non-string nor whitespace + # counts. + traces._capture_block(span, {"type": "text", "text": 5}, "finish") + traces._capture_block(span, {"type": "text", "text": " "}, "finish") + assert span.lines == ["TOOL Skill | finish"] + + +def test_span_buffer_only_captures_inside_a_matching_span() -> None: + """Lines outside a span, or inside another skill's span, are dropped.""" + span = traces._SpanBuffer() + span.add("TEXT | before any span") + span.start(matched=False, header="TOOL Skill | other") + span.add("TEXT | wrong skill") + span.start(matched=True, header="TOOL Skill | finish") + span.add("TEXT | inside") + span.close() + span.add("TEXT | after the span closed") + assert span.lines == ["TOOL Skill | finish", "TEXT | inside"] + + +def test_span_buffer_honours_the_per_span_budget( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The header is free; captured lines stop at the cap.""" + monkeypatch.setattr(traces, "MAX_LINES_PER_SPAN", 2) + span = traces._SpanBuffer() + span.start(matched=True, header="TOOL Skill | finish") + for index in range(5): + span.add(f"TEXT | line {index}") + assert span.lines == ["TOOL Skill | finish", "TEXT | line 0", "TEXT | line 1"] + # A fresh span gets a fresh budget. + span.start(matched=True, header="TOOL Skill | finish") + span.add("TEXT | second span") + assert span.lines[-2:] == ["TOOL Skill | finish", "TEXT | second span"] + + +def test_skill_traces_span_cap(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """The per-span line cap stops TOOL/TEXT/RESULT accumulation.""" + monkeypatch.setattr(traces, "MAX_LINES_PER_SPAN", 1) + lines: list[dict[str, object] | str] = [ + assistant_line(skill_block("finish")), + assistant_line(bash_block("one"), bash_block("two")), + assistant_line(text_block("long narration " * 20)), + tool_result_line("r1\nr2"), + ] + rec = _session(tmp_path, "cap", lines) + text = traces.render_traces(_candidate("skill", "finish", ["cap"]), {"cap": rec}, 5) + assert text.count("TOOL Bash") == 1 + assert "TEXT |" not in text + + +def test_render_traces_missing_and_cap(tmp_path: Path) -> None: + """Unknown ids are skipped, capture stops at max, empty → message.""" + lines: list[dict[str, object] | str] = [ + assistant_line(skill_block("finish")), + assistant_line(bash_block("x")), + ] + rec_a = _session(tmp_path, "a", lines) + rec_b = _session(tmp_path, "b", lines) + gone = record("gone", file=FileFacts(path=str(tmp_path / "nope.jsonl"))) + by_id = {"a": rec_a, "b": rec_b, "gone": gone} + capped = traces.render_traces( + _candidate("skill", "finish", ["missing", "a", "b"]), by_id, 1 + ) + assert "## session a" in capped + assert "## session b" not in capped + empty = traces.render_traces(_candidate("skill", "finish", ["gone"]), by_id, 5) + assert "no matching spans found" in empty + + +def test_error_traces(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Error hits pair the causing command with the line, capped.""" + monkeypatch.setattr(traces, "MAX_ERR_HITS_PER_SESSION", 1) + lines: list[dict[str, object] | str] = [ + assistant_line(tool_block("Bash", command=5)), + assistant_line(tool_block("Read", file_path="/x")), + assistant_line(bash_block("rm -rf /home/kuhy/.cache/yay/pkg")), + tool_result_line(f"unrelated noise line\n{DENIED}"), + tool_result_line({"stderr": DENIED}), + ] + rec = _session(tmp_path, "err", lines) + sig = "rm: cannot remove '': Permission denied" + text = traces.render_traces(_candidate("err", sig, ["err"]), {"err": rec}, 5) + assert text.count("CMD | rm -rf") == 1 + assert text.count("ERR |") == 1 + + +def test_ngram_traces(tmp_path: Path) -> None: + """The first window matching the gram renders its raw commands.""" + lines: list[dict[str, object] | str] = [ + assistant_line(text_block("hi")), + assistant_line(tool_block("Bash", command=7)), + assistant_line( + bash_block("git status"), bash_block("pytest -q"), bash_block("git push") + ), + ] + rec = _session(tmp_path, "ng", lines) + hit = traces.render_traces( + _candidate("ngram", "git → pytest → git", ["ng"]), {"ng": rec}, 5 + ) + assert "CMD | git status" in hit + assert "CMD | git push" in hit + miss = traces.render_traces( + _candidate("ngram", "adb → adb", ["ng"]), {"ng": rec}, 5 + ) + assert "no matching spans found" in miss + + +def test_prompt_traces(tmp_path: Path) -> None: + """Matching prompts render with the tools that followed.""" + lines: list[dict[str, object] | str] = [ + prompt_line("fix the tests please"), + assistant_line( + text_block("narration"), + bash_block("pytest -q"), + tool_block("Read", file_path="/x"), + ), + prompt_line("unrelated message here"), + assistant_line(bash_block("ls")), + prompt_line("fix the tests please"), + ] + rec = _session(tmp_path, "pr", lines) + text = traces.render_traces( + _candidate("prompt", "fix the tests please", ["pr"]), {"pr": rec}, 5 + ) + assert text.count("PROMPT | fix the tests please") == 2 + assert "THEN | Bash" in text + assert "THEN | Read" in text + assert text.count("THEN |") == 2 + + +def test_iter_lines_tolerates_junk(tmp_path: Path) -> None: + """Blank, malformed, and non-dict transcript lines are skipped.""" + lines: list[dict[str, object] | str] = [ + "", + "{bad", + "[1]", + assistant_line(bash_block("ls")), + ] + rec = _session(tmp_path, "junk", lines) + text = traces.render_traces(_candidate("ngram", "ls", ["junk"]), {"junk": rec}, 5) + assert "CMD | ls" in text + + +def test_result_texts_and_clip() -> None: + """toolUseResult text extraction and clipping edge cases.""" + assert traces._result_texts({"toolUseResult": {"stderr": "", "stdout": 3}}) == [] + assert traces._result_texts({"toolUseResult": 42}) == [] + assert traces._result_texts({"toolUseResult": {"stderr": "e", "stdout": "o"}}) == [ + "e", + "o", + ] + assert traces._clip("") == "" + assert traces._clip("x" * 500) == "x" * traces.MAX_LINE_CHARS diff --git a/python_pkg/session_autopsy/traces.py b/python_pkg/session_autopsy/traces.py new file mode 100644 index 00000000..43f45d8a --- /dev/null +++ b/python_pkg/session_autopsy/traces.py @@ -0,0 +1,392 @@ +"""Export compact transcript excerpts for a candidate. + +This is what keeps the LLM compile step cheap: instead of re-reading multi-MB +transcripts, /compile-candidate reads a few hundred normalized lines. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import TYPE_CHECKING + +from python_pkg.session_autopsy.signatures import command_signature, normalize_signature + +if TYPE_CHECKING: + from collections.abc import Iterator + + from python_pkg.session_autopsy.detectors import Candidate + from python_pkg.session_autopsy.records import SessionRecord + +MAX_LINES_PER_SPAN = 200 +MAX_TEXT_CHARS = 200 +MAX_RESULT_LINES = 3 +MAX_LINE_CHARS = 160 +MAX_ERR_HITS_PER_SESSION = 5 + + +def render_traces( + candidate: Candidate, + records_by_id: dict[str, SessionRecord], + max_invocations: int, +) -> str: + """Render trace excerpts for a candidate across its evidence sessions. + + Args: + candidate: The candidate to trace. + records_by_id: Stored records keyed by session id (for paths). + max_invocations: Stop after this many captured spans/hits. + + Returns: + The full trace document. + """ + lines = [f"# traces {candidate.id} — {candidate.title}", ""] + captured = 0 + for session_id in candidate.session_ids: + record = records_by_id.get(session_id) + if record is None: + continue + if captured >= max_invocations: + break + session_lines = _trace_session(candidate, record) + if session_lines: + captured += 1 + lines.append( + f"## session {session_id} ({record.meta.started_at or 'undated'})" + ) + lines.extend(session_lines) + lines.append("") + if captured == 0: + lines.append("no matching spans found (transcripts may have been deleted)") + return "\n".join(lines) + "\n" + + +def _trace_session(candidate: Candidate, record: SessionRecord) -> list[str]: + """Extract this candidate's excerpt lines from one session. + + Args: + candidate: The candidate to trace. + record: The session's stored record (holds the transcript path). + + Returns: + Excerpt lines, or an empty list when nothing matches. + """ + path = _existing_path(record) + if path is None: + return [] + extractors = { + "skill": _extract_skill, + "err": _extract_error, + "ngram": _extract_ngram, + "prompt": _extract_prompt, + } + return extractors[candidate.kind](path, candidate.sig) + + +def _existing_path(record: SessionRecord) -> Path | None: + """Resolve the record's transcript path if it still exists. + + Args: + record: The stored session record. + + Returns: + The path, or None when the transcript is gone. + """ + path = Path(record.transcript_path) + if path.is_file(): + return path + return None + + +def _iter_lines(path: Path) -> Iterator[dict[str, object]]: + """Yield parsed dict lines of a transcript, skipping malformed ones. + + Args: + path: Transcript path. + + Yields: + Each line's parsed dict. + """ + with path.open(encoding="utf-8", errors="replace") as handle: + for raw_line in handle: + stripped = raw_line.strip() + if not stripped: + continue + try: + obj = json.loads(stripped) + except json.JSONDecodeError: + continue + if isinstance(obj, dict): + yield obj + + +def _clip(text: str, limit: int = MAX_LINE_CHARS) -> str: + """Single-line clip for trace output. + + Args: + text: Raw text. + limit: Character cap. + + Returns: + The first line, capped. + """ + first = text.strip().splitlines()[0] if text.strip() else "" + return first[:limit] + + +def _content_blocks(obj: dict[str, object]) -> list[dict[str, object]]: + """Return a line's message content blocks. + + Args: + obj: A parsed transcript line. + + Returns: + The content block dicts, or an empty list. + """ + message = obj.get("message") + if not isinstance(message, dict): + return [] + content = message.get("content") + if not isinstance(content, list): + return [] + return [block for block in content if isinstance(block, dict)] + + +def _is_external_prompt(obj: dict[str, object]) -> bool: + """Whether a line is a real typed user prompt. + + Args: + obj: A parsed transcript line. + + Returns: + True for external, non-sidechain, string-content user lines. + """ + if ( + obj.get("type") != "user" + or obj.get("userType") != "external" + or obj.get("isSidechain") is True + ): + return False + message = obj.get("message") + return isinstance(message, dict) and isinstance(message.get("content"), str) + + +def _result_texts(obj: dict[str, object]) -> list[str]: + """Collect the textual parts of a line's toolUseResult. + + Args: + obj: A parsed transcript line. + + Returns: + Zero or more raw result texts. + """ + result = obj.get("toolUseResult") + texts = [] + if isinstance(result, str): + texts.append(result) + elif isinstance(result, dict): + for key in ("stderr", "stdout"): + value = result.get(key) + if isinstance(value, str) and value: + texts.append(value) + return texts + + +class _SpanBuffer: + """Excerpt lines for the skill spans found in one transcript. + + Owning ``in_span`` and the per-span line budget here is what keeps + :func:`_extract_skill` down to a readable walk: every "am I inside a span and + is there room left" question is answered by :meth:`add`. + """ + + def __init__(self) -> None: + """Start outside any span with nothing captured.""" + self.lines: list[str] = [] + self.in_span = False + self._span_lines = 0 + + def start(self, *, matched: bool, header: str) -> None: + """Open a span for a Skill invocation, or leave the one in progress. + + Args: + matched: Whether the invoked skill is the one being excerpted. + header: The span's first line, kept only when ``matched``. + """ + self.in_span = matched + self._span_lines = 0 + if matched: + self.lines.append(header) + + def close(self) -> None: + """End the current span, if any.""" + self.in_span = False + + def add(self, line: str) -> None: + """Capture one line, inside a span and within its budget. + + Args: + line: The formatted excerpt line. + """ + if self.in_span and self._span_lines < MAX_LINES_PER_SPAN: + self._span_lines += 1 + self.lines.append(line) + + +def _capture_results(span: _SpanBuffer, obj: dict[str, object]) -> None: + """Capture the head of one tool result. + + Args: + span: Buffer for the span in progress. + obj: A ``user`` transcript line carrying a tool result. + """ + for text in _result_texts(obj)[:1]: + for result_line in text.splitlines()[:MAX_RESULT_LINES]: + span.add(f"RESULT | {_clip(result_line)}") + + +def _capture_block( + span: _SpanBuffer, block: dict[str, object], skill_name: str +) -> None: + """Capture one assistant content block, opening a span when Skill runs. + + Args: + span: Buffer for the span in progress. + block: One content block of an ``assistant`` line. + skill_name: The skill being excerpted. + """ + block_type = block.get("type") + if block_type == "tool_use": + name = str(block.get("name") or "?") + if name == "Skill": + tool_input = block.get("input") + invoked = tool_input.get("skill") if isinstance(tool_input, dict) else None + span.start( + matched=invoked == skill_name, header=f"TOOL Skill | {skill_name}" + ) + return + payload = _clip(json.dumps(block.get("input", {}))[:MAX_LINE_CHARS]) + span.add(f"TOOL {name} | {payload}") + elif block_type == "text": + text = block.get("text") + if isinstance(text, str) and text.strip(): + span.add(f"TEXT | {_clip(text, MAX_TEXT_CHARS)}") + + +def _extract_skill(path: Path, skill_name: str) -> list[str]: + """Excerpt every span of one skill in a transcript. + + Args: + path: Transcript path. + skill_name: The skill to capture. + + Returns: + TOOL/TEXT/RESULT lines for each span, capped per span. + """ + span = _SpanBuffer() + for obj in _iter_lines(path): + if _is_external_prompt(obj): + span.close() + elif obj.get("type") == "user": + if span.in_span: + _capture_results(span, obj) + elif obj.get("type") == "assistant": + for block in _content_blocks(obj): + _capture_block(span, block, skill_name) + return span.lines + + +def _extract_error(path: Path, sig: str) -> list[str]: + """Excerpt occurrences of one error signature with the causing command. + + Args: + path: Transcript path. + sig: The normalized error signature to match. + + Returns: + CMD/ERR line pairs, capped per session. + """ + lines: list[str] = [] + last_bash = "" + hits = 0 + for obj in _iter_lines(path): + for block in _content_blocks(obj) if obj.get("type") == "assistant" else []: + if block.get("type") == "tool_use" and block.get("name") == "Bash": + tool_input = block.get("input") + command = ( + tool_input.get("command") if isinstance(tool_input, dict) else None + ) + if isinstance(command, str): + last_bash = command + if obj.get("type") != "user" or hits >= MAX_ERR_HITS_PER_SESSION: + continue + for text in _result_texts(obj): + for result_line in text.splitlines(): + if ( + normalize_signature(result_line) == sig + and hits < MAX_ERR_HITS_PER_SESSION + ): + hits += 1 + lines.append(f"CMD | {_clip(last_bash)}") + lines.append(f"ERR | {_clip(result_line)}") + return lines + + +def _extract_ngram(path: Path, sig: str) -> list[str]: + """Excerpt the first occurrence of a command sequence. + + Args: + path: Transcript path. + sig: The " → "-joined command-signature gram. + + Returns: + The raw commands of the first matching window, or empty. + """ + gram = sig.split(" → ") + window: list[tuple[str, str]] = [] + for obj in _iter_lines(path): + for block in _content_blocks(obj) if obj.get("type") == "assistant" else []: + if block.get("type") != "tool_use" or block.get("name") != "Bash": + continue + tool_input = block.get("input") + command = ( + tool_input.get("command") if isinstance(tool_input, dict) else None + ) + if not isinstance(command, str): + continue + window.append((command_signature(command), command)) + window = window[-len(gram) :] + if [entry[0] for entry in window] == gram: + return [f"CMD | {_clip(raw)}" for _, raw in window] + return [] + + +def _extract_prompt(path: Path, sig: str) -> list[str]: + """Excerpt occurrences of a repeated typed prompt and what followed. + + Args: + path: Transcript path. + sig: The normalized prompt signature. + + Returns: + PROMPT lines followed by the next few tool names. + """ + lines: list[str] = [] + following = 0 + for obj in _iter_lines(path): + if _is_external_prompt(obj): + message = obj.get("message") + content = str(message.get("content")) if isinstance(message, dict) else "" + if normalize_signature(content) == sig: + lines.append(f"PROMPT | {_clip(content, MAX_TEXT_CHARS)}") + following = MAX_RESULT_LINES + else: + following = 0 + continue + if following <= 0: + continue + for block in _content_blocks(obj) if obj.get("type") == "assistant" else []: + if block.get("type") == "tool_use" and following > 0: + following -= 1 + lines.append(f"THEN | {block.get('name')}") + return lines