Buckets:
| # One row per agent session (one file) across pi / Claude Code / Codex traces. | |
| # python -I sessions.py <scratchpad> > hub/sessions.jsonl | |
| import collections, glob, json, os, re, sys | |
| from datetime import datetime | |
| S = sys.argv[1] | |
| IDLE = 30 * 60 # gaps longer than this don't count as active time | |
| NOISE = re.compile(r"^\s*(<(command-|local-command|system-reminder|environment_context|user_instructions|user_shell_command|turn_aborted|task-notification)|" | |
| r"Caveat:|\[Request interrupted|# AGENTS\.md|This session is being continued)") | |
| def ts(s): | |
| try: | |
| return datetime.fromisoformat(str(s).replace("Z", "+00:00")).timestamp() | |
| except ValueError: | |
| return None | |
| def times_stats(t): | |
| t = sorted(x for x in t if x) | |
| if len(t) < 2: | |
| return 0, 0 | |
| gaps = [b - a for a, b in zip(t, t[1:])] | |
| return t[-1] - t[0], sum(g for g in gaps if g <= IDLE) | |
| def text_of(c): | |
| if isinstance(c, str): | |
| return c | |
| if isinstance(c, list): | |
| return "\n".join(x.get("text", "") for x in c if isinstance(x, dict) and x.get("type") in ("text", "input_text")) | |
| return "" | |
| def base(dataset, harness, f): | |
| return {"dataset": dataset, "harness": harness, "file": os.path.basename(f), "models": collections.Counter(), | |
| "providers": collections.Counter(), "version": None, "user_prompts": 0, "assistant_msgs": 0, "tool_calls": 0, | |
| "tools": collections.Counter(), "tool_errors": 0, "tok_in": 0, "tok_out": 0, "tok_cache": 0, "cost": 0.0, | |
| "subagent_msgs": 0, "compactions": 0, "aborts": 0, "cwd": None, "_t": []} | |
| def pi(f, dataset): | |
| r = base(dataset, "pi", f) | |
| for line in open(f, errors="replace"): | |
| try: | |
| d = json.loads(line) | |
| except ValueError: | |
| continue | |
| r["_t"].append(ts(d.get("timestamp"))) | |
| t = d.get("type") | |
| if t == "session": | |
| r["cwd"], r["version"] = d.get("cwd"), d.get("version") | |
| elif t == "compaction": | |
| r["compactions"] += 1 | |
| elif t == "message": | |
| m = d["message"] | |
| role = m.get("role") | |
| if role == "user": | |
| if text_of(m.get("content")).strip(): | |
| r["user_prompts"] += 1 | |
| elif role == "assistant": | |
| r["assistant_msgs"] += 1 | |
| if m.get("model"): | |
| r["models"][m["model"]] += 1 | |
| if m.get("provider"): | |
| r["providers"][m["provider"]] += 1 | |
| u = m.get("usage") or {} | |
| r["tok_in"] += u.get("input", 0) or 0 | |
| r["tok_out"] += u.get("output", 0) or 0 | |
| r["tok_cache"] += u.get("cacheRead", 0) or 0 | |
| r["cost"] += ((u.get("cost") or {}).get("total", 0) or 0) if isinstance(u.get("cost"), dict) else 0 | |
| r["aborts"] += m.get("stopReason") == "aborted" | |
| for c in m.get("content") or []: | |
| if isinstance(c, dict) and c.get("type") == "toolCall": | |
| r["tool_calls"] += 1 | |
| r["tools"][c.get("name")] += 1 | |
| elif role == "toolResult" and m.get("isError"): | |
| r["tool_errors"] += 1 | |
| return r | |
| def cc(f, dataset): | |
| r = base(dataset, "claude-code", f) | |
| seen_msg = set() | |
| for line in open(f, errors="replace"): | |
| try: | |
| d = json.loads(line) | |
| except ValueError: | |
| continue | |
| if not isinstance(d, dict): | |
| continue | |
| r["_t"].append(ts(d.get("timestamp"))) | |
| r["version"] = r["version"] or d.get("version") | |
| r["cwd"] = r["cwd"] or d.get("cwd") | |
| t = d.get("type") | |
| if t == "system" and d.get("subtype") == "compact_boundary": | |
| r["compactions"] += 1 | |
| m = d.get("message") or {} | |
| if t == "assistant": | |
| if d.get("isSidechain"): | |
| r["subagent_msgs"] += 1 | |
| mid = m.get("id") | |
| if mid not in seen_msg: | |
| seen_msg.add(mid) | |
| r["assistant_msgs"] += 1 | |
| if m.get("model") and m["model"] != "<synthetic>": | |
| r["models"][m["model"]] += 1 | |
| u = m.get("usage") or {} | |
| r["tok_in"] += (u.get("input_tokens", 0) or 0) + (u.get("cache_creation_input_tokens", 0) or 0) | |
| r["tok_out"] += u.get("output_tokens", 0) or 0 | |
| r["tok_cache"] += u.get("cache_read_input_tokens", 0) or 0 | |
| for c in m.get("content") or []: | |
| if isinstance(c, dict) and c.get("type") == "tool_use": | |
| r["tool_calls"] += 1 | |
| r["tools"][c.get("name")] += 1 | |
| elif t == "user" and not d.get("isMeta"): | |
| c = m.get("content") | |
| if isinstance(c, list) and any(isinstance(x, dict) and x.get("type") == "tool_result" for x in c): | |
| r["tool_errors"] += sum(1 for x in c if isinstance(x, dict) and x.get("is_error")) | |
| elif not d.get("isSidechain"): | |
| tx = text_of(c) | |
| if tx.strip() and not NOISE.match(tx): | |
| r["user_prompts"] += 1 | |
| if "[Request interrupted" in tx: | |
| r["aborts"] += 1 | |
| return r | |
| def codex(f, dataset): | |
| r = base(dataset, "codex", f) | |
| tot = None | |
| for line in open(f, errors="replace"): | |
| try: | |
| d = json.loads(line) | |
| except ValueError: | |
| continue | |
| if not isinstance(d, dict): | |
| continue | |
| r["_t"].append(ts(d.get("timestamp"))) | |
| p = d.get("payload") or {} | |
| t, pt = d.get("type"), p.get("type") | |
| if t == "session_meta": | |
| r["cwd"] = p.get("cwd") | |
| r["version"] = f"{p.get('originator')}/{p.get('cli_version')}" | |
| if p.get("model_provider"): | |
| r["providers"][p["model_provider"]] += 1 | |
| elif t == "turn_context" and p.get("model"): | |
| r["models"][p["model"]] += 1 | |
| elif t == "compacted" or pt == "context_compacted": | |
| r["compactions"] += 1 | |
| elif t == "event_msg": | |
| if pt == "user_message" and p.get("message", "").strip() and not NOISE.match(p["message"]): | |
| r["user_prompts"] += 1 | |
| elif pt == "agent_message": | |
| r["assistant_msgs"] += 1 | |
| elif pt == "token_count" and (p.get("info") or {}).get("total_token_usage"): | |
| tot = p["info"]["total_token_usage"] | |
| elif pt == "turn_aborted": | |
| r["aborts"] += 1 | |
| elif t == "response_item" and pt in ("function_call", "custom_tool_call", "web_search_call", "local_shell_call"): | |
| r["tool_calls"] += 1 | |
| r["tools"][p.get("name") or pt] += 1 | |
| elif t == "response_item" and pt in ("function_call_output", "custom_tool_call_output"): | |
| o = str(p.get("output", "")) | |
| m = re.search(r'"exit_code":\s*(-?\d+)|Exit code:? (-?\d+)|Process exited with code (-?\d+)', o) | |
| if m and next(g for g in m.groups() if g) != "0": | |
| r["tool_errors"] += 1 | |
| if tot: | |
| r["tok_in"] = (tot.get("input_tokens", 0) or 0) - (tot.get("cached_input_tokens", 0) or 0) | |
| r["tok_cache"] = tot.get("cached_input_tokens", 0) or 0 | |
| r["tok_out"] = tot.get("output_tokens", 0) or 0 | |
| return r | |
| def finish(r): | |
| wall, active = times_stats(r.pop("_t")) | |
| r["wall_s"], r["active_s"] = round(wall), round(active) | |
| for k in ("models", "providers", "tools"): | |
| r[k] = dict(r[k]) | |
| return r | |
| def main(): | |
| # pi: full datasets mirrored under ext/pi/<owner>_<name>/ | |
| pi_ids = {i.replace("/", "_"): i for i in open(f"{S}/pi_ds.txt").read().split()} | |
| for d in sorted(os.listdir(f"{S}/ext/pi")): | |
| ds = pi_ids.get(d, d.replace("_", "/", 1)) | |
| for f in glob.glob(f"{S}/ext/pi/{d}/**/*.jsonl", recursive=True): | |
| try: | |
| head = json.loads(open(f, errors="replace").readline()) | |
| except ValueError: | |
| continue | |
| if isinstance(head, dict) and head.get("type") == "session": | |
| print(json.dumps(finish(pi(f, ds)))) | |
| # Claude Code / Codex: sampled files in the HF cache (datasets--owner--name/snapshots/rev/...) | |
| kinds = {} | |
| for l in open(f"{S}/hub/prompts_raw.jsonl"): | |
| x = json.loads(l) | |
| kinds.setdefault(x["dataset"], x["format"]) | |
| for repo in glob.glob(f"{S}/hub/cache/datasets--*"): | |
| _, owner, name = os.path.basename(repo).split("--", 2) | |
| ds = f"{owner}/{name}" | |
| kind = kinds.get(ds) | |
| if kind not in ("claude-code", "codex"): | |
| continue | |
| for f in glob.glob(f"{repo}/snapshots/*/**/*.jsonl", recursive=True): | |
| try: | |
| r = (cc if kind == "claude-code" else codex)(f, ds) | |
| except Exception: # noqa: BLE001 | |
| continue | |
| if r["assistant_msgs"] or r["user_prompts"]: | |
| print(json.dumps(finish(r))) | |
| main() | |
Xet Storage Details
- Size:
- 8.98 kB
- Xet hash:
- 2cc67af12abc23b96eceab777f4e585c5ce23e5b554245bc0e335bcf994eed27
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.