#!/usr/bin/env python3
"""Sum Claude Code + Codex token usage since 2026-07-01. Read-only. stdlib only.

Claude Code sub-agent transcripts (a Task-tool call or a "fork" writes its own
~/.claude/projects/<project>/<session-id>/subagents/agent-*.jsonl, sibling to
the parent session's own <session-id>.jsonl) are scanned along with the
top-level session files. Their tokens count toward their parent session, not
as extra sessions, and every message id is deduped globally across parent and
sub-agent files, because a forked sub-agent's shared history can otherwise
show the same message id in more than one file. Codex has no equivalent
on-disk split -- forked/sub-agent Codex threads share the same session file
and are already handled by summing each turn's marginal delta.

Modes:
  python3 token_usage.py [--daily] [--last-hour] [--max-seconds N]
      Scan this host only. Prints one JSON object tagged with a "seat" label
      (from $LEDGER_SEAT, default "unknown") and no host-identifying paths.
      --max-seconds N stops the scan after N seconds with exit status 3,
      after caching the progress made (the next run carries on from there).
      --last-hour adds a "last_hour_total" per tool: the sum of each turn
      whose timestamp falls in the 3600 seconds before the scan started
      (second-precision; used by collect.py's rolling last_60_min figure).

  python3 token_usage.py --merge [--daily] [--last-hour] [--out ledger.json]
      Scan this host (seat = $LEDGER_LOCAL_SEAT, default "mac"), then for each
      "alias=seat" pair in $LEDGER_HOSTS (comma-separated, e.g.
      "vps=vps,pc=bigmem"), shell out to `ssh <alias> "LEDGER_SEAT=<seat>
      python3 -" < token_usage.py` to scan that host too, and merge every
      seat's result into the ledger.json shape. Writes --out (default
      ./ledger.json) and also prints it to stdout.

No hostnames, usernames, or local filesystem paths are baked into this
script; all host discovery is generic (walks whichever machine it runs on),
and remote hosts are supplied by the caller via $LEDGER_HOSTS.

Incremental cache: each transcript is reduced to compact rows once and kept in
~/.cache/token-ledger/scan-cache.json on the host being scanned (dir 700,
file 600), keyed by (path, size, mtime_ns); unchanged files are not re-read.
Totals are recomputed from rows every run, so they match a cache-less scan.
TOKEN_LEDGER_CACHE=0 disables it. See the comment above load_cache().
"""
import json, os, sys, glob, subprocess, datetime, hashlib, tempfile, time, signal
from collections import defaultdict

HOME = os.path.expanduser("~")
SINCE = "2026-07"
SINCE_DATE = "2026-07-01"

def month_of(ts):
    return ts[:7] if ts else None

def day_of(ts):
    return ts[:10] if ts else None

def epoch_of(ts):
    """First-19-chars ISO-8601 -> UTC epoch seconds. Second precision is enough
    for hour-bucketing; avoids fromisoformat's variable-fraction fragility."""
    if not ts or len(ts) < 19:
        return None
    try:
        dt = datetime.datetime.strptime(ts[:19], "%Y-%m-%dT%H:%M:%S")
        return dt.replace(tzinfo=datetime.timezone.utc).timestamp()
    except Exception:
        return None

def find_claude_bases():
    bases = []
    if os.path.isdir(os.path.join(HOME, ".claude", "projects")):
        bases.append(os.path.join(HOME, ".claude", "projects"))
    # any other users on this host with .claude/projects
    for home_root in ("/home", "/Users", "/root"):
        if os.path.isdir(home_root):
            try:
                for entry in os.listdir(home_root):
                    p = os.path.join(home_root, entry, ".claude", "projects")
                    if os.path.isdir(p) and p not in bases:
                        bases.append(p)
            except PermissionError:
                pass
    # Windows-side WSL mirror (native Windows Claude Code install visible from WSL)
    for p in glob.glob("/mnt/c/Users/*/.claude/projects"):
        if os.path.isdir(p) and p not in bases:
            bases.append(p)
    return sorted(set(bases))

