#!/usr/bin/env python3
"""meter.py — one usage meter for every AI plan, every session, every machine.

Jacob, 2026-09-19: *"a global usage metering and reporting system that all Claude sessions
use … just don't want the system to create too much overhead being sent in and out and
wasting tokens monitoring token usage."*

⭐⭐ THE DESIGN RULE THAT FOLLOWS FROM THAT: a session must never scan transcripts. All
three vendors already write usage to disk, so collection is **passive** — no model call, no
network, nothing sent anywhere. A timer collects; a session reads **one small summary**.

    meter.py                 print the summary (refreshes if stale). ~14 lines. THIS is
                             what a session runs.
    meter.py --collect       rescan the sources and rewrite state + ledger. The timer's job.
    meter.py --read <bucket> <pct> [note]
                             record a percentage a human read off an account page. The
                             vendors that hide their meter need this; it calibrates the rest.
                             Buckets: claude_session claude_week claude_fable grok codex
    meter.py --json          machine-readable state, including the normalised `buckets`
                             list that the website publisher (tools/shop-ai-usage.py) reads.
    meter.py --history       the 7-day hourly series per bucket, as JSON (what the plot gets).

## Where the numbers come from, and how much each can be trusted

| Vendor | Source on disk | Trust |
|---|---|---|
| **Codex** | `~/.codex/sessions/**/rollout-*.jsonl` — the server returns a `rate_limits` block on every request and the CLI writes it down. Every window it names (5-hour, weekly) is read | ⭐ **exact.** A real percentage and reset epoch |
| **Claude** | `~/.claude/projects/**/*.jsonl` — per-message `usage`, **subagent files included** (`<session>/subagents/agent-*.jsonl`); plus any 429's exact `resetsAt` | tokens exact, **percentage not exposed** — needs `--read` |
| **Grok** | `agent-bridge/exchanges/*.md` cost lines | dollars exact, **percentage not exposed** — needs `--read` |

⚠ **A percentage a human reads is the only ground truth for Claude and Grok.** Each
`--read` is stored with the consumption measured at that moment, so the ratio between them
becomes the estimator. Two Grok readings two days apart agreed to within a few cents per
point, which is why its estimate is now a budget rather than a guess.

## The three Claude buckets — 2026-09-19

`/usage` shows THREE meters and they are separate limits: the **5-hour session**, the
**week (all models)**, and **Fable's own week**. Each gets its own `--read`.

⭐ The session bucket is modelled the way Anthropic runs it: **the first message opens a
five-hour block, the block resets five hours after that first message, and the next
message after expiry opens a new one.** Consumption is counted from the block's first
message, not over a rolling five hours — the rolling figure (still logged as
`out_tokens_5h`) under-reads late in a block and never knows when the reset is.
⚠ Two things the transcripts alone get wrong, both found by checking the model against
the five genuine `You've hit your session limit · resets …` 429s on disk (2026-09-13/14/18):
**the block start is floored to ten minutes** (every real reset lands on a :x0), and
**the nightly compile anchors the chain** — it runs `claude -p --no-session-persistence`
at 03:3x, writes no transcript, and still opens the day's first block, which is why
2026-09-18's blocks ran 08:30→13:30 and not 07:12→12:12. Its start times come from
`~/.cache/garage-compile/logs/*.log`. With both, the model hits 5/5; without, 0/5.
⚠ The compile's own tokens are invisible, so a block it opens under-reads a little.
⭐ A genuine 429 whose reset is still in the future is exact truth and overrides the model.
⚠ A block that has expired shows 0% and no reset — the next message starts a fresh one.

## Windows that only a human can read — `windows.conf`

The weekly resets for Claude and Grok are not written to disk by either vendor. They are
read off the account page once and kept in `windows.conf` (`CLAUDE_WEEK_RESET=`,
`GROK_RESET=`, ISO local time); the meter rolls them forward a week at a time. ⚠ Setting
`CLAUDE_WEEK_RESET` changes the week figure from a rolling seven days to "since the window
opened", so take a fresh `--read claude_week` afterwards — the estimator uses the last two
readings, and the old ones were measured on the rolling basis.

## Adding a plan or a receipt

`plans.tsv` and `receipts.tsv` in this folder are plain TSV, edited by hand. A plan gives
the monthly price and the window; a receipt is a one-off purchase (API credits, a top-up).
Nothing here bills anything — it reports what has been consumed against what is paid for.
"""
import argparse, bisect, datetime, json, os, pathlib, re, socket, sys

HERE = pathlib.Path(__file__).resolve().parent
STATE = HERE / "state.json"
LEDGER = HERE / "ledger.tsv"
READS = HERE / "readings.tsv"
PLANS = HERE / "plans.tsv"
RECEIPTS = HERE / "receipts.tsv"
PRICES = HERE / "prices.tsv"
WINDOWS = HERE / "windows.conf"
LEGACY_GROK_CONF = HERE.parent / "grok-window.conf"
HOST = socket.gethostname()
NOW = lambda: datetime.datetime.now()
UTC = datetime.timezone.utc
SESSION_S = 5 * 3600
WEEK_S = 7 * 24 * 3600
# Where a Grok bridge logs its exchanges (see AI tips No. 2), and where a nightly headless
# `claude -p` job logs "compile start YYYY-MM-DD HH:MM" lines. Both optional.
EXCHANGES = pathlib.Path(os.environ.get("AI_METER_GROK_EXCHANGES",
                                        pathlib.Path.home() / "agent-bridge/exchanges"))
