usage_logger.py

#!/usr/bin/env python3
"""
Claude Code per-request usage + cost logger.

Runs as a Stop/StopFailure/SessionEnd hook. On each turn it reads the new part
of the session transcript, groups records into one row per *user request*,
prices the token usage, and appends a row to a per-session log file inside the
project:

    <project>/.claude/usage/<session_id>.jsonl

Usage:
    usage_logger.py hook              # hook mode: reads hook JSON on stdin
    usage_logger.py report [options]  # summarize the logs in the current project

Pure stdlib. Never raises into the session: hook mode always exits 0.
"""

import argparse
import datetime as _dt
import glob
import json
import os
import re
import shutil
import sys
from collections import defaultdict

# --------------------------------------------------------------------------
# Pricing -- USD per 1,000,000 tokens, Anthropic first-party list rates.
# --------------------------------------------------------------------------
PRICES = {
    "claude-fable-5":    (10.0, 50.0),
    "claude-mythos-5":   (10.0, 50.0),
    "claude-opus-5":     (5.0,  25.0),
    "claude-opus-4-8":   (5.0,  25.0),
    "claude-opus-4-7":   (5.0,  25.0),
    "claude-opus-4-6":   (5.0,  25.0),
    "claude-opus-4-5":   (5.0,  25.0),
    "claude-sonnet-5":   (3.0,  15.0),
    "claude-sonnet-4-6": (3.0,  15.0),
    "claude-sonnet-4-5": (3.0,  15.0),
    "claude-haiku-4-5":  (1.0,   5.0),
}

# Fast mode runs the same model at premium rates.
FAST_PRICES = {
    "claude-opus-5":   (10.0, 50.0),
    "claude-opus-4-8": (10.0, 50.0),
}

# Claude Sonnet 5 introductory pricing runs through 2026-08-31.
SONNET5_INTRO_END = _dt.date(2026, 8, 31)
SONNET5_INTRO = (2.0, 10.0)

CACHE_WRITE_5M_MULT = 1.25   # 5-minute TTL cache write
CACHE_WRITE_1H_MULT = 2.00   # 1-hour TTL cache write
CACHE_READ_MULT = 0.10       # cache read
WEB_SEARCH_USD_PER_REQ = 0.01  # $10 per 1,000 searches

PROMPT_MAX_CHARS = 300

# Context the IDE/harness injects into the user turn. Stripped from the logged
# prompt so the row shows what was actually typed.
NOISE_TAGS = ("ide_selection", "ide_opened_file", "system-reminder",
              "command-name", "command-message", "command-args",
              "local-command-stdout", "local-command-stderr")
NOISE_RE = re.compile(r"<(%s)\b.*?</\1>" % "|".join(NOISE_TAGS), re.DOTALL)


def rates_for(model, speed, day):
    """Return (input_rate, output_rate, known) per 1M tokens."""
    model = model or ""
    if speed == "fast" and model in FAST_PRICES:
        return FAST_PRICES[model] + (True,)
    if model.startswith("claude-sonnet-5") and day and day <= SONNET5_INTRO_END:
        return SONNET5_INTRO + (True,)
    if model in PRICES:
        return PRICES[model] + (True,)
    # Unknown/future model: fall back to the closest family so cost is not
    # silently zero, and flag it so the report can surface the guess.
    for fam, price in (("opus", (5.0, 25.0)), ("sonnet", (3.0, 15.0)),
                       ("haiku", (1.0, 5.0)), ("fable", (10.0, 50.0))):
        if fam in model:
            return price + (False,)
    return (0.0, 0.0, False)


# --------------------------------------------------------------------------
# Token bucket
# --------------------------------------------------------------------------
def new_bucket():
    return {
        "input": 0, "cache_write_5m": 0, "cache_write_1h": 0,
        "cache_read": 0, "output": 0, "thinking": 0,
        "web_search_requests": 0, "api_requests": 0,
    }


def add_usage(bucket, usage):
    """Accumulate one API response's usage into a bucket."""
    bucket["api_requests"] += 1
    bucket["input"] += usage.get("input_tokens") or 0
    bucket["output"] += usage.get("output_tokens") or 0
    bucket["cache_read"] += usage.get("cache_read_input_tokens") or 0

    # Prefer the explicit 5m/1h split; fall back to the flat total as 5m.
    cc = usage.get("cache_creation") or {}
    w5 = cc.get("ephemeral_5m_input_tokens")
    w1 = cc.get("ephemeral_1h_input_tokens")
    if w5 is None and w1 is None:
        bucket["cache_write_5m"] += usage.get("cache_creation_input_tokens") or 0
    else:
        bucket["cache_write_5m"] += w5 or 0
        bucket["cache_write_1h"] += w1 or 0

    det = usage.get("output_tokens_details") or {}
    bucket["thinking"] += det.get("thinking_tokens") or 0

    st = usage.get("server_tool_use") or {}
    bucket["web_search_requests"] += st.get("web_search_requests") or 0