def find_claude_session_files(bases):
    """A Claude Code project directory (base/<project>/) holds two kinds of
    transcript file:
      parent    base/<project>/<session-id>.jsonl -- the top-level session.
      subagent  base/<project>/<session-id>/subagents/agent-*.jsonl -- a
                Task-tool subagent or fork spawned by that session. Claude
                Code writes these to a directory that shares the parent
                session's UUID as its name, sibling to the parent .jsonl.
    Returns (parent_files, subagent_files), both sorted lists of paths.
    Tolerant of a project with no subagents directory at all (glob just
    returns nothing)."""
    parent_files = []
    subagent_files = []
    for base in bases:
        parent_files.extend(glob.glob(os.path.join(base, "*", "*.jsonl")))
        subagent_files.extend(
            glob.glob(os.path.join(base, "*", "*", "subagents", "*.jsonl"))
        )
    return sorted(set(parent_files)), sorted(set(subagent_files))

def find_codex_bases():
    bases = []
    if os.path.isdir(os.path.join(HOME, ".codex", "sessions")):
        bases.append(os.path.join(HOME, ".codex", "sessions"))
    for home_root in ("/home", "/Users", "/root"):
        if os.path.isdir(home_root):
            try:
                for entry in os.listdir(home_root):
                    p = os.path.join(home_root, entry, ".codex", "sessions")
                    if os.path.isdir(p) and p not in bases:
                        bases.append(p)
            except PermissionError:
                pass
    for p in glob.glob("/mnt/c/Users/*/.codex/sessions"):
        if os.path.isdir(p) and p not in bases:
            bases.append(p)
    return sorted(set(bases))

def _claude_session_key_for_parent(path):
    """base/<project>/<session-id>.jsonl -> (project, session-id)."""
    project = os.path.basename(os.path.dirname(path))
    session_id = os.path.basename(path)
    if session_id.endswith(".jsonl"):
        session_id = session_id[: -len(".jsonl")]
    return (project, session_id)

def _claude_session_key_for_subagent(path):
    """base/<project>/<session-id>/subagents/agent-*.jsonl -> (project, session-id)."""
    subagents_dir = os.path.dirname(path)
    session_dir = os.path.dirname(subagents_dir)
    project = os.path.basename(os.path.dirname(session_dir))
    session_id = os.path.basename(session_dir)
    return (project, session_id)

# ---------------------------------------------------------------------------
# Incremental scan cache.
#
# Reading every transcript on every run is what made a 5-minute collector
# re-read gigabytes. Each transcript file is reduced once to compact rows
# (one per usage-bearing message / token_count event) and those rows are kept
# in a cache file under the scanning user's own cache directory
# (~/.cache/token-ledger/, mode 700; the file itself 600), keyed by
# (path, size, mtime_ns). An unchanged file is served from its cached rows; a
# changed or new file is re-read; a file that is gone drops out of the cache.
#
# The cache holds rows, not per-file totals, on purpose: Claude Code message
# ids are deduped globally across files (first file in scan order wins), and
# the rolling last-hour figure moves with the clock. Both are recomputed from
# rows on every run, cached or not, so a scan with the cache gives exactly the
# totals a scan without it gives. A message id is stored as an 8-byte hash;
# the per-run id index is rebuilt in memory from those hashes.
#
# The cache never leaves the host it describes: it is not part of the JSON
# this script prints. Set TOKEN_LEDGER_CACHE=0 to scan without it. Any cache
# read or write failure (read-only home, corrupt file) falls back to a plain
# full scan; it never fails the scan.
# ---------------------------------------------------------------------------

CACHE_VERSION = 1


class ScanDeadline(BaseException):
    """Raised when a scan runs past its --max-seconds budget. scan() saves the
    cache progress made so far before re-raising, so the next run continues
    where this one stopped instead of starting over. A BaseException so that
    no `except Exception` in the read loops can swallow it (the SIGALRM
    backstop can fire anywhere)."""


def default_cache_path():
    return os.path.join(HOME, ".cache", "token-ledger", "scan-cache.json")


def _key_hash(text):
    return hashlib.blake2b(text.encode("utf-8", "replace"), digest_size=8).hexdigest()