COMPILE_LOGS = pathlib.Path(os.environ.get("AI_METER_HEADLESS_LOGS",
                                           pathlib.Path.home() / ".cache/headless-claude/logs"))
BLOCK_FLOOR_S = 600          # the server floors a block's start to ten minutes (5/5 real 429s)
LOOKBACK_DAYS = 14           # a 7-day plot needs a 7-day window at its LEFT edge too


def tsv(p):
    if not p.exists():
        return []
    rows = [l.split("\t") for l in p.read_text().splitlines() if l.strip() and not l.startswith("#")]
    return [dict(zip(rows[0], r)) for r in rows[1:]] if rows else []


def epoch(dt):
    """naive → local; aware → as is."""
    if dt.tzinfo is None:
        dt = dt.astimezone()
    return dt.timestamp()


# ── windows a human read off an account page ───────────────────────────────────────
def conf_reset(key, asof=None):
    """the first reset for `key` after `asof` (default now), rolled a week at a time from
    the recorded one — so a history point last Tuesday measures against last week's
    window, not this week's."""
    asof_dt = datetime.datetime.fromtimestamp(asof) if asof else NOW()
    for p in (WINDOWS, LEGACY_GROK_CONF):
        if not p.exists():
            continue
        m = re.search(rf"^{key}=(\S+)", p.read_text(), re.M)
        if not m:
            continue
        try:
            rs = datetime.datetime.fromisoformat(m.group(1))
        except ValueError:
            continue
        while rs <= asof_dt:
            rs += datetime.timedelta(days=7)
        while rs - datetime.timedelta(days=7) > asof_dt:
            rs -= datetime.timedelta(days=7)
        return rs
    return None


# ── collectors — all passive file reads ────────────────────────────────────────────
def claude_events(days=LOOKBACK_DAYS):
    """every assistant message in the last `days`, sorted:
    [(epoch, out_tokens, is_fable, in, cache_read, cache_write, model)], plus the newest
    genuine 429 per limit type: {"five_hour": (epoch_of_429, reset_epoch), "seven_day": …}.
    Genuine means the record carries a rejected quotaLimits block — a transcript that
    merely quotes the words "session limit" (a grep result, a pasted note) is not a 429
    and was fooling the first version of this."""
    cut = datetime.datetime.now(UTC) - datetime.timedelta(days=days)
    ev, reset = [], {}
    # ⚠ rglob, not glob: subagents write to projects/<proj>/<session>/subagents/agent-*.jsonl,
    # one level below the session files. Missed until 2026-09-19 — 1.04M Fable output
    # tokens in a day were invisible, and Fable read 28% against a meter that saw zero.
    for f in (pathlib.Path.home() / ".claude/projects").rglob("*.jsonl"):
        try:
            if datetime.datetime.fromtimestamp(f.stat().st_mtime, UTC) < cut:
                continue
        except OSError:
            continue
        for line in f.open(errors="ignore"):
            # A rejected request writes quotaLimits with an exact reset epoch and the
            # limit's type (five_hour / seven_day). Only 429s carry the block — and the
            # record also has a zeroed usage, so this check comes before the usage one.
            if '"quotaLimits"' in line and '"rejected"' in line:
                try:
                    d = json.loads(line)
                    q = d.get("quotaLimits") or {}
                    at = datetime.datetime.fromisoformat(d["timestamp"].replace("Z", "+00:00")).timestamp()
                    if q.get("status") == "rejected" and q.get("resetsAt"):
                        kind = q.get("rateLimitType") or "five_hour"
                        if kind not in reset or at > reset[kind][0]:
                            reset[kind] = (at, float(q["resetsAt"]))
                except (ValueError, KeyError, TypeError):
                    pass
                continue
            if '"usage"' not in line:
                continue
            try:
                d = json.loads(line)
            except Exception:
                continue
            ts = d.get("timestamp")
            if not ts:
                continue
            try:
                when = datetime.datetime.fromisoformat(ts.replace("Z", "+00:00"))
            except ValueError:
                continue
            if when < cut:
                continue
            msg = d.get("message") or {}
            u = msg.get("usage") or d.get("usage")
            if not isinstance(u, dict):
                continue
            mdl = (msg.get("model") or "?")
            ev.append((when.timestamp(), u.get("output_tokens", 0) or 0, "fable" in mdl,
                       u.get("input_tokens", 0) or 0, u.get("cache_read_input_tokens", 0) or 0,
                       u.get("cache_creation_input_tokens", 0) or 0, mdl.replace("claude-", "")))
    # The nightly compile opens a block and leaves no transcript: anchor on its log line.
    for p in COMPILE_LOGS.glob("*.log") if COMPILE_LOGS.exists() else []:
        m = re.search(r"compile start (\d{4}-\d{2}-\d{2} \d{2}:\d{2})", p.read_text(errors="ignore"))
        if m:
            t = epoch(datetime.datetime.strptime(m.group(1), "%Y-%m-%d %H:%M"))
            if t >= cut.timestamp():
                ev.append((t, 0, False, 0, 0, 0, "compile"))
    ev.sort()
    return ev, reset


