#!/usr/bin/env python3

"""Print one display-ready Claude Code usage record as JSON.

Everything the usage panel shows for Claude comes from here: local transcript
statistics from the Claude Code projects directory, the stats-cache and history
fallbacks for machines with no transcripts, and the authoritative rate limits
from Anthropic's OAuth usage endpoint. The panel reads only the JSON this
prints; it never learns a disk format or an endpoint shape.

Adapted from Omarchy (bin/omarchy-agent-usage-claude).

Copyright (c) David Heinemeier Hansson

Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

── What this deliberately does NOT do ────────────────────────────────────────

It never refreshes the OAuth token, and it never writes to the credentials file.

That token expires about hourly and Claude Code refreshes it on demand. If this
refreshed it too, two processes would be rotating one credential: a refresh that
rotates the refresh token invalidates the other holder's copy, and the failure
mode is being silently signed out of Claude Code by a status widget. No bar
indicator is worth that. So this reads the token, uses it while it is valid, and
reports an honest waiting state when it is not.

── The token ─────────────────────────────────────────────────────────────────

It never reaches argv. The request is made in this process with urllib, so there
is no child process whose /proc/<pid>/cmdline could carry a credential -- the
same rule panama-pick follows for passwords and panama-sudo for the MOK hash,
kept by having no subprocess at all rather than by hiding an argument.

It never reaches the output either. The record below carries percentages,
timestamps and token counts; the only thing from the credential store that may
travel into it is the display-safe plan label.

── Seams ─────────────────────────────────────────────────────────────────────

  CLAUDE_CONFIG_DIR             Claude Code's config directory (~/.claude)
  PANAMA_AGENT_CREDENTIALS      the credentials file, for tests
  PANAMA_AGENT_USAGE_ENDPOINT   the usage endpoint, for tests
  PANAMA_AGENT_USAGE_CACHE      the scan/probe cache directory

The output path is not a seam here: this prints, and panama-agent-usage-update
owns the atomic write into $XDG_STATE_HOME/panama/agents/usage/.
"""

from __future__ import annotations

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

AGENT_ID = "claude"
AGENT_NAME = "Claude Code"
AUTH_HELP = "Run `claude auth login` to restore authoritative usage."
DEFAULT_ENDPOINT = "https://api.anthropic.com/api/oauth/usage"

# A panel that is opened and shut repeatedly must not turn into a request per
# flick, so a recent probe result is reused for this long.
PROBE_MIN_INTERVAL_SECONDS = 15

# A normal run reuses a scan only long enough to dedup concurrent collectors
# (the fan-out backs one off per agent). --limits-only promises fresh limits and
# nothing else, so it may reuse a scan for far longer.
SCAN_REUSE_SECONDS = 20
LIMITS_ONLY_REUSE_SECONDS = 900


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


def config_dir() -> Path:
    return expand_path(os.environ.get("CLAUDE_CONFIG_DIR") or "~/.claude")


def credentials_path(claude_dir: Path) -> Path:
    override = os.environ.get("PANAMA_AGENT_CREDENTIALS")
    return expand_path(override) if override else claude_dir / ".credentials.json"


def endpoint() -> str:
    return os.environ.get("PANAMA_AGENT_USAGE_ENDPOINT") or DEFAULT_ENDPOINT


def cache_root() -> Path:
    override = os.environ.get("PANAMA_AGENT_USAGE_CACHE")
    root = expand_path(override) if override else (
        Path(os.environ.get("XDG_CACHE_HOME") or (Path.home() / ".cache")) / "panama" / "agent-usage"
    )
    root.mkdir(parents=True, exist_ok=True)
    return root


def date_string(value: dt.date) -> str:
    return value.strftime("%Y-%m-%d")


def recent_date_strings() -> list[str]:
    today = dt.datetime.now().date()
    return [date_string(today - dt.timedelta(days=offset)) for offset in range(6, -1, -1)]


def local_date_string() -> str:
    return date_string(dt.datetime.now().date())


