kuhyx/testsAndMisc
1
0
Fork 0
mirror of https://github.com/kuhyx/testsAndMisc.git synced 2026-11-10 21:07:03 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Extract session_autopsy and the MCP fleet to kuhyx/mcp-servers

Browse Source 
Two things that did not belong in a monorepo whose actual history is 48MB:

- third_party/mcp: 12GB of 24 third-party clones plus regenerable build output
  (node_modules, Rust target/, venvs). Moved out and gitignored -- committing it
  would have produced broken gitlinks from 24 nested .git directories. What is
  tracked instead is a pinned manifest (upstream URL + exact commit for each) and
  a setup.sh that reconstructs the fleet: 20KB in place of 12GB.
- python_pkg/session_autopsy: self-contained, imported nothing else here. Moved
  with history via git subtree split.

Verified before removing anything: the SessionEnd hook ingests a real transcript
cleanly from the new location, and all 12 absolute-path references across
~/.claude.json and the three mcp-optional configs were rewritten with every
command path confirmed to resolve.

third_party/agent-skills is unrelated to MCP and stays.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-07-26 20:44:23 +02:00
parent db9c8921a2
commit 627b9e78da
26 changed files with 0 additions and 5027 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
python_pkg/session_autopsy/__init__.py
View File

@ -1,7 +0,0 @@
"""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.
"""

8
python_pkg/session_autopsy/__main__.py
View File

@ -1,8 +0,0 @@
"""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())

390
python_pkg/session_autopsy/cli.py
View File

@ -1,390 +0,0 @@
"""Command-line entry point for the session autopsy analyzer.
Usage, with ``PYTHONPATH=~/testsAndMisc`` and ``python3 -m
python_pkg.session_autopsy``:
ingest <transcript.jsonl>
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 <uuid>.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:
handled = report.handled_ids(report.load_compiled(home))
pending = [
cid
for cid in report.significant_ids(result.candidates)
if cid not in handled
]
report.write_state(home, pending, _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)

30
python_pkg/session_autopsy/config.py
View File

@ -1,30 +0,0 @@
"""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()

467
python_pkg/session_autopsy/detectors.py
View File

@ -1,467 +0,0 @@
"""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],
)

487
python_pkg/session_autopsy/parse.py
View File

@ -1,487 +0,0 @@
"""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,
TurnEfficiency,
)
from python_pkg.session_autopsy.signatures import command_signature, normalize_signature
from python_pkg.session_autopsy.turn_tracker import TurnTracker
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"<command-name>/?([\w:-]+)</command-name>")
# Harness built-ins that appear as <command-name> 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 <N>",
"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
# One API response can span several transcript lines, so batching has to be
# grouped by requestId — counting per line would report every turn as
# single-tool and hide the real (much lower) batching rate.
self._turns = TurnTracker()
@property
def turns(self) -> TurnEfficiency:
"""Turn-efficiency totals accumulated so far."""
return self._turns.totals
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, skill span and API turn."""
self._flush_run()
self._flush_skill()
self._turns.flush()
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
self._turns.begin(obj.get("requestId") or _message_id(message))
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
self._turns.add_tool()
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
self._turns.add_command(command)
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 ``<uuid>.jsonl`` transcript.
Returns:
Sorted ``agent-*.jsonl`` paths under ``<uuid>/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 ``<uuid>.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,
turns=acc.turns,
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),
),
)
def _message_id(message: dict[str, object]) -> str | None:
"""Fallback turn key for lines that carry no ``requestId``.
Args:
message: The ``message`` object of an assistant line.
Returns:
The message id when present, else None.
"""
value = message.get("id")
return value if isinstance(value, str) else None

390
python_pkg/session_autopsy/records.py
View File

