#!/usr/bin/python3
# omarchy:summary=Print the Claude Code usage record as JSON
# omarchy:args=[--force] [--limits-only]
# omarchy:hidden=true
"""Collect Claude Code usage into one display-ready JSON record.

Everything the agents panel shows for Claude comes from this one
command: local transcript stats from ~/.claude/projects, the stats-cache and
history fallbacks for machines without transcripts, pi/omp and opencode
sessions that ran on an Anthropic provider, and the authoritative rate
limits from Anthropic's OAuth usage endpoint. The panel itself only ever
reads the JSON this prints; it never talks to disk formats or endpoints.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import os
import re
import sys
import time
import urllib.request
from pathlib import Path
from typing import Any

# The shared collector plumbing ships with the agents panel.
sys.dont_write_bytecode = True
sys.path.insert(0, os.path.join(os.environ["OMARCHY_PATH"], "shell", "plugins", "agents"))
from collectors import (  # noqa: E402
  LIMITS_ONLY_REUSE_SECONDS,
  SCAN_REUSE_SECONDS,
  HttpError,
  LimitsCache,
  Stats,
  account_entry,
  base_record,
  cache_root,
  cached_scan,
  claude_code_base_url,
  claude_code_dir,
  claude_code_index_path,
  claude_code_owner,
  claude_code_scan_match,
  emit,
  empty_bucket,
  number,
  obj,
  open_json,
  opencode_db,
  pi_session_roots,
  read_json,
  recent_dates,
  registered_accounts,
  claude_code_usage,
  scan_opencode,
  scan_pi_sessions,
  today as today_string,
  warn,
)

AGENT_ID = "claude"
AGENT_NAME = "Claude Code"
AUTH_HELP = "Run `claude auth login` to restore authoritative usage."
USAGE_ENDPOINT = "https://api.anthropic.com/api/oauth/usage"
# Anthropic rate-limits its usage endpoint readily. A refused re-check of
# numbers this recent still describes the account; only older ones are
# reported as stale.
CURRENT_ENOUGH_SECONDS = 900


def config_dir() -> Path:
  return claude_code_dir()


def expand_path(value: str) -> Path:
  return Path(os.path.expandvars(os.path.expanduser(value))).resolve()


# ---------------------------------------------------------------- local scan


# Claude Code's transcripts are read by the shared reader, through the index
# every collector that reads them shares. Claude Code fronts whatever
# ANTHROPIC_BASE_URL names, and Z.ai, Ollama, MiniMax, DeepSeek, Kimi,
# OpenRouter and Alibaba's Model Studio each document pointing it at their
# own Anthropic-compatible API;
# only what Anthropic served is Claude's, and claude_code_owner() hands the
# rest to the collector that bills it.


def scan_projects(projects_path: Path, index_file: Path | None = None, base_url: str = "", reread: bool = False) -> Stats:
  return scan_projects_read(projects_path, index_file, base_url, reread)[0]


def scan_projects_read(projects_path: Path, index_file: Path | None = None, base_url: str = "", reread: bool = False) -> tuple[Stats, bool]:
  """The transcripts' stats, and whether every transcript was read: a scan
  that missed one isn't cached."""
  stats = Stats()
  rows, read_all = claude_code_usage(AGENT_ID, projects_path, index_file, base_url, reread)
  for row in rows:
    stats.add_row(row, row.model)
  return stats, read_all


def anthropic_served(model: Any, base_url: str) -> bool:
  return claude_code_owner(model, base_url) == AGENT_ID


# ------------------------------------------------------------- local fallback
#
# A machine without transcripts on disk can still know its history: Claude
# Code keeps aggregate counters in stats-cache.json and per-prompt history in
# history.jsonl. Only consulted when projects/ holds no transcript at all: a
# scan that gave every reply to another collector has counted each prompt
# already, under whoever answered it. Neither file says who answered, so
# while ANTHROPIC_BASE_URL names another provider's host their prompts and
# sessions are that provider's, which reads no fallback, and count nowhere.