class Claude:
    """the event list with prefix sums and the block chain, so 168 hourly recomputes for
    the plot cost bisects rather than walks."""
    def __init__(self, ev):
        self.ev = ev
        self.t = [e[0] for e in ev]
        self.cum = [0]; self.cum_f = [0]
        for e in ev:
            self.cum.append(self.cum[-1] + e[1])
            self.cum_f.append(self.cum_f[-1] + (e[1] if e[2] else 0))
        # block starts: the first message opens one; the next message at or after
        # start+5h opens the next. Starts are floored to ten minutes, as the server does.
        self.starts, self.start_idx = [], []
        start = None
        for i, e in enumerate(ev):
            if start is None or e[0] >= start + SESSION_S:
                start = e[0] - (e[0] % BLOCK_FLOOR_S)
                self.starts.append(start); self.start_idx.append(i)

    def _hi(self, asof):
        return bisect.bisect_right(self.t, asof)

    def window_sum(self, asof, seconds, fable=None):
        lo, hi = bisect.bisect_right(self.t, asof - seconds), self._hi(asof)
        c = self.cum_f if fable else self.cum
        tot = c[hi] - c[lo]
        return (self.cum[hi] - self.cum[lo]) - tot if fable is False else tot

    def session_block(self, asof):
        """(block_start_epoch, out_tokens_since_start) for the block open at `asof`."""
        k = bisect.bisect_right(self.starts, asof) - 1
        if k < 0:
            return None, 0
        lo, hi = self.start_idx[k], self._hi(asof)
        return self.starts[k], self.cum[hi] - self.cum[lo]


def session_block(ev, asof):
    return Claude(ev).session_block(asof)


def window_sum(ev, asof, seconds, fable=None):
    return Claude(ev).window_sum(asof, seconds, fable)


def claude_totals(ev, since):
    tot = {"in": 0, "out": 0, "cache_read": 0, "cache_write": 0, "msgs": 0, "by_model": {}}
    for t, out, _, inp, cr, cw, mdl in ev:
        if t < since:
            continue
        tot["msgs"] += 1
        tot["in"] += inp; tot["out"] += out
        tot["cache_read"] += cr; tot["cache_write"] += cw
        tot["by_model"][mdl] = tot["by_model"].get(mdl, 0) + out
    return tot


def codex_snapshots(days=LOOKBACK_DAYS):
    """every rate-limit block Codex wrote in the last `days`, one row per window:
    sorted [(epoch, window_min, used_pct, resets_epoch)]."""
    cut = datetime.datetime.now(UTC) - datetime.timedelta(days=days)
    snaps = []
    for f in pathlib.Path.home().glob(".codex/sessions/*/*/*/rollout-*.jsonl"):
        try:
            if datetime.datetime.fromtimestamp(f.stat().st_mtime, UTC) < cut:
                continue
        except OSError:
            continue
        for line in f.open(errors="ignore"):
            if '"used_percent"' not in line:
                continue
            try:
                d = json.loads(line)
                when = datetime.datetime.fromisoformat(d["timestamp"].replace("Z", "+00:00"))
                rl = (d.get("payload") or {}).get("rate_limits") or d.get("rate_limits") or {}
            except Exception:
                continue
            if when < cut:
                continue
            for k in ("primary", "secondary"):
                w = rl.get(k)
                if isinstance(w, dict) and "used_percent" in w and w.get("window_minutes"):
                    snaps.append((when.timestamp(), int(w["window_minutes"]),
                                  float(w["used_percent"]), int(w.get("resets_at") or 0)))
    snaps.sort()
    return snaps


def codex_latest(snaps):
    """the newest snapshot per window, as {window_min: {pct, resets, at}}."""
    out = {}
    for t, wm, pct, rs in snaps:
        out[wm] = {"pct": pct, "resets": rs, "at": t, "window_min": wm}
    return out


def codex_usage(days=LOOKBACK_DAYS):
    """every turn's token usage Codex wrote: sorted [(epoch, input, cached_input, output)].
    cached_input is a subset of input, as OpenAI counts it."""
    cut = datetime.datetime.now(UTC) - datetime.timedelta(days=days)
    rows = []
    for f in pathlib.Path.home().glob(".codex/sessions/*/*/*/rollout-*.jsonl"):
        try:
            if datetime.datetime.fromtimestamp(f.stat().st_mtime, UTC) < cut:
                continue
        except OSError:
            continue
        for line in f.open(errors="ignore"):
            if '"last_token_usage"' not in line:
                continue
            try:
                d = json.loads(line)
                when = datetime.datetime.fromisoformat(d["timestamp"].replace("Z", "+00:00"))
                u = d["payload"]["info"]["last_token_usage"]
            except (KeyError, TypeError, ValueError):
                continue
            if when < cut:
                continue
            rows.append((when.timestamp(), u.get("input_tokens", 0) or 0,
                         u.get("cached_input_tokens", 0) or 0, u.get("output_tokens", 0) or 0))
    rows.sort()
    return rows


def grok_events():
    """sorted [(epoch, usd)] — one per consult that logged a cost."""
    ev = []
    for f in sorted(EXCHANGES.glob("*.md")) if EXCHANGES.exists() else []:
        try:
            when = datetime.datetime.strptime(f.name[:13], "%Y%m%d-%H%M")
        except ValueError:
            continue
        m = re.search(r"\*\*cost:\*\*\s*\$([\d.]+)", f.read_text(errors="ignore"))
        if m:
            ev.append((epoch(when), float(m.group(1))))
    ev.sort()
    return ev


