Skills › Compound Engineering › ce-compound › extract-skeleton.py
extract-skeleton.py
590 lines · skills/ce-compound/scripts/session-history/extract-skeleton.py at 3449230
1#!/usr/bin/env python3
2"""Extract the conversation skeleton from a Claude Code, Codex, Cursor, Pi, or oh-my-pi (omp) JSONL session file.
4Usage:
5 cat <session.jsonl> | python3 extract-skeleton.py
6 cat <session.jsonl> | python3 extract-skeleton.py --output PATH
8Auto-detects platform (Claude Code, Codex, Cursor, Pi, or oh-my-pi (omp)) from the JSONL structure.
9Extracts:
10 - User messages (text only, no tool results)
11 - Assistant text (no thinking/reasoning blocks)
12 - Collapsed tool call summaries (consecutive same-tool calls grouped)
14Consecutive tool calls of the same type are collapsed:
15 3+ Read calls -> "[tools] 3x Read (file1, file2, +1 more) -> all ok"
16Codex call/result pairs are deduplicated (only the result with status is kept).
18When --output PATH is given, the extracted skeleton is written to PATH and
19stdout receives only a one-line JSON status (_meta with wrote/bytes/stats).
20This lets callers route bulk content to a scratch file without round-tripping
21extraction bytes through orchestrator tool results.
23Without --output, extracted content goes to stdout and ends with a _meta line.
24"""
25import argparse
26import io
27import os
28import sys
29import json
30import re
32parser = argparse.ArgumentParser(add_help=True)
33parser.add_argument(
34 "--output",
35 metavar="PATH",
36 help="Write extracted skeleton to PATH instead of stdout. Stdout receives a one-line _meta status.",
37)
38args = parser.parse_args()
40# Session logs are UTF-8; force UTF-8 I/O so Windows' cp1252 default can't crash
41# on non-Latin-1 content like emoji, smart quotes, or box-drawing (#1258).
42sys.stdin.reconfigure(encoding="utf-8", errors="replace")
43sys.stdout.reconfigure(encoding="utf-8", errors="replace")
45# Capture-and-redirect when --output is set: prints in the rest of the script
46# go to the buffer; at the end the buffer is written to PATH and a status
47# line is emitted to the real stdout.
48_original_stdout = sys.stdout
49if args.output:
50 sys.stdout = io.StringIO()
52stats = {"lines": 0, "parse_errors": 0, "user": 0, "assistant": 0, "tool": 0, "summary": 0}
54# Claude Code wrapper tags to strip from user message content.
55# Strip entirely (tag + content): framework noise and raw command output.
56# Strip tags only (keep content): command-message, command-name, command-args, user_query.
57_STRIP_BLOCK = re.compile(
58 r"<(?:task-notification|local-command-caveat|local-command-stdout|local-command-stderr|system-reminder)[^>]*>.*?</(?:task-notification|local-command-caveat|local-command-stdout|local-command-stderr|system-reminder)>",
59 re.DOTALL,
60)
61_STRIP_TAG = re.compile(
62 r"</?(?:command-message|command-name|command-args|user_query)[^>]*>"
63)
66def clean_text(text):
67 """Strip framework wrapper tags from message text (Claude and Cursor)."""
68 text = _STRIP_BLOCK.sub("", text)
69 text = _STRIP_TAG.sub("", text)
70 text = re.sub(r"\n{3,}", "\n\n", text).strip()
71 return text
73# Buffer for pending tool entries: [{"ts", "name", "target", "status"}]
74pending_tools = []
77def flush_tools():
78 """Print buffered tool entries, collapsing consecutive same-name groups."""
79 if not pending_tools:
80 return
82 # Group consecutive entries by tool name
83 groups = []
84 for entry in pending_tools:
85 if groups and groups[-1][0]["name"] == entry["name"]:
86 groups[-1].append(entry)
87 else:
88 groups.append([entry])
90 for group in groups:
91 name = group[0]["name"]
92 if len(group) <= 2:
93 # Print individually
94 for e in group:
95 status = f" -> {e['status']}" if e.get("status") else ""
96 ts_prefix = f"[{e['ts']}] " if e.get("ts") else ""
97 print(f"{ts_prefix}[tool] {name} {e['target']}{status}")
98 stats["tool"] += 1
99 else:
100 # Collapse
101 ts = group[0].get("ts", "")
102 targets = [e["target"] for e in group if e.get("target")]
103 ok = sum(1 for e in group if e.get("status") == "ok")
104 err = sum(1 for e in group if e.get("status") and e["status"] != "ok")
105 no_status = len(group) - ok - err
107 # Show first 2 targets, then "+N more"
108 if len(targets) > 2:
109 target_str = ", ".join(targets[:2]) + f", +{len(targets) - 2} more"
110 elif targets:
111 target_str = ", ".join(targets)
112 else:
113 target_str = ""
115 if no_status == len(group):
116 status_str = ""
117 elif err == 0:
118 status_str = " -> all ok"
119 else:
120 status_str = f" -> {ok} ok, {err} error"
122 ts_prefix = f"[{ts}] " if ts else ""
123 print(f"{ts_prefix}[tools] {len(group)}x {name} ({target_str}){status_str}")
124 stats["tool"] += len(group)
126 pending_tools.clear()
129def _safe_slice(value, n):
130 """Slice value if it is a string; otherwise return ''.
132 Some Claude Code / MCP tool inputs put structured data (dicts, lists) in
133 fields like `query` or `prompt`. `dict[:N]` raises TypeError, so guard
134 every slice with an isinstance check.
135 """
136 return value[:n] if isinstance(value, str) else ""
139def summarize_claude_tool(block):
140 """Extract name and target from a Claude Code tool_use block."""
141 name = block.get("name", "unknown")
142 inp = block.get("input", {})
143 fp = inp.get("file_path")
144 p = inp.get("path")
145 target = (
146 (fp if isinstance(fp, str) else None)
147 or (p if isinstance(p, str) else None)
148 or _safe_slice(inp.get("command"), 120)
149 or _safe_slice(inp.get("pattern"), 200)
150 or _safe_slice(inp.get("query"), 80)
151 or _safe_slice(inp.get("prompt"), 80)
152 or ""
153 )
154 if isinstance(target, str) and len(target) > 120:
155 target = target[:120]
156 return name, target
159def handle_claude(obj):
160 msg_type = obj.get("type")
161 ts = obj.get("timestamp", "")[:19]
163 if msg_type == "user":
164 msg = obj.get("message", {})
165 content = msg.get("content", "")
167 if isinstance(content, list):
168 for block in content:
169 if block.get("type") == "tool_result":
170 is_error = block.get("is_error", False)
171 status = "error" if is_error else "ok"
172 tool_use_id = block.get("tool_use_id")
173 matched = False
174 if tool_use_id:
175 for entry in pending_tools:
176 if entry.get("id") == tool_use_id:
177 entry["status"] = status
178 matched = True
179 break
180 if not matched:
181 # Fallback: assign to earliest pending entry without a status
182 for entry in pending_tools:
183 if not entry.get("status"):
184 entry["status"] = status
185 break
187 texts = [
188 c.get("text", "")
189 for c in content
190 if c.get("type") == "text" and len(c.get("text", "")) > 10
191 ]
192 content = " ".join(texts)
194 if isinstance(content, str):
195 content = clean_text(content)
196 if len(content) > 15:
197 flush_tools()
198 print(f"[{ts}] [user] {content[:800]}")
199 print("---")
200 stats["user"] += 1
202 elif msg_type == "assistant":
203 msg = obj.get("message", {})
204 content = msg.get("content", [])
205 if isinstance(content, list):
206 has_text = False
207 for block in content:
208 if block.get("type") == "text":
209 text = clean_text(block.get("text", ""))
210 if len(text) > 20:
211 if not has_text:
212 flush_tools()
213 has_text = True
214 print(f"[{ts}] [assistant] {text[:800]}")
215 print("---")
216 stats["assistant"] += 1
217 elif block.get("type") == "tool_use":
218 name, target = summarize_claude_tool(block)
219 entry = {"ts": ts, "name": name, "target": target}
220 tool_id = block.get("id")
221 if tool_id:
222 entry["id"] = tool_id
223 pending_tools.append(entry)
226def handle_codex(obj):
227 msg_type = obj.get("type")
228 ts = obj.get("timestamp", "")[:19]
230 if msg_type == "event_msg":
231 p = obj.get("payload", {})
232 if p.get("type") == "user_message":
233 text = p.get("message", "")
234 if isinstance(text, str) and len(text) > 15:
235 parts = text.split("</system_instruction>")
236 user_text = parts[-1].strip() if parts else text
237 if len(user_text) > 15:
238 flush_tools()
239 print(f"[{ts}] [user] {user_text[:800]}")
240 print("---")
241 stats["user"] += 1
243 elif p.get("type") == "exec_command_end":
244 # This is the deduplicated result — has status info
245 command = p.get("command", [])
246 cmd_str = command[-1] if command else ""
247 output = p.get("aggregated_output", "")
249 status = "ok"
250 if "Process exited with code " in output:
251 try:
252 code = int(output.split("Process exited with code ")[1].split("\n")[0])
253 if code != 0:
254 status = f"error(exit {code})"
255 except (IndexError, ValueError):
256 pass
258 if cmd_str:
259 # Shorten common patterns for readability
260 short_cmd = cmd_str[:120]
261 pending_tools.append({"ts": ts, "name": "exec", "target": short_cmd, "status": status})
263 elif msg_type == "response_item":
264 p = obj.get("payload", {})
265 if p.get("type") == "message" and p.get("role") == "assistant":
266 for block in p.get("content", []):
267 if block.get("type") == "output_text" and len(block.get("text", "")) > 20:
268 flush_tools()
269 print(f"[{ts}] [assistant] {block['text'][:800]}")
270 print("---")
271 stats["assistant"] += 1
273 # Skip function_call — exec_command_end is the deduplicated version with status
276def _pi_text_content(content):
277 if isinstance(content, str):
278 return [content]
279 if not isinstance(content, list):
280 return []
281 return [
282 block.get("text", "")
283 for block in content
284 if isinstance(block, dict) and block.get("type") == "text"
285 ]
288def _pi_active_path_objects(objects):
289 """Return only entries on Pi's active leaf-to-root path."""
290 by_id = {
291 obj.get("id"): obj
292 for obj in objects
293 if isinstance(obj.get("id"), str) and obj.get("type") != "session"
294 }
295 leaf_id = None
296 for obj in objects:
297 if obj.get("type") != "session" and isinstance(obj.get("id"), str):
298 leaf_id = obj["id"]
299 if not leaf_id:
300 return objects
302 active_ids = set()
303 current = leaf_id
304 while isinstance(current, str) and current and current not in active_ids:
305 active_ids.add(current)
306 parent = by_id.get(current, {}).get("parentId")
307 current = parent if isinstance(parent, str) else None
308 return [
309 obj
310 for obj in objects
311 if obj.get("type") == "session" or obj.get("id") in active_ids
312 ]
315def _pi_context_objects(objects):
316 """Return Pi entries that participate in active LLM context."""
317 active = _pi_active_path_objects(objects)
318 compactions = [obj for obj in active if obj.get("type") == "compaction"]
319 if not compactions:
320 return active
322 first_kept = compactions[-1].get("firstKeptEntryId")
323 if not isinstance(first_kept, str):
324 return active
326 latest_compaction_id = compactions[-1].get("id")
327 started = False
328 found_first_kept = False
329 context = [obj for obj in active if obj.get("type") == "session"]
330 context.append(compactions[-1])
331 for obj in active:
332 if obj.get("type") == "session":
333 continue
334 if obj.get("id") == first_kept:
335 started = True
336 found_first_kept = True
337 if obj.get("id") == latest_compaction_id:
338 continue
339 if started:
340 context.append(obj)
341 return context if found_first_kept and len(context) > 1 else active
344def handle_pi(obj):
345 """Pi/omp sessions: type='message' with message.role and content blocks.
347 omp's physical type:'title' slot line lands here too and returns below
348 like any non-message entry. type:'title_change' entries are a different
349 pi entry type — they also carry no message payload and return the same
350 way; do not conflate the two when filtering."""
351 entry_type = obj.get("type")
352 ts = obj.get("timestamp", "")[:19]
354 if entry_type in ("compaction", "branch_summary"):
355 text = clean_text(obj.get("summary", ""))
356 if len(text) > 15:
357 flush_tools()
358 print(f"[{ts}] [summary] {text[:800]}")
359 print("---")
360 stats["summary"] += 1
361 return
363 if entry_type == "custom_message":
364 text = clean_text(" ".join(_pi_text_content(obj.get("content", []))))
365 if len(text) > 15:
366 flush_tools()
367 print(f"[{ts}] [summary] {text[:800]}")
368 print("---")
369 stats["summary"] += 1
370 return
372 if entry_type != "message":
373 return
375 msg = obj.get("message", {})
376 role = msg.get("role", "")
377 content = msg.get("content", [])
379 if role == "bashExecution":
380 exit_code = msg.get("exitCode")
381 if msg.get("cancelled"):
382 status = "cancelled"
383 elif exit_code in (None, 0):
384 status = "ok"
385 else:
386 status = f"error(exit {exit_code})"
387 command = _safe_slice(msg.get("command"), 120)
388 pending_tools.append({"ts": ts, "name": "bash", "target": command, "status": status})
389 return
391 if role == "custom":
392 text = clean_text(" ".join(_pi_text_content(content)))
393 if len(text) > 15:
394 flush_tools()
395 print(f"[{ts}] [summary] {text[:800]}")
396 print("---")
397 stats["summary"] += 1
398 return
400 if role == "user":
401 text = clean_text(" ".join(_pi_text_content(content)))
402 if len(text) > 15:
403 flush_tools()
404 print(f"[{ts}] [user] {text[:800]}")
405 print("---")
406 stats["user"] += 1
408 elif role == "assistant":
409 if isinstance(content, str):
410 text = clean_text(content)
411 if len(text) > 20:
412 flush_tools()
413 print(f"[{ts}] [assistant] {text[:800]}")
414 print("---")
415 stats["assistant"] += 1
416 return
418 has_text = False
419 for block in (content if isinstance(content, list) else []):
420 if not isinstance(block, dict):
421 continue
422 if block.get("type") == "text":
423 text = clean_text(block.get("text", ""))
424 if len(text) > 20:
425 if not has_text:
426 flush_tools()
427 has_text = True
428 print(f"[{ts}] [assistant] {text[:800]}")
429 print("---")
430 stats["assistant"] += 1
431 elif block.get("type") == "toolCall":
432 name = block.get("name", "unknown")
433 args = block.get("arguments", {})
434 if not isinstance(args, dict):
435 args = {}
436 target = (
437 _safe_slice(args.get("path"), 200)
438 or _safe_slice(args.get("file_path"), 200)
439 or _safe_slice(args.get("command"), 120)
440 or _safe_slice(args.get("pattern"), 200)
441 or _safe_slice(args.get("query"), 80)
442 or _safe_slice(args.get("prompt"), 80)
443 or ""
444 )
445 if isinstance(target, str) and len(target) > 120:
446 target = target[:120]
447 entry = {"ts": ts, "name": name, "target": target}
448 tool_id = block.get("id")
449 if tool_id:
450 entry["id"] = tool_id
451 pending_tools.append(entry)
453 elif role == "toolResult":
454 tool_call_id = msg.get("toolCallId")
455 is_error = bool(msg.get("isError"))
456 if isinstance(content, list):
457 is_error = is_error or any(
458 isinstance(block, dict) and block.get("type") == "toolError"
459 for block in content
460 )
461 status = "error" if is_error else "ok"
462 if tool_call_id:
463 for entry in pending_tools:
464 if entry.get("id") == tool_call_id:
465 entry["status"] = status
466 break
467 else:
468 for entry in pending_tools:
469 if not entry.get("status"):
470 entry["status"] = status
471 break
474def handle_cursor(obj):
475 """Cursor agent transcripts: role-based, no timestamps, same content structure as Claude."""
476 role = obj.get("role")
477 content = obj.get("message", {}).get("content", [])
479 if role == "user":
480 texts = []
481 for block in (content if isinstance(content, list) else []):
482 if block.get("type") == "text":
483 texts.append(block.get("text", ""))
484 text = clean_text(" ".join(texts))
485 if len(text) > 15:
486 flush_tools()
487 # No timestamps available in Cursor transcripts
488 print(f"[user] {text[:800]}")
489 print("---")
490 stats["user"] += 1
492 elif role == "assistant":
493 has_text = False
494 for block in (content if isinstance(content, list) else []):
495 if block.get("type") == "text":
496 text = block.get("text", "")
497 # Skip [REDACTED] placeholder blocks
498 if len(text) > 20 and text.strip() != "[REDACTED]":
499 if not has_text:
500 flush_tools()
501 has_text = True
502 print(f"[assistant] {text[:800]}")
503 print("---")
504 stats["assistant"] += 1
505 elif block.get("type") == "tool_use":
506 name = block.get("name", "unknown")
507 inp = block.get("input", {})
508 p = inp.get("path")
509 fp = inp.get("file_path")
510 target = (
511 (p if isinstance(p, str) else None)
512 or (fp if isinstance(fp, str) else None)
513 or _safe_slice(inp.get("command"), 120)
514 or _safe_slice(inp.get("pattern"), 200)
515 or _safe_slice(inp.get("glob_pattern"), 200)
516 or _safe_slice(inp.get("target_directory"), 200)
517 or ""
518 )
519 if isinstance(target, str) and len(target) > 120:
520 target = target[:120]
521 # No status info available — Cursor doesn't log tool results
522 pending_tools.append({"ts": "", "name": name, "target": target})
525# Auto-detect platform from first few lines, then process all
526detected = None
527buffer = []
528# omp files physically begin with a fixed-width type:'title' slot line before
529# the pi-shaped type:'session' header; bare pi files start with the header.
530seen_title_slot = False
532for line in sys.stdin:
533 line = line.strip()
534 if not line:
535 continue
536 buffer.append(line)
537 stats["lines"] += 1
539 # Scan until a decisive record appears (short-circuits once detected).
540 # Claude Code sessions front-load 10+ non-message metadata records
541 # (last-prompt, custom-title, attachments, ...), so capping detection to
542 # the first N lines silently misroutes them to the codex handler and
543 # produces an empty skeleton with parse_errors: 0 (issue #923).
544 if not detected:
545 try:
546 obj = json.loads(line)
547 if obj.get("type") == "session" and "cwd" in obj:
548 detected = "omp" if seen_title_slot else "pi"
549 elif obj.get("type") == "title" and len(buffer) == 1:
550 seen_title_slot = True
551 elif obj.get("type") in ("user", "assistant"):
552 detected = "claude"
553 elif obj.get("type") in ("session_meta", "turn_context", "response_item", "event_msg"):
554 detected = "codex"
555 elif obj.get("role") in ("user", "assistant") and "type" not in obj:
556 detected = "cursor"
557 except (json.JSONDecodeError, KeyError):
558 pass
560handlers = {"claude": handle_claude, "codex": handle_codex, "cursor": handle_cursor, "pi": handle_pi, "omp": handle_pi}
561handler = handlers.get(detected, handle_codex)
563objects = []
564for line in buffer:
565 try:
566 objects.append(json.loads(line))
567 except (json.JSONDecodeError, KeyError):
568 stats["parse_errors"] += 1
570if detected in ("pi", "omp"):
571 objects = _pi_context_objects(objects)
573for obj in objects:
574 try:
575 handler(obj)
576 except KeyError:
577 stats["parse_errors"] += 1
579# Flush any remaining buffered tools
580flush_tools()
582print(json.dumps({"_meta": True, **stats}))
584if args.output:
585 body = sys.stdout.getvalue()
586 sys.stdout = _original_stdout
587 with open(args.output, "w", encoding="utf-8") as f:
588 f.write(body)
589 bytes_written = os.path.getsize(args.output)
590 print(json.dumps({"_meta": True, "wrote": args.output, "bytes": bytes_written, **stats}))