def local_date_from_timestamp(value: Any) -> str:
    if value is None:
        return local_date_string()

    if isinstance(value, (int, float)):
        try:
            seconds = float(value) / 1000.0 if float(value) > 10_000_000_000 else float(value)
            return date_string(dt.datetime.fromtimestamp(seconds).date())
        except Exception:
            return local_date_string()

    raw = str(value).strip()
    if not raw:
        return local_date_string()

    # Claude JSONL timestamps are usually ISO-8601. Python accepts offsets but
    # not a trailing Z until it is normalized to +00:00.
    try:
        parsed = dt.datetime.fromisoformat(raw.replace("Z", "+00:00"))
        if parsed.tzinfo is not None:
            parsed = parsed.astimezone()
        return date_string(parsed.date())
    except Exception:
        return local_date_string()


def usage_token(usage: dict[str, Any], snake_key: str, camel_key: str) -> int:
    value = usage.get(snake_key, usage.get(camel_key, 0))
    try:
        return round(float(value or 0))
    except Exception:
        return 0


def number(value: Any) -> int:
    try:
        n = float(value or 0)
        return round(n) if n == n else 0
    except Exception:
        return 0


def empty_bucket() -> dict[str, int]:
    return {
        "inputTokens": 0,
        "outputTokens": 0,
        "cacheReadInputTokens": 0,
        "cacheCreationInputTokens": 0,
    }


# ─────────────────────────────────────────────────────────────── local scan ──


def scan_projects(projects_path: Path) -> dict[str, Any]:
    today = local_date_string()
    recent_dates = recent_date_strings()
    recent = {day: {"date": day, "messageCount": 0} for day in recent_dates}

    seen: set[str] = set()
    sessions: set[str] = set()
    active_days: set[str] = set()
    today_sessions: set[str] = set()
    today_tokens: dict[str, int] = {}
    usage_by_model: dict[str, dict[str, int]] = {}
    prompts = 0
    today_prompt_count = 0
    today_token_total = 0

    files = projects_path.rglob("*.jsonl") if projects_path.is_dir() else []
    for path in files:
        try:
            with path.open("r", encoding="utf-8", errors="replace") as handle:
                for line_number, line in enumerate(handle, 1):
                    # Cheap pre-filter before JSON parsing keeps files with
                    # unrelated lines inexpensive.
                    if '"usage":' not in line:
                        continue

                    try:
                        entry = json.loads(line)
                    except Exception:
                        continue

                    message = entry.get("message") if isinstance(entry.get("message"), dict) else {}
                    if entry.get("type") != "assistant" and message.get("role") != "assistant":
                        continue

                    usage = message.get("usage") or entry.get("usage")
                    if not isinstance(usage, dict):
                        continue

                    # One assistant message can be written more than once (a
                    # resumed session replays it). The message id is what tells
                    # a replay from a second answer.
                    message_id = message.get("id") or entry.get("messageId") or ""
                    unique_key = str(message_id) if message_id else (
                        f"{path}:{entry.get('uuid') or entry.get('requestId') or line_number}"
                    )
                    if unique_key in seen:
                        continue
                    seen.add(unique_key)

                    input_tokens = usage_token(usage, "input_tokens", "inputTokens")
                    output_tokens = usage_token(usage, "output_tokens", "outputTokens")
                    cache_read = usage_token(usage, "cache_read_input_tokens", "cacheReadInputTokens")
                    cache_write = usage_token(usage, "cache_creation_input_tokens", "cacheCreationInputTokens")
                    total = input_tokens + output_tokens + cache_read + cache_write
                    if total <= 0:
                        continue

                    model = str(message.get("model") or entry.get("model") or "claude")
                    day = local_date_from_timestamp(entry.get("timestamp") or message.get("timestamp"))
                    session_key = str(entry.get("sessionId") or path)
                    sessions.add(session_key)
                    active_days.add(day)
                    prompts += 1

                    bucket = usage_by_model.setdefault(model, empty_bucket())
                    bucket["inputTokens"] += input_tokens
                    bucket["outputTokens"] += output_tokens
                    bucket["cacheReadInputTokens"] += cache_read
                    bucket["cacheCreationInputTokens"] += cache_write

                    if day in recent:
                        # recentDays.messageCount is a token total despite the
                        # legacy name; the panel draws it as one.
                        recent[day]["messageCount"] += total

                    if day == today:
                        today_prompt_count += 1
                        today_sessions.add(session_key)
                        today_token_total += total
                        today_tokens[model] = today_tokens.get(model, 0) + total
        except Exception as exc:
            print(f"panama-agent-usage-claude: ignoring unreadable {path}: {exc}", file=sys.stderr)

    return {
        "todayPrompts": today_prompt_count,
        "todaySessions": len(today_sessions),
        "todayTotalTokens": today_token_total,
        "todayTokensByModel": today_tokens,
        "recentDays": [recent[day] for day in recent_dates],
        "modelUsage": usage_by_model,
        "totalPrompts": prompts,
        "totalSessions": len(sessions),
        "activeDays": len(active_days),
        "activeDates": sorted(active_days),
    }


