Skills › Compound Engineering › ce-compound › extract-errors.py
extract-errors.py
267 lines · skills/ce-compound/scripts/session-history/extract-errors.py at 3449230
1#!/usr/bin/env python3
2"""Extract error signals from a Claude Code, Codex, Cursor, Pi, or oh-my-pi (omp) JSONL session file.
4Usage:
5 cat <session.jsonl> | python3 extract-errors.py
6 cat <session.jsonl> | python3 extract-errors.py --output PATH
8Auto-detects platform from the JSONL structure.
9Note: Cursor agent transcripts do not log tool results, so no errors can be extracted.
10Finds failed tool calls / commands and outputs them with timestamps.
12When --output PATH is given, the extracted error log is written to PATH and
13stdout receives only a one-line JSON status (_meta with wrote/bytes/stats).
14This lets callers route bulk content to a scratch file without round-tripping
15extraction bytes through orchestrator tool results.
17Without --output, extracted content goes to stdout and ends with a _meta line.
18"""
19import argparse
20import io
21import os
22import sys
23import json
25parser = argparse.ArgumentParser(add_help=True)
26parser.add_argument(
27 "--output",
28 metavar="PATH",
29 help="Write extracted errors to PATH instead of stdout. Stdout receives a one-line _meta status.",
30)
31args = parser.parse_args()
33# Session logs are UTF-8; force UTF-8 I/O so Windows' cp1252 default can't crash
34# on non-Latin-1 content like emoji, smart quotes, or box-drawing (#1258).
35sys.stdin.reconfigure(encoding="utf-8", errors="replace")
36sys.stdout.reconfigure(encoding="utf-8", errors="replace")
38_original_stdout = sys.stdout
39if args.output:
40 sys.stdout = io.StringIO()
42stats = {"lines": 0, "parse_errors": 0, "errors_found": 0}
45def summarize_error(raw):
46 """Extract a short error summary instead of dumping the full payload."""
47 text = str(raw).strip()
48 # Take the first non-empty line as the error message
49 for line in text.split("\n"):
50 line = line.strip()
51 if line:
52 return line[:200]
53 return text[:200]
56def handle_claude(obj):
57 if obj.get("type") == "user":
58 content = obj.get("message", {}).get("content", [])
59 if isinstance(content, list):
60 for block in content:
61 if block.get("type") == "tool_result" and block.get("is_error"):
62 ts = obj.get("timestamp", "")[:19]
63 summary = summarize_error(block.get("content", ""))
64 print(f"[{ts}] [error] {summary}")
65 print("---")
66 stats["errors_found"] += 1
69def handle_codex(obj):
70 if obj.get("type") == "event_msg":
71 p = obj.get("payload", {})
72 if p.get("type") == "exec_command_end":
73 output = p.get("aggregated_output", "")
74 stderr = p.get("stderr", "")
75 command = p.get("command", [])
76 cmd_str = command[-1] if command else ""
78 exit_match = None
79 if "Process exited with code " in output:
80 try:
81 code_str = output.split("Process exited with code ")[1].split("\n")[0]
82 exit_code = int(code_str)
83 if exit_code != 0:
84 exit_match = exit_code
85 except (IndexError, ValueError):
86 pass
88 if exit_match is not None or stderr:
89 ts = obj.get("timestamp", "")[:19]
90 error_summary = summarize_error(stderr if stderr else output)
91 print(f"[{ts}] [error] exit={exit_match} cmd={cmd_str[:120]}: {error_summary}")
92 print("---")
93 stats["errors_found"] += 1
96def _pi_content_summary(content):
97 if isinstance(content, str):
98 return summarize_error(content)
99 if isinstance(content, list):
100 text = "\n".join(
101 block.get("text", "")
102 for block in content
103 if isinstance(block, dict) and block.get("type") in ("text", "toolError")
104 )
105 return summarize_error(text)
106 return summarize_error(content)
109def _pi_active_path_objects(objects):
110 """Return only entries on Pi's active leaf-to-root path."""
111 by_id = {
112 obj.get("id"): obj
113 for obj in objects
114 if isinstance(obj.get("id"), str) and obj.get("type") != "session"
115 }
116 leaf_id = None
117 for obj in objects:
118 if obj.get("type") != "session" and isinstance(obj.get("id"), str):
119 leaf_id = obj["id"]
120 if not leaf_id:
121 return objects
123 active_ids = set()
124 current = leaf_id
125 while isinstance(current, str) and current and current not in active_ids:
126 active_ids.add(current)
127 parent = by_id.get(current, {}).get("parentId")
128 current = parent if isinstance(parent, str) else None
129 return [
130 obj
131 for obj in objects
132 if obj.get("type") == "session" or obj.get("id") in active_ids
133 ]
136def _pi_context_objects(objects):
137 """Return Pi entries that participate in active LLM context."""
138 active = _pi_active_path_objects(objects)
139 compactions = [obj for obj in active if obj.get("type") == "compaction"]
140 if not compactions:
141 return active
143 first_kept = compactions[-1].get("firstKeptEntryId")
144 if not isinstance(first_kept, str):
145 return active
147 latest_compaction_id = compactions[-1].get("id")
148 started = False
149 found_first_kept = False
150 context = [obj for obj in active if obj.get("type") == "session"]
151 context.append(compactions[-1])
152 for obj in active:
153 if obj.get("type") == "session":
154 continue
155 if obj.get("id") == first_kept:
156 started = True
157 found_first_kept = True
158 if obj.get("id") == latest_compaction_id:
159 continue
160 if started:
161 context.append(obj)
162 return context if found_first_kept and len(context) > 1 else active
165def handle_pi(obj):
166 # omp's physical type:'title' slot line returns here like any non-message
167 # entry. type:'title_change' entries are a different pi entry type — they
168 # return the same way; do not conflate the two when filtering.
169 if obj.get("type") != "message":
170 return
171 msg = obj.get("message", {})
172 if msg.get("role") == "bashExecution":
173 exit_code = msg.get("exitCode")
174 if exit_code in (None, 0) and not msg.get("cancelled"):
175 return
176 ts = obj.get("timestamp", "")[:19]
177 command = msg.get("command", "")
178 output = msg.get("output", "")
179 summary = summarize_error(output)
180 status = "cancelled" if msg.get("cancelled") else f"exit={exit_code}"
181 print(f"[{ts}] [error] {status} cmd={command[:120]}: {summary}")
182 print("---")
183 stats["errors_found"] += 1
184 return
186 if msg.get("role") != "toolResult":
187 return
188 content = msg.get("content", [])
189 is_error = bool(msg.get("isError"))
190 if isinstance(content, list):
191 is_error = is_error or any(
192 isinstance(block, dict) and block.get("type") == "toolError"
193 for block in content
194 )
195 if not is_error:
196 return
198 ts = obj.get("timestamp", "")[:19]
199 tool = msg.get("toolName", "unknown")
200 summary = _pi_content_summary(content)
201 print(f"[{ts}] [error] tool={tool}: {summary}")
202 print("---")
203 stats["errors_found"] += 1
206# Auto-detect platform from first few lines, then process all
207detected = None
208buffer = []
209# omp files physically begin with a fixed-width type:'title' slot line before
210# the pi-shaped type:'session' header; bare pi files start with the header.
211seen_title_slot = False
213for line in sys.stdin:
214 line = line.strip()
215 if not line:
216 continue
217 buffer.append(line)
218 stats["lines"] += 1
220 if not detected and len(buffer) <= 10:
221 try:
222 obj = json.loads(line)
223 if obj.get("type") == "session" and "cwd" in obj:
224 detected = "omp" if seen_title_slot else "pi"
225 elif obj.get("type") == "title" and len(buffer) == 1:
226 seen_title_slot = True
227 elif obj.get("type") in ("user", "assistant"):
228 detected = "claude"
229 elif obj.get("type") in ("session_meta", "turn_context", "response_item", "event_msg"):
230 detected = "codex"
231 elif obj.get("role") in ("user", "assistant") and "type" not in obj:
232 detected = "cursor"
233 except (json.JSONDecodeError, KeyError):
234 pass
236# Cursor transcripts don't log tool results — no errors to extract
237def handle_noop(obj):
238 pass
240handlers = {"claude": handle_claude, "codex": handle_codex, "cursor": handle_noop, "pi": handle_pi, "omp": handle_pi}
241handler = handlers.get(detected, handle_noop)
243objects = []
244for line in buffer:
245 try:
246 objects.append(json.loads(line))
247 except (json.JSONDecodeError, KeyError):
248 stats["parse_errors"] += 1
250if detected in ("pi", "omp"):
251 objects = _pi_context_objects(objects)
253for obj in objects:
254 try:
255 handler(obj)
256 except KeyError:
257 stats["parse_errors"] += 1
259print(json.dumps({"_meta": True, **stats}))
261if args.output:
262 body = sys.stdout.getvalue()
263 sys.stdout = _original_stdout
264 with open(args.output, "w", encoding="utf-8") as f:
265 f.write(body)
266 bytes_written = os.path.getsize(args.output)
267 print(json.dumps({"_meta": True, "wrote": args.output, "bytes": bytes_written, **stats}))