@ -1,390 +0,0 @@
"""Dataclass schema for per-session autopsy records (schema v2)."""
from __future__ import annotations
from dataclasses import asdict, dataclass, field
SCHEMA_VERSION = 3
@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 TurnEfficiency:
"""API round-trips taken, and how many of them a script could have taken.
Cache-read tokens dominate a long session (one measured session: 235M read
against 800k output), and cache read is charged per round-trip on the whole
conversation so far. So the cost lever is the NUMBER of turns, not the length
of any message — and turns where the model decided nothing are pure waste.
"""
api_turns: int = 0
tool_calls: int = 0
batched_turns: int = 0
poll_turns: int = 0
lint_test_turns: int = 0
vcs_check_turns: int = 0
@property
def mechanical_turns(self) -> int:
"""Turns whose every action was observation a script could have done."""
return self.poll_turns + self.lint_test_turns + self.vcs_check_turns
@property
def batching_rate(self) -> float:
"""Share of tool-carrying turns that issued more than one call (0..1)."""
return self.batched_turns / self.api_turns if self.api_turns else 0.0
@property
def mechanical_rate(self) -> float:
"""Share of tool-carrying turns that were purely mechanical (0..1)."""
return self.mechanical_turns / self.api_turns if self.api_turns else 0.0
@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)
turns: TurnEfficiency = field(default_factory=TurnEfficiency)
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.
"""
data = asdict(self)
# asdict() only walks fields, so the derived rates are added by hand —
# readers of sessions.jsonl should not have to recompute them.
data["turns"] |= {
"mechanical_turns": self.turns.mechanical_turns,
"batching_rate": round(self.turns.batching_rate, 4),
"mechanical_rate": round(self.turns.mechanical_rate, 4),
}
return data
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 _turns_from_dict(data: object) -> TurnEfficiency:
"""Rebuild turn-efficiency counts from their serialized form.
Args:
data: The ``turns`` value of a stored record.
Returns:
The reconstructed counts (zeros on junk or on pre-v3 records).
"""
return TurnEfficiency(
api_turns=_as_int(_dict_get(data, "api_turns")),
tool_calls=_as_int(_dict_get(data, "tool_calls")),
batched_turns=_as_int(_dict_get(data, "batched_turns")),
poll_turns=_as_int(_dict_get(data, "poll_turns")),
lint_test_turns=_as_int(_dict_get(data, "lint_test_turns")),
vcs_check_turns=_as_int(_dict_get(data, "vcs_check_turns")),
)
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")),
turns=_turns_from_dict(data.get("turns")),
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()}

476
python_pkg/session_autopsy/report.py
View File

@ -1,476 +0,0 @@
"""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.report_cells import action_cell, verdict_label
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
# The ⚙ badge / SessionStart nudge count only candidates at least this big —
# nobody reviews 180 weak n-grams, so the count must mean "worth your time".
SIGNIFICANT_WEEKLY_TOKENS = 100_000
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 handled_ids(compiled: list[dict[str, object]]) -> set[str]:
"""Candidate ids that already have a verdict (compiled, keep-llm, dropped).
A handled candidate never counts toward the unreviewed nudge and its
table row stops suggesting a review — its story lives in the Reviewed
section and the scoreboard instead.
Args:
compiled: Entries from compiled.json.
Returns:
The candidate ids with any recorded verdict.
"""
return {
str(entry.get("candidate_id"))
for entry in compiled
if entry.get("candidate_id")
}
def significant_ids(candidates: list[Candidate]) -> list[str]:
"""Ids of the candidates big enough to surface in the badge.
Args:
candidates: Ranked candidates.
Returns:
Ids with at least :data:`SIGNIFICANT_WEEKLY_TOKENS` estimated weekly
tokens at stake, in rank order.
"""
return [
cand.id
for cand in candidates
if cand.est_weekly_savings >= SIGNIFICANT_WEEKLY_TOKENS
]
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.
"""
handled = handled_ids(compiled)
parts = [
_header(records, now),
_turn_efficiency(records),
_candidate_table(result.candidates, handled),
]
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 _turn_efficiency(records: list[SessionRecord]) -> str:
"""Render the turn-efficiency block.
Cache read is charged per API round-trip on the whole conversation so far, so
turns the model did not need to take are the single largest recoverable cost —
far larger than message length. This block prices them.
Args:
records: All session records.
Returns:
The turn-efficiency section.
"""
api = sum(r.turns.api_turns for r in records)
if not api:
return "## Turn efficiency\n\nno tool turns recorded yet\n"
batched = sum(r.turns.batched_turns for r in records)
poll = sum(r.turns.poll_turns for r in records)
lint = sum(r.turns.lint_test_turns for r in records)
vcs = sum(r.turns.vcs_check_turns for r in records)
calls = sum(r.turns.tool_calls for r in records)
cache_read = sum(r.tokens.cache_read for r in records)
mechanical = poll + lint + vcs
per_turn = cache_read // api if api else 0
return (
"## Turn efficiency\n\n"
f"{api} tool turns carrying {calls} calls | "
f"batched: {batched} ({batched / api:.0%}) | "
f"avg context re-read: {fmt_tokens(per_turn)}/turn\n\n"
"| mechanical turn kind | turns | est. cache-read |\n"
"| --- | ---: | ---: |\n"
f"| polling (progress, liveness, clocks) | {poll} "
f"| {fmt_tokens(poll * per_turn)} |\n"
f"| lint / test (exit-code adjudicated) | {lint} "
f"| {fmt_tokens(lint * per_turn)} |\n"
f"| read-only VCS checks | {vcs} | {fmt_tokens(vcs * per_turn)} |\n"
f"| **total** | **{mechanical}** ({mechanical / api:.0%}) | "
f"**{fmt_tokens(mechanical * per_turn)}** |\n\n"
"These turns decided nothing: a script could have taken them and reported "
"once. Batching independent calls cuts the rest.\n"
)
def _candidate_table(candidates: list[Candidate], handled: set[str]) -> str:
"""Render the ranked candidate table.
Args:
candidates: Ranked candidates.
handled: Ids with a recorded verdict — their action column stops
suggesting a review.
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 | "
f"{action_cell(cand, handled)} |"
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") in ("keep-llm", "dropped")
]
if not kept:
return "## Reviewed — keeping LLM\n\nnone reviewed yet\n"
lines = [
f"- {entry.get('candidate_id')} — "
f"reviewed {entry.get('compiled_at')} — {verdict_label(entry)}"
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"
)
pending = [
cid
for cid in significant_ids(result.candidates)
if cid not in handled_ids(compiled)
]
return write_state(home, pending, now)