# Claude Code's built-in commands that run on their own and ask the model
# nothing (Claude Code 2.1.308's `local` and `local-jsx` commands and their
# aliases), which history.jsonl keeps as typed. The transcript reader knows
# one by the local output answering it; history.jsonl keeps no output, so
# only the name tells. A custom command, a skill, and a built-in that sends a
# prompt (/init, /statusline) are prompts.
CLAUDE_CODE_LOCAL_COMMANDS = frozenset((
  "add-dir", "advisor", "agents", "allowed-tools", "android", "app", "artifacts", "autocompact", "autofix-pr",
  "auto-mode-setup", "background", "bashes", "bg", "branch", "break-reminder", "breaks", "brief", "btw", "bug", "cd",
  "checkpoint", "clear", "cloud-plugins", "color", "compact", "config", "context", "copy", "cost", "daemon",
  "design-consent", "design-login", "design-revoke", "desktop", "diff", "downtime", "effort", "exit", "export",
  "extra-usage", "fast", "feedback", "focus", "fork", "goal", "heapdump", "help", "hooks", "ide", "import", "install",
  "install-github-app", "install-slack-app", "ios", "list-agents", "login", "logout", "loops", "marketplace", "mcp",
  "memory", "memory-pause", "mobile", "model", "name", "new", "output-style", "passes", "pause-memory", "peers",
  "permissions", "plan", "plugin", "plugins", "powerup", "privacy-settings", "pro-trial-expired", "quit", "radio",
  "rate-limit-options", "rc", "recap", "reload-plugins", "reload-skills", "remote", "remote-control", "remote-env",
  "rename", "reset", "restart", "resume", "rewind", "scroll-speed", "session", "settings", "setup-bedrock",
  "setup-vertex", "skill-doctor", "skills", "stats", "status", "stickers", "stop", "subtask", "tasks", "teleport",
  "terminal-setup", "theme", "toggle-memory", "tp", "tui", "ultraplan", "ultrareview", "undo", "update", "upgrade",
  "usage", "usage-credits", "version", "voice", "web-setup", "wellbeing", "workflow-launch-exec", "workflows",
))


def history_prompt(display: Any) -> bool:
  """Whether a history.jsonl entry's text was a prompt: anything but a
  built-in local command, with or without arguments."""
  text = str(display or "").strip()
  if not text.startswith("/"):
    return True
  name = text[1:].split(None, 1)[0].lower() if len(text) > 1 else ""
  return name not in CLAUDE_CODE_LOCAL_COMMANDS


def has_transcripts(projects: Path) -> bool:
  try:
    return next(projects.rglob("*.jsonl"), None) is not None
  except OSError:
    return False


def stats_cache_fallback(claude_dir: Path, base_url: str = "") -> dict[str, Any] | None:
  try:
    data = json.loads((claude_dir / "stats-cache.json").read_text(encoding="utf-8"))
  except Exception:
    return None
  # A file or a field of the wrong shape reads as nothing, never a crash.
  if not isinstance(data, dict):
    return None

  def entries(key: str) -> list[dict[str, Any]]:
    value = data.get(key)
    return [entry for entry in value if isinstance(entry, dict)] if isinstance(value, list) else []

  # Tokens by model per day, the first entry for a day standing, as the
  # record's recentDays and today's tokens read them. Claude Code keys these
  # by UTC date (toISOString) and stops at yesterday's, so each is read as
  # the local day of the same date, the one it mostly overlaps: off by the
  # UTC offset at either end, and today's tokens are usually none.
  tokens_by_day: dict[str, dict[str, int]] = {}
  for entry in entries("dailyModelTokens"):
    date = entry.get("date")
    if isinstance(date, str) and date not in tokens_by_day:
      tokens_by_day[date] = {
        str(model): number(tokens) for model, tokens in obj(entry.get("tokensByModel")).items()
        if number(tokens) > 0 and anthropic_served(model, base_url)
      }
  today_tokens = tokens_by_day.get(today_string(), {})
  recent_days = [{"date": date, "messageCount": sum(tokens_by_day.get(date, {}).values())} for date in recent_dates()]

  # The four token kinds of each model, each a count number() trusts; the
  # file's other fields (costUSD, webSearchRequests) aren't the record's.
  model_usage = {
    str(model): {field: number(bucket.get(field)) for field in empty_bucket()}
    for model, bucket in obj(data.get("modelUsage")).items()
    if isinstance(bucket, dict) and anthropic_served(model, base_url)
  }

  # Prompts and sessions name no model, so they are Claude's unless Claude
  # Code is sent to another provider's host.
  prompts_here = anthropic_served(None, base_url)
  daily_activity = entries("dailyActivity") if prompts_here else []
  active_dates = sorted({str(day.get("date")) for day in daily_activity if number(day.get("messageCount")) > 0 and day.get("date")})
  today_prompts, today_sessions = today_prompts_from_history(claude_dir, base_url)

  return {
    "todayPrompts": today_prompts,
    "todaySessions": today_sessions,
    "todayTotalTokens": sum(today_tokens.values()),
    "todayTokensByModel": today_tokens,
    "recentDays": recent_days,
    "modelUsage": model_usage,
    "totalPrompts": number(data.get("totalMessages")) if prompts_here else 0,
    "totalSessions": number(data.get("totalSessions")) if prompts_here else 0,
    "activeDays": len(active_dates),
    "activeDates": active_dates,
  }


