naps62
f6b58e8849
Codifies the session-review process: read open issues for carry-over, scan transcripts, rank by how often the same correction repeated, check the current model docs before recommending, then apply or file. scripts/scan-sessions.py enumerates top-level sessions in a window, drops subagent transcripts, and flags swarm runs so a single 577-session security scan does not read as 40% of the week's work.
145 lines
4.7 KiB
Python
Executable File
145 lines
4.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Enumerate Claude Code sessions in a window and dump their human turns.
|
|
|
|
Usage: scan-sessions.py [--days N] [--since YYYY-MM-DD] [--out DIR]
|
|
|
|
Writes two files to --out (default: cwd):
|
|
sessions.json one record per top-level session, oldest first
|
|
userturns.txt every human turn, grouped by session, for reading
|
|
|
|
A "top-level" session is one with no isSidechain marker and at least one real
|
|
human turn, which drops subagent transcripts. Swarm runs still show up as many
|
|
sessions sharing one cwd and timestamp — the report should collapse those.
|
|
"""
|
|
import argparse
|
|
import datetime as dt
|
|
import glob
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
SKIP_PREFIXES = (
|
|
"<local-command", "<command-", "<task-notification", "<system-reminder",
|
|
"Caveat:", "Base directory for this skill:",
|
|
)
|
|
|
|
|
|
def human_turns(path):
|
|
"""Yield (timestamp, text) for each real human turn in a transcript."""
|
|
for line in open(path, errors="replace"):
|
|
try:
|
|
d = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if d.get("isSidechain") or d.get("type") != "user":
|
|
continue
|
|
c = (d.get("message") or {}).get("content")
|
|
if isinstance(c, list):
|
|
c = " ".join(
|
|
x.get("text", "") for x in c
|
|
if isinstance(x, dict) and x.get("type") == "text"
|
|
)
|
|
if not isinstance(c, str):
|
|
continue
|
|
c = " ".join(c.split())
|
|
if not c or c.startswith(SKIP_PREFIXES):
|
|
continue
|
|
if "This session is being continued" in c[:60]:
|
|
continue
|
|
yield d.get("timestamp"), c
|
|
|
|
|
|
def scan(root, cutoff):
|
|
out = []
|
|
for f in glob.glob(os.path.join(root, "*", "*.jsonl")):
|
|
try:
|
|
st = os.stat(f)
|
|
except OSError:
|
|
continue
|
|
if st.st_mtime < cutoff:
|
|
continue
|
|
turns, first_ts, sidechain, nlines = [], None, False, 0
|
|
try:
|
|
for line in open(f, errors="replace"):
|
|
nlines += 1
|
|
try:
|
|
d = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if d.get("isSidechain"):
|
|
sidechain = True
|
|
break
|
|
if first_ts is None and d.get("timestamp"):
|
|
first_ts = d["timestamp"]
|
|
if d.get("cwd") and "cwd" not in locals():
|
|
pass
|
|
if sidechain:
|
|
continue
|
|
turns = list(human_turns(f))
|
|
except OSError:
|
|
continue
|
|
if not turns:
|
|
continue
|
|
cwd = None
|
|
for line in open(f, errors="replace"):
|
|
try:
|
|
cwd = json.loads(line).get("cwd")
|
|
except ValueError:
|
|
continue
|
|
if cwd:
|
|
break
|
|
out.append({
|
|
"ts": first_ts or "",
|
|
"cwd": cwd or os.path.basename(os.path.dirname(f)),
|
|
"mb": round(st.st_size / 1048576, 1),
|
|
"lines": nlines,
|
|
"turns": [t for _, t in turns],
|
|
"file": f,
|
|
})
|
|
out.sort(key=lambda x: x["ts"])
|
|
return out
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--days", type=int, default=7)
|
|
ap.add_argument("--since")
|
|
ap.add_argument("--out", default=".")
|
|
ap.add_argument("--root", default=os.path.expanduser("~/.claude/projects"))
|
|
a = ap.parse_args()
|
|
|
|
if a.since:
|
|
cutoff = dt.datetime.fromisoformat(a.since).timestamp()
|
|
else:
|
|
cutoff = (dt.datetime.now() - dt.timedelta(days=a.days)).timestamp()
|
|
|
|
sessions = scan(a.root, cutoff)
|
|
os.makedirs(a.out, exist_ok=True)
|
|
|
|
with open(os.path.join(a.out, "sessions.json"), "w") as fh:
|
|
json.dump([{k: v for k, v in s.items() if k != "turns"} for s in sessions], fh, indent=1)
|
|
|
|
with open(os.path.join(a.out, "userturns.txt"), "w") as fh:
|
|
for s in sessions:
|
|
fh.write(f"\n===== {s['ts'][:16]} {s['cwd']} ({s['mb']}MB) =====\n")
|
|
for t in s["turns"]:
|
|
fh.write("- " + t[:500] + "\n")
|
|
|
|
by_cwd = {}
|
|
for s in sessions:
|
|
by_cwd[s["cwd"]] = by_cwd.get(s["cwd"], 0) + 1
|
|
swarms = {k: v for k, v in by_cwd.items() if v > 20}
|
|
|
|
print(f"{len(sessions)} top-level sessions, "
|
|
f"{sum(len(s['turns']) for s in sessions)} human turns, "
|
|
f"{sum(s['mb'] for s in sessions):.0f}MB")
|
|
if swarms:
|
|
print("likely swarm runs (collapse these to one line in the report):")
|
|
for k, v in sorted(swarms.items(), key=lambda x: -x[1]):
|
|
print(f" {v:4} sessions {k}")
|
|
print(f"wrote {a.out}/sessions.json and {a.out}/userturns.txt")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|