def scan_cache_paths(projects_path: Path) -> tuple[Path, Path]:
    digest = hashlib.sha1(str(projects_path).encode("utf-8")).hexdigest()[:16]
    root = cache_root()
    return root / f"claude-scan-{digest}.json", root / f"claude-scan-{digest}.lock"


def read_fresh_json(path: Path, max_age_seconds: float) -> dict[str, Any] | None:
    if max_age_seconds <= 0 or not path.exists():
        return None
    try:
        # A negative age means the mtime is in the future: the clock moved
        # backwards since the write, so the cache's freshness cannot be trusted.
        age = time.time() - path.stat().st_mtime
        if 0 <= age <= max_age_seconds:
            return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return None
    return None


def write_json(path: Path, payload: dict[str, Any]) -> None:
    # A temp name unique to this writer, not derived from the target: several
    # collectors can run at once (the fan-out backgrounds one per agent), and a
    # shared temp path means the second replace finds the first one's file
    # already moved away.
    handle_fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=path.name + ".", suffix=".tmp")
    tmp = Path(tmp_name)
    try:
        with os.fdopen(handle_fd, "w", encoding="utf-8") as handle:
            handle.write(json.dumps(payload, separators=(",", ":"), sort_keys=True) + "\n")
        # mkstemp opens at 0600; nothing in this cache is a secret.
        tmp.chmod(0o644)
        tmp.replace(path)
    except BaseException:
        tmp.unlink(missing_ok=True)
        raise


def cached_scan(projects_path: Path, max_age_seconds: float) -> dict[str, Any]:
    """Local stats, with the cache as a pure optimization.

    A cache-layer failure (unwritable cache root, lock errors, a full disk) must
    never take the collector down: it degrades to a direct scan. The printed
    record is the contract; the cache is not.
    """
    try:
        cache_file, lock_file = scan_cache_paths(projects_path)

        cached = read_cached_scan(cache_file, max_age_seconds)
        if cached is not None:
            return cached

        with lock_file.open("w") as lock:
            fcntl.flock(lock, fcntl.LOCK_EX)
            cached = read_cached_scan(cache_file, max_age_seconds)
            if cached is not None:
                return cached
            summary = scan_projects(projects_path)
            write_json(cache_file, {"schemaVersion": 1, "scanDate": local_date_string(), "stats": summary})
            return summary
    except Exception as exc:
        print(f"panama-agent-usage-claude: cache unavailable ({exc}); scanning directly", file=sys.stderr)
        return scan_projects(projects_path)


# The cache payload is a versioned envelope around the stats dict, so a
# corrupted or foreign-shaped file is a miss (rescan and rewrite) rather than a
# crash or a garbage record.
def read_cached_scan(cache_file: Path, max_age_seconds: float) -> dict[str, Any] | None:
    cached = read_fresh_json(cache_file, max_age_seconds)
    if not isinstance(cached, dict) or cached.get("schemaVersion") != 1:
        return None
    # today* fields only mean "today" on the day they were scanned. A cache from
    # another local date (midnight passed, or the clock moved) is a miss, not
    # merely old, whatever its mtime says.
    if cached.get("scanDate") != local_date_string():
        return None
    stats = cached.get("stats")
    if not isinstance(stats, dict):
        return None
    if not all(key in stats for key in ("todayPrompts", "todayTotalTokens", "recentDays", "activeDates", "modelUsage")):
        return None
    return stats


# ────────────────────────────────────────────────────────── 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 the project scan comes back empty.