def today_prompts_from_history(claude_dir: Path, base_url: str = "") -> tuple[int, int]:
  if not anthropic_served(None, base_url):
    return 0, 0
  prompts = 0
  sessions: set[str] = set()
  start_of_day = dt.datetime.combine(dt.datetime.now().date(), dt.time.min).timestamp() * 1000
  try:
    with (claude_dir / "history.jsonl").open("r", encoding="utf-8", errors="replace") as handle:
      lines = handle.readlines()
  except Exception:
    return 0, 0

  for line in reversed(lines):
    line = line.strip()
    if not line:
      continue
    try:
      entry = json.loads(line)
    except Exception:
      continue
    if not isinstance(entry, dict):
      continue
    if number(entry.get("timestamp")) < start_of_day:
      break
    if not history_prompt(entry.get("display")):
      continue
    prompts += 1
    if entry.get("sessionId"):
      sessions.add(str(entry.get("sessionId")))
  return prompts, len(sessions)



# ------------------------------------------------------- pi, omp, and opencode
#
# These agents can consume a Claude subscription without writing native
# Claude Code transcripts: pi's and omp's JSONL sessions, and opencode's
# message database, carry the provider, model, and token usage of every
# reply. The shared readers find them (pi and omp with `rg --no-ignore`, so a
# $HOME that is a git checkout hides nothing) and this keeps the Anthropic
# ones.

PI_PATTERN = r'"provider"\s*:\s*"anthropic"'


# Exact match: provider ids are free-form, and a custom "anthropic-proxy"
# gateway is not this subscription.
def billed_here(provider: str, model: str, entry: dict) -> bool:
  return provider == "anthropic"


def opencode_model(model: str) -> str:
  return str(model or "claude").rstrip("/").split("/")[-1] or "claude"


def add_shared_rows(stats: Stats, pi_rows: list, db_rows: list) -> None:
  for row in pi_rows:
    stats.add_row(row, row.model or "claude")
  for row in db_rows:
    stats.add_row(row, opencode_model(row.model))


def scan_local(claude_dir: Path, index_file: Path | None, reread: bool = False) -> tuple[dict[str, Any], bool]:
  """Local stats from every source, and whether every source was read in full
  (safe to cache). Transcripts come first; only a machine without any falls
  back on Claude Code's own aggregate counters."""
  base_url = claude_code_base_url(claude_dir)
  projects = claude_dir / "projects"
  stats, transcripts_complete = scan_projects_read(projects, index_file, base_url, reread)
  pi_rows, pi_complete = scan_pi_sessions(PI_PATTERN, billed_here)
  # The closing quote makes the database's prefilter match the whole id.
  db_rows, db_complete = scan_opencode(['anthropic"'], billed_here)
  complete = transcripts_complete and pi_complete and db_complete

  if stats.total_prompts > 0 or stats.model_usage or has_transcripts(projects):
    add_shared_rows(stats, pi_rows, db_rows)
    return stats.snapshot(), complete

  local = stats_cache_fallback(claude_dir, base_url)
  if local is None:
    local = stats.snapshot()
    # No transcripts and no aggregate cache, but history.jsonl alone can
    # still put numbers on today.
    today_prompts, today_sessions = today_prompts_from_history(claude_dir, base_url)
    if today_prompts or today_sessions:
      local.update(todayPrompts=today_prompts, todaySessions=today_sessions)
  extra = Stats()
  add_shared_rows(extra, pi_rows, db_rows)
  if extra.total_prompts > 0:
    local = merge_stats(local, extra.snapshot())
  return local, complete


