#!/usr/bin/env python3
"""omnigauge — one gauge for every AI agent and API account you run.

Reads plan quota straight from each CLI and token volume straight from the
transcripts those tools already write to disk. Keyless and offline for quota
and tokens. The optional spend panel is the exception: vendor keys, stored
0600 and never passed as arguments, used for the only network calls it makes.
No telemetry either way.

  omnigauge                 dashboard (cached quota + live token counts)
  omnigauge --refresh       re-scrape plan quota from every installed CLI
  omnigauge --refresh claude,codex
  omnigauge --lifetime      include all-time totals (incremental, cached)
  omnigauge --since 7d      24h | 7d | today | all
  omnigauge --watch         live redraw (tokens every tick, quota on a slower clock)
  omnigauge --watch 5 --quota-every 10m
  omnigauge --json          machine-readable
  omnigauge --scan-roots    hunt mounted drives for stores discovery misses
  omnigauge --no-color

STORES ON ANOTHER DRIVE
  Add a HOME-like path (the directory holding .claude/.codex/...) one per line
  to DATA/roots, or set OMNIGAUGE_ROOTS. Roots are scanned like a second home;
  when the drive is out, the board and --check SAY its history is excluded
  instead of quietly shrinking.

REFRESH RATES ARE NOT THE SAME FOR BOTH HALVES
  token volume  read from local files — cheap, so it can update every few seconds.
  plan quota    requires launching the vendor's TUI and reading its panel, which
                costs ~30s per agent and spawns a real session. It is cached and
                refreshed on a slow clock; anything faster would be pure waste.

TWO KINDS OF NUMBER, DELIBERATELY NEVER MERGED

  plan quota   how much of a subscription window is consumed. No vendor caches
               this to disk, so it is scraped from each CLI's own usage panel.
               Normalized to PERCENT CONSUMED — Codex reports percent REMAINING
               natively and is inverted here on purpose. Shown side by side raw,
               a 6%-left Codex looks healthier than a 23%-used Claude.

  token volume counted from local transcripts. Free, exact, never stale — and
               NOT comparable to the vendors' own figures. Different
               denominators: cache reads, per-turn context re-sends, total vs
               billed. Two honest numbers rather than one reconciled fiction.

Subscription plans have no dollar balance, so none is shown. API spend is a
separate product and is never blended into a "remaining" figure.

MIT licensed. https://github.com/omnigauge/omnigauge
Written by Claude Opus 5.0. Maintained by Claude Fable 5.
"""
import argparse, glob, io, json, os, re, shutil, sqlite3, subprocess, sys, time

APP = "omnigauge"

# The donation address, in one place. It appears on the board, on the site and
# in the README; when the vanity address finishes grinding, a wrong edit here
# sends money to a dead wallet quietly. One constant per file is the most that
# can be single-sourced across a Python script, a static page and a markdown
# document — so there are three edit points, not six, and each file has one.
DONATE_SOL = "HDDEfcYnLh4w8yG5Rn8chcm15xo1LavkvtRGeTRGAUGE"
DATA = os.environ.get("OMNIGAUGE_HOME") or os.path.join(
    os.environ.get("XDG_DATA_HOME") or os.path.expanduser("~/.local/share"), APP)
DB = os.path.join(DATA, "usage.db")
NOW = int(time.time())
SCAN_ERRORS = []

SCHEMA = """
CREATE TABLE IF NOT EXISTS snapshots (
  id INTEGER PRIMARY KEY,
  product TEXT NOT NULL, source TEXT NOT NULL, agent TEXT NOT NULL,
  window TEXT NOT NULL, model TEXT, usage_type TEXT NOT NULL,
  pct_used REAL, raw_value TEXT,
  tokens_in INTEGER, tokens_out INTEGER, tokens_cache_read INTEGER,
  tokens_cache_write INTEGER, tokens_think INTEGER, cost_usd REAL,
  reset_at TEXT, collected_at INTEGER NOT NULL, note TEXT
);
CREATE INDEX IF NOT EXISTS ix_snap ON snapshots(agent, usage_type, collected_at DESC);

-- last good answer per spend source. The X usage endpoint's budget is tiny
-- and a dashboard must never rate-limit itself out of its own data: fetches
-- respect a TTL, and a failed fetch serves the last good value WITH ITS AGE.
CREATE TABLE IF NOT EXISTS api_cache (
  name TEXT PRIMARY KEY, payload TEXT NOT NULL, fetched_at INTEGER NOT NULL
);

-- incremental lifetime cache: a 140GB rollup must never rescan unchanged files
CREATE TABLE IF NOT EXISTS filecache (
  path TEXT PRIMARY KEY, agent TEXT, mtime REAL, size INTEGER,
  tin INTEGER, tout INTEGER, cache_read INTEGER, cache_write INTEGER,
  think INTEGER, total INTEGER, msgs INTEGER, scanned_at INTEGER,
  models TEXT,  -- JSON {model: {out, total, msgs}}; models change over a corpus
  last_ts REAL, -- newest timestamp SEEN INSIDE the file; NULL = never scanned,
                -- 0 = scanned and nothing in it carries a timestamp
  scan_epoch INTEGER  -- which version of the counting arithmetic produced this
);
CREATE TABLE IF NOT EXISTS insights (
  id INTEGER PRIMARY KEY, agent TEXT, text TEXT, collected_at INTEGER
);
"""


# (table, column, type) added after the first release. CREATE TABLE IF NOT EXISTS
# silently does nothing when the table already exists, so new columns need ALTER.
MIGRATIONS = [("filecache", "models", "TEXT"),
              ("filecache", "last_ts", "REAL"),
              ("filecache", "scan_epoch", "INTEGER")]

# Bump whenever the ARITHMETIC of any scanner changes. The cache is keyed on
# mtime and size, which describe the file — nothing described the code. After the
# codex step-sum fix every unchanged file kept answering with the old numbers and
# the board sat 0.09% below an independent re-derivation, silently, until each
# file happened to be written again. A cached number has to know what made it.
SCAN_EPOCH = 3   # 3: claude rows now cache a real total (in+out+cache_read)

VERSION = "1.0.3"


def build_id():
    """A short fingerprint of THIS file's bytes - what a bug report needs and
    what a long-running board compares against to notice it is stale."""
    import hashlib
    try:
        with open(os.path.realpath(__file__), "rb") as fh:
            return hashlib.sha256(fh.read()).hexdigest()[:8]
    except OSError:
        return "unknown"


_BUILD_AT_START = None


def binary_changed():
    """True once the file this process was launched from has been rewritten.
    A --watch left running through an install kept drawing yesterday's code -
    a session reset rendered as a wall clock for hours after the fix landed.
    The board says so and re-execs itself rather than lying quietly."""
    global _BUILD_AT_START
    cur = build_id()
    if _BUILD_AT_START is None:
        _BUILD_AT_START = cur
        return False
    return cur != _BUILD_AT_START


def reexec():
    sys.stdout.write("\033[?25h\033[?1049l")
    sys.stdout.flush()
    os.execv(sys.executable, [sys.executable] + sys.argv)


def db():
    os.makedirs(DATA, exist_ok=True)
    c = sqlite3.connect(DB)
    c.executescript(SCHEMA)
    for table, col, typ in MIGRATIONS:
        have = {r[1] for r in c.execute(f"PRAGMA table_info({table})")}
        if col not in have:
            c.execute(f"ALTER TABLE {table} ADD COLUMN {col} {typ}")
    c.commit()
    return c


# ──────────────────────────────── presentation ────────────────────────────────

class S:
    """Themes.

    The first cut put bright green beside bright red; the second desaturated
    both but kept the pairing. Green-vs-red is the problem, not its saturation —
    complements at similar lightness vibrate whatever their chroma.

    `ink` (default) drops green altogether. Healthy reads as neutral blue-grey
    and colour is spent only where it means something: amber at 70%, rose at
    90%. Most of the screen is calm, and the one row that matters is the only
    coloured thing on it.
    """
    on = sys.stdout.isatty() and not os.environ.get("NO_COLOR")
    theme = "ink"

    THEMES = dict(
        # ok is HUELESS on purpose. 66 (#5f8787) was meant to read as slate but
        # it is teal, and teal reads as green — which put the red/green pair
        # straight back. A healthy bar now carries no hue at all, so the only
        # coloured thing on screen is whatever is actually in trouble.
        ink=dict(ok="\033[38;5;245m",    # neutral grey
                 warn="\033[38;5;179m",  # amber
                 crit="\033[38;5;167m",  # rose
                 accent="\033[38;5;110m",# soft blue, headings only
                 gry="\033[38;5;243m", faint="\033[38;5;237m", wht="\033[38;5;251m"),
        steel=dict(ok="\033[38;5;67m",   # steel blue — a hue, but never green
                   warn="\033[38;5;179m", crit="\033[38;5;167m",
                   accent="\033[38;5;110m",
                   gry="\033[38;5;243m", faint="\033[38;5;237m", wht="\033[38;5;251m"),
        muted=dict(ok="\033[38;5;72m", warn="\033[38;5;179m", crit="\033[38;5;167m",
                   accent="\033[38;5;109m",
                   gry="\033[38;5;242m", faint="\033[38;5;238m", wht="\033[38;5;252m"),
        mono=dict(ok="\033[38;5;245m", warn="\033[38;5;250m", crit="\033[38;5;255m",
                  accent="\033[38;5;248m",
                  gry="\033[38;5;240m", faint="\033[38;5;236m", wht="\033[38;5;253m"),
        vivid=dict(ok="\033[92m", warn="\033[93m", crit="\033[91m", accent="\033[96m",
                   gry="\033[90m", faint="\033[90m", wht="\033[97m"),
    )
    BASE = dict(r="\033[0m", b="\033[1m", dim="\033[2m")

    def __getattr__(self, k):
        if not S.on:
            return ""
        t = dict(S.BASE, **S.THEMES.get(S.theme, S.THEMES["ink"]))
        alias = dict(bgrn="ok", bred="crit", byel="warn", bcya="accent",
                     grn="ok", red="crit", yel="warn", cya="accent")
        return t.get(alias.get(k, k), "")


s = S()
# 120, not 100: the board grows a MARGIN column at 103 and a 24H trend at
# 111, so a wide terminal earns more instrument. Past 120 the tables would
# be stretching, not saying more - a board should sit in a screen, not
# fill it.
W = min(shutil.get_terminal_size((100, 30)).columns, 120)


def vis(t):
    return len(re.sub(r"\033\[[0-9;]*m", "", t))


def clip(text, width):
    """Hard-truncate to `width` VISIBLE columns, ANSI-aware. A row longer than
    its frame does not get to break the frame; it loses its tail instead. The
    reset is re-appended so a clipped colour cannot bleed into the border."""
    if vis(text) <= width:
        return text
    out, seen = [], 0
    for m in re.finditer(r"\033\[[0-9;]*m|.", text, re.S):
        tok = m.group(0)
        if tok.startswith("\033"):
            out.append(tok)
        elif seen < width:
            out.append(tok); seen += 1
        else:
            break
    return "".join(out) + s.r


def rule(ch="─"):
    return s.gry + ch * W + s.r


# Inner width of a framed panel. A frame line is ' │' + IN + '│' — one leading
# space and TWO borders, so IN must give back three columns, not two. W - 2 put
# every panel at W+1 and wrapped the right border on an exactly-W terminal.
IN = W - 3


def top(title, sub=""):
    """The site's frame, worn by the terminal: ╔══▌ TITLE ▐════ sub ═╗ with
    double-line sides and a blank row of air baked in after the head - John:
    'we have plenty of room to play with... i dont want everything as cramped
    looking as it is.'"""
    tabs = f"{s.accent}▌{s.r}{s.b}{s.wht} {title} {s.r}{s.accent}▐{s.r}"
    tail = f"{s.gry} {sub} {s.r}" if sub else ""
    # geometry: ' ╔══' (4) + tabs + ═-fill + tail + '╗' (1) must equal W,
    # and W = IN + 3.
    fill = IN - 2 - vis(tabs) - vis(tail)
    if fill < 1 and tail:            # the sub is decoration; the frame is not
        tail, fill = "", IN - 2 - vis(tabs)
    if fill < 1:
        tabs = (f"{s.accent}▌{s.r}{s.b}{s.wht} "
                f"{clip(title, IN - 8)} {s.r}{s.accent}▐{s.r}")
        fill = max(1, IN - 2 - vis(tabs))
    print(f"\n\n {s.accent}╔══{s.r}{tabs}{s.accent}{'═' * fill}{s.r}"
          f"{tail}{s.accent}╗{s.r}")
    mid()


def mid(text=""):
    text = clip(text, IN)
    print(f" {s.accent}║{s.r}{text}{' ' * max(0, IN - vis(text))}{s.accent}║{s.r}")


def bot(note=""):
    mid()
    if note:
        n = clip(f"{s.gry} {note} {s.r}", IN - 3)
        fill = IN - vis(n) - 2
        print(f" {s.accent}╚{'═' * 2}{s.r}{n}{s.accent}{'═' * max(1, fill)}╝{s.r}")
    else:
        print(f" {s.accent}╚{'═' * IN}╝{s.r}")


# Eighth blocks gave sub-character resolution — and cost seven rare glyphs.
# A font missing any one of them substitutes from a fallback face with a
# different advance, so that row alone renders wider. The count is identical, so
# no amount of measuring the string reveals it; it only shows on screen.
# Full block and middle dot are the two the terminals actually have.
BLOCKS = " █"


def sev(pct):
    return s.crit if pct >= 90 else s.warn if pct >= 70 else s.ok if pct >= 1 else s.faint


def bar(pct, w=22):
    """Whole cells only, rounded rather than truncated.

    Rounding keeps small readings visible — 1% still lights a cell instead of
    vanishing — and 24% vs 26% still differ because they round to different
    cells. Resolution drops from 1/8 of a cell to 1/2, which is the price of
    every row rendering at the same width in every font.
    """
    full = max(0, min(w, int(round(pct / 100 * w))))
    if pct > 0:
        full = max(1, full)
    # ▌ caps the fill with a crisp half-cell edge and ░ gives the empty track
    # real texture - the dot track read as absence, not as the rest of a
    # gauge. Both glyphs are already load-bearing (▌ in every panel head, ░
    # in the vendors' own panels), so no new fallback risk.
    cap = "▌" if 0 < full < w else ""
    return (sev(pct) + "█" * full + cap
            + s.faint + "░" * (w - full - len(cap)) + s.r)


def dur(sec):
    if sec is None: return "-"   # ASCII: an em dash is a font-fallback risk
    if sec < 3600: return f"{sec//60}m"
    if sec < 86400: return f"{sec//3600}h {(sec%3600)//60}m"
    return f"{sec//86400}d {(sec%86400)//3600}h"


def dot(pct):
    return f"{sev(pct)}{'●' if pct >= 1 else '○'}{s.r}"


_MONTHS = {m: i for i, m in enumerate(
    ["jan","feb","mar","apr","may","jun","jul","aug","sep","oct","nov","dec"], 1)}