def merge_bucket(dst, src):
    for k, v in src.items():
        dst[k] = dst.get(k, 0) + v


def price_bucket(bucket, model, speed, day):
    """Return (cost_usd, breakdown, priced_known)."""
    in_rate, out_rate, known = rates_for(model, speed, day)
    m = 1_000_000.0
    bd = {
        "input": bucket["input"] * in_rate / m,
        "cache_write": (bucket["cache_write_5m"] * in_rate * CACHE_WRITE_5M_MULT
                        + bucket["cache_write_1h"] * in_rate * CACHE_WRITE_1H_MULT) / m,
        "cache_read": bucket["cache_read"] * in_rate * CACHE_READ_MULT / m,
        "output": bucket["output"] * out_rate / m,
        "web_search": bucket["web_search_requests"] * WEB_SEARCH_USD_PER_REQ,
    }
    return round(sum(bd.values()), 6), {k: round(v, 6) for k, v in bd.items()}, known


# --------------------------------------------------------------------------
# Transcript parsing
# --------------------------------------------------------------------------
def extract_prompt_text(message):
    content = message.get("content")
    if isinstance(content, str):
        parts = [content]
    elif isinstance(content, list):
        parts = [b.get("text", "") for b in content
                 if isinstance(b, dict) and b.get("type") == "text"]
    else:
        parts = []
    keep = []
    for p in parts:
        s = NOISE_RE.sub(" ", p or "").strip()
        if s:
            keep.append(s)
    text = " ".join(" ".join(keep).split())
    if len(text) > PROMPT_MAX_CHARS:
        text = text[:PROMPT_MAX_CHARS - 1] + "…"
    return text


def is_user_prompt(rec):
    """True if this record is a real user request (not a tool result)."""
    if rec.get("type") != "user" or rec.get("isSidechain"):
        return False
    content = (rec.get("message") or {}).get("content")
    if isinstance(content, list):
        for b in content:
            if isinstance(b, dict) and b.get("type") == "tool_result":
                return False
    return True


def parse_ts(s):
    if not s:
        return None
    try:
        return _dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
    except Exception:
        return None


def new_turn(rec):
    msg = rec.get("message") or {}
    return {
        # promptId is NOT unique per turn -- an expanded skill/command reuses
        # the same promptId for a follow-up user message. The record uuid is
        # the stable per-turn identity.
        "turn_id": rec.get("uuid") or rec.get("promptId"),
        "prompt_id": rec.get("promptId"),
        "prompt": extract_prompt_text(msg),
        "ts_start": rec.get("timestamp"),
        "ts_end": rec.get("timestamp"),
        "cwd": rec.get("cwd"),
        "git_branch": rec.get("gitBranch"),
        "effort": None,
        "seen_msg_ids": [],
        "by_model": {},   # "model|speed" -> bucket
    }


def consume_assistant(turn, rec):
    """Fold one assistant record into the turn, deduping by message id."""
    msg = rec.get("message") or {}
    mid = msg.get("id")
    # Claude Code writes ONE transcript record per content block, each carrying
    # a full copy of the same usage object. Count each message id once.
    if mid:
        if mid in turn["seen_msg_ids"]:
            return
        turn["seen_msg_ids"].append(mid)

    usage = msg.get("usage") or {}
    if not usage:
        return
    key = "%s|%s" % (msg.get("model") or "unknown", usage.get("speed") or "standard")
    bucket = turn["by_model"].setdefault(key, new_bucket())
    add_usage(bucket, usage)
    if rec.get("timestamp"):
        turn["ts_end"] = rec["timestamp"]
    if rec.get("effort"):
        turn["effort"] = rec["effort"]