def stats_cache_fallback(claude_dir: Path) -> dict[str, Any] | None:
    try:
        data = json.loads((claude_dir / "stats-cache.json").read_text(encoding="utf-8"))
    except Exception:
        return None

    today = local_date_string()
    daily_model_tokens = data.get("dailyModelTokens") or []
    today_tokens = {}
    for entry in daily_model_tokens:
        if isinstance(entry, dict) and entry.get("date") == today:
            today_tokens = entry.get("tokensByModel") or {}
            break

    daily_activity = [day for day in (data.get("dailyActivity") or []) if isinstance(day, dict)]
    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)

    return {
        "todayPrompts": today_prompts,
        "todaySessions": today_sessions,
        "todayTotalTokens": sum(number(v) for v in today_tokens.values()),
        "todayTokensByModel": today_tokens,
        "recentDays": daily_activity[-7:],
        "modelUsage": data.get("modelUsage") or {},
        "totalPrompts": number(data.get("totalMessages")),
        "totalSessions": number(data.get("totalSessions")),
        "activeDays": len(active_dates),
        "activeDates": active_dates,
    }


def today_prompts_from_history(claude_dir: Path) -> tuple[int, int]:
    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 number(entry.get("timestamp")) < start_of_day:
            break
        prompts += 1
        if entry.get("sessionId"):
            sessions.add(str(entry.get("sessionId")))
    return prompts, len(sessions)


# ─────────────────────────────────────────────────────────────────  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(credentials: Path) -> tuple[str, int, str]:
    try:
        data = json.loads(credentials.read_text(encoding="utf-8"))
    except Exception:
        return "", 0, ""
    login = data.get("claudeAiOauth")
    if not isinstance(login, dict):
        return "", 0, ""
    plan = plan_label(str(login.get("rateLimitTier") or ""), str(login.get("subscriptionType") or ""))
    return str(login.get("accessToken") or ""), number(login.get("expiresAt")), plan


def plan_label(tier: str, subscription: str) -> str:
    if tier:
        match = re.search(r"max_(\d+x)", tier, re.IGNORECASE)
        if match:
            return "Max " + match.group(1)
    if subscription:
        return subscription[0].upper() + subscription[1:]
    return ""


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


def normalize_utilization(value: Any, percent_scale: bool) -> float:
    n = parse_utilization(value)
    if not (n >= 0):
        return -1.0
    # The OAuth usage endpoint currently reports percentages (37.0, or 1.0).
    # Older payloads sometimes used fractions (0.37). A payload containing any
    # value >= 1 is percent-scaled, so 1.0 renders as 1%, not 100%. Clamped as
    # well as normalized: a readout that can print 1500% is one you learn to
    # ignore.
    if percent_scale or n > 1:
        return min(1.0, n / 100.0)
    return min(1.0, n)


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"). Reading a window out of free text
# cannot survive a model name like "Opus 5 (1M context)" -- the "1M" reads as a
# one-minute window -- so the window is settled here and travels as an explicit
# title, 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
# only one model draws from, say. The matching legacy keys (`seven_day_opus`,
# `seven_day_sonnet`, ...) stayed behind at null, so a collector reading 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 dedupe key.
def scoped_limits(payload: dict[str, Any], percent_scale: bool) -> 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"), percent_scale)
        if percent < 0:
            continue
        seen.add((name, kind))
        window = scoped_window(kind)
        title = name + " " + window if window else name
        out.append({
            "label": title,
            "percent": percent,
            "resetsAt": normalize_reset_at(entry.get("resets_at")),
        })
    return out


def probe_limits(access_token: str) -> dict[str, Any]:
    # The token travels in a header on a request made in this process. There is
    # no child process, so there is no argv to leak it through.
    request = urllib.request.Request(
        endpoint(),
        headers={
            "Authorization": "Bearer " + access_token,
            "anthropic-beta": "oauth-2025-04-20",
            "Accept": "application/json",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            payload = json.loads(response.read().decode("utf-8", errors="replace"))
    except urllib.error.HTTPError as error:
        retry_after = error.headers.get("retry-after", "") if error.headers else ""
        if error.code == 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."
        else:
            help_text = (
                f"Anthropic's usage endpoint returned status {error.code}. "
                "Local Claude Code stats are still shown."
            )
        return {"ok": False, "helpText": help_text}
    except Exception:
        # A transport failure reached no server at all -- no route, no DNS. Any
        # real answer, including an error status, is a server worth not
        # 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.",
        }

    if not isinstance(payload, dict):
        return {"ok": False, "helpText": "Anthropic's usage endpoint returned an unfamiliar shape."}

    weekly = usage_bucket(payload, "seven_day_oauth_apps") or usage_bucket(payload, "seven_day")
    session = usage_bucket(payload, "five_hour")
    raw = [session.get("utilization") if session else None, weekly.get("utilization") if weekly else None]
    # One payload speaks one convention, so the scoped entries settle the scale
    # alongside the buckets rather than assuming their own.
    entries = payload.get("limits")
    if isinstance(entries, list):
        raw += [entry.get("percent") for entry in entries if isinstance(entry, dict)]
    percent_scale = any(parse_utilization(v) >= 1 for v in raw)

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

    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}