def grok_sum(ev, since, asof):
    n = usd = 0
    for t, u in ev:
        if since < t <= asof:
            usd += u; n += 1
    return round(usd, 4), n


# ── estimate a hidden percentage from readings ─────────────────────────────────────
_CALIB = {}


def calib(bucket):
    """(fn consumed→pct, per_percent) from the last two human readings, or None."""
    if bucket in _CALIB:
        return _CALIB[bucket]
    _CALIB[bucket] = _calib(bucket)
    return _CALIB[bucket]


def _calib(bucket):
    """Anchored on the newest reading — at the instant Jacob read the page the estimate
    IS the reading — with the slope (consumed per percent) fitted by least squares over
    the last four readings. Two readings a few points apart used to set the slope on
    their own, and a 3-point gap turned $0.22/pt into $0.36/pt; a fit over more of them
    is steadier. One reading alone falls back to the ratio through the origin."""
    rows = [r for r in tsv(READS) if r.get("vendor") == bucket]
    pts = []
    for r in sorted(rows, key=lambda r: r["when"]):
        try:
            pts.append((float(r["pct"]), float(r["consumed"])))
        except (ValueError, KeyError):
            pass
    if not pts:
        return None
    # ⚠ A MID-WINDOW RESET. On 2026-09-22 Anthropic zeroed the week for a promotion
    #   without moving the Friday reset: 34% at 9.0 M tokens, then 0% at 10.5 M. A
    #   percentage that FALLS while consumption RISES can only be a reset, so the fit
    #   restarts there. The slope is a property of the plan, not of the window, so the
    #   pre-reset fit is carried across until the new segment has two readings of its own.
    segs = [[pts[0]]]
    for a, b in zip(pts, pts[1:]):
        if b[0] < a[0] and b[1] >= a[1]:
            segs.append([])
        segs[-1].append(b)
    carried = None
    for seg in segs[:-1]:
        carried = _slope(seg) or carried
    pts = segs[-1]
    pa, ca = pts[-1]
    per = _slope(pts) or (carried if len(segs) > 1 else None)
    if per is None and pa > 0 and ca > 0:
        per = ca / pa
    if per is None:
        # One reading and nothing consumed on this basis yet — which happens when the
        # usage was somewhere this machine cannot see (the claude.ai app counts against
        # the same plan; Fable read 28% on 2026-09-19 with zero Fable tokens on the Pi or
        # the TUF since the window opened). The reading is still the truth at its
        # instant, so hold it flat until a second reading gives a slope.
        return (lambda c: pa), None
    return (lambda c: pa + (c - ca) / per), per


def _slope(pts):
    """consumed per percent, least squares over the last four readings, or None."""
    per = None
    if len(pts) >= 2:
        tail = pts[-4:]
        pm = sum(p for p, _ in tail) / len(tail); cm = sum(c for _, c in tail) / len(tail)
        sxx = sum((p - pm) ** 2 for p, _ in tail)
        sxy = sum((p - pm) * (c - cm) for p, c in tail)
        if sxx > 0 and sxy / sxx > 0:
            per = sxy / sxx
    return per


def estimate(bucket, consumed):
    c = calib(bucket)
    return (c[0](consumed), c[1]) if c else None


# ── the normalised view: one row per limit ─────────────────────────────────────────
def label_window(minutes):
    return {300: "5-hour session", 10080: "week"}.get(minutes, f"{minutes/60:.0f}-hour")