43
python_pkg/session_autopsy/report_cells.py
View File

@ -1,43 +0,0 @@
"""Cell labels for the candidate table in REPORT.md.
Split out of ``report.py`` to keep it inside the repo's 500-line-per-file
limit. Both functions are pure label logic with no formatting dependencies,
which is why these two came out and the rest of the table rendering did not:
the detail block needs ``fmt_tokens`` and would make the import circular.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from python_pkg.session_autopsy.detectors import Candidate
def action_cell(cand: Candidate, handled: set[str]) -> str:
"""The table's action column for one candidate.
Args:
cand: The candidate.
handled: Ids with a recorded verdict.
Returns:
The suggested action, or a pointer to where its verdict lives.
"""
if cand.id in handled:
return "already handled — see Reviewed/scoreboard"
return cand.action
def verdict_label(entry: dict[str, object]) -> str:
"""Human label for a reviewed entry's verdict.
Args:
entry: A compiled.json entry.
Returns:
``"dropped (workflow removed)"`` or ``"keep LLM"``.
"""
if entry.get("verdict") == "dropped":
return "dropped (workflow removed)"
return "keep LLM"

67
python_pkg/session_autopsy/signatures.py
View File

@ -1,67 +0,0 @@
"""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("<PATH>", first)
first = _HEX_RE.sub("<HEX>", first)
first = _NUM_RE.sub("<N>", 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 <PATH>"``, 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(("-", "<N>")):
return f"{head} {word}"
return head

120
python_pkg/session_autopsy/stats.py
View File

@ -1,120 +0,0 @@
"""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

136
python_pkg/session_autopsy/store.py
View File

@ -1,136 +0,0 @@
"""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)

1
python_pkg/session_autopsy/tests/__init__.py
View File

@ -1 +0,0 @@
"""Tests for the session_autopsy package."""

218
python_pkg/session_autopsy/tests/conftest.py
View File

@ -1,218 +0,0 @@
"""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(),
)

194
python_pkg/session_autopsy/tests/test_cli.py
View File

@ -1,194 +0,0 @@
"""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

240
python_pkg/session_autopsy/tests/test_detectors.py
View File