def local_stats(claude_dir: Path, max_age: float, use_index: bool = True) -> dict[str, Any]:
  # Without the index every transcript is read again, and the index rebuilt
  # in place: the other collectors reading Claude Code share it.
  try:
    index_file = claude_code_index_path(claude_dir / "projects")
  except Exception as exc:
    warn(f"scan index unavailable ({exc})")
    index_file = None
  # The digest covers every data path the scan reads: the transcripts, the
  # config home's fallbacks, the pi/omp session roots, and the opencode DB.
  sources = [claude_dir / "projects", claude_dir, *pi_session_roots(), opencode_db()]
  # Where Claude Code is sent decides whose its turns are, so a scan only
  # answers for the endpoint it was made under.
  return cached_scan(AGENT_ID, sources, max_age, lambda: scan_local(claude_dir, index_file, not use_index), match=claude_code_scan_match(claude_dir))


def merge_stats(base: dict[str, Any], extra: dict[str, Any]) -> dict[str, Any]:
  merged = dict(base)
  for key in ("todayPrompts", "todaySessions", "todayTotalTokens", "totalPrompts", "totalSessions"):
    merged[key] = number(base.get(key)) + number(extra.get(key))

  combined = dict(base.get("todayTokensByModel") or {})
  for model, count in (extra.get("todayTokensByModel") or {}).items():
    combined[model] = number(combined.get(model)) + number(count)
  merged["todayTokensByModel"] = combined

  usage = {model: dict(bucket) for model, bucket in (base.get("modelUsage") or {}).items()}
  for model, bucket in (extra.get("modelUsage") or {}).items():
    target = usage.setdefault(model, empty_bucket())
    for field, count in (bucket or {}).items():
      target[field] = number(target.get(field)) + number(count)
  merged["modelUsage"] = usage

  by_date: dict[str, int] = {}
  for source in (base.get("recentDays") or [], extra.get("recentDays") or []):
    for day in source:
      date = str((day or {}).get("date") or "")
      if date:
        by_date[date] = by_date.get(date, 0) + number((day or {}).get("messageCount"))
  merged["recentDays"] = [{"date": date, "messageCount": by_date[date]} for date in sorted(by_date)]

  # Sources overlap in time, so union dates rather than summing counts. A
  # fallback that only knows a count still bounds the answer from below.
  dates = set(base.get("activeDates") or []) | set(extra.get("activeDates") or [])
  merged["activeDates"] = sorted(dates)
  merged["activeDays"] = max(len(dates), number(base.get("activeDays")), number(extra.get("activeDays")))
  return merged


# ------------------------------------------------------------------- limits


# The access token, its expiry, and the display-safe plan label from the
# CLI's login. Nothing else leaves the credential store: the token goes
# nowhere but the Authorization header of the limits probe, and only the
# plan label may travel into the printed record.
def oauth_login(claude_dir: Path) -> tuple[str, int, int, str]:
  login = read_json(claude_dir / ".credentials.json").get("claudeAiOauth")
  if not isinstance(login, dict):
    return "", 0, 0, ""
  # The profile's tier is asked first, but only one that names a multiplier
  # outranks the credentials'. The subscription still leads the label, so a
  # Team seat keeps reading Team.
  tier = profile_tier(claude_dir)
  if not re.search(r"max_(\d+x)", tier, re.IGNORECASE):
    tier = str(login.get("rateLimitTier") or "")
  plan = plan_label(tier, str(login.get("subscriptionType") or ""))
  return (
    str(login.get("accessToken") or ""),
    number(login.get("expiresAt")),
    number(login.get("refreshTokenExpiresAt")),
    plan,
  )