def reset_epoch(text):
    """Vendor reset wording -> epoch, or None. Shares parsing with countdown()."""
    secs = countdown(text, _as_seconds=True)
    return None if secs is None else int(time.time()) + secs


def countdown(text, _as_seconds=False):
    """Best-effort 'in 4d 22h' from the vendor's own reset wording.

    Formats differ per vendor ('Aug 19, 6pm', '23:04 on 19 Aug', 'August 17,
    12:49'). Anything unparseable falls back to the raw string rather than
    guessing — a wrong countdown is worse than none.
    """
    if not text: return None
    t = text.lower()
    mon = day = None
    m = re.search(r"([a-z]{3,9})\s+(\d{1,2})", t)
    if m and m.group(1)[:3] in _MONTHS:
        mon, day = _MONTHS[m.group(1)[:3]], int(m.group(2))
    else:
        m = re.search(r"(\d{1,2})\s+([a-z]{3,9})", t)
        if m and m.group(2)[:3] in _MONTHS:
            day, mon = int(m.group(1)), _MONTHS[m.group(2)[:3]]
    hh = mm = 0
    have_time = False
    m = re.search(r"(\d{1,2}):(\d{2})", t)
    if m:
        hh, mm, have_time = int(m.group(1)), int(m.group(2)), True
    # A meridian applies whether or not minutes were given: '3:19pm' is
    # 15:19, not 03:19 - reading the HH:MM and stopping put every afternoon
    # session twelve hours off.
    m = re.search(r"(\d{1,2})(?::\d{2})?\s*(am|pm)", t)
    if m:
        hh = int(m.group(1)) % 12 + (12 if m.group(2) == "pm" else 0)
        have_time = True
    now = time.localtime()
    if not mon:
        # Time only ('3:19pm'): a session window resets at the next such
        # clock time - today if still ahead, else tomorrow. Renders as a
        # countdown like every other row instead of a wall clock.
        if not have_time:
            return None
        try:
            target = time.mktime((now.tm_year, now.tm_mon, now.tm_mday,
                                  hh, mm, 0, 0, 0, -1))
            if target < time.time():
                target += 86400
        except (ValueError, OverflowError, OSError):
            return None
    else:
        year = now.tm_year + (1 if mon < now.tm_mon - 6 else 0)
        try:
            target = time.mktime((year, mon, day, hh, mm, 0, 0, 0, -1))
        except (ValueError, OverflowError, OSError):
            return None
    d = int(target - time.time())
    if _as_seconds:
        return max(0, d)
    if d < 0: return "due"
    if d < 3600: return f"{d//60}m"
    if d < 86400: return f"{d//3600}h {(d%3600)//60}m"
    return f"{d//86400}d {(d%86400)//3600}h"


# P and E will never be reached by a token count. They are here so the ladder
# has no top: without them 999,999,999,999,999 printed "1000.00T", because T
# had nothing to promote into. A rule with one exception is a rule someone
# eventually meets.
_UNITS = (("E", 1e18), ("P", 1e15), ("T", 1e12), ("B", 1e9), ("M", 1e6), ("K", 1e3))


def human(n):
    """Pick the unit AFTER rounding, not before.

    Choosing on the raw value meant 999,999 was below 1e6, took K, and rounded
    999.999 into "1000.00K". The board could then print OUTPUT 1000.00M one
    column from TOTAL 1.00B — the same magnitude in two notations. A thousand of
    a unit is the next unit.

    (The old body ended `.rstrip("0").rstrip(".")`, which never fired once: the
    unit letter is appended before the strip, so the string never ends in a zero.
    Dead code that read like an intention.)
    """
    for i, (u, d) in enumerate(_UNITS):
        if n >= d:
            s = f"{n / d:.2f}"
            if s.startswith("1000.") and i > 0:
                u, d = _UNITS[i - 1]          # i-1 is the LARGER unit
                s = f"{n / d:.2f}"
            return s + u
    return f"{n:,}"


def short_model(agent, model):
    """Vendor labels are long and the prefix is already in the agent column."""
    if model in (None, "all"): return agent
    m = model
    m = re.sub(r"^GPT-[\d.]+-Codex-", "", m, flags=re.I)   # GPT-5.3-Codex-Spark -> Spark
    m = re.sub(r"\s*Weekly limit$", "", m, flags=re.I)
    return f"{agent}/{m.strip().lower()}"


def short_reset(t):
    """Drop the timezone parenthetical — it is the same for every row."""
    if not t: return "-"
    return re.sub(r"\s*\([^)]*\)\s*$", "", t).strip()


def age(ts):
    """Column-sized age of a reading: 'now', '7m', '3h', '2d'. The READ column
    exists so a stale number looks stale — a PARTIAL refresh leaves the missing
    window's previous reading on the board, and this is what says so."""
    d = max(0, NOW - ts)
    if d < 90: return "now"
    if d < 5400: return f"{d//60}m"
    if d < 172800: return f"{d//3600}h"
    return f"{d//86400}d"


# ──────────────────────────────── providers ──────────────────────────────────
#
# A provider teaches omnigauge about one source. It is a single file in providers/
# with no registration step — dropping it in is the whole install. See
# providers/README.md for the contract.

# Later directories WIN a name collision. The user-local data dir survives
# upgrades but must LOSE to a checkout sitting next to the script — editing
# providers/ in the repo has to beat the stale copy a previous install left in
# the data dir — and an explicit OMNIGAUGE_PROVIDERS path beats both.
PROVIDER_DIRS = [
    os.path.join(DATA, "providers"),
    os.path.join(os.path.dirname(os.path.realpath(__file__)), "providers"),
    *(os.environ.get("OMNIGAUGE_PROVIDERS", "").split(os.pathsep) if
      os.environ.get("OMNIGAUGE_PROVIDERS") else []),
]
PROVIDERS = {}


def load_providers():
    """Import every providers/*.py. A broken third-party provider must never
    take the dashboard down with it, so failures are collected and shown."""
    import importlib.util
    for d in PROVIDER_DIRS:
        if not d or not os.path.isdir(d):
            continue
        for f in sorted(glob.glob(os.path.join(d, "*.py"))):
            if os.path.basename(f).startswith("_"):
                continue
            try:
                sys.modules.setdefault("omnigauge", sys.modules[__name__])
                spec = importlib.util.spec_from_file_location(
                    "omnigauge_provider_" + os.path.basename(f)[:-3], f)
                mod = importlib.util.module_from_spec(spec)
                sys.modules.setdefault("omnigauge", sys.modules[__name__])
                spec.loader.exec_module(mod)
                name = getattr(mod, "NAME", None)
                if not name:
                    raise ValueError("provider defines no NAME")
                PROVIDERS[name] = mod
            except Exception as e:
                SCAN_ERRORS.append(f"provider {os.path.basename(f)}: "
                                   f"{type(e).__name__}: {e}")
    return PROVIDERS


# ─────────────────────────────── path discovery ───────────────────────────────

def windows_homes(sub):
    """WSL: the Windows-side install is a SEPARATE store. Missing it once cost a
    wrong 'no local session exists' answer, so it is discovered, not assumed."""
    return [p for p in glob.glob(f"/mnt/*/Users/*/{sub}") if os.path.isdir(p)]


def config_roots():
    """Extra HOME-like roots the user pointed at - a dev SSD, a second profile,
    a backup. One path per line in DATA/roots (# comments allowed), plus
    OMNIGAUGE_ROOTS (os.pathsep-separated). A root is a directory that CONTAINS
    agent stores (.claude, .codex, ...) - a home, wherever it is mounted."""
    out = []
    try:
        for line in io.open(os.path.join(DATA, "roots")):
            line = line.strip()
            if line and not line.startswith("#"):
                out.append(line)
    except OSError:
        pass
    env = os.environ.get("OMNIGAUGE_ROOTS", "")
    out += [x for x in env.split(os.pathsep) if x]
    return [os.path.expanduser(x) for x in out]


def missing_roots():
    """Configured roots that are not mounted right now. A removable drive that
    is out means its history is EXCLUDED this run - that must be said out loud,
    never left to read as 'usage went down'."""
    return [r for r in config_roots() if not os.path.isdir(r)]


def all_homes(sub):
    """Every place a store named `sub` may live: the real home, Windows-side
    homes on WSL, and every configured root. Existing directories only, and
    deduplicated by REALPATH - a root that reaches the same store through a
    symlink (or points back at the real home) must not count it twice."""
    out, seen = [], set()
    cands = []
    h = os.path.expanduser(f"~/{sub}")
    if os.path.isdir(h):
        cands.append(h)
    cands += windows_homes(sub)
    for r in config_roots():
        c = os.path.join(r, sub)
        if os.path.isdir(c):
            cands.append(c)
    for c in cands:
        key = os.path.realpath(c)
        if key not in seen:
            seen.add(key)
            out.append(c)
    return out


def claude_files():
    # Recursive, because transcripts do not all sit one level down:
    # subagent runs write project/<session>/subagents/agent-*.jsonl, and a
    # one-level glob silently dropped 39 files, ~1,500 messages and every
    # haiku token on this very machine. all_homes covers the Windows side
    # and every user-configured root the same way.
    out = []
    for h in all_homes(".claude"):
        out += glob.glob(f"{h}/projects/**/*.jsonl", recursive=True)
    return out


def codex_files():
    out = []
    for h in all_homes(".codex"):
        out += glob.glob(f"{h}/sessions/**/*.jsonl", recursive=True)
    return out


def grok_files():
    out = []
    for h in all_homes(".grok"):
        out += glob.glob(f"{h}/sessions/*/*/updates.jsonl")
    return out


def scrape_cwd(agent):
    """Pick a directory the CLI already trusts.

    Launching Claude in an unfamiliar directory raises a blocking workspace-trust
    dialog which swallows the keystrokes. omnigauge will NOT auto-accept that on
    the user's behalf — trusting a folder is a real security decision and it
    persists. Instead it reuses a cwd the tool has demonstrably run in before,
    taken from Claude's own session registry.
    """
    if agent == "claude":
        best, newest = None, 0
        for f in glob.glob(os.path.expanduser("~/.claude/sessions/*.json")):
            try:
                with io.open(f, encoding="utf-8") as fh:
                    d = json.load(fh)
                cwd = d.get("cwd")
                st = os.path.getmtime(f)
                if cwd and os.path.isdir(cwd) and st > newest:
                    best, newest = cwd, st
            except (OSError, ValueError):
                continue
        if best:
            return best
    return os.getcwd()


def installed(agent):
    return shutil.which(agent) is not None


# ─────────────────────────────── token counting ───────────────────────────────

def _dig(o, key):
    if isinstance(o, dict):
        if key in o: return o[key]
        for v in o.values():
            r = _dig(v, key)
            if r is not None: return r
    elif isinstance(o, list):
        for v in o:
            r = _dig(v, key)
            if r is not None: return r
    return None


def _epoch(ts):
    """Epoch seconds, or None when the stamp is unreadable. None, not 0: a
    message with a broken timestamp still happened, and 0 reads as "ancient",
    which makes a window filter silently drop real usage."""
    try:
        import datetime
        return datetime.datetime.fromisoformat(ts.replace("Z", "+00:00")).timestamp()
    except (ValueError, TypeError, AttributeError):
        return None


Z = dict(msgs=0, tin=0, tout=0, cache_read=0, cache_write=0, think=0, total=0, files=0)


def blank():
    t = dict(Z); t["models"] = {}
    return t


def add_model(t, model, out=0, total=0, msgs=0):
    if not model: return
    m = t["models"].setdefault(model, dict(out=0, total=0, msgs=0))
    m["out"] += out; m["total"] += total; m["msgs"] += msgs


def merge(dst, src):
    for k, v in src.items():
        if k == "models":
            for mo, mv in v.items():
                d = dst["models"].setdefault(mo, dict(out=0, total=0, msgs=0))
                for kk in mv: d[kk] += mv[kk]
        elif k == "last_ts":
            # A timestamp is not a quantity. Summing it across files produced a
            # number in the year 60,000 and nothing complained.
            if v: dst[k] = max(dst.get(k) or 0, v)
        elif k != "files":
            dst[k] += v


def scan_claude(path, since=0):
    t = blank()
    for line in io.open(path, errors="replace"):
        if '"usage"' not in line: continue
        try: d = json.loads(line)
        except ValueError: continue
        msg = d.get("message") or {}
        u = msg.get("usage")
        if not isinstance(u, dict): continue
        # Newest timestamp regardless of window — the core needs a fact about the
        # contents before it may skip this file on mtime alone. Kept identical to
        # providers/claude.py; the conformance tests lock the two together.
        ep = _epoch(d.get("timestamp", ""))
        if ep is not None and ep > (t.get("last_ts") or 0):
            t["last_ts"] = ep
        if since and ep is not None and ep < since: continue
        t["msgs"] += 1
        t["tin"] += u.get("input_tokens", 0)
        t["tout"] += u.get("output_tokens", 0)
        t["cache_read"] += u.get("cache_read_input_tokens", 0)
        t["cache_write"] += u.get("cache_creation_input_tokens", 0)
        t["think"] += (u.get("output_tokens_details") or {}).get("thinking_tokens", 0)
        # total = in + out + cache_read, the same formula the board and the
        # per-model rows use. Leaving it 0 made --json contradict the render.
        t["total"] += (u.get("input_tokens", 0) + u.get("output_tokens", 0)
                       + u.get("cache_read_input_tokens", 0))
        add_model(t, msg.get("model"), out=u.get("output_tokens", 0),
                  total=u.get("output_tokens", 0) + u.get("cache_read_input_tokens", 0)
                        + u.get("input_tokens", 0), msgs=1)
    return t


def tail_chunk(path, nbytes=1 << 20):
    """Read the last nbytes of a file as text, aligned to a line boundary."""
    size = os.path.getsize(path)
    with open(path, "rb") as fh:
        if size > nbytes:
            fh.seek(size - nbytes)
            fh.readline()          # discard the partial first line
        return fh.read().decode("utf-8", "replace")


_CODEX_FIELDS = (("tin", "input_tokens"), ("tout", "output_tokens"),
                 ("cache_read", "cached_input_tokens"),
                 ("think", "reasoning_output_tokens"), ("total", "total_tokens"))


def _codex_spend(seq, since, fields):
    """What was actually spent from `since` onward. Identical in intent to
    providers/codex.py::_spend — the conformance tests lock the two together.

    The cumulative counter is differenced STEP BY STEP, because it does not only
    climb. A rollout reused by a new session sends it backwards; the new value is
    then itself the spend since the restart. Endpoint arithmetic broke both ways:
    a reset inside the window went negative and max(0, ...) made it a confident
    zero, and a reset whose new run overtook the old total silently omitted
    everything before the restart.
    """
    acc = {k: 0 for k, _ in fields}
    prev = None
    for ep, u in seq:
        if ep is None or ep >= since:
            if prev is None:
                # No earlier reading to subtract, so this cumulative IS the spend
                # so far. That is the whole-file-inside-the-window case, and the
                # lifetime case where since is 0.
                for k, f in fields:
                    acc[k] += u.get(f, 0)
            else:
                for k, f in fields:
                    c, p = u.get(f, 0), prev.get(f, 0)
                    acc[k] += c if c < p else c - p   # c < p means it restarted
        prev = u
    return acc