@ -1,240 +0,0 @@
"""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: <PATH>": 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: <PATH>", "spread: x"}
assert errs["boom: <PATH>"].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"

356
python_pkg/session_autopsy/tests/test_parse.py
View File

@ -1,356 +0,0 @@
"""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("<command-name>/phone-deploy</command-name> args"),
assistant_line(bash_block("adb devices")),
prompt_line("<command-name>/model</command-name>"),
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("<local-command-caveat>stuff</local-command-caveat>"),
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 '<PATH>': Permission denied"] == 1
assert "Error: Exit code <N>" 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("<command-name>/finish</command-name>"))
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) == []
def test_turn_efficiency_groups_by_request_id() -> None:
"""One API response can span several transcript lines.
Counting per line would report every turn as single-tool and hide the real
batching rate, so batching is grouped by requestId.
"""
acc = _Accumulator()
first = assistant_line(bash_block("wc -l < /tmp/scrape.log"))
first["requestId"] = "req-1"
second = assistant_line(bash_block("tail -3 /tmp/scrape.log"))
second["requestId"] = "req-1" # same API response, two tool calls
third = assistant_line(bash_block("pytest -q"))
third["requestId"] = "req-2"
_feed(acc, first, second, third)
acc.finish()
assert acc.turns.api_turns == 2
assert acc.turns.tool_calls == 3
assert acc.turns.batched_turns == 1 # req-1 carried two calls
assert acc.turns.batching_rate == 0.5
assert acc.turns.poll_turns == 1 # req-1 was pure log watching
assert acc.turns.lint_test_turns == 1 # req-2 was the test gate
assert acc.turns.mechanical_turns == 2
def test_turn_efficiency_ignores_turns_without_tools() -> None:
acc = _Accumulator()
prose = assistant_line({"type": "text", "text": "just talking"})
prose["requestId"] = "req-9"
_feed(acc, prose)
acc.finish()
assert acc.turns.api_turns == 0
assert acc.turns.batching_rate == 0.0
assert acc.turns.mechanical_rate == 0.0
def test_turn_efficiency_counts_vcs_checks_and_real_work_apart() -> None:
acc = _Accumulator()
check = assistant_line(bash_block("git ls-remote origin refs/heads/main"))
check["requestId"] = "req-a"
work = assistant_line(bash_block("./run.sh --profile kcd index"))
work["requestId"] = "req-b"
_feed(acc, check, work)
acc.finish()
assert acc.turns.vcs_check_turns == 1
assert acc.turns.mechanical_turns == 1 # the index build is real work
assert acc.turns.api_turns == 2

150
python_pkg/session_autopsy/tests/test_records.py
View File

@ -1,150 +0,0 @@
"""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

240
python_pkg/session_autopsy/tests/test_report.py
View File

@ -1,240 +0,0 @@
"""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,
TokenTotals,
TurnEfficiency,
)
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, overflow note, and handled-action swap."""
assert "none yet" in report._candidate_table([], set())
table = report._candidate_table(
[_fake_candidate(index) for index in range(45)], {"err-00000001"}
)
assert "plus 5 below the fold" in table
assert "already handled — see Reviewed/scoreboard" 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"),
_compiled_entry(verdict="dropped", candidate_id="skill-y"),
]
)
assert "skill-x" in section
assert "skill-y — reviewed" in section
assert "dropped (workflow removed)" 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(output=500_000)] * 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 == 1
assert len(result.candidates) > 1
assert (tmp_path / report.REPORT_FILE).is_file()
(tmp_path / report.COMPILED_FILE).write_text(
json.dumps([_compiled_entry()]), encoding="utf-8"
)
assert report.write_report(tmp_path, records, result, NOW) == 0
assert "already handled" in (tmp_path / report.REPORT_FILE).read_text(
encoding="utf-8"
)
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")
def test_turn_efficiency_section_prices_mechanical_turns() -> None:
"""The point of the section: put a token number on turns nobody needed."""
rec = record("s1", tokens=TokenTotals(cache_read=100_000_000))
rec.turns = TurnEfficiency(
api_turns=100,
tool_calls=120,
batched_turns=12,
poll_turns=20,
lint_test_turns=10,
vcs_check_turns=5,
)
text = report._turn_efficiency([rec])
assert "100 tool turns carrying 120 calls" in text
assert "batched: 12 (12%)" in text
assert "**35** (35%)" in text # 20 + 10 + 5 mechanical turns
def test_turn_efficiency_section_without_tool_turns() -> None:
assert "no tool turns recorded yet" in report._turn_efficiency([record("s1")])

28
python_pkg/session_autopsy/tests/test_signatures.py
View File

@ -1,28 +0,0 @@
"""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 <PATH>"
assert normalize_signature("deadbeefcafe1234") == "<HEX>"
assert normalize_signature("retry 12 times") == "retry <N> 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 <PATH>"
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"

65
python_pkg/session_autopsy/tests/test_store.py
View File

@ -1,65 +0,0 @@
"""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)}

266
python_pkg/session_autopsy/tests/test_traces.py
View File

@ -1,266 +0,0 @@
"""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 '<PATH>': 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

71
python_pkg/session_autopsy/tests/test_turns.py
View File