# The credentials carry the tier their token was minted with, and an upgrade
# does not restamp it: a refresh rewrites the file and keeps the old value, so
# an account moved to Max 20x keeps reading Max 5x. The CLI's own profile is
# refreshed against the account, so the tier it states there is asked first.
def profile_tier(claude_dir: Path) -> str:
  account = read_json(profile_path(claude_dir)).get("oauthAccount")
  if not isinstance(account, dict):
    return ""
  # A seat with no tier of its own is limited by its organization, which is
  # the shape a personal Max subscription takes.
  for key in ("userRateLimitTier", "organizationRateLimitTier"):
    tier = str(account.get(key) or "")
    if tier:
      return tier
  return ""


# The CLI's profile for a config home: a legacy .config.json inside it, else
# ~/.claude.json for the default home and <home>/.claude.json for one it was
# pointed at with CLAUDE_CONFIG_DIR. A copy anywhere else is never refreshed,
# and another home's profile describes another account.
def profile_path(home: Path) -> Path:
  legacy = home / ".config.json"
  if legacy.exists():
    return legacy
  configured = os.environ.get("CLAUDE_CONFIG_DIR")
  if home == expand_path("~/.claude") and not (configured and expand_path(configured) == home):
    return Path.home() / ".claude.json"
  return home / ".claude.json"


# The rate-limit tier names the ceiling, not the plan: a Team premium seat
# runs on `default_claude_max_5x` while paying for Team. Lead with the
# subscription and keep the multiplier as its qualifier, so Max still reads
# "Max 5x" and a Team seat reads "Team 5x" rather than passing as Max.
def plan_label(tier: str, subscription: str) -> str:
  plan = subscription[0].upper() + subscription[1:] if subscription else ""
  match = re.search(r"max_(\d+x)", tier, re.IGNORECASE) if tier else None
  if not match:
    return plan
  return (plan or "Max") + " " + match.group(1)


def parse_utilization(value: Any) -> float:
  try:
    return float(str(value).strip().replace("%", ""))
  except Exception:
    return float("nan")


# Every utilization and percent this endpoint answers is a percentage: Claude
# Code's own /usage renders each as `${Math.floor(utilization)}% used`, so 0.6
# is 0.6%, not 60%.
def normalize_utilization(value: Any) -> float:
  n = parse_utilization(value)
  if not (n >= 0):
    return -1.0
  return min(1.0, n / 100.0)


def normalize_reset_at(value: Any) -> str:
  if value is None:
    return ""
  raw = str(value).strip()
  if raw == "":
    return ""
  if raw.isdigit():
    ts = int(raw)
    if ts < 1e12:
      ts *= 1000
    try:
      return dt.datetime.fromtimestamp(ts / 1000, dt.timezone.utc).isoformat()
    except Exception:
      return raw
  try:
    parsed = dt.datetime.fromisoformat(raw.replace("Z", "+00:00"))
    return parsed.isoformat()
  except Exception:
    return raw


def usage_bucket(payload: dict[str, Any], key: str) -> dict[str, Any] | None:
  bucket = payload.get(key)
  return bucket if isinstance(bucket, dict) else None


# An entry's `kind` names its window the way the flat buckets' keys do
# ("weekly_scoped", "five_hour_scoped"). The panel reads a window out of free
# text, which cannot survive a model name like "Opus 5 (1M context)" — the
# "1M" reads as a one-minute window — so the window is settled here instead
# and travels as an explicit title. It is capitalized the way the flat windows
# title themselves, so "Fable Weekly" sits beside "Weekly" rather than under it.
def scoped_window(kind: str) -> str:
  text = kind.lower()
  if "month" in text:
    return "Monthly"
  if "week" in text or "day" in text:
    return "Weekly"
  if "hour" in text or "session" in text:
    return "Session"
  return ""