def build_row(turn, session_id, cumulative_before):
    totals = new_bucket()
    by_model_out = {}
    cost_total = 0.0
    bd_total = defaultdict(float)
    unknown_model = False

    day = None
    ts = parse_ts(turn["ts_start"])
    if ts:
        day = ts.date()

    for key, bucket in sorted(turn["by_model"].items()):
        model, speed = key.split("|", 1)
        cost, bd, known = price_bucket(bucket, model, speed, day)
        if not known:
            unknown_model = True
        merge_bucket(totals, bucket)
        cost_total += cost
        for k, v in bd.items():
            bd_total[k] += v
        entry = {"api_requests": bucket["api_requests"],
                 "tokens": {k: v for k, v in bucket.items() if k != "api_requests"},
                 "cost_usd": cost}
        if speed != "standard":
            entry["speed"] = speed
        by_model_out[model if speed == "standard" else "%s (%s)" % (model, speed)] = entry

    billable_input = (totals["input"] + totals["cache_write_5m"]
                      + totals["cache_write_1h"] + totals["cache_read"])
    t0, t1 = parse_ts(turn["ts_start"]), parse_ts(turn["ts_end"])
    duration = round((t1 - t0).total_seconds(), 1) if (t0 and t1) else None
    cost_total = round(cost_total, 6)

    local = ""
    if t0:
        try:
            local = t0.astimezone().strftime("%Y-%m-%d %H:%M:%S")
        except Exception:
            local = t0.strftime("%Y-%m-%d %H:%M:%S")

    row = {
        "ts_start": turn["ts_start"],
        "ts_end": turn["ts_end"],
        "local_time": local,
        "duration_s": duration,
        "session_id": session_id,
        "turn_id": turn["turn_id"],
        "prompt_id": turn["prompt_id"],
        "cwd": turn["cwd"],
        "git_branch": turn["git_branch"],
        "prompt": turn["prompt"],
        "effort": turn["effort"],
        "api_requests": totals["api_requests"],
        "tokens": {
            "input": totals["input"],
            "cache_write_5m": totals["cache_write_5m"],
            "cache_write_1h": totals["cache_write_1h"],
            "cache_read": totals["cache_read"],
            "output": totals["output"],
            "thinking": totals["thinking"],
            "billable_input": billable_input,
            "total": billable_input + totals["output"],
        },
        "web_search_requests": totals["web_search_requests"],
        "cost_usd": cost_total,
        "cost_breakdown": {k: round(v, 6) for k, v in bd_total.items()},
        "by_model": by_model_out,
        "session_cost_usd": round(cumulative_before + cost_total, 6),
    }
    if unknown_model:
        row["price_estimated"] = True
    return row


def turn_signature(turn):
    """Cheap fingerprint so we only re-emit a turn when its totals changed."""
    return (len(turn["seen_msg_ids"]),
            sum(b["output"] for b in turn["by_model"].values()),
            sum(b["input"] + b["cache_read"] + b["cache_write_5m"] + b["cache_write_1h"]
                for b in turn["by_model"].values()))


# --------------------------------------------------------------------------
# Hook mode
# --------------------------------------------------------------------------
def usage_dir(cwd):
    return os.path.join(cwd, ".claude", "usage")


def transcripts_root():
    return os.path.join(os.path.expanduser("~"), ".claude", "projects")


def transcript_cwd(path):
    """Read the project directory a transcript belongs to.

    The directory name under ~/.claude/projects is a lossy slug (both ':' and
    '_' become '-'), so it cannot be inverted. The records carry the real cwd.
    """
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            for _ in range(200):
                line = f.readline()
                if not line:
                    break
                try:
                    rec = json.loads(line)
                except Exception:
                    continue
                if rec.get("cwd"):
                    return rec["cwd"]
    except Exception:
        pass
    return None


def discover_projects():
    """Map every known project dir -> its transcript paths."""
    found = defaultdict(list)
    for path in sorted(glob.glob(os.path.join(transcripts_root(), "*", "*.jsonl"))):
        cwd = transcript_cwd(path)
        if cwd and os.path.isdir(cwd):
            found[os.path.abspath(cwd)].append(path)
    return found