def load_cache(path):
    """-> {"claude": {path: entry}, "codex": {path: entry}}; empty on any problem."""
    empty = {"claude": {}, "codex": {}}
    if not path:
        return empty
    try:
        with open(path, "r") as fh:
            data = json.load(fh)
    except Exception:
        return empty
    if not isinstance(data, dict) or data.get("version") != CACHE_VERSION or data.get("since") != SINCE:
        return empty
    return {
        "claude": data.get("claude") if isinstance(data.get("claude"), dict) else {},
        "codex": data.get("codex") if isinstance(data.get("codex"), dict) else {},
    }


def save_cache(path, claude_entries, codex_entries):
    """Atomic write, dir 700, file 600. Returns True on success."""
    if not path:
        return False
    d = os.path.dirname(path)
    tmp = None
    try:
        os.makedirs(d, mode=0o700, exist_ok=True)
        if (os.stat(d).st_mode & 0o777) != 0o700:
            os.chmod(d, 0o700)
        fd, tmp = tempfile.mkstemp(prefix=".scan-cache-", dir=d)
        with os.fdopen(fd, "w") as fh:
            json.dump({"version": CACHE_VERSION, "since": SINCE,
                       "claude": claude_entries, "codex": codex_entries},
                      fh, separators=(",", ":"))
        os.replace(tmp, path)
        return True
    except Exception:
        if tmp:
            try:
                os.unlink(tmp)
            except OSError:
                pass
        return False


def _file_sig(path):
    st = os.stat(path)
    return st.st_size, st.st_mtime_ns


def _cached_or_read(path, reader, cache_in, cache_out, stats, errors, deadline=None):
    """Returns (entry, errored). entry is None only when the file can't be
    stat'ed. A file that errors while being read contributes whatever rows
    were read before the error (the old single-pass behavior) and is never
    cached, so the next run retries it. Raises ScanDeadline once past
    `deadline` (a time.monotonic() value)."""
    if deadline is not None and time.monotonic() > deadline:
        raise ScanDeadline()
    try:
        size, mtime_ns = _file_sig(path)
    except OSError as e:
        errors.append(f"{path}: {e}")
        return None, True
    if cache_in is not None:
        ent = cache_in.get(path)
        if isinstance(ent, dict) and ent.get("size") == size and ent.get("mtime_ns") == mtime_ns:
            stats["hits"] += 1
            if cache_out is not None:
                cache_out[path] = ent
            return ent, False
    stats["misses"] += 1
    ent, err = reader(path, deadline)
    if err is not None:
        errors.append(f"{path}: {err}")
        return ent, True
    ent["size"] = size
    ent["mtime_ns"] = mtime_ns
    if cache_out is not None:
        cache_out[path] = ent
    return ent, False


class _Budget:
    """Deadline check for the line loops, taken after every 1 MB read, so a
    large transcript read slowly (a niced scan on a busy seat, or a Windows
    path over WSL's drvfs) cannot overrun the budget. The partly-read file
    is dropped, not cached."""
    STEP = 1 << 20

    def __init__(self, deadline):
        self.deadline = deadline
        self.read = 0

    def add(self, nbytes):
        if self.deadline is None:
            return
        self.read += nbytes
        if self.read >= self.STEP:
            self.read = 0
            if time.monotonic() > self.deadline:
                raise ScanDeadline()