def buckets(cl, snaps, gev, asof=None, reset_429=None):
    """one dict per limit. `used` is a percentage or None; `exact` says whether the
    vendor wrote it down or a human reading is being extrapolated. `cl` is a Claude()."""
    now = asof or NOW().timestamp()
    out = []

    def iso(ts):
        return datetime.datetime.fromtimestamp(ts).astimezone().isoformat(timespec="seconds") if ts else None

    # Claude — session block. A genuine 429 whose reset is still ahead is the one exact
    # thing Anthropic tells us about this bucket, so it wins over the model.
    start, sess = cl.session_block(now)
    active = start is not None and now < start + SESSION_S
    e = estimate("claude_session", sess) if active else (0.0, None)
    row = {"id": "claude-session", "vendor": "Claude", "window": "5-hour session",
           "hours": 5, "used": e[0] if e else None, "exact": False,
           "resets": iso(start + SESSION_S) if active else None,
           "consumed": sess if active else 0, "unit": "out tokens",
           "per": e[1] if e else None, "block_start": iso(start) if start else None}
    r5 = (reset_429 or {}).get("five_hour")
    if r5 and r5[0] <= now < r5[1]:
        row.update({"used": 100.0, "exact": True, "resets": iso(r5[1]),
                    "note": "session limit hit — from the 429 itself"})
    out.append(row)

    # Claude — week, all models, and Fable's own week
    wk_reset = conf_reset("CLAUDE_WEEK_RESET", now)
    r7 = (reset_429 or {}).get("seven_day")
    if wk_reset is None and r7:                    # a weekly 429 pins the week exactly
        wk_reset = datetime.datetime.fromtimestamp(r7[1]).astimezone()
        while wk_reset.timestamp() <= now:
            wk_reset += datetime.timedelta(days=7)
    since = epoch(wk_reset) - WEEK_S if wk_reset else now - WEEK_S
    for bid, fable, win in (("claude-week", None, "week"), ("claude-fable", True, "Fable week")):
        c = cl.window_sum(now, now - since, fable=fable)
        e = estimate("claude_week" if fable is None else "claude_fable", c)
        out.append({"id": bid, "vendor": "Claude", "window": win, "hours": 168,
                    "used": e[0] if e else None, "exact": False,
                    "resets": wk_reset.astimezone().isoformat(timespec="seconds") if wk_reset else None,
                    "consumed": c, "unit": "out tokens", "per": e[1] if e else None,
                    "basis": "since window opened" if wk_reset else "rolling 7 days"})

    # Codex — every window the server names, exactly
    for wm, s in sorted(codex_latest([x for x in snaps if x[0] <= now]).items(), key=lambda kv: kv[0]):
        expired = s["resets"] and s["resets"] <= now
        out.append({"id": f"codex-{'session' if wm == 300 else 'week' if wm == 10080 else wm}",
                    "vendor": "Codex", "window": label_window(wm), "hours": wm / 60,
                    "used": 0.0 if expired else s["pct"], "exact": True,
                    "resets": None if expired else iso(s["resets"]),
                    "consumed": s["pct"], "unit": "%", "per": 1.0,
                    "note": "no use since the window reset" if expired else None})

    # Grok — the week, in dollars, against a human reading
    g_reset = conf_reset("GROK_RESET", now)
    g_since = epoch(g_reset) - WEEK_S if g_reset else now - WEEK_S
    usd, n = grok_sum(gev, g_since, now)
    e = estimate("grok", usd)
    out.append({"id": "grok-week", "vendor": "Grok", "window": "week", "hours": 168,
                "used": e[0] if e else None, "exact": False,
                "resets": g_reset.astimezone().isoformat(timespec="seconds") if g_reset else None,
                "consumed": usd, "unit": "usd", "per": e[1] if e else None, "consults": n})

    # A plan with an `ends` date in plans.tsv stops being a limit the day after it ends —
    # otherwise a cancelled Codex would sit on the website at "0% used" forever.
    ended = set()
    for pl in tsv(PLANS):
        e = (pl.get("ends") or "").strip()
        if e and e < datetime.datetime.fromtimestamp(now).strftime("%Y-%m-%d"):
            ended.add({"anthropic": "Claude", "xai": "Grok", "openai": "Codex"}.get(pl.get("vendor"), pl.get("vendor")))
    out = [b for b in out if b["vendor"] not in ended]
    for b in out:
        if b["used"] is not None:
            b["used"] = round(max(0.0, min(100.0, b["used"])), 1)
    return out


# Every Codex rollout on this Pi names this model (2450 of 2450, 2026-09-23). If a rollout ever
# names another, price per rollout instead of changing this constant.
CODEX_MODEL = "gpt-6-astra"


def price_for(model, table):
    """the prices.tsv row whose `match` is in the model id, longest match first."""
    best = None
    for r in table:
        m = r.get("match", "")
        if m and m in model and (best is None or len(m) > len(best["match"])):
            best = r
    return best


def value(cl, cu, gev, asof=None, days=7):
    """per plan, over the last `days`: what the subscription cost for that span, the tokens
    it bought, what the same tokens would have cost at API list prices, and the ratio.
    ⭐ Jacob 2026-09-19: "a $ per token figure … to help me maximize the money I'm spending
    on monthly subscriptions." Rolling seven days, not the plan's own window, so the figure
    is comparable across vendors regardless of where each one is in its week."""
    now = asof or NOW().timestamp()
    since = now - days * 86400
    plans = {p.get("vendor"): p for p in tsv(PLANS)}
    prices = tsv(PRICES)
    out = []

    def row(vendor, plan, tokens, api_usd, note=None):
        try:
            month = float(plan.get("price_month") or 0)
        except ValueError:
            month = 0.0
        paid = month * 12 / 365.25 * days
        total = sum(v for k, v in tokens.items() if k != "cached_pct") if tokens else 0
        r = {"vendor": vendor, "plan": plan.get("name"), "price_month": month, "paid": round(paid, 2),
             "days": days, "tokens": tokens, "api_usd": round(api_usd, 2) if api_usd is not None else None,
             "value_x": round(api_usd / paid, 2) if api_usd is not None and paid > 0 else None,
             "usd_per_mtok": round(paid / (total / 1e6), 4) if total and paid > 0 else None,
             "usd_per_mtok_out": round(paid / (tokens["out"] / 1e6), 2) if tokens and tokens.get("out") and paid > 0 else None,
             "api_usd_per_mtok": round(api_usd / (total / 1e6), 4) if api_usd is not None and total else None,
             "note": note}
        out.append(r)

    # Claude — every token type, priced per model
    if "anthropic" in plans:
        t = {"in": 0, "out": 0, "cache_read": 0, "cache_write": 0}
        usd, unpriced = 0.0, set()
        for e in cl.ev:
            if not (since < e[0] <= now) or e[6] == "compile":
                continue
            _, o, _, i, cr, cw, mdl = e
            t["in"] += i; t["out"] += o; t["cache_read"] += cr; t["cache_write"] += cw
            pr = price_for(mdl, prices)
            if pr:
                usd += (i * float(pr["in"]) + o * float(pr["out"]) + cr * float(pr["cache_read"])
                        + cw * float(pr["cache_write"])) / 1e6
            elif mdl not in ("?", "<synthetic>"):
                unpriced.add(mdl)
        tot = sum(t.values())
        t["cached_pct"] = round(100 * t["cache_read"] / tot, 1) if tot else None
        row("Claude", plans["anthropic"], t, usd,
            ("unpriced models ignored: " + ", ".join(sorted(unpriced))) if unpriced else None)

    # Codex — tokens exact. The usage rows carry no model id, so they are priced at the one
    # model every rollout on this Pi names (CODEX_MODEL); short-context rates, so a floor.
    if "openai" in plans:
        t = {"in": 0, "cache_read": 0, "out": 0}
        for ts, i, c, o in cu:
            if since < ts <= now:
                t["in"] += i - c; t["cache_read"] += c; t["out"] += o
        tot = sum(t.values())
        t["cached_pct"] = round(100 * t["cache_read"] / tot, 1) if tot else None
        pr = price_for(CODEX_MODEL, prices)
        usd = ((t["in"] * float(pr["in"]) + t["cache_read"] * float(pr["cache_read"])
                + t["out"] * float(pr["out"])) / 1e6) if pr else None
        row("Codex", plans["openai"], t, usd,
            f"priced as {CODEX_MODEL}, short-context rates — a floor" if pr
            else "no API price on file for its model")

    # Grok — the exchanges log xAI's own cost per consult; tokens are not logged
    if "xai" in plans:
        usd = sum(u for ts, u in gev if since < ts <= now)
        row("Grok", plans["xai"], None, usd, "tokens not logged by Grok Build; the $ is its own per-consult cost")
    return out