def run_hook(payload=None):
    if payload is None:
        raw = sys.stdin.read()
        payload = json.loads(raw) if raw.strip() else {}

    session_id = payload.get("session_id") or "unknown-session"
    transcript = payload.get("transcript_path")
    cwd = payload.get("cwd") or os.getcwd()
    if not transcript or not os.path.exists(transcript):
        return

    udir = usage_dir(cwd)
    sdir = os.path.join(udir, ".state")
    os.makedirs(sdir, exist_ok=True)
    log_path = os.path.join(udir, "%s.jsonl" % session_id)
    state_path = os.path.join(sdir, "%s.json" % session_id)

    state = {}
    if os.path.exists(state_path):
        try:
            with open(state_path, "r", encoding="utf-8") as f:
                state = json.load(f)
        except Exception:
            state = {}

    offset = state.get("offset", 0)
    turn = state.get("turn")
    cumulative = state.get("cumulative", 0.0)
    emitted = state.get("emitted", {})     # prompt_id -> [sig] (already logged)

    size = os.path.getsize(transcript)
    if size < offset:      # transcript rotated/truncated -> restart cleanly
        offset, turn, cumulative, emitted = 0, None, 0.0, {}

    with open(transcript, "rb") as f:
        f.seek(offset)
        blob = f.read()

    # Only advance past complete, newline-terminated lines; a partially
    # flushed final line is re-read on the next invocation.
    consumed = blob.rfind(b"\n") + 1
    lines = blob[:consumed].split(b"\n")
    offset += consumed

    pending = []   # turns completed during this scan, in order

    for bline in lines:
        if not bline.strip():
            continue
        try:
            rec = json.loads(bline.decode("utf-8", "replace"))
        except Exception:
            continue

        rtype = rec.get("type")
        if rtype == "user" and is_user_prompt(rec):
            if turn is not None:
                pending.append(turn)
            turn = new_turn(rec)
        elif rtype == "assistant":
            if turn is None:
                # Session resumed mid-stream: synthesize a container so usage
                # is still captured rather than dropped.
                turn = new_turn({"promptId": "orphan-%s" % rec.get("uuid"),
                                 "timestamp": rec.get("timestamp"),
                                 "cwd": rec.get("cwd"),
                                 "gitBranch": rec.get("gitBranch"),
                                 "message": {"content": []}})
                turn["prompt"] = "(resumed session - no prompt record)"
            consume_assistant(turn, rec)

    # The turn that just ended is emitted now; it stays in state so any
    # late-flushed record produces a corrected row on the next hook run.
    to_emit = list(pending)
    if turn is not None:
        to_emit.append(turn)

    rows = []
    for t in to_emit:
        if not t["by_model"]:
            continue
        sig = list(turn_signature(t))
        pid = t["turn_id"]
        prev = emitted.get(pid)
        if prev == sig:
            continue                      # nothing changed since last emit
        # A revision replaces the earlier row for this prompt_id; subtract the
        # superseded cost so the running session total stays correct.
        prior_cost = emitted.get(pid + "@cost", 0.0) if prev else 0.0
        base = cumulative - prior_cost
        row = build_row(t, session_id, base)
        if prev:
            row["revision"] = True
        rows.append(row)
        cumulative = row["session_cost_usd"]
        emitted[pid] = sig
        emitted[pid + "@cost"] = row["cost_usd"]

    if rows:
        os.makedirs(udir, exist_ok=True)
        with open(log_path, "a", encoding="utf-8") as f:
            for row in rows:
                f.write(json.dumps(row, ensure_ascii=False) + "\n")

    state = {"offset": offset, "turn": turn, "cumulative": cumulative,
             "emitted": emitted}
    tmp = state_path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(state, f)
    os.replace(tmp, state_path)


# --------------------------------------------------------------------------
# Report mode
# --------------------------------------------------------------------------
def fmt_tokens(n):
    if n >= 1_000_000:
        return "%.2fM" % (n / 1_000_000)
    if n >= 1_000:
        return "%.1fk" % (n / 1_000)
    return str(n)