def _read_claude_file(path, deadline=None):
    """One pass over a Claude Code transcript. -> (entry, error_or_None).
    entry = {"models": [...], "rows": [[key, ts, model_i, in, out, cw, cr], ...]}
    with one row per usage-bearing assistant message, in file order."""
    models, model_index, rows = [], {}, []
    err = None
    try:
        with open(path, "r", errors="replace") as fh:
            budget = _Budget(deadline)
            for line in fh:
                budget.add(len(line))
                line = line.strip()
                if not line:
                    continue
                try:
                    d = json.loads(line)
                except Exception:
                    continue
                if d.get("type") != "assistant":
                    continue
                msg = d.get("message") or {}
                usage = msg.get("usage")
                if not usage:
                    continue
                mid = msg.get("id")
                ts = d.get("timestamp")
                if mid:
                    key = _key_hash("id:" + str(mid))
                else:
                    key = _key_hash("anon:" + json.dumps([path, ts, json.dumps(usage, sort_keys=True)]))
                model = msg.get("model", "unknown")
                if not isinstance(model, (str, type(None))):
                    model = str(model)
                mk = json.dumps(model)
                if mk not in model_index:
                    model_index[mk] = len(models)
                    models.append(model)
                rows.append([
                    key, ts, model_index[mk],
                    usage.get("input_tokens", 0) or 0,
                    usage.get("output_tokens", 0) or 0,
                    usage.get("cache_creation_input_tokens", 0) or 0,
                    usage.get("cache_read_input_tokens", 0) or 0,
                ])
    except ScanDeadline:
        raise
    except Exception as e:
        err = str(e)
    return {"models": models, "rows": rows}, err


def process_claude(daily=False, now_cutoff=None, cache_in=None, cache_out=None, deadline=None):
    bases = find_claude_bases()
    parent_files, subagent_files = find_claude_session_files(bases)
    by_month = defaultdict(lambda: defaultdict(int))
    by_day = defaultdict(lambda: defaultdict(int))
    day_sessions = defaultdict(set)
    day_model = defaultdict(lambda: defaultdict(int))
    by_model = defaultdict(lambda: defaultdict(int))
    seen_ids = set()  # global across parent AND subagent files -- a fork's
                       # shared history can show the same message id in more
                       # than one subagent transcript, and rarely also in the
                       # parent transcript itself; every turn is counted once.
    session_has_usage = {}  # (project, session_id) -> bool
    turns = 0
    projects = set()
    earliest = None
    errors = []
    files_total = 0
    last_hour_total = 0
    stats = {"hits": 0, "misses": 0}

    def process_file(path, session_key):
        nonlocal turns, earliest, last_hour_total
        ent, errored = _cached_or_read(path, _read_claude_file, cache_in, cache_out, stats, errors, deadline)
        if ent is None:
            return False
        had_usage = False
        models = ent["models"]
        for key, ts, mi, inp, out, cw, cr in ent["rows"]:
            if key in seen_ids:
                continue
            seen_ids.add(key)
            had_usage = True
            turns += 1
            m = month_of(ts)
            dday = day_of(ts)
            if earliest is None or (ts and ts < earliest):
                earliest = ts
            if m and m >= SINCE:
                model = models[mi]
                tot = inp + out + cw + cr
                bm = by_month[m]
                bm["input"] += inp; bm["output"] += out
                bm["cache_write"] += cw; bm["cache_read"] += cr
                bm["total"] += tot
                mb = by_model[model]
                mb["input"] += inp; mb["output"] += out
                mb["cache_write"] += cw; mb["cache_read"] += cr
                mb["total"] += tot
                if now_cutoff is not None:
                    te = epoch_of(ts)
                    if te is not None and te >= now_cutoff:
                        last_hour_total += tot
                if daily and dday:
                    bd = by_day[dday]
                    bd["input"] += inp; bd["output"] += out
                    bd["cache_write"] += cw; bd["cache_read"] += cr
                    bd["total"] += tot
                    bd["turns"] += 1
                    day_sessions[dday].add(session_key)
                    dm = day_model[dday][model]
                    day_model[dday][model] = dm + tot
        return False if errored else had_usage

    for path in parent_files:
        files_total += 1
        key = _claude_session_key_for_parent(path)
        projects.add(key[0])
        had_usage = process_file(path, key)
        session_has_usage[key] = session_has_usage.get(key, False) or had_usage

    subagent_transcripts = len(subagent_files)
    for path in subagent_files:
        files_total += 1
        key = _claude_session_key_for_subagent(path)
        projects.add(key[0])
        had_usage = process_file(path, key)
        # A subagent's turns belong to its parent session for the "sessions"
        # figure -- this never adds an extra session per subagent file, and
        # it also picks up a session whose own top-level transcript logged no
        # usage of its own but whose subagents did real work.
        session_has_usage[key] = session_has_usage.get(key, False) or had_usage

    sessions = sum(1 for had in session_has_usage.values() if had)
    result = {
        "files_total": files_total,
        "sessions_with_usage": sessions,
        "subagent_transcripts": subagent_transcripts,
        "turns": turns,
        "project_count": len(projects),
        "earliest": earliest,
        "by_month": {k: dict(v) for k, v in sorted(by_month.items())},
        "by_model": {k: dict(v) for k, v in by_model.items()},
        "errors": errors,
        "cache_hits": stats["hits"],
        "cache_misses": stats["misses"],
    }
    if now_cutoff is not None:
        result["last_hour_total"] = last_hour_total
    if daily:
        result["by_day"] = {
            d: {**dict(v), "sessions": len(day_sessions[d])}
            for d, v in sorted(by_day.items())
        }
        result["by_day_model"] = {d: dict(v) for d, v in sorted(day_model.items())}
    return result