def history(cl, snaps, gev, hours=168, step=3600, asof=None):
    """per-bucket series of `used` at the end of each hourly bucket for the last 7 days.
    Estimated buckets are recomputed at every hour from the event lists; exact ones
    (Codex) carry the last snapshot forward inside a reset window and are None where no
    snapshot exists yet. Returns (t0, {bucket_id: [pct|None]})."""
    now = asof or NOW().timestamp()
    end = int(now) // step * step
    t0 = end - (hours - 1) * step
    series = {}
    n_pts = hours
    for i in range(n_pts):
        # The last point is NOW, not the top of this hour, so the plot's right-hand
        # end and the meters above it are the same number at the same instant.
        t = now if i == n_pts - 1 else min(t0 + i * step, now)
        for b in buckets(cl, snaps, gev, asof=t):
            if b["id"] == "claude-session":
                continue                      # a five-hour sawtooth is noise on a 7-day axis
            series.setdefault(b["id"], [None] * n_pts)
            series[b["id"]][i] = b["used"]
    # An estimated bucket is anchored on readings taken in THIS window; extrapolating the
    # previous window from them drew a week that never happened (a flat 100% before the
    # Friday reset, with no weekly 429 on disk to back it). So an estimate starts where
    # its current window opened; before that is a break, not a line.
    for b in buckets(cl, snaps, gev):
        if b["exact"] or not b.get("resets") or b["id"] not in series:
            continue
        opened = datetime.datetime.fromisoformat(b["resets"]).timestamp() - b["hours"] * 3600
        pts = series[b["id"]]
        for i in range(n_pts):
            t = now if i == n_pts - 1 else min(t0 + i * step, now)
            if t < opened:
                pts[i] = None
    # Codex is exact but only at the instants it was used. Between snapshots the level is
    # the last one seen (usage cannot change without a call, which writes a snapshot); a
    # window that has reset is 0 with no snapshot to prove it, so it stays None until the
    # next call.
    for wm, sid in ((300, "codex-session"), (10080, "codex-week")):
        pts = [None] * n_pts
        rows = [s for s in snaps if s[1] == wm]
        for i in range(n_pts):
            t = now if i == n_pts - 1 else min(t0 + i * step, now)
            last = None
            for s in rows:
                if s[0] <= t:
                    last = s
                else:
                    break
            if last is not None and not (last[3] and last[3] <= t):
                pts[i] = last[2]
        if rows:
            series[sid] = pts
    return t0, series


class Scan:
    """one pass over every source. The publisher and collect() share it so a run reads
    the transcripts once, not twice."""
    def __init__(self):
        self.ev, self.reset = claude_events()
        self.cl = Claude(self.ev)
        self.snaps = codex_snapshots()
        self.cu = codex_usage()
        self.gev = grok_events()

    def value(self, asof=None):
        return value(self.cl, self.cu, self.gev, asof=asof)

    def buckets(self, asof=None):
        return buckets(self.cl, self.snaps, self.gev, asof=asof, reset_429=self.reset)

    def history(self, **kw):
        return history(self.cl, self.snaps, self.gev, **kw)