def load_rows(udirs, since=None):
    """Latest revision per (session, turn)."""
    if isinstance(udirs, str):
        udirs = [udirs]
    latest = {}
    order = []
    files = []
    for udir in udirs:
        files.extend(sorted(glob.glob(os.path.join(udir, "*.jsonl"))))
    for path in files:
        with open(path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    row = json.loads(line)
                except Exception:
                    continue
                key = (row.get("session_id"),
                       row.get("turn_id") or row.get("prompt_id"))
                if key not in latest:
                    order.append(key)
                latest[key] = row
    rows = [latest[k] for k in order]
    if since:
        rows = [r for r in rows if (r.get("ts_start") or "") >= since]
    rows.sort(key=lambda r: r.get("ts_start") or "")
    return rows


def run_report(args):
    if args.all_projects:
        udirs = [usage_dir(p) for p in discover_projects()]
        udirs = [d for d in udirs if os.path.isdir(d)]
        if not udirs:
            print("No usage logs in any project yet. Run: backfill --all-projects")
            return
        scope = "%d project(s)" % len(udirs)
    else:
        udirs = [usage_dir(args.project or os.getcwd())]
        if not os.path.isdir(udirs[0]):
            print("No usage logs found at %s" % udirs[0])
            return
        scope = udirs[0]
    rows = load_rows(udirs, args.since)
    if not rows:
        print("No matching requests.")
        return

    if args.by == "request":
        print("%-19s  %8s  %5s  %9s  %9s  %8s  %s" %
              ("TIME", "COST$", "REQ", "IN(bill)", "CACHE-RD", "OUT", "PROMPT"))
        print("-" * 110)
        for r in rows:
            t = r.get("tokens", {})
            print("%-19s  %8.4f  %5d  %9s  %9s  %8s  %s" % (
                r.get("local_time", "")[:19], r.get("cost_usd", 0),
                r.get("api_requests", 0),
                fmt_tokens(t.get("billable_input", 0)),
                fmt_tokens(t.get("cache_read", 0)),
                fmt_tokens(t.get("output", 0)),
                (r.get("prompt") or "")[:60]))
    else:
        groups = defaultdict(lambda: {"cost": 0.0, "req": 0, "turns": 0,
                                      "in": 0, "out": 0, "read": 0})
        for r in rows:
            if args.by == "day":
                key = (r.get("local_time") or r.get("ts_start") or "")[:10]
            elif args.by == "session":
                key = r.get("session_id", "")[:8]
            elif args.by == "model":
                key = ", ".join(sorted(r.get("by_model", {}))) or "unknown"
            elif args.by == "project":
                key = os.path.basename((r.get("cwd") or "").rstrip("\\/")) or "?"
            else:
                key = "all"
            g = groups[key]
            t = r.get("tokens", {})
            g["cost"] += r.get("cost_usd", 0)
            g["req"] += r.get("api_requests", 0)
            g["turns"] += 1
            g["in"] += t.get("billable_input", 0)
            g["out"] += t.get("output", 0)
            g["read"] += t.get("cache_read", 0)
        print("%-26s  %10s  %7s  %7s  %9s  %9s  %8s" %
              (args.by.upper(), "COST$", "TURNS", "APIREQ", "IN(bill)", "CACHE-RD", "OUT"))
        print("-" * 90)
        for key in sorted(groups):
            g = groups[key]
            print("%-26s  %10.4f  %7d  %7d  %9s  %9s  %8s" % (
                key[:26], g["cost"], g["turns"], g["req"],
                fmt_tokens(g["in"]), fmt_tokens(g["read"]), fmt_tokens(g["out"])))

    total = sum(r.get("cost_usd", 0) for r in rows)
    treq = sum(r.get("api_requests", 0) for r in rows)
    tout = sum(r.get("tokens", {}).get("output", 0) for r in rows)
    tin = sum(r.get("tokens", {}).get("billable_input", 0) for r in rows)
    print("-" * 90)
    print("TOTAL  %d requests over %d API calls   in(billable) %s   out %s   $%.4f"
          % (len(rows), treq, fmt_tokens(tin), fmt_tokens(tout), total))
    print("Scope: %s" % scope)
    print("Note: cost is Anthropic API list price. On a Pro/Max subscription "
          "this is the equivalent value consumed, not an amount billed.")


def run_backfill(args):
    """Process this project's existing transcripts through the same pipeline.

    Idempotent: reuses the per-session state files, so re-running only picks up
    transcript records that were not already logged.
    """
    if args.all_projects:
        projects = discover_projects()
        if not projects:
            print("No transcripts found under %s" % transcripts_root())
            return
        print("Backfilling %d project(s)\n" % len(projects))
    else:
        cwd = os.path.abspath(args.project or os.getcwd())
        slug = re.sub(r"[^A-Za-z0-9]", "-", cwd)
        tdir = os.path.join(transcripts_root(), slug)
        paths = sorted(glob.glob(os.path.join(tdir, "*.jsonl")))
        if not paths:
            # The slug rule is a guess; fall back to reading the cwd recorded
            # inside each transcript, which is platform-independent.
            paths = discover_projects().get(cwd, [])
        if not paths:
            print("No transcripts for this project (looked in %s)" % tdir)
            return
        projects = {cwd: paths}

    for cwd in sorted(projects):
        paths = projects[cwd]
        ok = fail = 0
        for p in paths:
            sid = os.path.splitext(os.path.basename(p))[0]
            try:
                run_hook({"session_id": sid, "transcript_path": p, "cwd": cwd,
                          "hook_event_name": "Backfill"})
                ok += 1
            except Exception as exc:
                fail += 1
                print("  FAIL %s (%s)" % (sid, exc))
        print("%-55s %3d session(s)%s" % (cwd, ok, "  %d FAILED" % fail if fail else ""))
    print("\nLogs written to <project>/.claude/usage/ in each project above.")


# --------------------------------------------------------------------------
# Install mode -- makes this script portable across devices
# --------------------------------------------------------------------------
def default_hook_command(script_path):
    """Build the hook command line for THIS device.

    Paths and interpreter names differ per machine ('py -3' is Windows-only),
    so the command is resolved at install time rather than shipped hardcoded.
    """
    exe = sys.executable
    # A venv interpreter can vanish when the project is deleted; prefer the
    # base installation it was built from.
    if sys.prefix != sys.base_prefix:
        cand = (os.path.join(sys.base_prefix, "python.exe") if os.name == "nt"
                else os.path.join(sys.base_prefix, "bin", "python3"))
        if os.path.exists(cand):
            exe = cand
    return '"%s" "%s" hook' % (exe.replace("\\", "/"),
                               script_path.replace("\\", "/"))


def run_install(args):
    home = os.path.expanduser("~")
    hooks_dir = os.path.join(home, ".claude", "hooks")
    os.makedirs(hooks_dir, exist_ok=True)
    target = os.path.join(hooks_dir, "usage_logger.py")

    src = os.path.abspath(__file__)
    if os.path.abspath(target) != src:
        shutil.copyfile(src, target)
        print("Installed script -> %s" % target)
    else:
        print("Script already at %s" % target)

    settings = os.path.join(home, ".claude", "settings.json")
    data = {}
    if os.path.exists(settings):
        text = open(settings, encoding="utf-8").read().strip()
        try:
            data = json.loads(text) if text else {}
        except ValueError as exc:
            print("ERROR: %s is not valid JSON (%s). Fix it first; "
                  "nothing was changed." % (settings, exc))
            return 1
        shutil.copyfile(settings, settings + ".bak")
        print("Backed up existing settings -> %s.bak" % settings)

    cmd = default_hook_command(target)
    hooks = data.setdefault("hooks", {})
    for event in ("Stop", "StopFailure", "SessionEnd"):
        groups = hooks.setdefault(event, [])
        # Drop a previous install of THIS logger; leave every other hook alone.
        for g in groups:
            g["hooks"] = [h for h in g.get("hooks", [])
                          if "usage_logger.py" not in str(h.get("command", ""))]
        groups[:] = [g for g in groups if g.get("hooks")]
        groups.append({"hooks": [{"type": "command", "command": cmd,
                                  "timeout": 30}]})

    tmp = settings + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
    os.replace(tmp, settings)
    print("Registered hooks in %s: Stop, StopFailure, SessionEnd" % settings)
    print("Hook command: %s" % cmd)
    print("\nDone. Logs will appear in <project>/.claude/usage/ from the next turn.")
    return 0


def main():
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd")
    sub.add_parser("hook", help="hook mode (reads hook JSON on stdin)")
    rp = sub.add_parser("report", help="summarize this project's usage logs")
    rp.add_argument("--by", default="request",
                    choices=["request", "day", "session", "model", "project", "all"])
    rp.add_argument("--since", help="ISO date, e.g. 2026-08-01")
    rp.add_argument("--project", help="project dir (default: cwd)")
    rp.add_argument("--all-projects", action="store_true",
                    help="aggregate across every project")
    sub.add_parser("install",
                    help="install this script + hooks on this device")
    bp = sub.add_parser("backfill",
                        help="log existing transcripts")
    bp.add_argument("--project", help="project dir (default: cwd)")
    bp.add_argument("--all-projects", action="store_true",
                    help="backfill every project found in ~/.claude/projects")
    args = ap.parse_args()

    if args.cmd == "hook":
        try:
            run_hook()
        except Exception:
            # Never let logging break the session.
            if os.environ.get("CLAUDE_USAGE_DEBUG"):
                import traceback
                traceback.print_exc(file=sys.stderr)
        sys.exit(0)
    elif args.cmd == "report":
        run_report(args)
    elif args.cmd == "backfill":
        run_backfill(args)
    elif args.cmd == "install":
        sys.exit(run_install(args) or 0)
    else:
        ap.print_help()


if __name__ == "__main__":
    main()