def _read_codex_file(path, deadline=None):
    """One pass over a Codex session file. -> (entry, error_or_None).
    entry = {"meta": {forked, cwd, earliest, had_events, turns}, "models": [...],
             "rows": [[ts, model_i, in, out, cached, cache_write], ...]}
    with one row per token_count event on or after SINCE, in file order."""
    meta = {"forked": False, "cwd": None, "earliest": None, "had_events": False, "turns": 0}
    models, model_index, rows = [], {}, []
    model_seen = "unknown"
    err = None
    try:
        with open(path, "r", errors="replace") as fh:
            budget = _Budget(deadline)
            for line in fh:
                budget.add(len(line))
                line = line.strip()
                if not line:
                    continue
                try:
                    d = json.loads(line)
                except Exception:
                    continue
                t = d.get("type")
                ts = d.get("timestamp")
                if t == "session_meta":
                    payload = d.get("payload", {})
                    if payload.get("forked_from_id"):
                        meta["forked"] = True
                    meta["cwd"] = payload.get("cwd")
                    if ts and (meta["earliest"] is None or ts < meta["earliest"]):
                        meta["earliest"] = ts
                elif t == "turn_context":
                    model_seen = d.get("payload", {}).get("model", model_seen)
                elif t == "event_msg" and d.get("payload", {}).get("type") == "token_count":
                    info = d["payload"].get("info") or {}
                    last_usage = info.get("last_token_usage") or {}
                    meta["had_events"] = True
                    meta["turns"] += 1
                    m = month_of(ts)
                    if m and m >= SINCE:
                        mk = json.dumps(model_seen)
                        if mk not in model_index:
                            model_index[mk] = len(models)
                            models.append(model_seen)
                        rows.append([
                            ts, model_index[mk],
                            last_usage.get("input_tokens", 0) or 0,
                            last_usage.get("output_tokens", 0) or 0,
                            last_usage.get("cached_input_tokens", 0) or 0,
                            last_usage.get("cache_write_input_tokens", 0) or 0,
                        ])
    except ScanDeadline:
        raise
    except Exception as e:
        err = str(e)
    return {"meta": meta, "models": models, "rows": rows}, err