@ -1,71 +0,0 @@
"""Tests for classifying turns a script could have taken instead of the model."""
from __future__ import annotations
import pytest
from python_pkg.session_autopsy.turns import (
LINT_TEST,
POLL,
SUBSTANTIVE,
VCS_CHECK,
classify_command,
classify_turn,
)
@pytest.mark.parametrize(
("command", "expected"),
[
# Waiting on something.
("kill -0 262023 2>/dev/null && echo ALIVE", POLL),
("pgrep -af claude | head", POLL),
("ps -eo pid,pgid,stat,args | grep python", POLL),
("wc -l < /tmp/scrape.log", POLL),
("tail -5 /tmp/scrape.log", POLL),
("nvidia-smi --query-gpu=utilization.gpu --format=csv", POLL),
("sleep 30", POLL),
("until grep -q DONE log; do sleep 5; done", POLL),
("date '+%H:%M'", POLL),
# Adjudicated by an exit code.
("pytest -q --cov=wikikb", LINT_TEST),
(".venv/bin/python -m ruff check wikikb", LINT_TEST),
("pre-commit run --all-files", LINT_TEST),
("shellcheck scripts/safepush.sh", LINT_TEST),
# Read-only VCS inspection.
("git status --short", VCS_CHECK),
("git ls-remote origin refs/heads/main", VCS_CHECK),
("git rev-parse HEAD", VCS_CHECK),
("gh api repos/kuhyx/wiki-kb --jq .visibility", VCS_CHECK),
# Real work: anything that changes state, fetches, or builds.
("./run.sh --profile kcd index", SUBSTANTIVE),
("git commit -q -m 'thing'", SUBSTANTIVE),
("git push origin main", SUBSTANTIVE),
("curl -sS https://example.com/api.php", SUBSTANTIVE),
("", SUBSTANTIVE),
(" ", SUBSTANTIVE),
],
)
def test_classify_command(command: str, expected: str) -> None:
assert classify_command(command) == expected
def test_commit_and_push_are_substantive_even_though_scriptable() -> None:
# They change state, so a turn carrying them was not pure observation. Counting
# them as waste would overstate the finding, and an overstated report is worse
# than none.
assert classify_command("git add -A && git commit -m x") == SUBSTANTIVE
def test_classify_turn_requires_every_command_to_be_mechanical() -> None:
assert classify_turn(["git status", "git rev-parse HEAD"]) == VCS_CHECK
assert classify_turn(["pytest -q"]) == LINT_TEST
# Mixed kinds: still mechanical individually, but the turn is not one thing.
assert classify_turn(["git status", "pytest -q"]) == SUBSTANTIVE
# One real action makes the round-trip necessary regardless of what else it did.
assert classify_turn(["wc -l < log", "./run.sh index"]) == SUBSTANTIVE
def test_classify_turn_with_no_bash_is_substantive() -> None:
# A turn that ran Edit/Write/Read rather than Bash is not a polling turn.
assert classify_turn([]) == SUBSTANTIVE

392
python_pkg/session_autopsy/traces.py
View File

@ -1,392 +0,0 @@
"""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

76
python_pkg/session_autopsy/turn_tracker.py
View File

@ -1,76 +0,0 @@
"""Turn-efficiency accounting for a single transcript.
Split out of ``parse.py``: the tracker is self-contained — it knows only about
turn totals and the classification of the commands in a turn — and lifting it
out keeps the parser inside the repo's 500-line-per-file limit.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from python_pkg.session_autopsy.records import TurnEfficiency
from python_pkg.session_autopsy.turns import LINT_TEST, POLL, VCS_CHECK, classify_turn
@dataclass
class OpenTurn:
"""The API turn currently being accumulated, keyed by requestId."""
request_id: str | None = None
tools: int = 0
commands: list[str] = field(default_factory=list)
class TurnTracker:
"""Running turn-efficiency totals plus the turn currently open.
The two belong together: the open turn only exists to be folded into the
totals, and keeping them as one object is also what keeps the parser's
accumulator inside its instance-attribute budget.
"""
def __init__(self) -> None:
"""Start with empty totals and no open turn."""
self.totals = TurnEfficiency()
self._open = OpenTurn()
def begin(self, request_id: object) -> None:
"""Open a new turn if this line belongs to a different API response.
Args:
request_id: The line's ``requestId`` (or message-id fallback).
"""
if request_id != self._open.request_id:
self.flush()
self._open.request_id = request_id if isinstance(request_id, str) else None
def add_tool(self) -> None:
"""Count one tool_use block in the open turn."""
self._open.tools += 1
def add_command(self, command: str) -> None:
"""Record one Bash command issued in the open turn.
Args:
command: The raw Bash tool input.
"""
self._open.commands.append(command)
def flush(self) -> None:
"""Fold the open turn into the totals and start a fresh one."""
open_turn = self._open
self._open = OpenTurn()
if open_turn.tools <= 0:
return
self.totals.api_turns += 1
self.totals.tool_calls += open_turn.tools
if open_turn.tools > 1:
self.totals.batched_turns += 1
label = classify_turn(open_turn.commands)
if label == POLL:
self.totals.poll_turns += 1
elif label == LINT_TEST:
self.totals.lint_test_turns += 1
elif label == VCS_CHECK:
self.totals.vcs_check_turns += 1

