wehub-resource-sync
255 lines
8.2 KiB
Python
255 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Extract error signals from a Claude Code, Codex, Cursor, or Pi JSONL session file.
|
|
|
|
Usage:
|
|
cat <session.jsonl> | python3 extract-errors.py
|
|
cat <session.jsonl> | python3 extract-errors.py --output PATH
|
|
|
|
Auto-detects platform from the JSONL structure.
|
|
Note: Cursor agent transcripts do not log tool results, so no errors can be extracted.
|
|
Finds failed tool calls / commands and outputs them with timestamps.
|
|
|
|
When --output PATH is given, the extracted error log is written to PATH and
|
|
stdout receives only a one-line JSON status (_meta with wrote/bytes/stats).
|
|
This lets callers route bulk content to a scratch file without round-tripping
|
|
extraction bytes through orchestrator tool results.
|
|
|
|
Without --output, extracted content goes to stdout and ends with a _meta line.
|
|
"""
|
|
import argparse
|
|
import io
|
|
import os
|
|
import sys
|
|
import json
|
|
|
|
parser = argparse.ArgumentParser(add_help=True)
|
|
parser.add_argument(
|
|
"--output",
|
|
metavar="PATH",
|
|
help="Write extracted errors to PATH instead of stdout. Stdout receives a one-line _meta status.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
_original_stdout = sys.stdout
|
|
if args.output:
|
|
sys.stdout = io.StringIO()
|
|
|
|
stats = {"lines": 0, "parse_errors": 0, "errors_found": 0}
|
|
|
|
|
|
def summarize_error(raw):
|
|
"""Extract a short error summary instead of dumping the full payload."""
|
|
text = str(raw).strip()
|
|
# Take the first non-empty line as the error message
|
|
for line in text.split("\n"):
|
|
line = line.strip()
|
|
if line:
|
|
return line[:200]
|
|
return text[:200]
|
|
|
|
|
|
def handle_claude(obj):
|
|
if obj.get("type") == "user":
|
|
content = obj.get("message", {}).get("content", [])
|
|
if isinstance(content, list):
|
|
for block in content:
|
|
if block.get("type") == "tool_result" and block.get("is_error"):
|
|
ts = obj.get("timestamp", "")[:19]
|
|
summary = summarize_error(block.get("content", ""))
|
|
print(f"[{ts}] [error] {summary}")
|
|
print("---")
|
|
stats["errors_found"] += 1
|
|
|
|
|
|
def handle_codex(obj):
|
|
if obj.get("type") == "event_msg":
|
|
p = obj.get("payload", {})
|
|
if p.get("type") == "exec_command_end":
|
|
output = p.get("aggregated_output", "")
|
|
stderr = p.get("stderr", "")
|
|
command = p.get("command", [])
|
|
cmd_str = command[-1] if command else ""
|
|
|
|
exit_match = None
|
|
if "Process exited with code " in output:
|
|
try:
|
|
code_str = output.split("Process exited with code ")[1].split("\n")[0]
|
|
exit_code = int(code_str)
|
|
if exit_code != 0:
|
|
exit_match = exit_code
|
|
except (IndexError, ValueError):
|
|
pass
|
|
|
|
if exit_match is not None or stderr:
|
|
ts = obj.get("timestamp", "")[:19]
|
|
error_summary = summarize_error(stderr if stderr else output)
|
|
print(f"[{ts}] [error] exit={exit_match} cmd={cmd_str[:120]}: {error_summary}")
|
|
print("---")
|
|
stats["errors_found"] += 1
|
|
|
|
|
|
def _pi_content_summary(content):
|
|
if isinstance(content, str):
|
|
return summarize_error(content)
|
|
if isinstance(content, list):
|
|
text = "\n".join(
|
|
block.get("text", "")
|
|
for block in content
|
|
if isinstance(block, dict) and block.get("type") in ("text", "toolError")
|
|
)
|
|
return summarize_error(text)
|
|
return summarize_error(content)
|
|
|
|
|
|
def _pi_active_path_objects(objects):
|
|
"""Return only entries on Pi's active leaf-to-root path."""
|
|
by_id = {
|
|
obj.get("id"): obj
|
|
for obj in objects
|
|
if isinstance(obj.get("id"), str) and obj.get("type") != "session"
|
|
}
|
|
leaf_id = None
|
|
for obj in objects:
|
|
if obj.get("type") != "session" and isinstance(obj.get("id"), str):
|
|
leaf_id = obj["id"]
|
|
if not leaf_id:
|
|
return objects
|
|
|
|
active_ids = set()
|
|
current = leaf_id
|
|
while isinstance(current, str) and current and current not in active_ids:
|
|
active_ids.add(current)
|
|
parent = by_id.get(current, {}).get("parentId")
|
|
current = parent if isinstance(parent, str) else None
|
|
return [
|
|
obj
|
|
for obj in objects
|
|
if obj.get("type") == "session" or obj.get("id") in active_ids
|
|
]
|
|
|
|
|
|
def _pi_context_objects(objects):
|
|
"""Return Pi entries that participate in active LLM context."""
|
|
active = _pi_active_path_objects(objects)
|
|
compactions = [obj for obj in active if obj.get("type") == "compaction"]
|
|
if not compactions:
|
|
return active
|
|
|
|
first_kept = compactions[-1].get("firstKeptEntryId")
|
|
if not isinstance(first_kept, str):
|
|
return active
|
|
|
|
latest_compaction_id = compactions[-1].get("id")
|
|
started = False
|
|
found_first_kept = False
|
|
context = [obj for obj in active if obj.get("type") == "session"]
|
|
context.append(compactions[-1])
|
|
for obj in active:
|
|
if obj.get("type") == "session":
|
|
continue
|
|
if obj.get("id") == first_kept:
|
|
started = True
|
|
found_first_kept = True
|
|
if obj.get("id") == latest_compaction_id:
|
|
continue
|
|
if started:
|
|
context.append(obj)
|
|
return context if found_first_kept and len(context) > 1 else active
|
|
|
|
|
|
def handle_pi(obj):
|
|
if obj.get("type") != "message":
|
|
return
|
|
msg = obj.get("message", {})
|
|
if msg.get("role") == "bashExecution":
|
|
exit_code = msg.get("exitCode")
|
|
if exit_code in (None, 0) and not msg.get("cancelled"):
|
|
return
|
|
ts = obj.get("timestamp", "")[:19]
|
|
command = msg.get("command", "")
|
|
output = msg.get("output", "")
|
|
summary = summarize_error(output)
|
|
status = "cancelled" if msg.get("cancelled") else f"exit={exit_code}"
|
|
print(f"[{ts}] [error] {status} cmd={command[:120]}: {summary}")
|
|
print("---")
|
|
stats["errors_found"] += 1
|
|
return
|
|
|
|
if msg.get("role") != "toolResult":
|
|
return
|
|
content = msg.get("content", [])
|
|
is_error = bool(msg.get("isError"))
|
|
if isinstance(content, list):
|
|
is_error = is_error or any(
|
|
isinstance(block, dict) and block.get("type") == "toolError"
|
|
for block in content
|
|
)
|
|
if not is_error:
|
|
return
|
|
|
|
ts = obj.get("timestamp", "")[:19]
|
|
tool = msg.get("toolName", "unknown")
|
|
summary = _pi_content_summary(content)
|
|
print(f"[{ts}] [error] tool={tool}: {summary}")
|
|
print("---")
|
|
stats["errors_found"] += 1
|
|
|
|
|
|
# Auto-detect platform from first few lines, then process all
|
|
detected = None
|
|
buffer = []
|
|
|
|
for line in sys.stdin:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
buffer.append(line)
|
|
stats["lines"] += 1
|
|
|
|
if not detected and len(buffer) <= 10:
|
|
try:
|
|
obj = json.loads(line)
|
|
if obj.get("type") == "session" and "cwd" in obj:
|
|
detected = "pi"
|
|
elif obj.get("type") in ("user", "assistant"):
|
|
detected = "claude"
|
|
elif obj.get("type") in ("session_meta", "turn_context", "response_item", "event_msg"):
|
|
detected = "codex"
|
|
elif obj.get("role") in ("user", "assistant") and "type" not in obj:
|
|
detected = "cursor"
|
|
except (json.JSONDecodeError, KeyError):
|
|
pass
|
|
|
|
# Cursor transcripts don't log tool results — no errors to extract
|
|
def handle_noop(obj):
|
|
pass
|
|
|
|
handlers = {"claude": handle_claude, "codex": handle_codex, "cursor": handle_noop, "pi": handle_pi}
|
|
handler = handlers.get(detected, handle_noop)
|
|
|
|
objects = []
|
|
for line in buffer:
|
|
try:
|
|
objects.append(json.loads(line))
|
|
except (json.JSONDecodeError, KeyError):
|
|
stats["parse_errors"] += 1
|
|
|
|
if detected == "pi":
|
|
objects = _pi_context_objects(objects)
|
|
|
|
for obj in objects:
|
|
try:
|
|
handler(obj)
|
|
except KeyError:
|
|
stats["parse_errors"] += 1
|
|
|
|
print(json.dumps({"_meta": True, **stats}))
|
|
|
|
if args.output:
|
|
body = sys.stdout.getvalue()
|
|
sys.stdout = _original_stdout
|
|
with open(args.output, "w") as f:
|
|
f.write(body)
|
|
bytes_written = os.path.getsize(args.output)
|
|
print(json.dumps({"_meta": True, "wrote": args.output, "bytes": bytes_written, **stats}))
|