# A cached percentage outlives the probe that measured it, but only until its
# window rolls over: once a window has reset the figure describes a period that
# is over, and a stale 78% would misreport an allowance that is now untouched. A
# window with no reset time, or one that will not parse, is kept -- an
# unreadable timestamp is no reason to throw away a real number.
def limit_window_open(entry: dict[str, Any], now: dt.datetime) -> bool:
    raw = str(entry.get("resetsAt") or "")
    if raw == "":
        return True
    try:
        resets_at = dt.datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except Exception:
        return True
    if resets_at.tzinfo is None:
        resets_at = resets_at.replace(tzinfo=dt.timezone.utc)
    return resets_at > now


def usable_cached_limits(cached: dict[str, Any]) -> list[dict[str, Any]]:
    entries = cached.get("limits")
    if not isinstance(entries, list):
        return []
    now = dt.datetime.now(dt.timezone.utc)
    return [entry for entry in entries if isinstance(entry, dict) and limit_window_open(entry, now)]


def collect_limits(access_token: str, expires_at_ms: int, force: bool) -> dict[str, Any]:
    result: dict[str, Any] = {"limits": [], "usageStatusText": "", "authHelpText": AUTH_HELP}

    probe_cache = cache_root() / "claude-limits.json"
    cached = read_fresh_json(probe_cache, float("inf")) or {}
    fallback = usable_cached_limits(cached)

    # 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 whose window has not since reset.
    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
        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 absorbs repeated panel opens, it does not overrule
    # someone who pressed refresh.
    fetched_at = number(cached.get("fetchedAtMs")) / 1000
    if fallback and not force and time.time() - fetched_at < PROBE_MIN_INTERVAL_SECONDS:
        result["limits"] = fallback
        return result

    probe = probe_limits(access_token)
    if probe["ok"]:
        result["limits"] = probe["limits"]
        try:
            write_json(probe_cache, {"fetchedAtMs": round(time.time() * 1000), "limits": probe["limits"]})
        except Exception as exc:
            print(f"panama-agent-usage-claude: could not cache limits ({exc})", file=sys.stderr)
        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
    else:
        result["usageStatusText"] = "Claude limits unavailable"
        result["authHelpText"] = probe["helpText"]
    return result


# ─────────────────────────────────────────────────────────────────  record ──


def main() -> int:
    parser = argparse.ArgumentParser(description="Print the Claude Code usage record as JSON")
    parser.add_argument("--force", action="store_true",
                        help="rescan transcripts 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")
    args = parser.parse_args()

    claude_dir = config_dir()
    scan_age = 0 if args.force else (LIMITS_ONLY_REUSE_SECONDS if args.limits_only else SCAN_REUSE_SECONDS)
    stats = cached_scan(claude_dir / "projects", scan_age)

    if number(stats.get("totalPrompts")) <= 0:
        fallback = stats_cache_fallback(claude_dir)
        if fallback is not None:
            stats = fallback
        else:
            # 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)
            if today_prompts or today_sessions:
                stats = dict(stats, todayPrompts=today_prompts, todaySessions=today_sessions)

    access_token, expires_at_ms, plan = oauth_login(credentials_path(claude_dir))
    limits = collect_limits(access_token, expires_at_ms, args.force)

    record = {
        "schemaVersion": 1,
        "id": AGENT_ID,
        "name": AGENT_NAME,
        "updatedAt": dt.datetime.now(dt.timezone.utc).isoformat(),
        "ready": number(stats.get("totalPrompts")) > 0 or len(limits["limits"]) > 0,
        "hasLocalStats": True,
        "tierLabel": plan,
        "usageStatusText": limits["usageStatusText"],
        "authHelpText": limits["authHelpText"],
        "limits": limits["limits"],
    }
    if limits.get("retryAdvised"):
        record["retryAdvised"] = True
    record.update(stats)
    print(json.dumps(record, separators=(",", ":"), sort_keys=True))
    return 0


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