109
python_pkg/session_autopsy/turns.py
View File

@ -1,109 +0,0 @@
"""Classify assistant turns by whether a model was needed to take them.
Cache-read tokens dominate the cost of a long session — one measured session spent
235M cache-read against 800k output, a ratio of roughly 290:1. Cache read is paid per
API round-trip on the whole conversation so far, so the lever is the NUMBER OF TURNS,
not the length of any message. A short turn taken late is as expensive as a long one.
That makes it worth counting the turns where the model decided nothing: polling a
counter, running a linter, checking whether a push landed. Those are pass/fail
mechanisms that code can adjudicate, which is exactly what belongs in a script rather
than in a round-trip (see ``rules/token-spend.instructions.md``).
Classification is deliberately conservative: a command is only called mechanical when
its whole purpose is observation. Anything that edits, fetches, builds or decides
counts as substantive, so the reported waste is a floor rather than a guess.
"""
from __future__ import annotations
import re
POLL = "poll"
LINT_TEST = "lint_test"
VCS_CHECK = "vcs_check"
SUBSTANTIVE = "substantive"
MECHANICAL = frozenset({POLL, LINT_TEST, VCS_CHECK})
# Ordered: the first pattern to match wins, so put the narrow ones first.
_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
# Waiting on something: process liveness, log tails, progress counters, clocks.
(
POLL,
re.compile(
r"""(?x)
\bkill\s+-0\b # is that pid still alive
| \bpgrep\b | \bpidof\b
| \bps\s+-[eo] # ps -eo / ps -o : process listing
| \bnvidia-smi\b
| \bwc\s+-l\s*< # counting lines in a log
| \btail\s+(-[nf]?\d*\s+)?\S*\.log
| ^\s*sleep\s
| \buntil\s+.*\bdo\b.*\bdone\b
| \bdate\s+['\"]?\+
"""
),
),
# Adjudicated by an exit code, never by a model.
(
LINT_TEST,
re.compile(
r"\b(pytest|ruff|mypy|shellcheck|pre-commit\s+run|npm\s+test|cargo\s+test)\b"
),
),
# Read-only VCS inspection. `git commit`/`push`/`add` are excluded on purpose:
# they change state, so they are substantive even though they are scriptable.
(
VCS_CHECK,
re.compile(
r"""(?x)
\bgit\s+(status|log|diff|ls-remote|rev-parse|remote\s+-v|branch|show)\b
| \bgh\s+(api|repo\s+view|pr\s+view)\b
"""
),
),
)
def classify_command(command: str) -> str:
"""Classify one Bash command.
Args:
command: The raw Bash tool input (may be multi-line).
Returns:
One of :data:`POLL`, :data:`LINT_TEST`, :data:`VCS_CHECK` or
:data:`SUBSTANTIVE`.
"""
text = command.strip()
if not text:
return SUBSTANTIVE
for label, pattern in _PATTERNS:
if pattern.search(text):
return label
return SUBSTANTIVE
def classify_turn(commands: list[str]) -> str:
"""Classify a whole turn from the Bash commands it issued.
A turn only counts as mechanical when *every* command in it is — one real piece
of work makes the round-trip necessary regardless of what else it carried, and
over-counting waste would make the report untrustworthy.
Args:
commands: Every Bash command issued in this turn, in order.
Returns:
The shared mechanical label, or :data:`SUBSTANTIVE` if the turn mixed kinds,
carried real work, or ran no Bash at all.
"""
if not commands:
return SUBSTANTIVE
labels = {classify_command(c) for c in commands}
if len(labels) == 1:
only = labels.pop()
if only in MECHANICAL:
return only
return SUBSTANTIVE