def collect(sc=None):
    sc = sc or Scan()
    ev, reset, cl, snaps, gev = sc.ev, sc.reset, sc.cl, sc.snaps, sc.gev
    now = NOW().timestamp()
    bk = sc.buckets()
    by = {b["id"]: b for b in bk}
    cx = codex_latest(snaps)
    start, sess = cl.session_block(now)
    wk = claude_totals(ev, now - WEEK_S)
    wk["last_429"] = {k: {"at": datetime.datetime.fromtimestamp(v[0]).astimezone().isoformat(timespec="seconds"),
                          "resets": datetime.datetime.fromtimestamp(v[1]).astimezone().isoformat(timespec="seconds")}
                      for k, v in reset.items()}
    g_reset = conf_reset("GROK_RESET")
    st = {
        "host": HOST,
        "at": NOW().isoformat(timespec="seconds"),
        # legacy keys, kept so anything reading the old shape still works
        "codex": ({"pct": cx[10080]["pct"], "window_min": 10080, "resets": cx[10080]["resets"]}
                  if 10080 in cx else (next(iter(cx.values())) if cx else None)),
        "codex_windows": cx,
        "claude_week": wk,
        "claude_session": {"out": sess, "block_start": by["claude-session"]["block_start"],
                           "resets": by["claude-session"]["resets"],
                           "out_rolling_5h": cl.window_sum(now, SESSION_S)},
        "grok": {"usd": by["grok-week"]["consumed"], "consults": by["grok-week"]["consults"]},
        "grok_window_start": (g_reset - datetime.timedelta(days=7)).strftime("%Y%m%d%H%M") if g_reset else None,
        "grok_resets": g_reset.isoformat() if g_reset else None,
        "buckets": bk,
        "value": sc.value(),
    }
    STATE.write_text(json.dumps(st, indent=1))
    hdr = not LEDGER.exists()
    with LEDGER.open("a") as f:
        if hdr:
            f.write("ts\thost\tvendor\tmetric\tvalue\n")
        rows = [("codex", "pct", (st["codex"] or {}).get("pct")),
                ("claude", "out_tokens_week", by["claude-week"]["consumed"]),
                ("claude", "out_tokens_5h", st["claude_session"]["out_rolling_5h"]),
                ("claude", "out_tokens_session", sess),
                ("claude", "out_tokens_fable_week", by["claude-fable"]["consumed"]),
                ("grok", "usd_window", st["grok"]["usd"])]
        if 300 in cx:
            rows.append(("codex", "pct_5h", cx[300]["pct"]))
        for v, k, val in rows:
            if val is not None:
                f.write(f"{st['at']}\t{HOST}\t{v}\t{k}\t{val}\n")
    return st


def money(plan_rows, vendor, pct):
    for p in plan_rows:
        if p.get("vendor") == vendor or p.get("name", "").startswith(vendor):
            try:
                price = float(p["price_month"])
                return price * (pct / 100.0) if pct is not None else None
            except (ValueError, KeyError):
                return None
    return None


def fresh_state(max_age_s=1200):
    if not STATE.exists():
        return collect()
    st = json.loads(STATE.read_text())
    if "buckets" not in st or (NOW() - datetime.datetime.fromisoformat(st["at"])).total_seconds() > max_age_s:
        return collect()
    return st


def report(as_json=False):
    st = fresh_state()
    if as_json:
        print(json.dumps(st, indent=1)); return 0
    plans = tsv(PLANS)
    B, D, R, Y, RED = "\033[1m", "\033[2m", "\033[0m", "\033[33m", "\033[31m"
    out = [f"{B}AI USAGE{R} {D}· {st['host']} · {st['at'][:16]}{R}"]
    by = {b["id"]: b for b in st["buckets"]}

    def when(iso):
        if not iso:
            return ""
        dt = datetime.datetime.fromisoformat(iso)
        hrs = (dt - datetime.datetime.now(dt.tzinfo)).total_seconds() / 3600
        return f"resets {dt:%a %H:%M} ({hrs:.0f}h)"

    for bid in sorted(by):
        if bid.startswith("codex"):
            b = by[bid]
            left = 100 - b["used"]
            col = RED if left < 10 else Y if left < 25 else ""
            spend = money(plans, "openai", b["used"])
            out.append(f"  codex   {col}{left:5.1f}% left{R}  {b['window']:<15} {when(b['resets']) or b.get('note') or ''}"
                       + (f"  {D}≈${spend:.0f} of plan{R}" if spend else ""))

    s, w, fb = by["claude-session"], by["claude-week"], by["claude-fable"]
    pct = lambda b, tag: f"{b['used']:.0f}% {tag}" if b["used"] is not None else f"{tag} ?"
    spend = money(plans, "anthropic", w["used"])
    out.append(f"  claude  {pct(w, 'wk'):>10}  {w['consumed']/1e6:.1f}M out {w.get('basis', 'this week')}"
               + (f"  {D}≈${spend:.0f} of plan{R}" if spend else "")
               + (f"  {D}{when(w['resets'])}{R}" if w["resets"] else f"  {D}(week reset unknown — set CLAUDE_WEEK_RESET){R}"))
    out.append(f"  {'':8}{pct(s, '5h'):>10}  {s['consumed']/1e6:.2f}M out since {s['block_start'][11:16] if s['block_start'] else '—'}"
               + (f"  {D}{when(s['resets'])}{R}" if s["resets"] else f"  {D}block expired — next message opens one{R}"))
    out.append(f"  {'':8}{pct(fb, 'Fable'):>10}  {fb['consumed']/1e6:.1f}M out")

    g = by["grok-week"]
    if g["used"] is not None and g["per"]:
        gp, per = g["used"], g["per"]
        # ⚠ Jacob raised the stop from 90 to 95 on 2026-09-19. It is HIS line, not a
        # vendor limit — the plan does not cut off, so crossing it is a decision, not an error.
        STOP = 95
        stop = (STOP - gp) * per
        col = RED if gp >= STOP else Y if gp >= STOP - 15 else ""
        out.append(f"  grok    {col}{gp:5.0f}% used{R}  ${g['consumed']:.2f} over {g['consults']} consults  {D}{when(g['resets'])}{R}")
        out.append(f"          {RED + f'⛔ AT THE {STOP}% STOP LINE' + R if stop <= 0 else Y + f'${stop:.2f} to the {STOP}% stop' + R}")
    else:
        out.append(f"  grok    ${g['consumed']:.2f} over {g['consults']} consults  {D}(no reading yet){R}")

    for v in st.get("value") or []:
        tk = v.get("tokens") or {}
        tot = sum(x for k, x in tk.items() if k != "cached_pct") if tk else 0
        bits = [f"${v['paid']:.0f} paid this week"]
        if v.get("api_usd") is not None:
            bits.append(f"${v['api_usd']:.0f} at API prices → {B}{v['value_x']:.1f}×{R}")
        if v.get("usd_per_mtok") is not None:
            bits.append(f"${v['usd_per_mtok']:.3f}/Mtok on {tot/1e9:.1f}B tokens" if tot >= 1e9 else f"${v['usd_per_mtok']:.3f}/Mtok on {tot/1e6:.0f}M tokens")
        if v.get("usd_per_mtok_out") is not None:
            bits.append(f"${v['usd_per_mtok_out']:.2f}/M out")
        out.append(f"  {D}{v['vendor'].lower():<7} {' · '.join(bits)}{R}")
    rec = tsv(RECEIPTS)
    if rec:
        tot = sum(float(r.get("amount_usd", 0) or 0) for r in rec)
        out.append(f"  {D}receipts on file: {len(rec)} · ${tot:.2f}{R}")
    out.append(f"  {D}meter.py --read <claude_session|claude_week|claude_fable|grok|codex> <pct> to calibrate{R}")
    print("\n".join(out))
    return 0