def process_codex(daily=False, now_cutoff=None, cache_in=None, cache_out=None, deadline=None):
    bases = find_codex_bases()
    by_month = defaultdict(lambda: defaultdict(int))
    by_day = defaultdict(lambda: defaultdict(int))
    day_sessions = defaultdict(set)
    day_model = defaultdict(lambda: defaultdict(int))
    by_model = defaultdict(lambda: defaultdict(int))
    sessions = 0
    turns = 0
    projects = set()
    earliest = None
    errors = []
    files_total = 0
    forked_count = 0
    root_count = 0
    last_hour_total = 0
    stats = {"hits": 0, "misses": 0}
    for base in bases:
        for path in glob.glob(os.path.join(base, "**", "*.jsonl"), recursive=True):
            files_total += 1
            ent, errored = _cached_or_read(path, _read_codex_file, cache_in, cache_out, stats, errors, deadline)
            if ent is None:
                continue
            meta = ent["meta"]
            models = ent["models"]
            turns += meta["turns"]
            fe = meta["earliest"]
            if fe and (earliest is None or fe < earliest):
                earliest = fe
            for ts, mi, inp, out, cached, cwrite in ent["rows"]:
                m = month_of(ts)
                dday = day_of(ts)
                model_seen = models[mi]
                tot = inp + out + cwrite
                # NOTE: OpenAI/Codex convention: cached_input_tokens is a SUBSET
                # of input_tokens (unlike Anthropic where cache fields are
                # additive). Do not add cached into total to avoid double count.
                bm = by_month[m]
                bm["input"] += inp; bm["output"] += out
                bm["cache_write"] += cwrite
                bm["cache_read_subset_of_input"] += cached
                bm["total"] += tot
                mb = by_model[model_seen]
                mb["input"] += inp; mb["output"] += out
                mb["cache_write"] += cwrite
                mb["cache_read_subset_of_input"] += cached
                mb["total"] += tot
                if now_cutoff is not None:
                    te = epoch_of(ts)
                    if te is not None and te >= now_cutoff:
                        last_hour_total += tot
                if daily and dday:
                    bd = by_day[dday]
                    bd["input"] += inp; bd["output"] += out
                    bd["cache_write"] += cwrite
                    bd["cache_read_subset_of_input"] += cached
                    bd["total"] += tot
                    bd["turns"] += 1
                    day_sessions[dday].add(path)
                    dm = day_model[dday][model_seen]
                    day_model[dday][model_seen] = dm + tot
            if errored:
                continue
            if meta["cwd"]:
                projects.add(meta["cwd"])
            if meta["had_events"]:
                sessions += 1
            if meta["forked"]:
                forked_count += 1
            else:
                root_count += 1
    result = {
        "files_total": files_total,
        "sessions_with_usage": sessions,
        "turns": turns,
        "project_count": len(projects),
        "earliest": earliest,
        "by_month": {k: dict(v) for k, v in sorted(by_month.items())},
        "by_model": {k: dict(v) for k, v in by_model.items()},
        "forked_count": forked_count,
        "root_count": root_count,
        "errors": errors,
        "cache_hits": stats["hits"],
        "cache_misses": stats["misses"],
    }
    if now_cutoff is not None:
        result["last_hour_total"] = last_hour_total
    if daily:
        result["by_day"] = {
            d: {**dict(v), "sessions": len(day_sessions[d])}
            for d, v in sorted(by_day.items())
        }
        result["by_day_model"] = {d: dict(v) for d, v in sorted(day_model.items())}
    return result


def scan(daily=False, last_hour=False, cache_path="default", max_seconds=None):
    """cache_path: "default" -> ~/.cache/token-ledger/scan-cache.json unless
    TOKEN_LEDGER_CACHE=0; None -> no cache; anything else -> that file.
    max_seconds: stop with ScanDeadline once this much time has passed,
    after saving the cache progress made so far (None = no limit)."""
    seat = os.environ.get("LEDGER_SEAT") or os.environ.get("LEDGER_LOCAL_SEAT") or "unknown"
    now_cutoff = (datetime.datetime.now(datetime.timezone.utc).timestamp() - 3600) if last_hour else None
    deadline = (time.monotonic() + max_seconds) if max_seconds is not None else None
    if cache_path == "default":
        cache_path = None if os.environ.get("TOKEN_LEDGER_CACHE", "1") == "0" else default_cache_path()
    if not cache_path:
        claude = process_claude(daily=daily, now_cutoff=now_cutoff, deadline=deadline)
        codex = process_codex(daily=daily, now_cutoff=now_cutoff, deadline=deadline)
        return {"seat": seat, "claude": claude, "codex": codex, "cache_saved": False}
    cache = load_cache(cache_path)
    out_claude, out_codex = {}, {}
    try:
        claude = process_claude(daily=daily, now_cutoff=now_cutoff, cache_in=cache["claude"],
                                cache_out=out_claude, deadline=deadline)
        codex = process_codex(daily=daily, now_cutoff=now_cutoff, cache_in=cache["codex"],
                              cache_out=out_codex, deadline=deadline)
    except ScanDeadline:
        # Keep what this run read AND every entry it did not reach yet.
        save_cache(cache_path, {**cache["claude"], **out_claude}, {**cache["codex"], **out_codex})
        raise
    saved = save_cache(cache_path, out_claude, out_codex)
    return {"seat": seat, "claude": claude, "codex": codex, "cache_saved": saved}


