feat: add the week-review skill

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.
2026-08-01 15:15:14 +00:00
parent ad4d14e078
commit f6b58e8849
3 changed files with 278 additions and 0 deletions
@@ -0,0 +1,144 @@
+#!/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())