778 lines
32 KiB
Python
Executable File
778 lines
32 KiB
Python
Executable File
#!/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())
|