def run_remote(alias, seat, daily, script_path, last_hour=False, timeout=1700):
    # nice: the seats are working machines (bigmem trains models); a ledger
    # scan should only ever use idle CPU there. --max-seconds: the remote scan
    # stops itself (saving its cache progress) before this side gives up, so
    # a timed-out ssh never leaves an orphaned scan running on the seat.
    budget = max(5, int(timeout) - 10)
    cmd = ["ssh", "-o", "ServerAliveInterval=15", "-o", "ServerAliveCountMax=8",
           alias, f"LEDGER_SEAT={seat} nice -n 19 python3 - --max-seconds {budget}"]
    if daily:
        cmd[-1] += " --daily"
    if last_hour:
        cmd[-1] += " --last-hour"
    with open(script_path, "rb") as fh:
        proc = subprocess.run(cmd, stdin=fh, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout)
    if proc.returncode != 0:
        raise RuntimeError(f"ssh {alias} failed: {proc.stderr.decode(errors='replace')[:2000]}")
    return json.loads(proc.stdout.decode(errors="replace"))

def merge(results, daily):
    """results: list of {'seat':..., 'claude':{...}, 'codex':{...}} -> ledger.json shape."""
    by_tool = {t: defaultdict(int) for t in ("claude", "codex")}
    by_tool_sess_turns = {t: {"sessions": 0, "turns": 0} for t in ("claude", "codex")}
    by_model = defaultdict(int)  # (model,tool) -> tokens
    by_seat = []
    subagent_transcripts_total = 0
    daily_merged = defaultdict(lambda: {
        "claude": {"in": 0, "out": 0, "cw": 0, "cr": 0, "total": 0},
        "codex": {"in": 0, "out": 0, "total": 0},
    })
    caveats = []
    seats_seen = set()

    for r in results:
        seat = r.get("seat", "unknown")
        seats_seen.add(seat)
        for tool in ("claude", "codex"):
            d = r[tool]
            tokens = sum(v.get("total", 0) for v in d["by_month"].values())
            by_tool_sess_turns[tool]["sessions"] += d["sessions_with_usage"]
            by_tool_sess_turns[tool]["turns"] += d["turns"]
            subagent_transcripts_total += d.get("subagent_transcripts", 0)
            for k in ("input", "output", "cache_write"):
                by_tool[tool][k] += sum(v.get(k, 0) for v in d["by_month"].values())
            cr_key = "cache_read" if tool == "claude" else "cache_read_subset_of_input"
            by_tool[tool]["cache_read"] += sum(v.get(cr_key, 0) for v in d["by_month"].values())
            by_tool[tool]["total"] += tokens
            by_seat.append({
                "seat": seat, "tool": tool, "tokens": tokens,
                "sessions": d["sessions_with_usage"], "turns": d["turns"],
                "first": d.get("earliest"),
                "last_hour_total": d.get("last_hour_total", 0),
            })
            for model, v in d["by_model"].items():
                by_model[(model, tool)] += v.get("total", 0)
            if d.get("errors"):
                caveats.append(f"{seat}/{tool}: {len(d['errors'])} unreadable file(s)")
            if daily:
                for day, v in d.get("by_day", {}).items():
                    if tool == "claude":
                        c = daily_merged[day]["claude"]
                        c["in"] += v.get("input", 0); c["out"] += v.get("output", 0)
                        c["cw"] += v.get("cache_write", 0); c["cr"] += v.get("cache_read", 0)
                        c["total"] += v.get("total", 0)
                    else:
                        c = daily_merged[day]["codex"]
                        c["in"] += v.get("input", 0); c["out"] += v.get("output", 0)
                        c["total"] += v.get("total", 0)

    grand_tokens = by_tool["claude"]["total"] + by_tool["codex"]["total"]
    grand_sessions = by_tool_sess_turns["claude"]["sessions"] + by_tool_sess_turns["codex"]["sessions"]
    grand_turns = by_tool_sess_turns["claude"]["turns"] + by_tool_sess_turns["codex"]["turns"]

    caveats = [
        "Claude Code cache fields (cache_creation/cache_read) are additive to input_tokens per "
        "Anthropic's accounting and are included in totals.",
        "Claude Code sub-agent transcripts (~/.claude/projects/<project>/<session-id>/subagents/"
        "*.jsonl) are scanned and included in the totals above, counted toward their parent "
        "session rather than as extra sessions; message ids are deduped globally across every "
        "parent and sub-agent file so a forked sub-agent's shared history is never counted twice.",
        "Codex's cached_input_tokens is a SUBSET of input_tokens (OpenAI convention); it is "
        "reported informationally and excluded from codex totals to avoid double counting.",
        "Codex subagent/forked threads inherit the parent thread's cumulative token counter at "
        "fork time; totals here sum each turn's marginal (last_token_usage) delta instead of "
        "cumulative per-session totals, to avoid overcounting shared history.",
        "Daily buckets use each event's UTC timestamp truncated to the day; a session spanning "
        "midnight UTC contributes to more than one day.",
        "Reflects only sessions still present on disk at scan time; anything rotated or deleted "
        "before the scan is not counted.",
    ] + caveats

    daily_list = [
        {"date": d, **v} for d, v in sorted(daily_merged.items())
    ] if daily else []

    ledger = {
        "generated_at": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "since": SINCE_DATE,
        "totals": {
            "tokens": grand_tokens,
            "sessions": grand_sessions,
            "turns": grand_turns,
            "seats": len(seats_seen),
            "subagent_transcripts": subagent_transcripts_total,
        },
        "by_tool": {
            "claude": dict(by_tool["claude"]),
            "codex": dict(by_tool["codex"]),
        },
        "by_model": [
            {"model": model, "tool": tool, "tokens": tok}
            for (model, tool), tok in sorted(by_model.items(), key=lambda x: -x[1])
        ],
        "by_seat": sorted(by_seat, key=lambda x: (x["seat"], x["tool"])),
        "daily": daily_list,
        "caveats": caveats,
    }
    return ledger

