#!/usr/bin/env python3 """Collect de-identified Codex CLI usage for the shared-quota reconciliation. Reads local Codex CLI rollout files (~/.codex/sessions/YYYY/MM/DD/*.jsonl) for token_count events since the current weekly cycle start, aggregates per (day, model) with exact token components, and emits a JSON file that contains NO prompts, responses, project paths, or account identifiers. Token accounting uses payload.info.last_token_usage (the per-request delta) — NOT total_token_usage, which is cumulative per session and would overcount. Pure Python 3 stdlib. No network. Never reads message content. Usage: python3 codex_collect_usage.py --cycle-start-epoch 1791694111 --out my_codex_usage.json python3 codex_collect_usage.py --cycle-start "2026-10-11T04:08:31Z" --out my_codex_usage.json If --cycle-start is omitted, uses 7 days before the newest rollout's date. Output schema: codex-usage.collect/v1 """ import argparse import datetime import glob import json import os UTC = datetime.timezone.utc # Codex subscription credit rates (credits per 1M tokens): (input, cached, output) CREDIT_RATES = { "gpt-6-astra": (250.0, 25.0, 1250.0), "gpt-6-sol": (50.0, 5.0, 250.0), "gpt-6.1-sol": (50.0, 2.5, 250.0), "gpt-6-luna": (2.5, 0.25, 12.5), "gpt-5.6-sol": (100.0, 10.0, 500.0), "gpt-5.6-terra": (50.0, 5.0, 250.0), "gpt-5.6-luna": (5.0, 0.5, 30.0), "gpt-5.5": (100.0, 10.0, 500.0), } def find_rollouts(): return sorted(glob.glob(os.path.expanduser( "~/.codex/sessions/*/*/*/rollout-*.jsonl"))) def parse_ts(s): try: return datetime.datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() except Exception: return None def main(): ap = argparse.ArgumentParser() ap.add_argument("--cycle-start-epoch", type=float, default=None) ap.add_argument("--cycle-start", default=None, help="ISO time, e.g. 2026-10-11T04:08:31Z") ap.add_argument("--out", default="my_codex_usage.json") args = ap.parse_args() if args.cycle_start_epoch is not None: w0 = args.cycle_start_epoch elif args.cycle_start: w0 = parse_ts(args.cycle_start) else: w0 = None files = find_rollouts() if not files: print("No rollout files found under ~/.codex/sessions/") return 1 # auto-detect cycle start if not given: newest file date - 7d if w0 is None: newest = max(os.path.getmtime(f) for f in files) w0 = newest - 7 * 86400 print(f"[auto] cycle start = newest rollout - 7d = " f"{datetime.datetime.fromtimestamp(w0, UTC).isoformat()}") agg = {} # (day, model) -> token totals n_files = n_events = 0 for f in files: if os.path.getmtime(f) < w0 - 3600: continue n_files += 1 session_model = None try: fh = open(f, errors="ignore") except OSError: continue current_model = session_model with fh: for line in fh: if '"token_count"' not in line and '"turn_context"' not in line and '"session_meta"' not in line: continue try: e = json.loads(line) except Exception: continue et = e.get("type") if et == "turn_context": m = (e.get("payload") or {}).get("model") if m: current_model = m continue if et == "session_meta": if current_model is None: prov = ((e.get("payload") or {}).get("base_instructions") or {}).get("provenance") or {} if prov.get("model"): current_model = prov["model"] continue if et != "event_msg": continue p = e.get("payload") or {} if p.get("type") != "token_count": continue ts = parse_ts(e.get("timestamp") or "") if ts is None or ts < w0: continue info = p.get("info") or {} model = info.get("model") or current_model or "unknown" # per-request delta, NOT cumulative total u = info.get("last_token_usage") or {} if not u: continue inp = u.get("input_tokens") or 0 cached = u.get("cached_input_tokens") or 0 cwrite = u.get("cache_write_input_tokens") or 0 out = u.get("output_tokens") or 0 if inp == 0 and out == 0: continue # status-only events with zero deltas key = (datetime.datetime.fromtimestamp(ts, UTC).date().isoformat(), model) d = agg.setdefault(key, {"input_tokens": 0, "cached_input_tokens": 0, "cache_write_tokens": 0, "output_tokens": 0, "reasoning_output_tokens": 0, "requests": 0, "credits": 0.0}) d["input_tokens"] += inp d["cached_input_tokens"] += cached d["cache_write_tokens"] += cwrite d["output_tokens"] += out d["reasoning_output_tokens"] += u.get("reasoning_output_tokens") or 0 d["requests"] += 1 r = CREDIT_RATES.get(model) if r: d["credits"] += (inp * r[0] + cached * r[1] + cwrite * r[0] * 1.25 + out * r[2]) / 1e6 n_events += 1 records = [] for (day, model), d in sorted(agg.items()): records.append({"date": day, "model": model, **d}) payload = { "schema": "codex-usage.collect/v1", "cycle_start_utc": datetime.datetime.fromtimestamp(w0, UTC).isoformat(), "generated_at_utc": datetime.datetime.now(UTC).isoformat(), "usage_records": len(records), "total_credits": round(sum(r["credits"] for r in records), 2), "records": records, } with open(args.out, "w") as f: json.dump(payload, f, indent=1, ensure_ascii=False) models = sorted({r["model"] for r in records}) print(f"files scanned: {n_files}, token_count events: {n_events}") print(f"usage_records: {len(records)}, models: {models}") print(f"total credits: {payload['total_credits']}") if records: print(f"date range: {records[0]['date']} .. {records[-1]['date']}") print(f"written: {args.out}") return 0 if __name__ == "__main__": raise SystemExit(main())