def _codex_events(text):
    """(epoch, totals) for every token_count event in a text block."""
    out = []
    for line in text.splitlines():
        if '"total_token_usage"' not in line: continue
        try: d = json.loads(line)
        except ValueError: continue
        u = _dig(d, "total_token_usage")
        if u: out.append((_epoch(d.get("timestamp", "")), u))
    return out


def _codex_block(fh, size, off, nbytes=1 << 18):
    """A line-aligned text block starting at off."""
    fh.seek(max(0, off))
    if off > 0: fh.readline()
    return fh.read(min(nbytes, size)).decode("utf-8", "replace")


def scan_codex(path, since=0):
    """Last cumulative total per rollout. input_tokens counts context re-sent
    each turn, so it runs far above the vendor's own 'tokens used'.

    With `since`, the value is the DELTA of that cumulative across the window
    edge. The old behaviour attributed a rollout's ENTIRE history to the window
    whenever its mtime was recent - one real file put 1.13B tokens accumulated
    over 25 days into a panel labelled '24h'. A cumulative counter's growth
    inside the window is last_total minus the last total recorded BEFORE the
    window opened.

    Rollout logs are append-only and chronological, so the window edge is found
    by bisecting BYTE OFFSETS - a handful of 256KB probes even on a multi-GB
    file. Scanning forward once took ~18s per redraw across a 140GB corpus,
    and an early version of this delta grew its tail until it swallowed whole
    files and hung the suite. If no event carries a parseable timestamp the
    whole cumulative is attributed, as before: overstating is visible on the
    board, silently dropping is not.
    """
    t = blank()
    try: size = os.path.getsize(path)
    except OSError: return t
    tail = tail_chunk(path)
    events = _codex_events(tail)
    if not events and size > (1 << 20):
        try: whole = io.open(path, errors="replace").read()
        except OSError: return t
        tail, events = whole, _codex_events(whole)
    if not events:
        return t
    # Newest stamped event, recorded whatever the window is. Kept identical to
    # providers/codex.py; the conformance tests lock the two together.
    _st = [ep for ep, _ in events if ep is not None]
    if _st:
        t["last_ts"] = max(_st)
    model = None
    mm = re.findall(r'"model":"([^"]+)"', tail)
    if mm: model = mm[-1]          # rollouts can span models; attribute to the last
    last = events[-1][1]
    base = None
    seq = None
    if since:
        stamped = [(ep, u) for ep, u in events if ep is not None]
        if not stamped:
            # Nothing here can be placed in time; mtime is the only evidence.
            try:
                if os.path.getmtime(path) < since:
                    return t
            except OSError:
                return t
        if stamped and not any(ep >= since for ep, _ in stamped):
            return t               # tail is authoritative: nothing in the window
        pre = [u for ep, u in stamped if ep < since]
        if pre:
            base = pre[-1]         # window edge sits inside the tail
            seq = stamped          # and every in-window event is here too
        elif stamped and size > (1 << 20):
            # Edge is deeper than the tail: bisect offsets for the last event
            # before the cutoff. Probes that see no event fall back toward 0.
            with open(path, "rb") as fh:
                lo, hi = 0, size   # invariant: first event at lo is < since
                while hi - lo > (1 << 18):
                    mid = (lo + hi) // 2
                    ev = _codex_events(_codex_block(fh, size, mid))
                    ts = next((ep for ep, _ in ev if ep is not None), None)
                    if ts is None or ts >= since: hi = mid
                    else: lo = mid
                ev = _codex_events(_codex_block(fh, size, lo, (1 << 19)))
                pre = [u for ep, u in ev if ep is not None and ep < since]
                if pre:
                    base = pre[-1]
                    # The gap between the located edge and the tail was never
                    # read - a counter restart hidden there was invisible to
                    # the step-sum (Assay's documented boundary). Walk ONLY
                    # that in-window stretch forward in 512KB steps: bounded
                    # by what the window actually holds, never the whole file.
                    # Blocks are line-aligned, so a walk can overlap the
                    # tail by up to one block: stop the gap at the tail's
                    # FIRST stamped event, or those events count twice.
                    gap, first_tail = [], stamped[0][0]
                    off, tail_off = lo, size - (1 << 20)
                    while off < tail_off:
                        blk = _codex_block(fh, size, off, (1 << 19))
                        if not blk: break
                        gap += [(ep, u) for ep, u in _codex_events(blk)
                                if ep is not None and since <= ep < first_tail]
                        off += (1 << 19)
                    seq = [(since - 1, base)] + gap + stamped
    t["msgs"] = 1
    # Always step-sum. Falling back to `last - base` when no baseline was found
    # meant a rollout sitting entirely inside the window reported only its final
    # cumulative - the spend since its LAST restart, everything before it dropped.
    # Window and lifetime were wrong in the same direction, so the
    # window-cannot-exceed-lifetime invariant never noticed.
    for key, val in _codex_spend(seq if seq else events, since, _CODEX_FIELDS).items():
        t[key] = val
    add_model(t, model or "unknown", out=t["tout"], total=t["total"], msgs=1)
    return t


def scan_grok(path, since=0):
    """Session-cumulative totalTokens; with `since`, the window's growth of it.
    Lines carry epoch timestamps, so the baseline is the highest total recorded
    before the window opened - a session merely TOUCHED in the window no longer
    donates its whole history to it. Untimestamped lines count toward the
    baseline: understating the window is visible against LIFETIME, inflating it
    is not."""
    t = blank()
    best, before, in_window = 0, 0, False
    model = None
    try:
        smry = os.path.join(os.path.dirname(path), "summary.json")
        if os.path.exists(smry):
            with io.open(smry, encoding="utf-8") as fh:
                model = json.load(fh).get("current_model_id")
    except (OSError, ValueError):
        pass
    for line in io.open(path, errors="replace"):
        if '"totalTokens"' not in line: continue
        try: d = json.loads(line)
        except ValueError: continue
        v = _dig(d, "totalTokens")
        if not isinstance(v, int): continue
        best = max(best, v)
        ts = d.get("timestamp")
        if since:
            if isinstance(ts, (int, float)) and ts >= since: in_window = True
            else: before = max(before, v)
    if since:
        got = max(0, best - before) if in_window else 0
    else:
        got = best
    if got:
        t["msgs"], t["total"] = 1, got
        add_model(t, model or "unknown", total=got, msgs=1)
    return t


SCANNERS = dict(claude=(claude_files, scan_claude), codex=(codex_files, scan_codex),
                grok=(grok_files, scan_grok))


def register_provider_quota():
    """A provider may bring its own QUOTA spec, parser, insights and cwd —
    overriding a built-in of the same name. That is how a contributor fixes a
    vendor UI change without touching the core."""
    for name, mod in PROVIDERS.items():
        q = getattr(mod, "QUOTA", None)
        if not q or not hasattr(mod, "parse_quota"):
            continue
        AGENTS[name] = dict(
            argv=q["argv"], keys=q["keys"], ready=q["ready"], done=q["done"],
            expect=q.get("expect", []), parse=mod.parse_quota,
            source=q.get("source", "cli_usage"),
            insights=getattr(mod, "insights", None),
            cwd=getattr(mod, "scrape_cwd", None),
        )


def all_agents():
    """Built-ins first, then providers — a provider may also override a built-in."""
    out = list(SCANNERS)
    for name, mod in PROVIDERS.items():
        if getattr(mod, "KIND", "agent") != "agent":
            continue
        if name not in out:
            out.append(name)
        SCANNERS[name] = (mod.files, mod.scan)
    return out


# ─────────────────────────────── capabilities ────────────────────────────────
#
# Three states, not two. "obtained" is on your board now. "available" means
# the vendor exposes it and this provider does not read it yet. "unavailable"
# means the vendor does not expose it - with the reason stated, because the
# reason is the useful part: "the vendor writes the schema and zeroes every
# value" saves the next person a day, where a blank cell says nothing.
# A state may carry a note after a colon: "obtained: session totals only".

CAPABILITIES = ("tokens", "quota", "reset", "models", "lifetime", "spend", "burn")
CAP_STATES = ("obtained", "available", "unavailable")

_NO_SPEND = "unavailable: subscription plans have no dollar balance"
_API_ACCT = "unavailable: an API account, not a local agent"

BUILTIN_CAPS = {
    "claude": dict(tokens="obtained", quota="obtained", reset="obtained",
                   models="obtained", lifetime="obtained", spend=_NO_SPEND,
                   burn="obtained: derived from the quota series"),
    "codex": dict(tokens="obtained: last cumulative total per rollout",
                  quota="obtained: percent remaining, inverted", reset="obtained",
                  models="obtained: attributed to the last model in the file",
                  lifetime="obtained", spend=_NO_SPEND,
                  burn="obtained: derived from the quota series"),
    "grok": dict(tokens="obtained: session totals only", quota="obtained",
                 reset="obtained", models="obtained: from the session summary",
                 lifetime="obtained", spend=_NO_SPEND,
                 burn="obtained: derived from the quota series"),
    # the two built-in spend sources are part of the board and belong here too
    "openai": dict(tokens="obtained: 30-day organization totals",
                   quota="unavailable: admin keys expose spend, not plan windows",
                   reset="unavailable: no window to reset",
                   models="unavailable: not broken out by the usage API",
                   lifetime="available: the API takes wider date ranges",
                   spend="obtained: organization costs, 30 days",
                   burn="unavailable: no quota series to derive from"),
    "x": dict(tokens=_API_ACCT,
              quota="obtained: posts against the project cap",
              reset="unavailable: the usage endpoint does not state the window end",
              models=_API_ACCT, lifetime=_API_ACCT,
              spend="unavailable: dollar balances are console-only, verified",
              burn="unavailable: no series is kept for post caps"),
}


# Verified absences are worth as much as presences: each of these was
# concluded by reading real data on a real install, not from a directory
# listing. providers/README.md carries the full forensics.
NEGATIVES = [
    ("cursor", "the IDE writes token fields and zeroes every value"),
    ("antigravity", "records trajectories; no token accounting at all"),
    ("copilot", "no token accounting in its logs"),
]


def cap_state(v):
    return (v or "").split(":", 1)[0].strip()


def caps_for(name):
    """A provider's declared CAPS wins; built-ins carry theirs here. The
    legend walks the LOADED registry, so a third-party provider that declares
    CAPS appears automatically - that is the whole promise of the plugin
    system, and hardcoding this table in a view would quietly break it."""
    mod = PROVIDERS.get(name)
    if mod is not None and isinstance(getattr(mod, "CAPS", None), dict):
        return mod.CAPS
    return BUILTIN_CAPS.get(name, {})


def legend_rows():
    """(name, kind, caps) for every source that ships, agents first."""
    rows = [(a, "agent", caps_for(a)) for a in agent_order()]
    seen = {n for n, _, _ in rows}
    for name in ("openai", "x"):
        rows.append((name, "api", caps_for(name)))
    for name, mod in sorted(PROVIDERS.items()):
        if getattr(mod, "KIND", "") == "api" and name not in seen:
            rows.append((name, "api", caps_for(name)))
    return rows


def legend():
    """The providers legend: what each source actually gets, could get, and
    cannot get. ● obtained · ○ available, not read yet · - unavailable."""
    top("PROVIDERS", "what ships, what it gets, and what it cannot get")
    short = {"lifetime": "LIFE", "models": "MODEL"}
    hdr = "  " + f"{'SOURCE':<11}{'KIND':<6}" + "".join(
        f"{short.get(c, c).upper():>8}" for c in CAPABILITIES)
    mid(f"{s.gry}{hdr}{s.r}")
    notes = []
    for name, kind, caps in legend_rows():
        cells = ""
        for c in CAPABILITIES:
            v = caps.get(c, "")
            st = cap_state(v)
            mark = {"obtained": f"{s.ok}●{s.r}", "available": f"{s.warn}○{s.r}",
                    "unavailable": f"{s.faint}-{s.r}"}.get(st, f"{s.crit}?{s.r}")
            cells += f"{'':>7}" + mark
            if ":" in (v or "") or st not in CAP_STATES:
                notes.append((name, c, v))
        mid(f"  {s.b}{name:<11.11}{s.r}{s.gry}{kind:<6}{s.r}{cells}")
    bot("● obtained · ○ available, not read yet · - unavailable")
    if notes:
        top("THE REASONS", "the third state is the honest one")
        last = None
        for name, c, v in notes:
            st, _, why = (v or "").partition(":")
            st, why = st.strip(), why.strip()
            if name != last:
                if last is not None:
                    mid()
                mid(f"  {s.b}{s.wht}{name}{s.r}")
                last = name
            mark = {"obtained": f"{s.ok}●{s.r}", "available": f"{s.warn}○{s.r}"
                    }.get(st, f"{s.faint}-{s.r}")
            first, rest = _wrap_note(why or st, IN - 15)
            mid(f"  {mark} {s.gry}{c:<9}{s.r}{s.gry}{first}{s.r}")
            for cont in rest:
                mid(f"{'':<15}{s.gry}{cont}{s.r}")
        bot("the glyph is the state; the words are why")
    if NEGATIVES:
        top("VERIFIED NEGATIVES", "read on real installs, not guessed")
        for name, why in NEGATIVES:
            mid(f"  {s.faint}-{s.r} {s.b}{name:<12}{s.r}{s.gry}{why}{s.r}")
        bot("a missing provider with a reason beats a broken one")
    print()


def _wrap_note(text, width):
    """(first_line, [continuations]), broken on WORDS. A note that snaps
    mid-word reads like a rendering bug; a test caps note length so current
    notes never wrap at all, and this is the net for future ones."""
    words, lines, cur = text.split(), [], ""
    for w in words:
        cand = f"{cur} {w}".strip()
        if len(cand) <= width or not cur:
            cur = cand
        else:
            lines.append(cur); cur = w
    lines.append(cur)
    return lines[0], lines[1:]


def agent_order():
    """Every agent the board renders: built-ins in their canonical order, then
    provider agents alphabetically. The render loops iterate THIS — a provider
    whose data is collected but never drawn is a plugin system in name only."""
    rest = sorted(a for a in SCANNERS if a not in ("claude", "codex", "grok"))
    return [a for a in ("claude", "codex", "grok") if a in SCANNERS] + rest


def api_providers():
    """KIND="api" provider modules. Their detect() answers --doctor's
    "configured?" and their api_usage(creds) feeds the spend panel — the two
    halves of the contract providers/README promises them."""
    return {n: m for n, m in PROVIDERS.items()
            if getattr(m, "KIND", "") == "api"}


# Memo for the interactive loop. Cycling a theme must not re-read 200 files;
# only a window change or a refresh invalidates the numbers.
_MEMO = {}


def memo_clear():
    _MEMO.clear()