if __name__ == "__main__":
    args = sys.argv[1:]
    daily = "--daily" in args
    last_hour = "--last-hour" in args
    if "--merge" in args:
        script_path = os.path.abspath(__file__)
        out_path = "ledger.json"
        if "--out" in args:
            out_path = args[args.index("--out") + 1]
        local_seat = os.environ.get("LEDGER_LOCAL_SEAT", "mac")
        os.environ["LEDGER_SEAT"] = local_seat
        results = [scan(daily=daily, last_hour=last_hour)]
        hosts_spec = os.environ.get("LEDGER_HOSTS", "")
        for pair in [p for p in hosts_spec.split(",") if p.strip()]:
            alias, _, seat = pair.partition("=")
            alias = alias.strip()
            seat = (seat or alias).strip()
            sys.stderr.write(f"scanning seat={seat} via ssh {alias} ...\n")
            results.append(run_remote(alias, seat, daily, script_path, last_hour=last_hour))
        ledger = merge(results, daily)
        with open(out_path, "w") as fh:
            json.dump(ledger, fh, indent=2)
        print(json.dumps(ledger, indent=2))
    else:
        max_seconds = float(args[args.index("--max-seconds") + 1]) if "--max-seconds" in args else None
        if max_seconds is not None and hasattr(signal, "SIGALRM"):
            # Hard backstop for a read that blocks for a long time between
            # budget checks: the alarm raises ScanDeadline wherever the scan is.
            def _alarm(signum, frame):
                raise ScanDeadline()
            signal.signal(signal.SIGALRM, _alarm)
            signal.alarm(max(1, int(max_seconds) + 5))
        try:
            result = scan(daily=daily, last_hour=last_hour, max_seconds=max_seconds)
        except ScanDeadline:
            sys.stderr.write(f"scan stopped at its {max_seconds:g}s budget; progress is cached and "
                             "the next run continues from there\n")
            sys.exit(3)
        print(json.dumps(result, indent=2, default=str))