# Alongside the flat buckets, the payload carries a `limits` array, and that
# array is the only place a model-scoped allowance shows up — a weekly window
# that only Fable draws from, say. The matching legacy keys
# (`seven_day_opus`, `seven_day_sonnet`, …) stayed behind at null, so a
# collector that reads buckets alone silently drops a limit the account is
# actually spending against. A model can hold more than one scoped window, and
# only the pair of model and window tells them apart, so both make the title
# and both make the key that keeps a repeat out.
def scoped_limits(payload: dict[str, Any]) -> list[dict[str, Any]]:
  entries = payload.get("limits")
  if not isinstance(entries, list):
    return []
  out: list[dict[str, Any]] = []
  seen: set[tuple[str, str]] = set()
  for entry in entries:
    if not isinstance(entry, dict):
      continue
    scope = entry.get("scope")
    model = scope.get("model") if isinstance(scope, dict) else None
    if not isinstance(model, dict):
      continue
    # A display name is what the panel wants, but an entry carrying only an id
    # still names a window worth showing.
    name = str(model.get("display_name") or model.get("id") or "").strip()
    kind = str(entry.get("kind") or "").strip()
    if name == "" or (name, kind) in seen:
      continue
    percent = normalize_utilization(entry.get("percent"))
    if percent < 0:
      continue
    seen.add((name, kind))
    window = scoped_window(kind)
    title = name + " " + window if window else name
    out.append({
      "label": title,
      "title": title,
      "percent": percent,
      "resetsAt": normalize_reset_at(entry.get("resets_at")),
    })
  return out



def probe_limits(access_token: str) -> dict[str, Any]:
  request = urllib.request.Request(
    USAGE_ENDPOINT,
    headers={
      "Authorization": "Bearer " + access_token,
      "anthropic-beta": "oauth-2025-04-20",
      "Accept": "application/json",
    },
  )
  try:
    payload = open_json(request, timeout=10)
  except HttpError as error:
    if error.status:
      retry_after = error.headers.get("retry-after", "") if error.headers else ""
      if error.status == 429:
        help_text = "Anthropic's usage endpoint is rate limiting checks right now" + (
          f" (retry after {retry_after}s)" if retry_after else ""
        ) + ". Local Claude Code stats are still shown."
        return {"ok": False, "rateLimited": True, "helpText": help_text}
      if error.status in (401, 403):
        return {
          "ok": False,
          "refused": True,
          "helpText": "Anthropic refused Claude Code's sign-in. Start Claude Code, or run `claude auth login`, to sign in again.",
        }
      help_text = f"Anthropic's usage endpoint returned status {error.status}. Local Claude Code stats are still shown."
      return {"ok": False, "helpText": help_text}
    if error.transport:
      # A transport failure reached no server at all — no route, no DNS. Any
      # real answer, including an error status, is a server we should stop
      # pestering; this is not.
      return {
        "ok": False,
        "transport": True,
        "helpText": "Couldn't reach Anthropic's usage endpoint. Retrying shortly. Local Claude Code stats are still shown.",
      }
    return {"ok": False, "helpText": "Anthropic's usage endpoint returned an invalid answer. Local Claude Code stats are still shown."}
  if not isinstance(payload, dict):
    payload = {}

  # The account's week is `seven_day`, the row Claude Code's own /usage titles
  # "Current week (all models)". `seven_day_oauth_apps` is a side bucket it
  # groups with cowork and omelette and never shows as the weekly limit.
  weekly = usage_bucket(payload, "seven_day")
  session = usage_bucket(payload, "five_hour")

  limits = []
  if session is not None:
    percent = normalize_utilization(session.get("utilization"))
    if percent >= 0:
      limits.append({"label": "Session (5-hour)", "percent": percent, "resetsAt": normalize_reset_at(session.get("resets_at"))})
  if weekly is not None:
    percent = normalize_utilization(weekly.get("utilization"))
    if percent >= 0:
      limits.append({"label": "Weekly (7-day)", "percent": percent, "resetsAt": normalize_reset_at(weekly.get("resets_at"))})
  limits.extend(scoped_limits(payload))

  if not limits:
    return {"ok": False, "helpText": "Anthropic's usage endpoint returned no limits. Local Claude Code stats are still shown."}
  return {"ok": True, "limits": limits}


