Skills › Compound Engineering › ce-compound › extract-metadata.py
extract-metadata.py
551 lines · skills/ce-compound/scripts/session-history/extract-metadata.py at 3449230
1#!/usr/bin/env python3
2"""Extract session metadata from Claude Code, Codex, Cursor, Pi, and oh-my-pi (omp) JSONL files.
4Batch mode (preferred — one invocation for all files):
5 python3 extract-metadata.py /path/to/dir/*.jsonl
6 python3 extract-metadata.py file1.jsonl file2.jsonl file3.jsonl
8Single-file mode (stdin):
9 head -20 <session.jsonl> | python3 extract-metadata.py
11Auto-detects platform from the JSONL structure.
12Outputs one JSON object per file, one per line.
13Includes a final _meta line with processing stats.
14"""
15import sys
16import json
17import os
19MAX_LINES = 200 # Resumed Claude sessions front-load dozens of non-message records before the first user record
22def try_claude(lines):
23 result = None
24 cwd = ""
25 for line in lines:
26 try:
27 obj = json.loads(line.strip())
28 except (json.JSONDecodeError, KeyError):
29 continue
30 if not cwd and obj.get("cwd"):
31 cwd = obj["cwd"]
32 if result is None and obj.get("type") == "user" and "gitBranch" in obj:
33 result = {
34 "platform": "claude",
35 "branch": obj["gitBranch"],
36 "ts": obj.get("timestamp", ""),
37 "session": obj.get("sessionId", ""),
38 }
39 if obj.get("cwd"):
40 cwd = obj["cwd"]
41 if result is not None and cwd:
42 break
43 if result is not None and cwd:
44 result["cwd"] = cwd
45 return result
48def try_codex(lines):
49 meta = {}
50 for line in lines:
51 try:
52 obj = json.loads(line.strip())
53 if obj.get("type") == "session_meta":
54 p = obj.get("payload", {})
55 meta["platform"] = "codex"
56 meta["cwd"] = p.get("cwd", "")
57 meta["session"] = p.get("id", "")
58 meta["ts"] = p.get("timestamp", obj.get("timestamp", ""))
59 meta["source"] = p.get("source", "")
60 meta["cli_version"] = p.get("cli_version", "")
61 elif obj.get("type") == "turn_context":
62 p = obj.get("payload", {})
63 meta["model"] = p.get("model", "")
64 meta["cwd"] = meta.get("cwd") or p.get("cwd", "")
65 except (json.JSONDecodeError, KeyError):
66 pass
67 return meta if meta else None
70def try_omp(lines):
71 """oh-my-pi (omp) sessions: a fixed-width type='title' slot line physically
72 first, then a pi-shaped type='session' header with cwd. Checked before Pi:
73 a bare pi file has no title slot and must still detect as pi."""
74 seen_first = False
75 for line in lines:
76 stripped = line.strip()
77 if not stripped:
78 continue
79 try:
80 obj = json.loads(stripped)
81 except (json.JSONDecodeError, KeyError):
82 if not seen_first:
83 return None
84 continue
85 if not seen_first:
86 seen_first = True
87 if obj.get("type") != "title":
88 return None
89 continue
90 if obj.get("type") == "session" and "cwd" in obj:
91 return {
92 "platform": "omp",
93 "cwd": obj.get("cwd", ""),
94 "session": obj.get("id", ""),
95 "ts": obj.get("timestamp", ""),
96 }
97 return None
100def try_pi(lines):
101 """Pi sessions: type='session' header with cwd, followed by message entries."""
102 for line in lines:
103 try:
104 obj = json.loads(line.strip())
105 if obj.get("type") == "session" and "cwd" in obj:
106 return {
107 "platform": "pi",
108 "cwd": obj.get("cwd", ""),
109 "session": obj.get("id", ""),
110 "ts": obj.get("timestamp", ""),
111 }
112 except (json.JSONDecodeError, KeyError):
113 pass
114 return None
117def try_cursor(lines):
118 """Cursor agent transcripts: role-based entries, no timestamps or metadata fields."""
119 for line in lines:
120 try:
121 obj = json.loads(line.strip())
122 # Cursor entries have 'role' at top level but no 'type'
123 if obj.get("role") in ("user", "assistant") and "type" not in obj:
124 return {"platform": "cursor"}
125 except (json.JSONDecodeError, KeyError):
126 pass
127 return None
130def extract_from_lines(lines):
131 return try_claude(lines) or try_codex(lines) or try_omp(lines) or try_pi(lines) or try_cursor(lines)
134TAIL_BYTES = 16384 # Read last 16KB to find final timestamp past trailing metadata
137def get_last_timestamp(filepath, size):
138 """Read the tail of a file to find the last message with a timestamp."""
139 try:
140 with open(filepath, "rb") as f:
141 f.seek(max(0, size - TAIL_BYTES))
142 tail = f.read().decode("utf-8", errors="ignore")
143 lines = tail.strip().split("\n")
144 for line in reversed(lines):
145 try:
146 obj = json.loads(line.strip())
147 if "timestamp" in obj:
148 return obj["timestamp"]
149 except (json.JSONDecodeError, KeyError):
150 pass
151 except (OSError, IOError):
152 pass
153 return None
156def _pi_active_path_objects(objects):
157 """Return only entries on Pi's active leaf-to-root path.
159 Pi session files are append-only trees. The final non-session entry is the
160 active leaf; abandoned branches remain in the file but are not in context.
161 """
162 by_id = {
163 obj.get("id"): obj
164 for obj in objects
165 if isinstance(obj.get("id"), str) and obj.get("type") != "session"
166 }
167 leaf_id = None
168 for obj in objects:
169 if obj.get("type") != "session" and isinstance(obj.get("id"), str):
170 leaf_id = obj["id"]
171 if not leaf_id:
172 return objects
174 active_ids = set()
175 current = leaf_id
176 while isinstance(current, str) and current and current not in active_ids:
177 active_ids.add(current)
178 parent = by_id.get(current, {}).get("parentId")
179 current = parent if isinstance(parent, str) else None
180 return [
181 obj
182 for obj in objects
183 if obj.get("type") == "session" or obj.get("id") in active_ids
184 ]
187def _pi_context_objects(objects):
188 """Return Pi entries that participate in active LLM context."""
189 active = _pi_active_path_objects(objects)
190 compactions = [obj for obj in active if obj.get("type") == "compaction"]
191 if not compactions:
192 return active
194 # Pi emits compaction summary first, then entries from firstKeptEntryId
195 # onward. Exclude older ancestors so keyword search mirrors context.
196 first_kept = compactions[-1].get("firstKeptEntryId")
197 if not isinstance(first_kept, str):
198 return active
200 latest_compaction_id = compactions[-1].get("id")
201 started = False
202 found_first_kept = False
203 context = [obj for obj in active if obj.get("type") == "session"]
204 context.append(compactions[-1])
205 for obj in active:
206 if obj.get("type") == "session":
207 continue
208 if obj.get("id") == first_kept:
209 started = True
210 found_first_kept = True
211 if obj.get("id") == latest_compaction_id:
212 continue
213 if started:
214 context.append(obj)
215 return context if found_first_kept and len(context) > 1 else active
218def _append_pi_content_text(chunks, content):
219 if isinstance(content, str):
220 chunks.append(content)
221 elif isinstance(content, list):
222 for block in content:
223 if isinstance(block, dict) and block.get("type") == "text":
224 chunks.append(block.get("text", ""))
227def _append_pi_tool_call_targets(chunks, content):
228 """Append searchable Pi toolCall targets without indexing tool output."""
229 if not isinstance(content, list):
230 return
231 for block in content:
232 if not isinstance(block, dict) or block.get("type") != "toolCall":
233 continue
234 args = block.get("arguments", {})
235 if not isinstance(args, dict):
236 continue
237 for key in ("path", "file_path", "command", "pattern", "query", "prompt"):
238 value = args.get(key)
239 if isinstance(value, str):
240 chunks.append(value)
243def _extract_user_assistant_text(filepath):
244 """Return concatenated user + assistant text content from a session JSONL.
246 Skips JSONL metadata field names and values (sessionId, gitBranch, uuid,
247 timestamps, type tags), tool_use blocks (tool names + tool inputs),
248 tool_result blocks (tool outputs), and thinking/reasoning blocks. Only
249 content the user or assistant actually said is included.
251 Without this filtering, common topic words like "session" would match every
252 JSONL file via the sessionId field, drowning out real content matches.
253 """
254 chunks = []
255 try:
256 objects = []
257 with open(filepath, "r", encoding="utf-8", errors="replace") as f:
258 for line in f:
259 try:
260 objects.append(json.loads(line.strip()))
261 except (json.JSONDecodeError, ValueError):
262 continue
264 # omp files share the pi-shaped session header, so this covers both.
265 is_pi = any(
266 obj.get("type") == "session" and "cwd" in obj for obj in objects
267 )
268 if is_pi:
269 objects = _pi_context_objects(objects)
271 for obj in objects:
272 # Claude Code: type-tagged top-level
273 t = obj.get("type")
274 if t == "user":
275 msg = obj.get("message", {})
276 content = msg.get("content")
277 if isinstance(content, str):
278 chunks.append(content)
279 elif isinstance(content, list):
280 for block in content:
281 if isinstance(block, dict) and block.get("type") == "text":
282 chunks.append(block.get("text", ""))
283 # Skip tool_result blocks — tool outputs are not user content.
284 continue
285 if t == "assistant":
286 msg = obj.get("message", {})
287 content = msg.get("content", [])
288 if isinstance(content, list):
289 for block in content:
290 if isinstance(block, dict) and block.get("type") == "text":
291 chunks.append(block.get("text", ""))
292 # Skip tool_use and thinking blocks.
293 continue
295 # Codex: payload-typed events
296 if t == "event_msg":
297 p = obj.get("payload", {})
298 if p.get("type") == "user_message":
299 # Strip Codex/Conductor `<system_instruction>...</system_instruction>`
300 # wrapper before counting. Without this, generic wrapper terms
301 # (e.g., "Conductor", environment labels) false-match against
302 # boilerplate the user did not author. Mirrors the same split
303 # used in ce-session-extract/scripts/extract-skeleton.py.
304 msg = p.get("message", "")
305 if isinstance(msg, str):
306 parts = msg.split("</system_instruction>")
307 chunks.append(parts[-1] if parts else msg)
308 continue
309 if t == "response_item":
310 p = obj.get("payload", {})
311 if p.get("type") == "message" and p.get("role") == "assistant":
312 for block in p.get("content", []):
313 if isinstance(block, dict) and block.get("type") == "output_text":
314 chunks.append(block.get("text", ""))
315 continue
317 # Pi: type='message' envelope with AgentMessage under message.
318 if t == "message" and "message" in obj:
319 msg = obj.get("message", {})
320 role = msg.get("role", "")
321 if role == "bashExecution":
322 command = msg.get("command", "")
323 if isinstance(command, str):
324 chunks.append(command)
325 # Search command text only. Output is tool output and can
326 # be large/noisy in the same way as toolResult content.
327 continue
328 content = msg.get("content", [])
329 if role == "custom":
330 _append_pi_content_text(chunks, content)
331 continue
332 if role not in ("user", "assistant"):
333 continue
334 _append_pi_content_text(chunks, content)
335 if role == "assistant":
336 _append_pi_tool_call_targets(chunks, content)
337 continue
339 if t in ("compaction", "branch_summary"):
340 summary = obj.get("summary", "")
341 if isinstance(summary, str):
342 chunks.append(summary)
343 continue
345 if t == "custom_message":
346 _append_pi_content_text(chunks, obj.get("content", []))
347 continue
349 # Cursor: role-tagged with no top-level type
350 if obj.get("role") in ("user", "assistant") and "type" not in obj:
351 msg = obj.get("message", {})
352 for block in msg.get("content", []) if isinstance(msg.get("content"), list) else []:
353 if isinstance(block, dict) and block.get("type") == "text":
354 chunks.append(block.get("text", ""))
355 continue
356 except (OSError, IOError):
357 pass
358 return "\n".join(chunks)
361def count_keyword_matches(filepath, keywords):
362 """Case-insensitive substring count for each keyword in user/assistant text.
364 Returns a dict {original_keyword: count}. Scans only content the user or
365 assistant said — not JSONL metadata, tool calls, tool outputs, or thinking
366 blocks — so common topic words like "session" do not false-match against
367 the sessionId field.
368 """
369 text_lower = _extract_user_assistant_text(filepath).lower()
370 return {kw: text_lower.count(kw.lower()) for kw in keywords}
373def process_file(filepath):
374 """Extract metadata only. Keyword scanning is done separately so callers
375 can apply cheap filters (e.g. --cwd-filter) before paying the full-file
376 content scan cost.
378 Returns (result, error): (dict, None) on success, (None, None) when the
379 file is readable but matches no platform, (None, filepath) on read error."""
380 try:
381 size = os.path.getsize(filepath)
382 with open(filepath, "r", encoding="utf-8", errors="replace") as f:
383 lines = []
384 for i, line in enumerate(f):
385 if i >= MAX_LINES:
386 break
387 lines.append(line)
388 result = extract_from_lines(lines)
389 if result:
390 result["file"] = filepath
391 result["size"] = size
392 return result, None
393 else:
394 return None, None
395 except (OSError, IOError) as e:
396 return None, filepath
399def _normalize_cwd(path):
400 path = path.replace("\\", "/").strip()
401 if len(path) > 1 and path.endswith("/") and not (
402 len(path) == 3 and path[1] == ":"
403 ):
404 path = path.rstrip("/")
405 # Windows drive paths are case-insensitive. Do not rewrite POSIX /d/...
406 # to D:/ — those are different filesystems.
407 if len(path) >= 2 and path[1] == ":":
408 path = path[0].upper() + ":" + path[2:].casefold()
409 return path
412def _is_abs_cwd(path):
413 if os.path.isabs(path):
414 return True
415 return len(path) >= 2 and path[1] == ":"
418def _cwd_paths_related(session_cwd, cwd_filter):
419 if session_cwd == cwd_filter:
420 return True
421 prefix = cwd_filter if cwd_filter.endswith("/") else cwd_filter + "/"
422 if session_cwd.startswith(prefix):
423 return True
424 prefix = session_cwd if session_cwd.endswith("/") else session_cwd + "/"
425 return cwd_filter.startswith(prefix)
428def _attach_timestamps(result, filepath):
429 if result["platform"] == "cursor":
430 # Cursor transcripts have no timestamps in JSONL.
431 # Use file modification time as the best available signal.
432 # Derive session ID from the parent directory name (UUID).
433 mtime = os.path.getmtime(filepath)
434 from datetime import datetime, timezone
436 result["ts"] = datetime.fromtimestamp(mtime, tz=timezone.utc).isoformat()
437 result["session"] = os.path.basename(os.path.dirname(filepath))
438 return
439 last_ts = get_last_timestamp(filepath, result["size"])
440 if last_ts:
441 result["last_ts"] = last_ts
444def cwd_matches_filter(session_cwd, cwd_filter):
445 if not session_cwd or not cwd_filter:
446 return True
447 session = _normalize_cwd(session_cwd)
448 filt = _normalize_cwd(cwd_filter)
449 if not session or not filt:
450 return True
451 if _is_abs_cwd(filt):
452 return _cwd_paths_related(session, filt)
453 return filt in session.split("/")
456# Parse arguments: files and optional --cwd-filter / --keyword
457files = []
458cwd_filter = None
459keywords = None
460args = sys.argv[1:]
461i = 0
462while i < len(args):
463 if args[i] == "--cwd-filter" and i + 1 < len(args):
464 cwd_filter = args[i + 1]
465 i += 2
466 elif args[i] == "--keyword" and i + 1 < len(args):
467 keywords = [k for k in args[i + 1].split(",") if k]
468 i += 2
469 elif not args[i].startswith("-"):
470 files.append(args[i])
471 i += 1
472 else:
473 i += 1
475if files:
476 # Batch mode: process all files
477 processed = 0
478 parse_errors = 0
479 no_metadata = 0
480 filtered = 0
481 matched = 0
482 for filepath in files:
483 if not filepath.endswith(".jsonl"):
484 continue
485 result, error = process_file(filepath)
486 processed += 1
487 if result:
488 # Apply CWD filter first: cheap metadata-only check. Skip
489 # sessions from other repos before paying the full-file keyword
490 # scan cost — Claude and Codex discovery list across projects,
491 # so without this ordering --keyword would scan files that are
492 # immediately discarded.
493 if cwd_filter:
494 session_cwd = result.get("cwd")
495 if session_cwd:
496 if not cwd_matches_filter(session_cwd, cwd_filter):
497 filtered += 1
498 continue
499 elif result.get("platform") in ("claude", "codex", "pi", "omp"):
500 filtered += 1
501 continue
502 _attach_timestamps(result, filepath)
503 # Apply keyword scan only after cheap filters pass.
504 if keywords:
505 matches = count_keyword_matches(filepath, keywords)
506 result["keyword_matches"] = matches
507 result["match_count"] = sum(matches.values())
508 if result["match_count"] == 0:
509 continue
510 matched += 1
511 print(json.dumps(result))
512 elif error:
513 parse_errors += 1
514 else:
515 no_metadata += 1
517 meta = {"_meta": True, "files_processed": processed, "parse_errors": parse_errors}
518 if no_metadata:
519 meta["files_without_metadata"] = no_metadata
520 if filtered:
521 meta["filtered_by_cwd"] = filtered
522 if keywords:
523 meta["files_matched"] = matched
524 print(json.dumps(meta))
525else:
526 # No file arguments: either single-file stdin mode or empty xargs invocation.
527 # When xargs runs us with no input (e.g., discover found no files), stdin is
528 # empty or a TTY — emit a clean zero-file result instead of a false parse error.
529 if sys.stdin.isatty():
530 lines = []
531 else:
532 lines = list(sys.stdin)
534 if not lines:
535 # No input at all — zero-file result (clean exit for empty pipelines).
536 # When --keyword was supplied, emit files_matched: 0 so callers relying
537 # on its presence to terminate quickly in zero-match scans see a
538 # consistent shape with the batch-mode no-match case.
539 meta = {"_meta": True, "files_processed": 0, "parse_errors": 0}
540 if keywords:
541 meta["files_matched"] = 0
542 print(json.dumps(meta))
543 else:
544 # Genuine single-file stdin mode (backward compatible)
545 result = extract_from_lines(lines)
546 meta = {"_meta": True, "files_processed": 1, "parse_errors": 0}
547 if result:
548 print(json.dumps(result))
549 else:
550 meta["files_without_metadata"] = 1
551 print(json.dumps(meta))