#!/usr/bin/env python3
"""Deterministic aggregator for a single Claude Code session.
Reads the main transcript JSONL for one session plus its subagent
sidechain logs, and emits a compact JSON summary on stdout that the
post-mortem SKILL interprets into a written report.
Why a script and not skill prose: transcripts run to >1 MB and a
session can spawn dozens of subagent logs. Token accounting has two
traps that only deterministic code gets right every time:
1. Streamed assistant lines repeat the SAME requestId with the SAME
usage object. Counting per-line double-counts tokens. We dedup
per requestId (max per field within a request, then sum across
requests).
2. The transcript carries NO costUSD and NO durationMs (both null).
Cost is derived from tokens x a per-model price table; wall-clock
from timestamp deltas. The price table is a LIST-PRICE ESTIMATE
and is surfaced in the output so the report never presents a
derived dollar figure as if it were metered.
Stdlib only. No network. Read-only against ~/.claude.
"""
import argparse
import glob
import json
import os
import re
import sys
from datetime import datetime, timezone
# --- Pricing -------------------------------------------------------------
# USD per million tokens. Anthropic public list prices; treated as an
# ESTIMATE, not a metered cost. Override wholesale via the
# POSTMORTEM_PRICING env var (JSON of the same shape) or --pricing.
# Keys are matched as substrings against message.model, longest first.
DEFAULT_PRICING = {
"opus": {"input": 15.0, "output": 75.0,
"cache_write_5m": 18.75, "cache_write_1h": 30.0, "cache_read": 1.5},
"sonnet": {"input": 3.0, "output": 15.0,
"cache_write_5m": 3.75, "cache_write_1h": 6.0, "cache_read": 0.30},
"haiku": {"input": 1.0, "output": 5.0,
"cache_write_5m": 1.25, "cache_write_1h": 2.0, "cache_read": 0.10},
}
PRICING_AS_OF = "2026-01 list prices (estimate, override via POSTMORTEM_PRICING)"
STATUS_TOKENS = ["DONE_WITH_CONCERNS", "NEEDS_CONTEXT", "BLOCKED", "PARTIAL", "DONE"]
def load_pricing(override_json):
if override_json:
return json.loads(override_json)
env = os.environ.get("POSTMORTEM_PRICING")
if env:
return json.loads(env)
return DEFAULT_PRICING
def rate_for(model, pricing):
if not model:
return pricing.get("opus"), True
for key in sorted(pricing, key=len, reverse=True):
if key in model:
return pricing[key], False
# Unknown model: fall back to the most expensive known tier so the
# estimate errs high rather than silently undercounting.
return pricing.get("opus", next(iter(pricing.values()))), True
def parse_ts(s):
if not s:
return None
try:
return datetime.fromisoformat(s.replace("Z", "+00:00"))
except ValueError:
return None
def empty_tokens():
return {"input": 0, "output": 0, "cache_creation": 0, "cache_read": 0,
"cache_creation_5m": 0, "cache_creation_1h": 0}
def add_tokens(acc, usage):
"""Add one request's deduped usage into an accumulator."""
acc["input"] += usage.get("input_tokens", 0) or 0
acc["output"] += usage.get("output_tokens", 0) or 0
cc = usage.get("cache_creation_input_tokens", 0) or 0
acc["cache_creation"] += cc
acc["cache_read"] += usage.get("cache_read_input_tokens", 0) or 0
detail = usage.get("cache_creation") or {}
f5 = detail.get("ephemeral_5m_input_tokens")
f1 = detail.get("ephemeral_1h_input_tokens")
if f5 is None and f1 is None:
acc["cache_creation_5m"] += cc # no split available -> assume 5m
else:
acc["cache_creation_5m"] += f5 or 0
acc["cache_creation_1h"] += f1 or 0
def cost_of(tokens, rate):
return (tokens["input"] * rate["input"]
+ tokens["output"] * rate["output"]
+ tokens["cache_creation_5m"] * rate["cache_write_5m"]
+ tokens["cache_creation_1h"] * rate["cache_write_1h"]
+ tokens["cache_read"] * rate["cache_read"]) / 1_000_000.0
def iter_lines(path):
with open(path, "r", encoding="utf-8", errors="replace") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
yield json.loads(line)
except json.JSONDecodeError:
continue
def aggregate_transcript(path, pricing):
"""Return token totals, cost, model mix, and per-request count for one
JSONL transcript (main or subagent). Dedups usage by requestId."""
# requestId -> {field: max seen}, plus model per request.
req_usage = {}
req_model = {}
for rec in iter_lines(path):
if rec.get("type") != "assistant":
continue
msg = rec.get("message") or {}
usage = msg.get("usage")
if not usage:
continue
key = rec.get("requestId") or rec.get("uuid")
model = msg.get("model")
if model:
req_model[key] = model
cur = req_usage.setdefault(key, {})
# Max-per-field guards against both identical-repeat and
# cumulative-within-request streaming shapes.
for f in ("input_tokens", "output_tokens",
"cache_creation_input_tokens", "cache_read_input_tokens"):
v = usage.get(f, 0) or 0
if v > cur.get(f, 0):
cur[f] = v
det = usage.get("cache_creation") or {}
cd = cur.setdefault("cache_creation", {})
for f in ("ephemeral_5m_input_tokens", "ephemeral_1h_input_tokens"):
v = det.get(f, 0) or 0
if v > cd.get(f, 0):
cd[f] = v
st = usage.get("server_tool_use") or {}
sc = cur.setdefault("_server", {})
for f in ("web_search_requests", "web_fetch_requests"):
v = st.get(f, 0) or 0
if v > sc.get(f, 0):
sc[f] = v
tokens = empty_tokens()
server = {"web_search": 0, "web_fetch": 0}
models = {}
cost = 0.0
unknown_model = False
for key, usage in req_usage.items():
add_tokens(tokens, usage)
model = req_model.get(key)
models[model] = models.get(model, 0) + 1
rate, unk = rate_for(model, pricing)
unknown_model = unknown_model or unk
# per-request cost so each request bills at its own model's rate
per = empty_tokens()
add_tokens(per, usage)
cost += cost_of(per, rate)
sv = usage.get("_server", {})
server["web_search"] += sv.get("web_search_requests", 0)
server["web_fetch"] += sv.get("web_fetch_requests", 0)
return {
"requests": len(req_usage),
"tokens": tokens,
"cost_usd": round(cost, 4),
"models": models,
"server_tools": server,
"unknown_model": unknown_model,
}
def cache_hit_ratio(tokens):
denom = tokens["input"] + tokens["cache_read"] + tokens["cache_creation"]
return round(tokens["cache_read"] / denom, 4) if denom else None
def analyze_main(path):
"""Tool mix, error rate, skills, slash commands, prompts, wall-clock."""
tool_calls = {}
tool_errors = {}
error_total = 0
interrupted = 0
skills = []
slash = []
tasks = []
user_prompts = 0
first_ts = last_ts = None
cwd = project = version = None
for rec in iter_lines(path):
ts = parse_ts(rec.get("timestamp"))
if ts:
if first_ts is None or ts < first_ts:
first_ts = ts
if last_ts is None or ts > last_ts:
last_ts = ts
cwd = rec.get("cwd") or cwd
version = rec.get("version") or version
typ = rec.get("type")
msg = rec.get("message") or {}
content = msg.get("content")
if typ == "assistant" and isinstance(content, list):
for block in content:
if block.get("type") != "tool_use":
continue
name = block.get("name", "?")
tool_calls[name] = tool_calls.get(name, 0) + 1
if name == "Skill":
skills.append({"skill": (block.get("input") or {}).get("skill"),
"ts": rec.get("timestamp")})
elif name in ("Task", "Agent"):
inp = block.get("input") or {}
tasks.append({"subagent_type": inp.get("subagent_type"),
"description": inp.get("description"),
"ts": rec.get("timestamp")})
if typ == "user":
# tool_result blocks carry the authoritative is_error flag
if isinstance(content, list):
for block in content:
if block.get("type") == "tool_result":
if block.get("is_error"):
error_total += 1
elif block.get("type") == "text":
for m in re.findall(r"(.*?)",
block.get("text", "")):
slash.append({"cmd": m, "ts": rec.get("timestamp")})
elif isinstance(content, str):
for m in re.findall(r"(.*?)", content):
slash.append({"cmd": m, "ts": rec.get("timestamp")})
# genuine user turns (not tool results, not slash plumbing)
is_tool_result = (isinstance(content, list)
and any(b.get("type") == "tool_result" for b in content))
if not is_tool_result and rec.get("toolUseResult") is None:
user_prompts += 1
tur = rec.get("toolUseResult")
if isinstance(tur, dict) and tur.get("interrupted"):
interrupted += 1
# error attribution by tool name needs pairing result->call; we report
# the session-wide error_total and leave per-tool to the weak signal.
return {
"tool_calls": tool_calls,
"total_tool_calls": sum(tool_calls.values()),
"tool_error_total": error_total,
"interrupted": interrupted,
"skills_invoked": skills,
"slash_commands": slash,
"tasks_dispatched": tasks,
"user_prompts": user_prompts,
"wall_clock_sec": round((last_ts - first_ts).total_seconds(), 1)
if first_ts and last_ts else None,
"started": first_ts.isoformat() if first_ts else None,
"ended": last_ts.isoformat() if last_ts else None,
"cwd": cwd,
"version": version,
}
def subagent_terminal_status(path):
last_text = ""
for rec in iter_lines(path):
if rec.get("type") != "assistant":
continue
content = (rec.get("message") or {}).get("content")
if isinstance(content, list):
for block in content:
if block.get("type") == "text" and block.get("text", "").strip():
last_text = block["text"]
for tok in STATUS_TOKENS:
if re.search(r"\b" + tok + r"\b", last_text):
return tok
return "unknown"
def analyze_subagents(session_dir, pricing):
out = []
sub_dir = os.path.join(session_dir, "subagents")
for jl in sorted(glob.glob(os.path.join(sub_dir, "agent-*.jsonl"))):
meta = {}
mp = jl.replace(".jsonl", ".meta.json")
if os.path.exists(mp):
try:
with open(mp) as fh:
meta = json.load(fh)
except (json.JSONDecodeError, OSError):
pass
agg = aggregate_transcript(jl, pricing)
out.append({
"agent_id": os.path.basename(jl)[len("agent-"):-len(".jsonl")],
"agent_type": meta.get("agentType"),
"description": meta.get("description"),
"dispatched_by_tool": meta.get("toolUseId"),
"models": agg["models"],
"requests": agg["requests"],
"tokens": agg["tokens"],
"cost_usd": agg["cost_usd"],
"terminal_status": subagent_terminal_status(jl),
})
return out
def resolve_session(args):
"""Return (main_jsonl_path, session_dir_or_None)."""
if args.file:
path = os.path.abspath(args.file)
return path, path[:-len(".jsonl")] if path.endswith(".jsonl") else None
cwd = args.cwd or os.getcwd()
proj_dir = args.project_dir or os.path.join(
os.path.expanduser("~/.claude/projects"), cwd.replace("/", "-"))
if not os.path.isdir(proj_dir):
raise SystemExit(f"no project log dir for cwd {cwd}: {proj_dir} not found")
if args.session:
path = os.path.join(proj_dir, args.session + ".jsonl")
if not os.path.exists(path):
raise SystemExit(f"session {args.session} not found in {proj_dir}")
return path, os.path.join(proj_dir, args.session)
# default: newest .jsonl in the project dir = current/last session
candidates = sorted(glob.glob(os.path.join(proj_dir, "*.jsonl")),
key=os.path.getmtime, reverse=True)
if not candidates:
raise SystemExit(f"no session transcripts in {proj_dir}")
path = candidates[0]
return path, path[:-len(".jsonl")]
def main():
ap = argparse.ArgumentParser(description="Single-session post-mortem aggregator.")
ap.add_argument("--session", help="explicit session id (UUID)")
ap.add_argument("--file", help="explicit path to a main transcript .jsonl")
ap.add_argument("--cwd", help="project cwd to resolve logs for (default: $PWD)")
ap.add_argument("--project-dir", help="explicit ~/.claude/projects/ dir")
ap.add_argument("--pricing", help="JSON price table override")
args = ap.parse_args()
pricing = load_pricing(args.pricing)
main_path, session_dir = resolve_session(args)
session_id = os.path.basename(main_path)[:-len(".jsonl")]
main_agg = aggregate_transcript(main_path, pricing)
main_meta = analyze_main(main_path)
subagents = []
if session_dir and os.path.isdir(session_dir):
subagents = analyze_subagents(session_dir, pricing)
sub_tokens = empty_tokens()
sub_cost = 0.0
for s in subagents:
for k in sub_tokens:
sub_tokens[k] += s["tokens"][k]
sub_cost += s["cost_usd"]
total_tokens = empty_tokens()
for k in total_tokens:
total_tokens[k] = main_agg["tokens"][k] + sub_tokens[k]
total_cost = round(main_agg["cost_usd"] + sub_cost, 4)
warnings = []
is_active = False
if main_meta["ended"]:
ended = parse_ts(main_meta["ended"])
if ended and (datetime.now(timezone.utc) - ended).total_seconds() < 120:
is_active = True
warnings.append("Session appears ACTIVE (last event <2 min ago); "
"totals are partial and will grow.")
if main_agg["unknown_model"] or any(s.get("models") and
any(rate_for(m, pricing)[1] for m in s["models"])
for s in subagents):
warnings.append("One or more models were not in the price table; "
"billed at the opus tier (estimate errs high).")
warnings.append("Cost is DERIVED from tokens x list prices, not metered. "
"Price basis: " + PRICING_AS_OF)
report = {
"session": {
"id": session_id,
"transcript": main_path,
"cwd": main_meta["cwd"],
"version": main_meta["version"],
"started": main_meta["started"],
"ended": main_meta["ended"],
"wall_clock_sec": main_meta["wall_clock_sec"],
"active": is_active,
},
"main": {
"requests": main_agg["requests"],
"models": main_agg["models"],
"tokens": main_agg["tokens"],
"cost_usd": main_agg["cost_usd"],
"cache_hit_ratio": cache_hit_ratio(main_agg["tokens"]),
"server_tools": main_agg["server_tools"],
"user_prompts": main_meta["user_prompts"],
},
"tools": {
"calls": main_meta["tool_calls"],
"total": main_meta["total_tool_calls"],
"error_total": main_meta["tool_error_total"],
"error_ratio": round(main_meta["tool_error_total"]
/ main_meta["total_tool_calls"], 4)
if main_meta["total_tool_calls"] else None,
"interrupted": main_meta["interrupted"],
},
"skills_invoked": main_meta["skills_invoked"],
"slash_commands": main_meta["slash_commands"],
"tasks_dispatched": main_meta["tasks_dispatched"],
"subagents": subagents,
"subagent_totals": {
"count": len(subagents),
"tokens": sub_tokens,
"cost_usd": round(sub_cost, 4),
},
"totals": {
"tokens": total_tokens,
"cache_hit_ratio": cache_hit_ratio(total_tokens),
"cost_usd": total_cost,
},
"pricing_used": pricing,
"pricing_basis": PRICING_AS_OF,
"warnings": warnings,
}
json.dump(report, sys.stdout, indent=2)
sys.stdout.write("\n")
if __name__ == "__main__":
main()