def collect_limits(access_token: str, expires_at_ms: int, refresh_expires_at_ms: int, force: bool, cache_name: str = "claude-limits.json") -> dict[str, Any]:
  # `live` says the numbers came from Anthropic just now (or within the reuse
  # window) rather than from a cache kept past a failed or impossible probe.
  # `fetchedAtMs` says when they were measured, so a stale account can say
  # how old they are.
  result = {"limits": [], "usageStatusText": "", "authHelpText": AUTH_HELP, "live": False, "fetchedAtMs": 0}

  # A panel that is opened and shut repeatedly must not turn into a request
  # per flick, so recent probe results are reused for a short window — and
  # kept as the answer of record when a later probe fails. Every account
  # keeps its own, so one account's allowance never stands in for another's.
  # Elapsed windows come back at zero. A home that says who it's signed in as
  # keeps claude-limits-<subscription>.json, and one that doesn't the
  # single-account claude-limits.json.
  cache = LimitsCache.at(cache_root() / cache_name)
  fallback, result["fetchedAtMs"] = cache.kept()

  # Probing needs a live token and only the Claude Code CLI can mint one: it
  # refreshes the credential file when it runs, so a machine left alone long
  # enough finds the saved token lapsed. Say so — an empty limits list with
  # nothing else set hides the whole section and explains nothing — and keep
  # showing the last numbers, resetting elapsed windows to zero.
  if access_token == "":
    result["limits"] = fallback
    result["usageStatusText"] = "Waiting for auth"
    return result
  if expires_at_ms > 0 and expires_at_ms <= time.time() * 1000:
    result["limits"] = fallback
    # A lapsed access token is routine, not a sign-out: it lives about eight
    # hours, and the CLI mints a new one from the refresh token the next time
    # it runs. Only a lapsed *refresh* token is a sign-in a person must redo,
    # and saying so either way sends people to `claude auth login` nightly --
    # which does not restore anything the CLI would not have restored itself.
    if refresh_expires_at_ms > time.time() * 1000:
      result["usageStatusText"] = "Limits paused"
      result["authHelpText"] = (
        "Claude Code's access token has lapsed"
        + (" — showing the last known limits." if fallback else ".")
        + " Start Claude Code to refresh it; signing in again is not needed."
      )
      return result
    result["usageStatusText"] = "Sign-in expired"
    result["authHelpText"] = (
      "Claude Code's saved sign-in expired"
      + (" — showing the last known limits." if fallback else ".")
      + " Start Claude Code, or run `claude auth login`, to refresh it."
    )
    return result

  # --force is a person asking for fresh numbers, so it skips the reuse window
  # entirely; the interval is there to absorb repeated panel opens, not to
  # overrule someone who pressed refresh.
  if cache.fresh(force):
    result["limits"] = fallback
    result["live"] = True
    result["authHelpText"] = ""
    return result

  probe = probe_limits(access_token)
  if probe["ok"]:
    result["limits"] = probe["limits"]
    result["live"] = True
    result["authHelpText"] = ""
    result["fetchedAtMs"] = cache.save(probe["limits"])
    return result

  # The first probe after login often fires before DHCP has handed out a
  # route. Ask the shell to try again sooner than its regular interval.
  if probe.get("transport"):
    result["retryAdvised"] = True
  if fallback:
    result["limits"] = fallback
    result["authHelpText"] = probe["helpText"]
    # Anthropic rate-limits this endpoint readily, and a 429 only says "ask
    # later": numbers measured minutes ago still describe the account. Any
    # other failure leaves them stale, as the collector contract has it.
    result["live"] = bool(probe.get("rateLimited")) and cache.younger_than(CURRENT_ENOUGH_SECONDS)
  else:
    result["usageStatusText"] = "Claude limits unavailable"
    result["authHelpText"] = probe["helpText"]
  if probe.get("refused"):
    result["usageStatusText"] = "Sign-in expired"
  return result


# ---------------------------------------------------------------- accounts


def current_account_id(home: Path) -> str:
  account = read_json(profile_path(home)).get("oauthAccount")
  return str(account.get("accountUuid") or "") if isinstance(account, dict) else ""


