Skills › BMAD Method › bmad-toolsmith › read_session_log.py
read_session_log.py
247 lines · skills/bmad-toolsmith/scripts/read_session_log.py at bd70c64
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# ///
5"""read_session_log: digest Claude Code transcripts and memlogs into a compact JSON for the from-logs approach.
7Sources:
8 claude-code a session transcript, one JSON object per line, as Claude Code keeps them under
9 ~/.claude/projects/<encoded-cwd>/*.jsonl (the cwd with every "/" and "." turned into "-";
10 $CLAUDE_CONFIG_DIR replaces ~/.claude when set). Read: user prompts, tool calls per turn,
11 `Skill` tool calls and /command invocations, files named in tool inputs. Meta and sidechain
12 (subagent) records are skipped. Tool results and assistant prose are never copied out.
13 memlog a .memlog.md: entries typed direction or decision, or marked `by user`, count as requests;
14 entries typed gap or correction, or reading like a correction, count as corrections.
16A correction is a user message that opens with a word of refusal or redirection (no, don't, stop, wrong,
17wait, actually, instead, undo, revert, ...) or a request interrupted by the user. It is a heuristic; read the
18digest as leads, not verdicts.
20Usage:
21 read_session_log.py [<path> ...] [--project <cwd>] [--format auto|claude-code|memlog] [--max-items N]
23 A path may be a file or a folder (every *.jsonl and *.memlog.md inside). --project resolves the
24 transcript folder for that working directory. --format auto picks by file name.
26Output, one JSON object on stdout; every text is cut to 300 characters and every list to --max-items:
27 {"sources": [{"path", "format", "entries"}], "user_requests": [{"text", "count"}],
28 "tool_sequences": [{"tools": [...], "count"}], "corrections": [text, ...],
29 "files_touched": [path, ...], "skills_invoked": [{"name", "count"}]}
30"""
32from __future__ import annotations
34import argparse
35import json
36import os
37import re
38import sys
39from collections import Counter
40from pathlib import Path
42TEXT_LIMIT = 300
43CORRECTION_RE = re.compile(
44 r"^(?:no|nope|not|don'?t|do not|stop|wrong|wait|actually|instead|undo|revert|never|hold on|"
45 r"that'?s (?:not|wrong)|that is (?:not|wrong)|why did you|i said|i asked|not what)\b",
46 re.IGNORECASE,
47)
48INTERRUPTED_RE = re.compile(r"^\[Request interrupted by user")
49COMMAND_RE = re.compile(r"<command-name>\s*/?([^<\s]+)\s*</command-name>")
50MEMLOG_ENTRY_RE = re.compile(r"^- (?:\((?P<tag>[^)]*)\) )?(?P<text>.*)$")
51FILE_INPUT_KEYS = ("file_path", "notebook_path", "path")
54def truncate(text: str) -> str:
55 text = " ".join(text.split())
56 return text if len(text) <= TEXT_LIMIT else text[: TEXT_LIMIT - 3] + "..."
59def encode_cwd(cwd: str) -> str:
60 return re.sub(r"[/.]", "-", cwd)
63def projects_dir() -> Path:
64 base = os.environ.get("CLAUDE_CONFIG_DIR")
65 return (Path(base) if base else Path.home() / ".claude") / "projects"
68class Digest:
69 def __init__(self) -> None:
70 self.sources: list[dict] = []
71 self.requests: Counter[str] = Counter()
72 self.request_text: dict[str, str] = {}
73 self.sequences: Counter[tuple[str, ...]] = Counter()
74 self.corrections: list[str] = []
75 self.files: set[str] = set()
76 self.skills: Counter[str] = Counter()
78 def add_request(self, text: str) -> None:
79 key = " ".join(text.lower().split())
80 if not key:
81 return
82 self.requests[key] += 1
83 self.request_text.setdefault(key, truncate(text))
84 if CORRECTION_RE.match(text.strip()) or INTERRUPTED_RE.match(text.strip()):
85 self.add_correction(text)
87 def add_correction(self, text: str) -> None:
88 short = truncate(text)
89 if short not in self.corrections:
90 self.corrections.append(short)
92 def add_sequence(self, tools: list[str]) -> None:
93 collapsed: list[str] = []
94 for tool in tools:
95 if not collapsed or collapsed[-1] != tool:
96 collapsed.append(tool)
97 if collapsed:
98 self.sequences[tuple(collapsed)] += 1
100 def render(self, max_items: int) -> dict:
101 return {
102 "sources": self.sources,
103 "user_requests": [
104 {"text": self.request_text[key], "count": count} for key, count in self.requests.most_common(max_items)
105 ],
106 "tool_sequences": [
107 {"tools": list(seq), "count": count} for seq, count in self.sequences.most_common(max_items)
108 ],
109 "corrections": self.corrections[:max_items],
110 "files_touched": sorted(self.files)[:max_items],
111 "skills_invoked": [{"name": name, "count": count} for name, count in self.skills.most_common(max_items)],
112 }
115def user_text(content) -> str | None:
116 """Return the prompt text of a user record, or None when it is only tool results."""
117 if isinstance(content, str):
118 return content
119 if isinstance(content, list):
120 texts = [b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"]
121 return "\n".join(t for t in texts if t) if texts else None
122 return None
125def read_claude_code(path: Path, digest: Digest) -> int:
126 entries = 0
127 turn_tools: list[str] = []
128 for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
129 try:
130 record = json.loads(line)
131 except json.JSONDecodeError:
132 continue
133 if not isinstance(record, dict) or record.get("isMeta") or record.get("isSidechain"):
134 continue
135 message = record.get("message") if isinstance(record.get("message"), dict) else None
136 if message is None:
137 continue
138 content = message.get("content")
139 kind = record.get("type")
140 if kind == "user":
141 text = user_text(content)
142 if text is None:
143 continue
144 entries += 1
145 digest.add_sequence(turn_tools)
146 turn_tools = []
147 for command in COMMAND_RE.findall(text):
148 digest.skills[command] += 1
149 stripped = text.strip()
150 if INTERRUPTED_RE.match(stripped):
151 digest.add_correction(stripped)
152 elif stripped and not stripped.startswith("<"):
153 digest.add_request(stripped)
154 elif kind == "assistant" and isinstance(content, list):
155 for block in content:
156 if not isinstance(block, dict) or block.get("type") != "tool_use":
157 continue
158 entries += 1
159 name = str(block.get("name", ""))
160 turn_tools.append(name)
161 inputs = block.get("input") if isinstance(block.get("input"), dict) else {}
162 if name == "Skill" and inputs.get("skill"):
163 digest.skills[str(inputs["skill"])] += 1
164 for key in FILE_INPUT_KEYS:
165 if isinstance(inputs.get(key), str) and inputs[key]:
166 digest.files.add(inputs[key])
167 digest.add_sequence(turn_tools)
168 return entries
171def read_memlog(path: Path, digest: Digest) -> int:
172 entries = 0
173 for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
174 match = MEMLOG_ENTRY_RE.match(line)
175 if not match:
176 continue
177 entries += 1
178 tag = (match.group("tag") or "").strip().lower()
179 kind = tag.split(" by ")[0].strip()
180 by_user = " by user" in f" {tag}" or tag.startswith("by user")
181 text = match.group("text").strip()
182 if kind in ("gap", "correction"):
183 digest.add_correction(text)
184 elif kind in ("direction", "decision") or by_user:
185 digest.add_request(text)
186 elif CORRECTION_RE.match(text):
187 digest.add_correction(text)
188 return entries
191def detect_format(path: Path, requested: str) -> str | None:
192 if requested != "auto":
193 return requested
194 if path.suffix == ".jsonl":
195 return "claude-code"
196 if path.name.endswith(".memlog.md") or path.name == ".memlog.md":
197 return "memlog"
198 return None
201def expand(paths: list[Path]) -> list[Path]:
202 files: list[Path] = []
203 for path in paths:
204 if path.is_dir():
205 files.extend(sorted(path.glob("*.jsonl"), key=lambda p: p.stat().st_mtime))
206 files.extend(sorted(path.glob("*.memlog.md")))
207 else:
208 files.append(path)
209 return files
212def main(argv: list[str] | None = None) -> int:
213 p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
214 p.add_argument("paths", nargs="*", type=Path, help="transcript files, memlogs, or folders of them")
215 p.add_argument("--project", help="working directory whose Claude Code transcripts to read")
216 p.add_argument("--format", choices=("auto", "claude-code", "memlog"), default="auto")
217 p.add_argument("--max-items", type=int, default=20, help="cap for every list (default 20)")
218 args = p.parse_args(argv)
220 paths = list(args.paths)
221 if args.project:
222 folder = projects_dir() / encode_cwd(str(Path(args.project).expanduser().resolve()))
223 if not folder.is_dir():
224 p.error(f"no transcripts for {args.project} at {folder}")
225 paths.append(folder)
226 if not paths:
227 p.error("give at least one path or --project")
228 missing = [str(path) for path in paths if not path.exists()]
229 if missing:
230 p.error(f"not found: {', '.join(missing)}")
232 digest = Digest()
233 for path in expand(paths):
234 fmt = detect_format(path, args.format)
235 if fmt is None:
236 print(f"read_session_log: skipping {path}, unknown format", file=sys.stderr)
237 continue
238 reader = read_claude_code if fmt == "claude-code" else read_memlog
239 entries = reader(path, digest)
240 digest.sources.append({"path": str(path), "format": fmt, "entries": entries})
242 print(json.dumps(digest.render(args.max_items), indent=2))
243 return 0
246if __name__ == "__main__":
247 sys.exit(main())