READ_BUCKETS = ("claude_session", "claude_week", "claude_fable", "grok", "codex")   # what --read accepts


def record(vendor, pct, note):
    # Validate BEFORE collect(): a bad bucket or percentage used to cost a full transcript
    # scan (a minute) and only then fail, with the error coming out of the lookup instead.
    if vendor not in READ_BUCKETS:
        sys.exit(f"unknown bucket {vendor} — {', '.join(READ_BUCKETS)} are what --read accepts")
    if not 0 <= pct <= 100:
        sys.exit(f"percentage out of range: {pct} — must be 0–100")
    st = collect()
    by = {b["id"]: b for b in st["buckets"]}
    consumed = {"claude_week": by["claude-week"]["consumed"],
                "claude_session": by["claude-session"]["consumed"],
                "claude_fable": by["claude-fable"]["consumed"],
                "grok": by["grok-week"]["consumed"],
                "codex": (st["codex"] or {}).get("pct", 0)}[vendor]
    hdr = not READS.exists()
    with READS.open("a") as f:
        if hdr:
            f.write("when\tvendor\tpct\tconsumed\tnote\n")
        f.write(f"{NOW():%Y-%m-%d %H:%M}\t{vendor}\t{pct}\t{consumed}\t{note}\n")
    print(f"recorded: {vendor} {pct}% at consumed={consumed}")
    _CALIB.clear()                                # the first collect() cached the OLD calibration
    collect()                                     # so the estimate reflects the new point at once


def undo(bucket):
    """remove the most recent reading for `bucket` from readings.tsv — the inverse of a
    mistaken `--read`. A pure file edit: no transcript scan, no collect(), back instantly.
    The header and every other line are left byte-for-byte unchanged."""
    if bucket not in READ_BUCKETS:
        sys.exit(f"unknown bucket {bucket} — {', '.join(READ_BUCKETS)} are what --read accepts")
    if not READS.exists():
        sys.exit(f"nothing to undo — no readings on file for {bucket}")
    lines = READS.read_text().splitlines(keepends=True)
    hdr = lines[0].split("\t") if lines else []
    v = hdr.index("vendor") if "vendor" in hdr else None
    i = None
    if v is not None:
        for k in range(len(lines) - 1, 0, -1):      # newest is the last one in the file
            cols = lines[k].split("\t")
            if len(cols) > v and cols[v] == bucket:
                i = k
                break
    if i is None:
        sys.exit(f"nothing to undo — no readings on file for {bucket}")
    removed = lines[i]
    READS.write_text("".join(lines[:i] + lines[i + 1:]))
    print(f"undid: {removed.rstrip()}")


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--collect", action="store_true")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--history", action="store_true", help="7-day hourly series per bucket, JSON")
    ap.add_argument("--read", nargs="+", metavar=("BUCKET PCT", "NOTE"))
    ap.add_argument("--undo", metavar="BUCKET", help="remove the most recent reading for BUCKET")
    a = ap.parse_args()
    if a.read:
        if len(a.read) < 2:
            sys.exit("usage: --read BUCKET PCT [note]")
        try:
            pct = float(a.read[1])
        except ValueError:
            sys.exit(f"not a percentage: {a.read[1]}")
        record(a.read[0], pct, " ".join(a.read[2:]) or "read off the account page")
    elif a.undo:
        undo(a.undo)
    elif a.collect:
        collect(); print("collected")
    elif a.history:
        t0, s = Scan().history()
        print(json.dumps({"t0": t0, "step": 3600, "series": s}, indent=1))
    else:
        sys.exit(report(a.json))