def account_limits(account: dict[str, Any], force: bool) -> dict[str, Any]:
  home = expand_path(account["home"]) if account.get("home") else expand_path("~/.claude")
  access_token, expires_at_ms, refresh_expires_at_ms, plan = oauth_login(home)
  # Keyed by the subscription, not the label: forget "Work" and add a different
  # "Work", and the new one must never fall back on the old one's allowance.
  # Who the home is signed in as now, not when it was added: a `claude auth
  # login` run in it by hand makes it a different subscription.
  identity = current_account_id(home)
  key = re.sub(r"[^A-Za-z0-9_-]", "", identity or str(account.get("accountId") or "")) or account["id"]
  # The primary home is keyed the same way once it says who it's signed in
  # as, so signing it in to another subscription never inherits the last
  # one's numbers; until then it keeps the single-account cache.
  if account.get("primary") and not identity:
    cache_name = "claude-limits.json"
  else:
    cache_name = f"claude-limits-{key}.json"
  limits = collect_limits(access_token, expires_at_ms, refresh_expires_at_ms, force, cache_name)
  return account_entry(account, limits, plan)


# -------------------------------------------------------------------- record


def main() -> int:
  parser = argparse.ArgumentParser()
  parser.add_argument("--force", action="store_true", help="rescan transcripts from scratch and re-probe limits, ignoring caches")
  parser.add_argument("--limits-only", action="store_true", help="reuse any recent transcript scan; only the limits probe must be fresh")
  parser.add_argument("--cache-seconds", type=float, default=SCAN_REUSE_SECONDS)
  args = parser.parse_args()

  claude_dir = config_dir()
  scan_age = 0 if args.force else (LIMITS_ONLY_REUSE_SECONDS if args.limits_only else args.cache_seconds)
  stats = local_stats(claude_dir, scan_age, use_index=not args.force)

  registered = registered_accounts(AGENT_ID)
  accounts = [account_limits(account, args.force) for account in registered]
  current = next((a for a in accounts if a["active"]), None)
  if current:
    # The top-level fields keep describing the account new sessions use, so
    # the bar icon and anything else reading one set of limits stays right.
    plan = current["plan"]
    limits = {key: current[key] for key in ("limits", "usageStatusText", "authHelpText", "retryAdvised")}
    stale, fetched_at = current["stale"], current["fetchedAt"]
  else:
    access_token, expires_at_ms, refresh_expires_at_ms, plan = oauth_login(claude_dir)
    # Keyed by who the home is signed in as, as a registered primary is, so
    # signing in to another subscription never shows the last one's numbers
    # and registering the home later keeps its cache.
    identity = re.sub(r"[^A-Za-z0-9_-]", "", current_account_id(claude_dir))
    cache_name = f"claude-limits-{identity}.json" if identity else "claude-limits.json"
    limits = collect_limits(access_token, expires_at_ms, refresh_expires_at_ms, args.force, cache_name)
    stale, fetched_at = not limits["live"] and bool(limits["limits"]), limits["fetchedAtMs"]
    # Asking for a sign-in only makes sense where Claude Code is in use: its
    # config home or profile exists, or there is local usage. A machine that
    # never ran it prints the unused record, which the panel skips.
    in_use = (
      claude_dir.exists()
      or profile_path(claude_dir).exists()
      or number(stats.get("totalPrompts")) > 0
      or bool(stats.get("modelUsage"))
      or number(stats.get("todayPrompts")) > 0
      or len(limits["limits"]) > 0
    )
    if not in_use:
      limits = {"limits": [], "usageStatusText": "", "authHelpText": ""}
      stale, fetched_at = False, 0

  record = base_record(
    AGENT_ID,
    AGENT_NAME,
    ready=number(stats.get("totalPrompts")) > 0 or bool(stats.get("modelUsage")) or len(limits["limits"]) > 0,
    hasLocalStats=True,
    tierLabel=plan,
    usageStatusText=limits["usageStatusText"],
    authHelpText=limits["authHelpText"],
    limits=limits["limits"],
    # Numbers kept past a failed probe, and when they were measured, so the
    # panel can tell them apart from ones checked just now.
    limitsStale=stale,
    limitsFetchedAt=fetched_at,
  )
  if limits.get("retryAdvised"):
    record["retryAdvised"] = True
  if accounts:
    record["accountSwitch"] = registered[0]["switch"]
    record["accounts"] = accounts
  record.update(stats)
  emit(record, sort_keys=True)
  return 0


if __name__ == "__main__":
  raise SystemExit(main())