def window_totals(agent, since):
    # Quantize: `since` is derived from now() on every render, so an exact key
    # changed every second and the memo never hit — every theme press rescanned
    # the whole corpus.
    key = ("w", agent, int(since // 300))
    if key in _MEMO:
        return _MEMO[key]
    files, scan = SCANNERS[agent]
    t = blank()
    # What the last full scan actually FOUND in each file, keyed by the identity
    # of the file at that moment. mtime alone is a claim about when something was
    # written; last_ts is a fact about what is inside it.
    known = {}
    try:
        with db() as _c:
            for _p, _m, _z, _l in _c.execute(
                    "SELECT path,mtime,size,last_ts FROM filecache WHERE agent=?", (agent,)):
                known[_p] = (_m, _z, _l)
    except Exception:
        pass
    for f in files():
        try:
            if os.path.getmtime(f) < since:
                # mtime says this predates the window. Skip it ONLY with proof:
                # the file is byte-identical to the one we scanned, and that scan
                # found nothing newer than `since`. Without proof, read it.
                #
                # mtime and the timestamps inside the file come from different
                # clocks. A restored backup, rsync -a, tar -x, or a container with
                # a skewed clock makes mtime older than the content, and the
                # window silently loses every token in the file.
                st = os.stat(f)
                k = known.get(f)
                if (k and k[2] is not None
                        and abs((k[0] or 0) - st.st_mtime) < 1e-6
                        and k[1] == st.st_size and k[2] < since):
                    continue
            # since goes to EVERY scanner. Passing 0 to the rest attributed a
            # rollout's whole cumulative history to the window whenever its
            # mtime was recent - and quietly disabled the correct since
            # handling goose and aider already had.
            r = scan(f, since)
            if r["msgs"]:
                t["files"] += 1
                merge(t, r)
        except Exception as e:
            # A blanket `continue` here once hid a NameError on EVERY claude file
            # and reported a confident row of zeros. Failures are counted and
            # surfaced; zero should mean zero, not "it crashed".
            SCAN_ERRORS.append(f"{agent}: {type(e).__name__}: {e}")
    _MEMO[key] = t
    return t


def prune_filecache(con):
    """Drop cache rows no known home can reach - a deleted store, a renamed
    directory. Rows under a configured root are kept even while the root is
    unmounted: a pulled drive's history must survive its absence. Assay's
    finding: unreachable rows were never pruned - no wrong number, but
    unbounded growth."""
    keeps = [os.path.expanduser("~") + os.sep]
    keeps += [p if p.endswith(os.sep) else p + os.sep
              for p in glob.glob("/mnt/*/Users/*/")]
    keeps += [r.rstrip(os.sep) + os.sep for r in config_roots()]
    rows = con.execute("SELECT path FROM filecache").fetchall()
    dead = [p for (p,) in rows if not any(p.startswith(k) for k in keeps)]
    if dead:
        con.executemany("DELETE FROM filecache WHERE path=?",
                        [(p,) for p in dead])
        con.commit()
    return len(dead)


def lifetime_totals(agent, con, progress=False):
    """Incremental: a file is rescanned only when mtime or size changed."""
    key = ("l", agent)
    if key in _MEMO:
        return _MEMO[key]
    files, scan = SCANNERS[agent]
    cached = {r[0]: r for r in con.execute(
        "SELECT path,mtime,size,tin,tout,cache_read,cache_write,think,total,msgs,models,"
        "scan_epoch FROM filecache WHERE agent=?", (agent,))}
    t, fresh, n = blank(), 0, 0
    all_files = files()
    for f in all_files:
        n += 1
        try:
            st = os.stat(f)
        except Exception:
            continue
        c = cached.get(f)
        # c[10] is the models JSON. Rows written before per-model tracking existed
        # still match on mtime+size, so they must be revalidated or the BY MODEL
        # panel silently reports nothing for an unchanged corpus.
        # Exact mtime, not a one-second window. The tolerance was wide enough
        # to hide a whole rewrite: same byte count, new contents, written within
        # a second of the last scan, and the cache kept answering with the old
        # numbers. Append-only transcripts grow, so size usually catches a
        # change — a file rewritten in place, restored, or rotated to a
        # coincidentally equal length does not grow, and nothing else was
        # checked. A float second survives the round trip through SQLite REAL
        # intact; the epsilon is for the representation, not for the clock.
        if (c and abs(c[1] - st.st_mtime) < 1e-6 and c[2] == st.st_size
                and c[10] not in (None, "", "{}") and c[11] == SCAN_EPOCH):
            vals = dict(zip(("tin", "tout", "cache_read", "cache_write", "think", "total", "msgs"), c[3:10]))
            try: vals["models"] = json.loads(c[10] or "{}")
            except Exception: vals["models"] = {}
        else:
            fresh += 1
            if progress and fresh % 25 == 1:
                print(f"\r  {s.gry}scanning {agent}: {n:,}/{len(all_files):,} "
                      f"({fresh} changed)…{s.r}", end="", flush=True)
            try:
                r = scan(f, 0)
            except Exception as e:
                SCAN_ERRORS.append(f"{agent}: {type(e).__name__}: {e}")
                continue
            vals = {k: r[k] for k in ("tin", "tout", "cache_read", "cache_write", "think", "total", "msgs")}
            vals["models"] = r.get("models", {})
            con.execute(
                "INSERT OR REPLACE INTO filecache(path,agent,mtime,size,tin,tout,"
                "cache_read,cache_write,think,total,msgs,scanned_at,models,last_ts,"
                "scan_epoch) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
                (f, agent, st.st_mtime, st.st_size, vals["tin"], vals["tout"],
                 vals["cache_read"], vals["cache_write"], vals["think"],
                 vals["total"], vals["msgs"], NOW, json.dumps(vals["models"]),
                 r.get("last_ts"), SCAN_EPOCH))
        if vals["msgs"]:
            t["files"] += 1
            merge(t, vals)
    con.commit()
    if progress and fresh:
        print("\r" + " " * (W - 1) + "\r", end="", flush=True)
    _MEMO[key] = t
    return t


# ─────────────────────────────────  alerts  ──────────────────────────────────
#
# The incumbents track and report. None of them tell you BEFORE it hurts —
# tokscale's own docs say it "cannot set usage budgets or trigger alerts". This
# is that. It is also the only part designed to run headless, from cron.

ALERTS = os.path.join(DATA, "alerts.json")
DEFAULT_ALERTS = {
    "pct_used": 85,            # warn once any window crosses this
    "dry_before_reset": True,  # warn when the forecast says it empties early
    "notify": True,            # desktop notification if a notifier exists
    "webhook": "",             # optional POST target
    "quiet_hours": [],         # e.g. [23, 7] — no desktop popups overnight
}


def load_alerts():
    cfg = dict(DEFAULT_ALERTS)
    if os.path.exists(ALERTS):
        try:
            cfg.update(json.load(io.open(ALERTS, encoding="utf-8")))
        except Exception as e:
            SCAN_ERRORS.append(f"alerts.json unreadable: {e}")
    return cfg


def evaluate_alerts(con, cfg):
    """Return (severity, [messages]). Severity drives the process exit code so
    cron and CI can act on it: 0 fine, 1 warning, 2 critical."""
    fired, sev = [], 0
    for agent, rows in quota_rows(con).items():
        for r in rows:
            name = short_model(agent, r["model"])
            reset_s = countdown(r["reset"], _as_seconds=True) if r["reset"] else None
            if cfg.get("dry_before_reset") and r.get("dry_in") is not None \
               and reset_s is not None and r["dry_in"] < reset_s:
                fired.append(f"{name} runs dry in {dur(r['dry_in'])} - "
                             f"{dur(reset_s - r['dry_in'])} BEFORE its window resets")
                sev = 2
            elif r["pct"] >= cfg.get("pct_used", 85):
                fired.append(f"{name} at {r['pct']:.0f}% of its {r['window']} window"
                             + (f", resets in {dur(reset_s)}" if reset_s else ""))
                sev = max(sev, 1)
    return sev, fired


def notify(cfg, messages, sev):
    """Best-effort, never fatal. A monitor that crashes the cron job it runs in
    is worse than no monitor."""
    if not messages:
        return
    title = "OmniGauge — quota critical" if sev >= 2 else "OmniGauge — quota warning"
    body = "\n".join(messages[:6])
    hours = cfg.get("quiet_hours") or []
    quiet = False
    if len(hours) == 2:
        h = time.localtime().tm_hour
        a, b = hours
        quiet = (a <= h or h < b) if a > b else (a <= h < b)
    if cfg.get("notify") and not quiet:
        for argv in (["notify-send", "-u", "critical" if sev >= 2 else "normal", title, body],
                     ["osascript", "-e",
                      f'display notification "{body[:180]}" with title "{title}"'],
                     ["wsl-notify-send.exe", title, body]):
            if shutil.which(argv[0]):
                try:
                    subprocess.run(argv, capture_output=True, timeout=10)
                    break
                except Exception:
                    pass
    if cfg.get("webhook"):
        import urllib.request
        payload = json.dumps(dict(source="omnigauge", severity=sev,
                                  text=f"{title}\n{body}", messages=messages,
                                  host=os.uname().nodename, at=int(time.time()))).encode()
        try:
            req = urllib.request.Request(cfg["webhook"], data=payload,
                                         headers={"Content-Type": "application/json"})
            urllib.request.urlopen(req, timeout=15)
        except Exception as e:
            SCAN_ERRORS.append(f"webhook failed: {type(e).__name__}: {e}")


# ──────────────────────────── API spend & credits ─────────────────────────────
#
# SEPARATE from plan quota, always. A subscription window and a dollar balance
# are different products; blending them into one "remaining" number would be
# fiction. These are real money and they deplete.

CREDS = os.path.join(DATA, "credentials.json")


def load_creds():
    """Keys come from a 0600 file or the environment — never from argv, which
    would put them in shell history and in the process table."""
    c = {}
    if os.path.exists(CREDS):
        try:
            st = os.stat(CREDS)
            if st.st_mode & 0o077:
                print(f"  {s.crit}refusing to read {CREDS}: mode {oct(st.st_mode & 0o777)} "
                      f"is group/world readable. chmod 600 it.{s.r}", file=sys.stderr)
            else:
                c = json.load(io.open(CREDS, encoding="utf-8"))
        except Exception as e:
            print(f"  {s.crit}credentials unreadable: {e}{s.r}", file=sys.stderr)
    for k, env in (("openai_admin_key", "OPENAI_ADMIN_KEY"),
                   ("x_acct1_bearer", "X_ACCT1_BEARER"),
                   ("x_acct2_bearer", "X_ACCT2_BEARER"),
                   ("openrouter_api_key", "OPENROUTER_API_KEY"),
                   ("moonshot_api_key", "MOONSHOT_API_KEY"),
                   ("deepseek_api_key", "DEEPSEEK_API_KEY")):
        if os.environ.get(env):
            c[k] = os.environ[env]
    return c


def _get(url, token, timeout=20):
    import urllib.request, urllib.error
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return json.loads(r.read().decode()), None
    except urllib.error.HTTPError as e:
        body = e.read().decode()[:200]
        hint = ""
        if "api.usage.read" in body:
            hint = " — needs an ADMIN key with the api.usage.read scope"
        return None, f"HTTP {e.code}{hint}"
    except Exception as e:
        return None, f"{type(e).__name__}: {e}"


def openai_spend(key, days=30):
    """Real dollars. /v1/organization/costs is authoritative for reconciling to
    an invoice; usage/completions gives requests and tokens."""
    start = int(time.time()) - days * 86400
    out = dict(source="openai", spend_usd=None, requests=None, tokens=None, err=None)
    d, err = _get(f"https://api.openai.com/v1/organization/costs?start_time={start}&limit=180", key)
    if err:
        out["err"] = err
        return out
    total = 0.0
    for bucket in (d.get("data") or []):
        for r in (bucket.get("results") or []):
            amt = (r.get("amount") or {}).get("value")
            if isinstance(amt, (int, float)):
                total += amt
    out["spend_usd"] = round(total, 4)
    d, err = _get(f"https://api.openai.com/v1/organization/usage/completions"
                  f"?start_time={start}&limit=180", key)
    if not err and d:
        req = tok = 0
        for bucket in (d.get("data") or []):
            for r in (bucket.get("results") or []):
                req += r.get("num_model_requests") or 0
                tok += (r.get("input_tokens") or 0) + (r.get("output_tokens") or 0)
        out["requests"], out["tokens"] = req, tok
    return out


def x_usage(token, label):
    """X post-cap consumption.

    VERIFIED 2026-08-14: calling this does NOT draw against the post cap — two
    consecutive calls left project_usage unchanged. It has its own limit of 50
    per window, so it must not be polled aggressively.

    It reports POSTS, not the dollar balance shown in the developer console.
    No public endpoint for that balance has been found, so it is not invented
    here; the console remains the source for money.
    """
    out = dict(source="x", label=label, used=None, cap=None, pct=None,
               spend_usd=None, requests=None, tokens=None, err=None)
    d, err = _get("https://api.x.com/2/usage/tweets", token)
    if err:
        out["err"] = err
        return out
    u = (d or {}).get("data") or {}
    try:
        out["used"] = int(u.get("project_usage") or 0)
        out["cap"] = int(u.get("project_cap") or 0)
        if out["cap"]:
            out["pct"] = out["used"] / out["cap"] * 100
    except Exception as e:
        out["err"] = f"unexpected shape: {e}"
    return out


API_TTL = int(os.environ.get("OMNIGAUGE_SPEND_TTL_SEC", "900"))


def cached_api(con, name, fetch):
    """Fetch through the TTL cache. Within TTL the stored answer is served
    and the vendor is not called at all. On a failed fetch the last good
    answer is served WITH ITS AGE and the error alongside - a sourced stale
    number beats a red row that threw the data away. Only a failure with no
    history ever renders as an error."""
    row = con.execute("SELECT payload, fetched_at FROM api_cache WHERE name=?",
                      (name,)).fetchone()
    if row and NOW - row[1] < API_TTL:
        r = json.loads(row[0])
        r["age"] = NOW - row[1]
        return r
    r = fetch()
    if not r.get("err"):
        con.execute("INSERT OR REPLACE INTO api_cache(name,payload,fetched_at) "
                    "VALUES(?,?,?)", (name, json.dumps(r), NOW))
        con.commit()
        return r
    if row:
        stale = json.loads(row[0])
        stale["age"] = NOW - row[1]
        stale["stale"] = r["err"]
        return stale
    return r


def collect_api(con):
    creds = load_creds()
    rows = []
    if creds.get("openai_admin_key"):
        r = cached_api(con, "openai",
                       lambda: openai_spend(creds["openai_admin_key"]))
        rows.append(("openai", "org · 30d", r))
    # Any x_<label>_bearer in the credentials file is an X account on the
    # board - the fixed two-slot design predates operators with three.
    for k in sorted(creds):
        m = re.fullmatch(r"x_(\w+)_bearer", k)
        if m and creds[k]:
            label = f"x/{m.group(1)}"
            rows.append((label, "posts · cap",
                         cached_api(con, label,
                                    lambda b=creds[k], l=label: x_usage(b, l))))
    for name, mod in api_providers().items():
        if not hasattr(mod, "api_usage"):
            continue
        # Unconfigured is absence, not an error: a vendor the user never gave
        # a key does not belong on the board as a red row. --doctor carries
        # the not-set hint; detect() is exactly this question.
        if hasattr(mod, "detect"):
            try:
                if not mod.detect():
                    continue
            except Exception as e:
                SCAN_ERRORS.append(f"{name}: detect() raised {type(e).__name__}: {e}")
                continue
        def _fetch(mod=mod, name=name):
            try:
                return mod.api_usage(creds)
            except Exception as e:
                return dict(source=name, err=f"{type(e).__name__}: {e}")
        rows.append((name, "api", cached_api(con, name, _fetch)))
    return rows, creds


# ─────────────────────────── CLI scrape: plan quota ───────────────────────────

def _cap(session):
    return subprocess.run(["tmux", "capture-pane", "-p", "-t", session],
                          capture_output=True, text=True).stdout


def _wait_for(session, pattern, timeout, poll=0.7):
    """Poll the rendered pane until pattern appears. Returns (ok, last_screen).

    Fixed sleeps were the original design and they broke the first time a CLI
    shipped a slower splash screen: keystrokes landed before the input box was
    live, so the command was silently swallowed and the scrape returned a
    welcome screen. Readiness is now observed, not assumed.
    """
    rx = re.compile(pattern, re.I)
    deadline = time.time() + timeout
    screen = ""
    while time.time() < deadline:
        screen = _cap(session)
        if rx.search(screen):
            return True, screen
        time.sleep(poll)
    return False, screen


TRUST_RX = r"(Quick safety check|Is this a project you created|Yes, I trust this folder)"


def tmux_scrape(session, argv, keys, ready, done, boot=40, render=35, cwd=None):
    """Drive the TUI under tmux and return the rendered screen.

    tmux is a real terminal emulator; raw pty output from these CLIs is drawn
    character-by-character with cursor moves and cannot be regex-stripped.
    """
    if not shutil.which("tmux"):
        return None, "tmux not installed"
    run = lambda *a: subprocess.run(a, capture_output=True, text=True)
    run("tmux", "kill-session", "-t", session)
    r = run("tmux", "new-session", "-d", "-x", "210", "-y", "55",
            "-s", session, "-c", cwd or os.getcwd(), *argv)
    if r.returncode:
        return None, f"tmux: {r.stderr.strip()[:110]}"
    try:
        ok, screen = _wait_for(session, f"{ready}|{TRUST_RX}", boot)
        if re.search(TRUST_RX, screen or "", re.I):
            return screen, ("workspace-trust dialog is blocking input — run omnigauge "
                            "from a directory this CLI already trusts, or use --cwd")
        if not ok:
            return screen, f"CLI never became ready within {boot}s"
        run("tmux", "send-keys", "-t", session, keys)
        # confirm the text actually landed in the input box before pressing Enter
        typed, screen = _wait_for(session, re.escape(keys), 8)
        if not typed:
            if re.search(TRUST_RX, screen or "", re.I):
                return screen, ("workspace-trust dialog is blocking input — run omnigauge "
                                "from a directory this CLI already trusts, or use --cwd")
            return screen, "command did not reach the input box"
        run("tmux", "send-keys", "-t", session, "Enter")
        ok, screen = _wait_for(session, done, render)
        if not ok:
            return screen, f"panel did not render within {render}s"
        time.sleep(1.5)          # let the last frame settle
        return _cap(session), None
    finally:
        run("tmux", "kill-session", "-t", session)


def parse_claude(screen):
    """% USED. Line-based: a collapsed-whitespace regex silently dropped the
    weekly row when a promo line appeared between it and the next heading."""
    rows, pend = [], None
    for raw in screen.splitlines():
        line = raw.strip()
        m = re.match(r"Current (session|week)\s*(?:\(([^)]*)\))?\s*$", line, re.I)
        if m:
            if pend and pend["pct_used"] is not None: rows.append(pend)
            model = (m.group(2) or "all").strip()
            if model.lower() in ("all models", ""): model = "all"
            pend = dict(window=m.group(1).lower(), model=model, pct_used=None,
                        raw_value=None, reset_at=None)
            continue
        if pend is None: continue
        m = re.search(r"(\d+(?:\.\d+)?)\s*%\s*used", line, re.I)
        if m and pend["pct_used"] is None:
            pend["pct_used"] = float(m.group(1)); pend["raw_value"] = f"{m.group(1)}% used"; continue
        m = re.match(r"Resets\s+(.+?)\s*$", line, re.I)
        if m and pend["reset_at"] is None:
            pend["reset_at"] = m.group(1).strip()
    if pend and pend["pct_used"] is not None: rows.append(pend)
    return rows


def parse_codex(screen):
    """% LEFT natively — inverted to % USED here."""
    rows = []
    for line in screen.splitlines():
        m = re.search(r"(.*?)Weekly limit:\s*\[[^\]]*\]\s*(\d+)%\s*left\s*\(resets\s+([^)]+)\)", line)
        if not m: continue
        label = re.sub(r"[│|]", "", m.group(1)).strip() or "all"
        left = float(m.group(2))
        rows.append(dict(window="week", model=label, pct_used=100.0 - left,
                         raw_value=f"{int(left)}% left", reset_at=m.group(3).strip()))
    return rows


def parse_grok(screen):
    rows, txt = [], re.sub(r"\s+", " ", screen)
    m = re.search(r"Weekly limit\s*\(([^)]*)\).*?(\d+)\s*%", txt)
    if m:
        r = re.search(r"Resets:\s*([A-Za-z0-9 ,:]{4,30})", txt)
        rows.append(dict(window="week", model=m.group(1).strip() or "all",
                         pct_used=float(m.group(2)), raw_value=f"{m.group(2)}%",
                         reset_at=r.group(1).strip() if r else None))
    return rows


def claude_insights(screen):
    return [l.strip() for l in screen.splitlines()
            if re.match(r"\d+% of your usage", l.strip(), re.I)]


AGENTS = {
    "claude": dict(argv=["claude"], keys="/usage", parse=parse_claude,
                   source="cli_usage", expect=[("week", "all"), ("session", "all")],
                   insights=claude_insights,
                   ready=r"(Try \"|❯|>\s*$|for shortcuts)", done=r"Current (week|session)"),
    "codex":  dict(argv=["codex"], keys="/status", parse=parse_codex,
                   source="cli_status", expect=[("week", "all")],
                   ready=r"(»|Explain this codebase|/model to change)", done=r"Weekly limit:"),
    "grok":   dict(argv=["grok"], keys="/usage", parse=parse_grok,
                   source="cli_usage", expect=[("week", None)],
                   ready=r"(❯|»|Enter:send|shortcuts)", done=r"Weekly limit"),
}


CWD_OVERRIDE = None


def _scrape_one(agent):
    """The tmux half of a refresh - no DB, no printing, safe to run in a
    thread. Returns (screen, err)."""
    spec = AGENTS[agent]
    return tmux_scrape(f"am-{agent}", spec["argv"], spec["keys"],
                       spec["ready"], spec["done"],
                       cwd=(CWD_OVERRIDE
                            or (spec["cwd"]() if spec.get("cwd") else None)
                            or scrape_cwd(agent)))


def refresh(which):
    """Scrape every installed agent AT ONCE, then parse and store in order.
    Serial scraping cost ~30s per agent - a three-agent refresh was a
    ninety-second wait for what is three independent tmux sessions (each
    already has its own name, so they never collided). Parsing and the DB
    stay single-threaded and deterministic; only the waiting is parallel."""
    import threading
    con = db()
    todo = [a for a in which if installed(a)]
    for agent in which:
        if agent not in todo:
            print(f"  {s.gry}{agent:<8} not installed - skipped{s.r}")
    if not todo:
        con.close(); return
    print(f"  {s.gry}scraping {', '.join(todo)} in parallel…{s.r}", flush=True)
    results = {}
    def run(a):
        try:
            results[a] = _scrape_one(a)
        except Exception as e:
            results[a] = (None, f"{type(e).__name__}: {e}")
    threads = [threading.Thread(target=run, args=(a,), daemon=True) for a in todo]
    for th in threads: th.start()
    for th in threads: th.join()
    for agent in todo:
        spec = AGENTS[agent]
        screen, err = results.get(agent, (None, "no result"))
        print(f"  {agent:<8}", end="", flush=True)
        if err:
            print(f" {s.bred}FAILED{s.r} - {err}"); continue
        try:
            rows = spec["parse"](screen)
        except Exception as e:
            rows, err = [], f"parser raised {e}"
        dump = os.path.join(DATA, f"last-scrape-{agent}.txt")
        if not rows:
            with io.open(dump, "w", encoding="utf-8") as fh:
                fh.write(screen or "")
            print(f" {s.bred}NO QUOTA PARSED{s.r}{' - ' + err if err else ''}")
            print(f"           raw screen -> {dump}")
            continue
        got = {(r["window"], r["model"]) for r in rows}
        missing = [e for e in spec["expect"]
                   if (e[1] is not None and e not in got)
                   or (e[1] is None and not any(w == e[0] for w, _ in got))]
        # History is KEPT. Delete-then-insert left exactly one point per window,
        # which is enough for a status display and useless for a burn rate.
        # Forecasting is the whole differentiator, so the series is the asset.
        for r in rows:
            con.execute("INSERT INTO snapshots(product,source,agent,window,model,"
                        "usage_type,pct_used,raw_value,reset_at,collected_at) "
                        "VALUES(?,?,?,?,?,?,?,?,?,?)",
                        (f"{agent}_plan", spec["source"], agent, r["window"], r["model"],
                         "plan_quota", r["pct_used"], r["raw_value"], r.get("reset_at"), NOW))
        if spec.get("insights"):
            con.execute("DELETE FROM insights WHERE agent=?", (agent,))
            for t in spec["insights"](screen):
                con.execute("INSERT INTO insights(agent,text,collected_at) VALUES(?,?,?)",
                            (agent, t, NOW))
        con.commit()
        if missing:
            with io.open(dump, "w", encoding="utf-8") as fh:
                fh.write(screen)
            print(f" {s.byel}PARTIAL{s.r} - {len(rows)} row(s), missing {missing}")
            print(f"           raw screen -> {dump}")
        else:
            print(f" {s.bgrn}ok{s.r} - {len(rows)} window(s)")
    con.close()


# ────────────────────────────────── render ────────────────────────────────────

def quota_rows(con):
    """Latest reading per (agent, window, model), plus a burn rate from history."""
    q = {}
    seen = set()
    for r in con.execute(
            "SELECT agent,window,model,pct_used,raw_value,reset_at,collected_at "
            "FROM snapshots WHERE usage_type='plan_quota' ORDER BY collected_at DESC"):
        k = (r[0], r[1], r[2])
        if k in seen:
            continue
        seen.add(k)
        row = dict(window=r[1], model=r[2], pct=r[3], raw=r[4], reset=r[5], at=r[6])
        row.update(burn(con, r[0], r[1], r[2], r[3], r[6]))
        row["trend"] = trend(con, r[0], r[1], r[2], r[6])
        q.setdefault(r[0], []).append(row)
    return q


def trend(con, agent, window, model, at_now, span=24 * 3600, cells=6):
    """USED as a tiny glyph strip - burn is a slope averaged flat; the trend
    shows acceleration. Six equal time buckets over the LAST span seconds of
    the series (up to 24h) - the strip covers what history exists, so a
    fresh install shows shape after an hour of readings instead of a blank
    day. Latest reading per bucket, scaled to the strip's own min..max so a
    slow week still shows its shape. Only glyphs already load-bearing on the
    board: ░ ▒ ▓ █ (a bucket with no reading is a space). None when fewer
    than two buckets have data or the series is under ten minutes long."""
    rows = list(con.execute(
        "SELECT pct_used, collected_at FROM snapshots WHERE usage_type='plan_quota' "
        "AND agent=? AND window=? AND IFNULL(model,'')=IFNULL(?,'') AND collected_at>=? "
        "ORDER BY collected_at", (agent, window, model, at_now - span)))
    if len(rows) < 2:
        return None
    t0 = rows[0][1]
    # A strip over ten minutes of readings is flat because there is no
    # history, and looks identical to flat because nothing changed. Under an
    # hour of series the honest strip is no strip.
    if at_now - t0 < 3600:
        return None
    span = at_now - t0
    buckets = [None] * cells
    for pct, at in rows:
        i = min(cells - 1, int((at - t0) / span * cells))
        buckets[i] = pct                       # latest reading wins
    have = [b for b in buckets if b is not None]
    if len(have) < 2:
        return None
    lo, hi = min(have), max(have)
    ramp = "░▒▓█"
    out = []
    for b in buckets:
        if b is None:
            out.append(" ")
        elif hi == lo:
            out.append("▒")
        else:
            out.append(ramp[min(3, int((b - lo) / (hi - lo) * 4))])
    return "".join(out)


def burn(con, agent, window, model, pct_now, at_now, lookback=6 * 3600):
    """Percent-per-hour and time-to-empty from the snapshot series.

    Deliberately conservative:
      · needs two readings at least 10 minutes apart, else no claim is made
      · a reset (percentage went DOWN) truncates the series — burning through a
        reset boundary would otherwise read as negative usage
      · returns None rather than a guess when the rate is flat or falling
    """
    out = dict(rate=None, dry_in=None, dry_before_reset=None)
    rows = list(con.execute(
        "SELECT pct_used, collected_at FROM snapshots WHERE usage_type='plan_quota' "
        "AND agent=? AND window=? AND IFNULL(model,'')=IFNULL(?,'') AND collected_at>=? "
        "ORDER BY collected_at", (agent, window, model, at_now - lookback)))
    if len(rows) < 2:
        return out
    # truncate at the most recent reset
    start = 0
    for i in range(1, len(rows)):
        if rows[i][0] < rows[i - 1][0] - 1:
            start = i
    rows = rows[start:]
    if len(rows) < 2:
        return out
    dt = rows[-1][1] - rows[0][1]
    dp = rows[-1][0] - rows[0][0]
    if dt < 600 or dp <= 0:
        return out
    rate = dp / (dt / 3600.0)
    out["rate"] = rate
    remaining = max(0.0, 100.0 - pct_now)
    out["dry_in"] = int(remaining / rate * 3600) if rate > 0 else None
    return out


def api_row_text(name, window, r):
    """One API SPEND panel row. Four shapes, in precedence order: an error
    (loud, never a zero), a balance (prepaid vendors), usage against a cap,
    or plain spend. Factored so the widths are assertable like quota rows."""
    lead = f"  {s.accent}▸{s.r} {s.b}{name:<16.16}{s.r} {s.gry}{window:<12.12}{s.r}"
    # A stale row is a SERVED row: the last good answer, with its age and the
    # reason it could not be refreshed. Only failure-with-no-history is red.
    suff = (f"  {s.warn}· {dur(r['age'])} old · {r['stale'][:24]}{s.r}"
            if r.get("stale") else "")
    if r.get("err"):
        return (f"  {s.crit}▸{s.r} {s.b}{name:<16.16}{s.r} {s.gry}{window:<12.12}{s.r}"
                f"{s.crit}{r['err'][:39]}{s.r}")
    if r.get("balance") is not None:
        cur = f" {r['currency']}" if r.get("currency") else ""
        note = f"  {r['note']}" if r.get("note") else ""
        return lead + f"{s.wht}{r['balance']:>10}{s.r}{s.gry}{cur} balance{note}{s.r}" + suff
    if r.get("cap"):
        return (lead + f"{bar(r['pct'] or 0, 14)} {s.wht}{r['pct']:>5.2f}%{s.r}"
                f"{s.gry}  {r['used']:,.0f} / {human(r['cap'])} {r.get('unit', 'posts')}{s.r}"
                + suff)
    spend = f"${r['spend_usd']:.2f}" if r.get("spend_usd") is not None else "-"
    return (lead + f"{s.wht}{spend:>10}{s.r}{s.gry}"
            f"{(f'  {r["requests"]:,} req' if r.get('requests') else ''):<14}"
            f"{(f'  {human(r["tokens"])} tok' if r.get('tokens') else '')}{s.r}" + suff)


def since_epoch(name):
    """Window name -> epoch floor. 'today' is local midnight; 'all' is simply
    far enough back to precede every file on disk."""
    if name == "today":
        return int(time.mktime(time.strptime(time.strftime("%Y-%m-%d"), "%Y-%m-%d")))
    return int(time.time()) - {"24h": 86400, "7d": 604800,
                               "30d": 2592000, "all": 10**10}.get(name, 86400)


def _quota_layout():
    """Column plan for PLAN QUOTA at the current width. The full board needs
    94 columns of content; below that the forecast columns come out (the doom
    sub-row and --check still carry the forecast) and the bar keeps its size
    where it fits. 80-column terminals are the default case, not an edge case.
    From 103 columns a MARGIN column joins: the verdict of DRY IN against
    RESETS IN in one signed number, so the tightest plan reads at a glance."""
    full = IN >= 94
    margin = IN >= 103
    trend_col = IN >= 111
    bar_w = min(22, max(8, IN - (89 if trend_col else 81 if margin else 72 if full else 53)))
    return full, bar_w, margin, trend_col


def margin_text(dry_in, reset_s):
    """Signed headroom between running dry and the reset. Negative and red
    when the plan runs out first; positive and green when the reset comes
    first; '-' when either half is unknown."""
    if dry_in is None or reset_s is None:
        return f"{s.gry}{'-':>9}{s.r}"
    d = dry_in - reset_s
    if d < 0:
        return f"{s.crit}{'-' + dur(-d):>9}{s.r}"
    return f"{s.ok}{'+' + dur(d):>9}{s.r}"


def quota_row_text(agent, model, window, pct, rate, dry_in, reset, at, trend=None):
    """One PLAN QUOTA row, laid out for the current width. The READ column is
    the reading's age — a PARTIAL refresh leaves the missing window's previous
    reading on the board, and READ is what keeps that visible."""
    full, bar_w, margin, trend_col = _quota_layout()
    nm = short_model(agent, model)
    cd = countdown(reset) or (short_reset(reset)[:11] if reset else "-")
    pc = sev(pct) if pct >= 70 else s.wht
    reset_s = countdown(reset, _as_seconds=True) if reset else None
    doom = dry_in is not None and reset_s is not None and dry_in < reset_s
    row = (f"  {dot(pct)}  {s.b}{nm:<17.17}{s.r}{s.gry}{window:<8.8}{s.r}"
           f"{bar(pct, bar_w)}  {pc}{pct:>4.0f}%{s.r}")
    if full:
        rate_s = f"{rate:.2f}%" if rate else "-"
        dcol = s.crit if doom else s.gry
        row += f"{s.gry}{rate_s:>9}{s.r}{dcol}{dur(dry_in):>10}{s.r}"
    row += f"{s.gry}{cd:>10}{s.r}"
    if margin:
        row += margin_text(dry_in, reset_s)
    if trend_col:
        row += f"  {s.gry}{(trend or ''):<6}{s.r}"
    row += f"{s.faint}{age(at):>6}{s.r}"
    return row


def render(args):
    global NOW
    NOW = int(time.time())
    con = db()
    since = since_epoch(args.since)

    # ── masthead
    logo = f"{s.bcya}▐▌{s.r}"
    name = f"{s.b}{s.wht}OMNIGAUGE{s.r}  {s.gry}one gauge, every provider{s.r}"
    right = f"{s.gry}{os.uname().nodename} · {time.strftime('%H:%M %Z')}{s.r}"
    head = f" {logo} {name}"
    print(f"\n{head}{' ' * max(1, W - vis(head) - vis(right) - 1)}{right}")
    print(f" {s.faint}{'━' * (W - 2)}{s.r}")

    prune_filecache(con)
    q = quota_rows(con)
    flat = [(a_, r) for a_ in agent_order() for r in q.get(a_, [])]
    if not flat and not any(SCANNERS[a_][0]() for a_ in SCANNERS):
        print(f"\n  {s.b}Nothing to show yet.{s.r}")
        print(f"  {s.gry}OmniGauge reads what your agent CLIs already write to disk.{s.r}")
        print(f"  {s.accent}omnigauge --doctor{s.r}{s.gry}  shows what is connected and what to do next{s.r}\n")
        con.close(); return

    # ── headline: the single thing worth knowing
    if flat:
        worst = max(flat, key=lambda x: x[1]["pct"])
        ag, r = worst
        cd = countdown(r["reset"])
        when = f"{cd} to reset" if cd else (short_reset(r["reset"]) or "")
        if r["pct"] >= 90:
            print(f" {s.bred}▲  {short_model(ag, r['model'])} is at {r['pct']:.0f}% "
                  f"- {when} · tightest window{s.r}")
        else:
            print(f" {s.bgrn}▲  headroom everywhere · tightest is "
                  f"{short_model(ag, r['model'])} at {r['pct']:.0f}% ({when}){s.r}")

    # A configured root that is not mounted (a dev SSD pulled out) means its
    # history is missing from every panel below. Say so, or the board reads
    # as "usage went down" - a silent wrong answer.
    for _r in missing_roots():
        print(f" {s.warn}▲  configured root not mounted: {_r} "
              f"- its history is excluded this run{s.r}")

    # ── quota panel
    top("PLAN QUOTA", "normalized to % consumed")
    full_cols, bar_w, margin_col, trend_col = _quota_layout()
    hdr = (f"  {s.gry}{'':<3}{'AGENT':<17}{'WINDOW':<8}{'':<{bar_w + 2}}{'USED':>5}"
           + (f"{'BURN/h':>9}{'DRY IN':>10}" if full_cols else "")
           + f"{'RESETS IN':>10}"
           + (f"{'MARGIN':>9}" if margin_col else "")
           + (f"  {'TREND':<6}" if trend_col else "")
           + f"{'READ':>6}{s.r}")
    mid(hdr)
    if not flat:
        mid(f"  {s.gry}nothing collected - run{s.r} {s.bcya}omnigauge --refresh{s.r}")
    # A blank line between gauges: six solid bars stacked line-on-line read
    # as one color mass, red mashed straight onto green. Each row gets its
    # own air; the vendor-said and dry-warning notes stay tucked under the
    # row they belong to.
    first_row = True
    for agent in agent_order():
        rows = q.get(agent)
        if not rows:
            if agent not in AGENTS:
                continue        # no quota concept (key-based agents): no row to miss
            state = "not collected" if installed(agent) else "not installed"
            if not first_row:
                mid()
            first_row = False
            mid(f"  {s.gry}○  {agent:<17}{state}{s.r}")
            continue
        for r in sorted(rows, key=lambda x: (x["window"] != "week", x["model"] != "all")):
            if not first_row:
                mid()
            first_row = False
            mid(quota_row_text(agent=agent, model=r["model"], window=r["window"],
                               pct=r["pct"], rate=r.get("rate"), dry_in=r.get("dry_in"),
                               reset=r["reset"], at=r["at"], trend=r.get("trend")))
            reset_s = countdown(r["reset"], _as_seconds=True) if r["reset"] else None
            if (r.get("dry_in") is not None and reset_s is not None
                    and r["dry_in"] < reset_s):
                short = dur(reset_s - r["dry_in"])
                mid(f"  {s.crit}{'':<6}▲ runs dry {short} BEFORE the window resets{s.r}")
            if r["raw"] and "left" in (r["raw"] or ""):
                mid(f'  {s.gry}{"":<21}vendor said "{r["raw"]}" - inverted{s.r}')
    bot("subscription windows · no dollar balance exists for these plans")

    # ── volume panel
    def vol_table(title, note, getter):
        # Gather BEFORE drawing: the lifetime scan prints progress, and doing that
        # between mid() calls tears the frame apart.
        data = {a_: (getter(a_) if SCANNERS[a_][0]() else None)
                for a_ in agent_order()}
        # ELASTIC: the seven numeric columns split the whole inner width, so
        # the table fills its box at any terminal - fixed widths left a third
        # of a wide frame empty and CLIPPED at a strict 80.
        # THINK% = reasoning tokens as a share of output - '4.45M / 140.86M'
        # says nothing at a glance; '3.2%' says which agent over-thinks.
        # It joins from 96 columns; below that the seven columns keep the box.
        ratio = IN >= 96
        ncol = 8 if ratio else 7
        cw = max(8, (IN - 12 - (7 if ratio else 0)) // 7)
        lw = max(cw, (IN - 13 - (7 if ratio else 0)) - cw * 6)  # TOTAL absorbs
                                              # the remainder, minus one breath
        top(title, note)
        mid(f"  {s.gry}{'AGENT':<10}{'FILES':>{cw}}{'MSGS':>{cw}}{'OUTPUT':>{cw}}"
            f"{'THINK':>{cw}}" + (f"{'THINK%':>7}" if ratio else "")
            + f"{'CACHE-RD':>{cw}}{'INPUT':>{cw}}{'TOTAL':>{lw}}{s.r}")
        for agent in data:
            t = data[agent]
            if t is None:
                mid(f"  {s.gry}{agent:<10}no local transcripts{s.r}"); continue
            tot = t["total"] or (t["tin"] + t["tout"] + t["cache_read"])
            tr = (f"{t['think'] / t['tout'] * 100:>6.1f}%" if t["tout"] else f"{'-':>7}")
            mid(f"  {s.b}{s.bcya}{agent:<10}{s.r}{t['files']:>{cw},}{t['msgs']:>{cw},}"
                f"{s.bgrn}{human(t['tout']):>{cw}}{s.r}{s.wht}{human(t['think']):>{cw}}{s.r}"
                + (f"{s.gry}{tr}{s.r}" if ratio else "")
                + f"{s.gry}{human(t['cache_read']):>{cw}}{human(t['tin']):>{cw}}{s.r}"
                f"{s.b}{human(tot):>{lw}}{s.r}")
        bot("not comparable to vendor counters · different denominators")

    vol_table("TOKEN VOLUME", f"local transcripts · {args.since}",
              lambda a: window_totals(a, since))

    if not args.brief:
        life = {a_: (lifetime_totals(a_, con, progress=True) if SCANNERS[a_][0]() else None)
                for a_ in agent_order()}
        vol_table("LIFETIME", "every transcript on this machine", lambda a: life[a])

        # ── per model, across the whole corpus
        rows = []
        for a_ in agent_order():
            t = life.get(a_)
            if not t: continue
            for mo, mv in t["models"].items():
                rows.append((a_, mo, mv))
        if rows:
            grand = sum(r[2]["total"] for r in rows) or 1
            # Group by agent, agents ordered by their biggest model, models
            # by size inside the group - so the panel reads as blocks.
            best = {}
            for a_, mo, mv in rows:
                best[a_] = max(best.get(a_, 0), mv["total"])
            rows.sort(key=lambda r: (-best[r[0]], -r[2]["total"]))
            # ELASTIC: MODEL gets real room and the share bar absorbs the
            # rest of the inner width - the fixed 90-column layout clipped
            # at 80 and stranded a wide frame's right third.
            mw = min(24, max(12, IN - 53 - 16))
            sw = max(8, IN - 53 - mw)
            top("BY MODEL", "lifetime · every version you have run")
            mid(f"  {s.gry}{'AGENT':<9}{'MODEL':<{mw}}{'MSGS':>9}{'OUTPUT':>11}"
                f"{'TOTAL':>11}{'SHARE':>8}{s.r}")
            # A log axis for the bar: one 95% leader flattened every other
            # row to nothing on a linear scale, and the comparison the panel
            # exists for lives among the small ones. Four decades, so 0.01%
            # is the first cell and 100% the last. The number beside it is
            # still linear truth; the foot says which axis the bar is on.
            import math as _m
            # Air between AGENT groups, not between every bar: four gaps in a
            # sixteen-row panel. Rows inside a group are one agent's models,
            # so their closeness means something; the gap marks the change.
            # (A dot track alone read the same as none at John's font size.)
            first = True
            for a_, mo, mv in rows:
                # The quota design, bar and air alike: █ fill, ▌ cap, ░ track,
                # and a line between rows so a zero-line-gap terminal cannot
                # fuse them - the same treatment John chose for the gauges.
                if not first:
                    mid()
                first = False
                share = mv["total"] / grand * 100
                w = 0
                # Below the axis floor the bar stays EMPTY. Lighting a cell
                # for anything above zero gave a 3K-token model the same bar
                # as a 0.01% one - a gauge asserting presence where the linear
                # truth is rounding error. The number beside it says 0.0%.
                if share >= 0.01:
                    w = int(round((_m.log10(share) + 2) / 4 * sw))
                    w = max(1, min(sw, w))
                # Lower-half blocks: the bar owns the bottom of its cell, so
                # consecutive rows carry their own air without a blank line
                # between them - neither clustered nor padded.
                # A ▌ cap and a DOT track, not ░: fifteen filled tracks stacked
                # line-on-line fused into one slab; the dotted track leaves
                # each bar reading as its own row without a blank line - the
                # middle between clustered and padded.
                # ■ not █: a full block fills its cell to the line edges, so
                # rows one line apart FUSE into a slab whatever the track is
                # (John's terminal has zero line gap). The filled square sits
                # inside its cell with air above and below by design, so each
                # row's bar stands alone. Already load-bearing (the quota dot).
                cap = "▌" if 0 < w < sw else ""
                spark = (s.accent + "█" * w + cap
                         + s.faint + "░" * (sw - w - len(cap)) + s.r)
                mid(f"  {s.b}{s.accent}{a_:<9}{s.r}{s.wht}{mo[:mw - 1]:<{mw}}{s.r}"
                    f"{mv['msgs']:>9,}{s.ok}{human(mv['out']):>11}{s.r}"
                    f"{s.b}{human(mv['total']):>11}{s.r}{share:>7.1f}%  {spark}")
            bot("share bar is LOG scale, 0.01% to 100% · codex rollups attributed to the LAST model in the file")

    api_rows, creds = collect_api(con)
    if api_rows or not creds:
        top("API SPEND & CREDITS", "real dollars · never merged with plan quota")
        if not api_rows:
            mid(f"  {s.gry}no API credentials configured{s.r}")
            mid(f"  {s.gry}add one with{s.r} {s.accent}omnigauge --setup{s.r}"
                f"{s.gry} - needs an OpenAI ADMIN key (scope api.usage.read){s.r}")
        for name, window, r in api_rows:
            mid(api_row_text(name, window, r))
        bot("X dollar balances are console-only · post-cap checks do not consume quota")

    cfg = load_alerts()
    # NOT `sev` — assigning that name anywhere in render() makes it local for the
    # whole function and shadows the sev() severity helper used above.
    alert_sev, msgs = evaluate_alerts(con, cfg)
    if msgs:
        top("ALERTS", f"threshold {cfg['pct_used']}% · forecast {'on' if cfg['dry_before_reset'] else 'off'}")
        for m in msgs[:6]:
            col = s.crit if alert_sev >= 2 else s.warn
            mid(f"  {col}▲{s.r} {m}")
        bot("omnigauge --check runs this headless and exits 0/1/2 for cron")

    ins = list(con.execute("SELECT agent,text FROM insights ORDER BY id"))
    if ins:
        top("WHAT IS DRIVING USAGE", "reported by the vendor")
        for a_, t in ins:
            mid(f"  {s.bcya}▸{s.r} {t}")
        bot()
    if SCAN_ERRORS:
        uniq = sorted({e for e in SCAN_ERRORS})
        top("SCAN ERRORS", f"{len(SCAN_ERRORS)} file(s) failed to parse")
        for e in uniq[:6]:
            mid(f"  {s.crit}▸{s.r} {e[:W-8]}")
        bot("counts above are INCOMPLETE - these files were skipped")
    print()
    con.close()


# ──────────────────────────────── interactive ─────────────────────────────────

def alt_screen_ok():
    """Whether it is safe to take over the screen.

    Screen control bytes in a pipe corrupt whatever is reading them, so this is
    false for every non-interactive path. --once is the deliberate escape hatch:
    it prints a board that STAYS in scrollback, for anyone who wants to scroll
    back to it or pipe it somewhere.
    """
    if os.environ.get("OMNIGAUGE_NO_ALTSCREEN"):
        return False
    if not (sys.stdout.isatty() and sys.stdin.isatty()):
        return False
    term = os.environ.get("TERM", "")
    if not term or term == "dumb":
        return False
    return True


class RawTTY:
    """Hold raw mode for the WHOLE session.

    Toggling per-keypress looked correct and was not: between renders the
    terminal fell back to cooked mode, so anything typed during a render sat in
    the line buffer waiting for Enter and appeared dropped. Three rapid presses
    advanced one step. Entered once, released on exit.
    """
    def __enter__(self):
        import termios, tty
        self.fd = sys.stdin.fileno()
        self.old = termios.tcgetattr(self.fd)
        # cbreak, NOT raw: raw clears OPOST so "\n" no longer returns the
        # carriage and every line stair-steps to the right. cbreak still gives
        # character-at-a-time input.
        tty.setcbreak(self.fd)
        # The alternate screen, the way less/vim/htop use it. \033[2J clears what
        # you can SEE; everything it wipes has already scrolled into the
        # scrollback buffer, so a session of redraws and refreshes leaves a stack
        # of dead boards behind. On the alternate screen nothing enters history
        # at all, and on exit the terminal is restored to exactly what it looked
        # like before the program ran.
        self.alt = False
        if alt_screen_ok():
            sys.stdout.write("\033[?1049h\033[H")
            self.alt = True
        sys.stdout.write("\033[?25l")      # hide cursor
        sys.stdout.flush()
        return self

    def __exit__(self, *a):
        # Every path restores, including an unhandled exception. Leaving a user
        # on the alternate screen with no scrollback and no prompt they
        # recognise is the one thing in this tool that can damage their session,
        # and the usual reaction is to kill the terminal.
        try:
            import termios
            termios.tcsetattr(self.fd, termios.TCSADRAIN, self.old)
        finally:
            sys.stdout.write("\033[?25h")  # show cursor
            if getattr(self, "alt", False):
                sys.stdout.write("\033[?1049l")
            sys.stdout.flush()


def getkey(timeout=None):
    """One keypress. '' on timeout, None if stdin is not a tty. Assumes raw."""
    import select as _sel
    if not sys.stdin.isatty():
        return None
    fd = sys.stdin.fileno()
    r, _, _ = _sel.select([fd], [], [], timeout)
    if not r:
        return ""
    ch = os.read(fd, 1).decode("utf-8", "replace")
    if ch == "\x1b":
        r, _, _ = _sel.select([fd], [], [], 0.05)
        if r:
            os.read(fd, 8)
    return ch


def drain():
    """Discard keys buffered while a slow render was running."""
    import select as _sel
    fd = sys.stdin.fileno()
    while True:
        r, _, _ = _sel.select([fd], [], [], 0)
        if not r:
            return
        os.read(fd, 64)


# ── content panels - the site mirrors these; the doctrine lives here ─────────

def why_panel():
    """The dialect table is the USER'S OWN vendors in their own words - the
    raw_value every parser keeps beside its converted figure, exactly so the
    conversion can be checked. A static illustration was shipped here once and
    three of its four rows contradicted the shipped parsers (codex "5h rolling"
    when only the week is read; grok "raw tokens · month" when a week
    percentage is read; a gemini row for a vendor that does not ship). Static
    rows survive only for the empty state, and say what the parsers read."""
    con = db()
    seen, rows = set(), []
    for r in con.execute(
            "SELECT agent, window, model, raw_value, pct_used FROM snapshots "
            "WHERE usage_type='plan_quota' AND raw_value IS NOT NULL "
            "ORDER BY collected_at DESC"):
        k = (r[0], r[1], r[2])
        if k in seen: continue
        seen.add(k); rows.append(r)
    con.close()
    top("EVERY VENDOR SPEAKS ITS OWN DIALECT",
        "your readings, in their words" if rows else "what the parsers read")
    mid(f"  {s.gry}Ask each one how much you have left. Every answer is honest,{s.r}")
    mid(f"  {s.gry}and useless beside the others:{s.r}")
    mid()
    if rows:
        for agent, window, model, raw, pct in rows[:6]:
            name = short_model(agent, model)
            note = ("counts down · inverted" if "left" in (raw or "")
                    else "counts up") + f" · {window}"
            mid(f"    {s.warn}{name:<17.17}{s.r}{s.wht}{(raw or ''):<13.13}{s.r}"
                f"{s.faint}{note:<32.32}{s.r}{s.gry}-> {pct:.0f}% used{s.r}")
    else:
        # Empty state: an illustration - and it matches what ships. Codex is
        # read as a WEEK ("% left", inverted); grok as a week percentage;
        # claude as week and session, "% used". No vendor is named that
        # does not ship.
        for name, said, note in (("codex", "38% left", "counts down · week · inverted"),
                                 ("claude", "87% used", "counts up · week"),
                                 ("grok", "41%", "counts up · week")):
            mid(f"    {s.warn}{name:<9}{s.r}{s.wht}{said:<13}{s.r}{s.faint}{note}{s.r}")
        mid()
        mid(f"  {s.faint}(illustration - run  omnigauge --refresh  to see your own){s.r}")
    mid()
    mid(f"  {s.gry}Some count down. Some count up. Here, every window is percent{s.r}")
    mid(f"  {s.gry}consumed - and the question that matters gets answered: do you{s.r}")
    mid(f"  {s.gry}run dry BEFORE the reset does.{s.r}")
    bot("it is usually not the plan you were worried about")


def privacy_panel():
    top("THINGS IT WILL NOT DO")
    mid(f"  {s.wht}It will not ask for a key to read your agents.{s.r}")
    mid(f"  {s.gry}Plan quota and token volume come from files your agents{s.r}")
    mid(f"  {s.gry}already write to your own disk. Nothing to paste, nothing{s.r}")
    mid(f"  {s.gry}to leak.{s.r}")
    mid()
    mid(f"  {s.wht}The optional spend panel is the exception.{s.r}")
    mid(f"  {s.gry}Dollar spend needs a vendor key: stored 0600, never passed{s.r}")
    mid(f"  {s.gry}as an argument, refused outright if group- or world-readable.{s.r}")
    mid(f"  {s.gry}Leave spend off and no credential is involved at all.{s.r}")
    mid()
    mid(f"  {s.wht}It will not phone home.{s.r}")
    mid(f"  {s.gry}No telemetry, no account, no server. What you spend is{s.r}")
    mid(f"  {s.gry}nobody's business.{s.r}")
    mid()
    mid(f"  {s.wht}It will not guess.{s.r}")
    mid(f"  {s.gry}When a parse fails you get{s.r} {s.crit}NO QUOTA PARSED{s.r}{s.gry} and the raw{s.r}")
    mid(f"  {s.gry}screen - never a plausible number with nothing behind it.{s.r}")
    bot("the optional spend panel takes keys - yours, 0600, your own billing")


def donate_panel():
    top("DONATE", "optional, and it changes nothing")
    mid(f"  {s.gry}MIT, no paid tier, nothing gated on this.{s.r}")
    mid()
    mid(f"  {s.wht}Solana{s.r}")
    mid(f"  {s.warn}" + DONATE_SOL + f"{s.r}")
    mid()
    mid(f"  {s.gry}There will be no OmniGauge token from the developer of{s.r}")
    mid(f"  {s.gry}OmniGauge. No presale, no airdrop, no Phase 3.{s.r}")
    bot("if it saved you something, the trade already worked")


def about_panel():
    top("ABOUT OMNIGAUGE")
    mid(f"  {s.b}{s.wht}OmniGauge 1.0{s.r} {s.gry}- every AI plan and API account you pay for{s.r}")
    mid()
    mid(f"  {s.gry}Reads what your agent CLIs already write. Normalizes every{s.r}")
    mid(f"  {s.gry}vendor's counter to percent consumed. Invents nothing.{s.r}")
    mid()
    mid(f"  {s.gry}Written by Claude Opus 5.0. Maintained by Claude Fable 5.{s.r}")
    mid(f"  {s.gry}MIT · one Python file · github.com/omnigauge/omnigauge{s.r}")
    bot("read it before you run it")


KEYS = [("r", "refresh quota - all agents"),
        ("1 2 3", "refresh claude / codex / grok only"),
        ("w", "watch mode - auto redraw"),
        ("t", "cycle theme (ink · steel · mono · muted · vivid)"),
        ("s", "cycle window (24h · 7d · 30d · today · all)"),
        ("b", "brief - hide lifetime and by-model"),
        ("l", "legend - what each provider gets, and cannot"),
        ("d", "doctor - what is connected, what is missing"),
        ("y", "why this exists - the four dialects"),
        ("p", "privacy - what it refuses to do"),
        ("a", "about"),
        ("g", "donate"),
        ("?", "this help"),
        ("q", "quit")]


def page(fn):
    """Clear, render one content panel, wait for a key, return to the board -
    the CLI's version of the site's pull-down menus."""
    sys.stdout.write("\033[2J\033[H")
    fn()
    print(f"\n {s.gry}any key to return{s.r}")
    getkey()


def keybar(args, watching=False):
    # raw mode does not translate \n to \r\n. Two lines: board controls,
    # then the content panels - a key nobody can see is a guessing game.
    board = [f"{s.accent}r{s.r} refresh", f"{s.accent}w{s.r} watch" + ("*" if watching else ""),
             f"{s.accent}t{s.r} {args.theme}", f"{s.accent}s{s.r} {args.since}",
             f"{s.accent}b{s.r} " + ("brief" if args.brief else "full")]
    panels = [f"{s.accent}l{s.r} legend", f"{s.accent}d{s.r} doctor",
              f"{s.accent}y{s.r} why", f"{s.accent}p{s.r} privacy",
              f"{s.accent}a{s.r} about", f"{s.accent}g{s.r} donate",
              f"{s.accent}?{s.r} help", f"{s.accent}q{s.r} quit"]
    print(f" {s.gry}" + f"{s.gry} · ".join(board) + s.r)
    print(f" {s.gry}" + f"{s.gry} · ".join(panels) + s.r)


def helpscreen():
    sys.stdout.write("\033[2J\033[H")
    print(f"\n {s.b}{s.wht}OMNIGAUGE - keys{s.r}\n")
    for k, d in KEYS:
        print(f"   {s.accent}{k:<7}{s.r}{d}")
    print(f"\n {s.gry}Flags still work for scripting:{s.r}")
    print(f"   {s.gry}omnigauge --json | --brief | --since 7d | --theme mono{s.r}")
    print(f"   {s.gry}omnigauge --refresh [claude,codex,grok] | --watch [SEC]{s.r}")
    print(f"\n {s.gry}any key to return{s.r}")
    getkey()


def interactive(args):
    watching = False
    last_quota = 0
    themes = ["ink", "steel", "mono", "muted", "vivid"]
    spans = ["24h", "7d", "30d", "today", "all"]
    with RawTTY():
      try:
        while True:
            if binary_changed():
                sys.stdout.write("\033[2J\033[H")
                print(f"\n  {s.warn}omnigauge was updated on disk - restarting with the new build{s.r}")
                time.sleep(1.2)
                reexec()
            sys.stdout.write("\033[2J\033[H")
            if not _MEMO:
                # Say it is working instead of leaving a live-looking key bar on
                # screen while input is not being read.
                sys.stdout.write(f"\r {s.gry}reading transcripts…{s.r}")
                sys.stdout.flush()
            render(args)
            keybar(args, watching)
            if watching and args.quota_every and int(time.time()) - last_quota >= args.quota_every:
                which = [a for a in AGENTS if installed(a)]
                print(f"\n {s.gry}auto-refreshing quota…{s.r}")
                refresh(which); last_quota = int(time.time()); memo_clear()
                continue
            k = getkey(timeout=(args.watch or 10) if watching else None)
            if k is None: return
            k = k.lower()
            if k in ("q", "\x03", "\x1b"): break
            elif k == "r":
                which = [a for a in AGENTS if installed(a)]
                print(f"\n {s.b}refreshing{s.r} {s.gry}({', '.join(which)} · ~30s each){s.r}")
                refresh(which); last_quota = int(time.time()); memo_clear()
            elif k in "123":
                a_ = ["claude", "codex", "grok"][int(k) - 1]
                print(f"\n {s.b}refreshing {a_}{s.r}")
                refresh([a_]); last_quota = int(time.time()); memo_clear()
            elif k == "w": watching = not watching; last_quota = int(time.time())
            elif k == "t": args.theme = themes[(themes.index(args.theme) + 1) % len(themes)]; S.theme = args.theme
            elif k == "s":
                args.since = spans[(spans.index(args.since) + 1) % len(spans)]
                memo_clear()
            elif k == "b": args.brief = not args.brief
            elif k == "l": page(legend)
            elif k == "d": page(doctor)
            elif k == "y": page(why_panel)
            elif k == "p": page(privacy_panel)
            elif k == "a": page(about_panel)
            elif k == "g": page(donate_panel)
            elif k == "?": helpscreen()
      except KeyboardInterrupt:
        pass
      finally:
        print()


def scan_roots():
    """Hunt likely mount points for agent stores that discovery does not
    already cover, and print exactly how to add each find. Read-only, bounded
    to shallow globs, and user-invoked - never part of a render."""
    print(f"\n {s.b}{s.wht}OMNIGAUGE - store scan{s.r}")
    print(f" {s.faint}{'━' * (W - 2)}{s.r}\n")
    sigs = ((".claude", "projects"), (".codex", "sessions"), (".grok", "sessions"))
    known = {h for sub, _ in sigs for h in all_homes(sub)}
    pats = []
    for base in ("/mnt/*", "/media/*", "/media/*/*", "/run/media/*/*", "/Volumes/*"):
        for depth in ("", "*", "*/*"):
            pats.append(os.path.join(base, depth) if depth else base)
    hits = {}
    for pat in pats:
        for sub, sig in sigs:
            for d in glob.glob(os.path.join(pat, sub, sig)):
                store = os.path.dirname(d)
                if store in known or not os.path.isdir(d):
                    continue
                hits.setdefault(os.path.dirname(store), set()).add(sub)
    if not hits:
        print(f"  {s.gry}No stores found outside the ones already scanned.{s.r}")
        print(f"  {s.gry}If yours lives somewhere unusual, add its HOME-like parent "
              f"(the directory holding .claude/.codex/...) as a line in{s.r}")
        print(f"  {s.accent}{os.path.join(DATA, 'roots')}{s.r}\n")
        return
    for home, subs in sorted(hits.items()):
        print(f"  {s.ok}●{s.r} {s.b}{home}{s.r}  {s.gry}holds {', '.join(sorted(subs))}{s.r}")
        print(f"    {s.gry}add it:{s.r} {s.accent}echo '{home}' >> {os.path.join(DATA, 'roots')}{s.r}")
    print(f"\n  {s.gry}Roots are scanned like a second home. When the drive is out, the "
          f"board says so instead of quietly shrinking.{s.r}\n")


def doctor():
    """What is connected, what is not, and the exact next command for each gap.

    Written for someone who has never seen this tool: no step assumes knowledge
    of where a vendor hides its keys or why quota has to be scraped.
    """
    creds = load_creds()
    print(f"\n {s.b}{s.wht}OMNIGAUGE - setup check{s.r}")
    print(f" {s.faint}{'━' * (W - 2)}{s.r}\n")

    print(f"  {s.b}1. Agent CLIs{s.r} {s.gry}- token counts come from files these already write{s.r}")
    any_cli = False
    for a in agent_order():
        n = len(SCANNERS[a][0]())
        if installed(a):
            any_cli = True
            print(f"     {s.ok}●{s.r} {a:<8} installed · {n:,} transcript file(s) readable")
        elif n:
            print(f"     {s.warn}●{s.r} {a:<8} not on PATH, but {n:,} transcript(s) found - counts still work")
        else:
            print(f"     {s.faint}○ {a:<8} not installed - skipped{s.r}")
    if not any_cli:
        print(f"     {s.warn}no agent CLIs found. Install at least one, then re-run.{s.r}")

    roots = config_roots()
    if roots:
        print(f"\n     {s.gry}extra roots ({os.path.join(DATA, 'roots')} + OMNIGAUGE_ROOTS):{s.r}")
        for r in roots:
            if os.path.isdir(r):
                print(f"     {s.ok}●{s.r} {r}  mounted, scanned")
            else:
                print(f"     {s.warn}●{s.r} {r}  {s.warn}NOT MOUNTED{s.r} - its history is "
                      f"excluded until it returns")
    else:
        print(f"     {s.faint}○ no extra roots configured. Stores on another drive?{s.r}")
        print(f"       {s.accent}omnigauge --scan-roots{s.r}{s.gry} hunts mounts, or add a home-like "
              f"path per line to {os.path.join(DATA, 'roots')}{s.r}")

    print(f"\n  {s.b}2. Plan quota{s.r} {s.gry}- scraped from each CLI; no vendor offers an API for it{s.r}")
    if not shutil.which("tmux"):
        print(f"     {s.crit}● tmux missing{s.r} - required to read the CLIs' usage panels")
        print(f"       {s.accent}sudo apt install tmux{s.r}")
    else:
        con = db()
        have = {r[0] for r in con.execute("SELECT DISTINCT agent FROM snapshots "
                                          "WHERE usage_type='plan_quota'")}
        con.close()
        for a in agent_order():
            if not installed(a):
                print(f"     {s.faint}○ {a:<8} n/a{s.r}"); continue
            if a in have:
                print(f"     {s.ok}●{s.r} {a:<8} collected")
            else:
                print(f"     {s.warn}●{s.r} {a:<8} never collected - {s.accent}omnigauge --refresh {a}{s.r}")

    print(f"\n  {s.b}3. API spend{s.r} {s.gry}- optional; real dollars, separate from plan quota{s.r}")
    checks = [("openai_admin_key", "OpenAI",
               "platform.openai.com -> Settings -> Organization -> Admin keys -> Restricted, "
               "Usage API Scope = Read")]
    for k, label, how in checks:
        if creds.get(k):
            print(f"     {s.ok}●{s.r} {label:<14} configured")
        else:
            print(f"     {s.faint}○ {label:<14} not set{s.r}")
            print(f"       {s.gry}{how}{s.r}")
    x_named = sorted(k[2:-7] for k in creds
                     if re.fullmatch(r"x_(\w+)_bearer", k) and creds[k])
    if x_named:
        for n in x_named:
            print(f"     {s.ok}●{s.r} {('x/' + n):<14} configured")
    else:
        print(f"     {s.faint}○ {'x accounts':<14} none - name one in "
              f"{s.accent}omnigauge --setup{s.r}")
    for name, mod in api_providers().items():
        ok = False
        if hasattr(mod, "detect"):
            try:
                ok = bool(mod.detect())
            except Exception as e:
                SCAN_ERRORS.append(f"{name}: detect() raised {type(e).__name__}: {e}")
        if ok:
            print(f"     {s.ok}●{s.r} {name:<14} configured (provider)")
        else:
            print(f"     {s.faint}○ {name:<14} not set (provider){s.r}")
    if not all(creds.get(k) for k, _, _ in checks):
        print(f"       {s.accent}omnigauge --setup{s.r} {s.gry}(hidden input, stored 0600){s.r}")

    if os.path.exists(CREDS):
        mode = os.stat(CREDS).st_mode & 0o777
        ok = not (mode & 0o077)
        print(f"\n  {s.b}4. Credential file{s.r}")
        print(f"     {s.ok if ok else s.crit}●{s.r} {CREDS}")
        print(f"       mode {oct(mode)} - {'owner only' if ok else 'TOO OPEN, run: chmod 600 ' + CREDS}")

    print(f"\n  {s.b}Next{s.r}")
    print(f"     {s.accent}omnigauge{s.r}          {s.gry}the board - press ? for keys{s.r}")
    print(f"     {s.accent}omnigauge --refresh{s.r} {s.gry}pull fresh quota (~30s per agent){s.r}\n")


def parse_every(v):
    m = re.match(r"^(\d+)\s*([smh]?)$", str(v).strip(), re.I)
    if not m: raise argparse.ArgumentTypeError("use e.g. 30, 90s, 15m, 1h")
    return int(m.group(1)) * dict(s=1, m=60, h=3600).get(m.group(2).lower(), 1)


def watch(args):
    """Redraw on a fast clock; re-scrape quota on a slow one."""
    last_quota = 0
    try:
        while True:
            if args.quota_every and NOW_fn() - last_quota >= args.quota_every:
                which = [a for a in AGENTS if installed(a)]
                if which:
                    sys.stdout.write("\033[2J\033[H")
                    print(f"\n  {s.b}refreshing quota{s.r} {s.gry}({', '.join(which)} · ~30s each){s.r}")
                    refresh(which)
                last_quota = NOW_fn()
            if binary_changed():
                sys.stdout.write("\033[2J\033[H")
                print(f"\n  {s.warn}omnigauge was updated on disk - restarting with the new build{s.r}")
                time.sleep(1.2)
                reexec()
            sys.stdout.write("\033[2J\033[H")
            render(args)
            nxt = args.quota_every - (NOW_fn() - last_quota) if args.quota_every else None
            note = f" · next quota scrape in {max(0, nxt)//60}m" if nxt is not None else ""
            print(f"  {s.gry}watching · redraw every {args.watch}s{note} · Ctrl-C to stop{s.r}\n")
            time.sleep(args.watch)
    except KeyboardInterrupt:
        print()


def NOW_fn():
    return int(time.time())


def main():
    binary_changed()          # arm: remember the build we launched from
    load_providers()
    register_provider_quota()
    all_agents()
    p = argparse.ArgumentParser(prog=APP, description="Usage dashboard for Claude, Codex and Grok CLIs.")
    p.add_argument("--refresh", nargs="?", const="@all", metavar="AGENTS",
                   help="re-scrape quota; bare --refresh takes every installed agent, "
                        "providers included")
    p.add_argument("--since", default="24h", choices=["24h", "7d", "30d", "today", "all"])
    p.add_argument("--lifetime", action="store_true",
                   help="(now on by default; kept for compatibility)")
    p.add_argument("--brief", action="store_true",
                   help="quota + recent volume only, skip lifetime and per-model")
    p.add_argument("--json", action="store_true",
                   help="machine-readable; lifetime is null unless --lifetime is also given")
    p.add_argument("--no-color", action="store_true")
    p.add_argument("--theme", default=os.environ.get("OMNIGAUGE_THEME", "ink"),
                   choices=sorted(S.THEMES), help="ink (calm, default) · muted · mono · vivid")
    p.add_argument("--watch", nargs="?", const=10, type=int, metavar="SEC",
                   help="live redraw every SEC seconds (default 10)")
    p.add_argument("--quota-every", default="15m", type=parse_every, metavar="DUR",
                   help="how often --watch re-scrapes quota (default 15m; 0 = never)")
    p.add_argument("--check", action="store_true",
                   help="headless: evaluate alerts, notify, exit 0/1/2 (for cron)")
    p.add_argument("--quiet", action="store_true", help="with --check, print only on alert")
    p.add_argument("--providers", action="store_true",
                   help="the providers legend: what each source gets, could get, "
                        "and cannot get - with reasons")
    p.add_argument("--version", action="store_true",
                   help="print version and build fingerprint")
    p.add_argument("--scan-roots", action="store_true",
                   help="hunt mounted drives for agent stores discovery does not cover")
    p.add_argument("--doctor", action="store_true",
                   help="what is connected, what is missing, and how to fix it")
    p.add_argument("--setup", action="store_true",
                   help="store API credentials (hidden input, 0600 file)")
    p.add_argument("--once", action="store_true",
                   help="print once and exit; the board STAYS in scrollback "
                        "(the interactive board draws on the alternate screen "
                        "and leaves no trace)")
    p.add_argument("--cwd", metavar="DIR",
                   help="directory to launch the CLIs in (must already be trusted by them)")
    a = p.parse_args()
    global CWD_OVERRIDE
    CWD_OVERRIDE = a.cwd
    if a.no_color: S.on = False
    S.theme = a.theme

    if a.check:
        con = db()
        cfg = load_alerts()
        sev, msgs = evaluate_alerts(con, cfg)
        con.close()
        # A pulled drive is worth a line in cron mail, but not an alarm:
        # exit stays as the thresholds decided.
        for _r in missing_roots():
            print(f"NOTICE: configured root not mounted: {_r} "
                  "- its history is excluded")
        if msgs:
            for m in msgs:
                print(("CRITICAL: " if sev >= 2 else "WARNING: ") + m)
            notify(cfg, msgs, sev)
        elif not a.quiet:
            print("ok - every window within thresholds")
        # exit code IS the interface: 0 fine, 1 warning, 2 critical
        sys.exit(sev)

    if a.providers:
        legend(); return

    if a.doctor:
        doctor(); return
    if a.version:
        print(f"omnigauge {VERSION} build {build_id()} epoch {SCAN_EPOCH}"); return
    if a.scan_roots:
        scan_roots(); return

    if a.setup:
        import getpass
        os.makedirs(DATA, exist_ok=True)
        cur = load_creds()
        print("\n  Credentials are stored 0600 in", CREDS)
        print("  Leave blank to keep the current value. Input is hidden.\n")
        fields = [("openai_admin_key", "OpenAI ADMIN key (sk-admin-…, scope api.usage.read)"),
                  ("openrouter_api_key", "OpenRouter API key (credits + usage)"),
                  ("moonshot_api_key", "Moonshot/Kimi API key (balance)"),
                  ("deepseek_api_key", "DeepSeek API key (balance)")]
        for k, prompt in fields:
            have = " [set]" if cur.get(k) else ""
            v = getpass.getpass(f"  {prompt}{have}: ").strip()
            if v:
                cur[k] = v
        # X accounts are named, not numbered - the name is what the board
        # shows, so make it the one you want to read.
        have_x = sorted(k[2:-7] for k in cur if re.fullmatch(r"x_(\w+)_bearer", k))
        if have_x:
            print(f"  X accounts set: {', '.join(have_x)} "
                  "(re-enter a name to update its bearer)")
        while True:
            label = input("  add another X account? short NAME only, e.g. acct3 - "
                          "the bearer comes next, hidden (blank to finish): ").strip().lower()
            if not label:
                break
            if len(label) > 24 or not re.fullmatch(r"\w+", label):
                # A pasted secret lands here on reflex after a run of hidden
                # prompts. Say what happened; do not repeat it back.
                if len(label) > 24:
                    print("  that looked like a KEY, not a name - it was NOT stored, "
                          "but it DID echo: clear your scrollback. Name first.")
                else:
                    print("  letters, digits and _ only")
                continue
            v = getpass.getpass(f"  X developer bearer - {label} (hidden): ").strip()
            if v:
                cur[f"x_{label}_bearer"] = v
        fd = os.open(CREDS, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w") as fh:
            json.dump(cur, fh, indent=2)
        os.chmod(CREDS, 0o600)
        print(f"\n  saved {CREDS} (mode 600)\n")
        return

    if a.refresh:
        which = ([w for w in AGENTS if installed(w)] if a.refresh == "@all"
                 else [w.strip() for w in a.refresh.split(",") if w.strip() in AGENTS])
        print(f"\n  {s.b}refreshing quota{s.r} {s.gry}({', '.join(which)} · ~30s each){s.r}")
        refresh(which)

    if a.json:
        con = db()
        since = NOW - 86400
        out = dict(collected_at=NOW, quota=quota_rows(con),
                   window={k: window_totals(k, since) for k in SCANNERS},
                   lifetime=({k: lifetime_totals(k, con) for k in SCANNERS} if a.lifetime else None))
        print(json.dumps(out, indent=2, default=str)); con.close(); return
    if a.watch:
        watch(a); return
    # A bare invocation in a terminal is interactive; piped or flagged output
    # stays one-shot so scripts keep working.
    explicit = any([a.refresh, a.json, a.brief, a.no_color, a.since != "24h",
                    a.theme != os.environ.get("OMNIGAUGE_THEME", "ink")])
    if sys.stdout.isatty() and sys.stdin.isatty() and not explicit and not a.once:
        interactive(a); return
    render(a)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(130)
