Add the agents usage bar widget

bin/blob-agent-usage-claude

@@ -0,0 +1,728 @@
+#!/usr/bin/python3
+# blob:summary=Print the Claude Code usage record as JSON
+# blob:args=[--force] [--limits-only]
+# blob:hidden=true
+
+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 usage limits."
+USAGE_ENDPOINT = "https://api.anthropic.com/api/oauth/usage"
+PROBE_MIN_INTERVAL_SECONDS = 15
+CURRENT_ENOUGH_SECONDS = 900
+INDEX_VERSION = 4
+INDEX_HEAD_BYTES = 4096
+
+
+def config_dir() -> Path:
+  return expand_path(os.environ.get("CLAUDE_CONFIG_DIR") or "~/.claude")
+
+
+def expand_path(value: str) -> Path:
+  return Path(os.path.expandvars(os.path.expanduser(value))).resolve()
+
+
+def cache_root() -> Path:
+  root = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "blob" / "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()
+
+  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,
+  }
+
+
+def read_json(path: Path) -> dict[str, Any] | None:
+  try:
+    return json.loads(path.read_text(encoding="utf-8"))
+  except Exception:
+    return None
+
+
+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:
+    age = time.time() - path.stat().st_mtime
+    if 0 <= age <= max_age_seconds:
+      return read_json(path)
+  except Exception:
+    return None
+  return None
+
+
+def write_json(path: Path, payload: dict[str, Any]) -> None:
+  handle_fd, temporary_name = tempfile.mkstemp(dir=path.parent, prefix=path.name + ".", suffix=".tmp")
+  temporary_path = Path(temporary_name)
+  try:
+    with os.fdopen(handle_fd, "w", encoding="utf-8") as handle:
+      handle.write(json.dumps(payload, separators=(",", ":"), sort_keys=True) + "\n")
+    temporary_path.chmod(0o644)
+    temporary_path.replace(path)
+  except BaseException:
+    temporary_path.unlink(missing_ok=True)
+    raise
+
+
+def index_zone() -> list[Any]:
+  return [os.environ.get("TZ"), list(time.tzname), time.timezone, time.altzone]
+
+
+def load_index(index_file: Path) -> dict[str, Any]:
+  data = read_json(index_file)
+  if (
+    isinstance(data, dict)
+    and data.get("version") == INDEX_VERSION
+    and data.get("zone") == index_zone()
+    and isinstance(data.get("files"), dict)
+  ):
+    return data["files"]
+  return {}
+
+
+def read_digest(path: Path, length: int) -> str:
+  digest = hashlib.sha1()
+  with path.open("rb") as handle:
+    digest.update(handle.read(min(length, INDEX_HEAD_BYTES)))
+    handle.seek(max(0, length - INDEX_HEAD_BYTES))
+    digest.update(handle.read(min(length, INDEX_HEAD_BYTES)))
+  return digest.hexdigest()
+
+
+def parse_usage_records(path: Path, entry: dict[str, Any] | None) -> dict[str, Any]:
+  stat = path.stat()
+  size = stat.st_size
+  mtime = stat.st_mtime_ns
+
+  if entry and entry.get("size") == size and entry.get("mtime") == mtime:
+    return entry
+
+  if (
+    entry
+    and size > number(entry.get("size"))
+    and entry.get("file") == [stat.st_dev, stat.st_ino]
+    and entry.get("head") == read_digest(path, number(entry.get("offset")))
+  ):
+    records = list(entry.get("records") or [])
+    offset = number(entry.get("offset"))
+    line_number = number(entry.get("lines"))
+  else:
+    records = []
+    offset = 0
+    line_number = 0
+
+  positions = {record[0]: index for index, record in enumerate(records)}
+  with path.open("rb") as handle:
+    handle.seek(offset)
+    for raw in handle:
+      if not raw.endswith(b"\n"):
+        break
+      offset += len(raw)
+      line_number += 1
+
+      if b'"usage":' not in raw:
+        continue
+
+      try:
+        parsed = json.loads(raw.decode("utf-8", errors="replace"))
+      except Exception:
+        continue
+
+      message = parsed.get("message") if isinstance(parsed.get("message"), dict) else {}
+      if parsed.get("type") != "assistant" and message.get("role") != "assistant":
+        continue
+
+      usage = message.get("usage") or parsed.get("usage")
+      if not isinstance(usage, dict):
+        continue
+
+      message_id = message.get("id") or parsed.get("messageId") or ""
+      key = str(message_id) if message_id else f"{path}:{parsed.get('uuid') or parsed.get('requestId') or line_number}"
+
+      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")
+      if input_tokens + output_tokens + cache_read + cache_write <= 0:
+        continue
+
+      held = positions.get(key)
+      if held is not None and output_tokens < records[held][4]:
+        continue
+
+      timestamp = parsed.get("timestamp") or message.get("timestamp")
+      record = [
+        key,
+        str(message.get("model") or parsed.get("model") or "claude"),
+        local_date_from_timestamp(timestamp) if timestamp is not None else None,
+        input_tokens,
+        output_tokens,
+        cache_read,
+        cache_write,
+        str(parsed.get("sessionId") or path),
+      ]
+      if held is None:
+        positions[key] = len(records)
+        records.append(record)
+      else:
+        records[held] = record
+
+  return {
+    "size": size,
+    "file": [stat.st_dev, stat.st_ino],
+    "mtime": mtime,
+    "head": read_digest(path, offset),
+    "offset": offset,
+    "lines": line_number,
+    "records": records,
+  }
+
+
+def scan_projects(projects_path: Path, index_file: Path) -> dict[str, Any]:
+  previous = load_index(index_file)
+  current: dict[str, Any] = {}
+  order: list[str] = []
+  unreadable_count = 0
+
+  files = projects_path.rglob("*.jsonl") if projects_path.is_dir() else []
+  for path in files:
+    key = str(path)
+    try:
+      current[key] = parse_usage_records(path, previous.get(key))
+    except Exception:
+      unreadable_count += 1
+      if key in previous:
+        current[key] = previous[key]
+    if key in current:
+      order.append(key)
+
+  if unreadable_count:
+    print(f"Ignoring {unreadable_count} unreadable Claude project files", file=sys.stderr)
+
+  if current != previous:
+    try:
+      write_json(index_file, {"version": INDEX_VERSION, "zone": index_zone(), "files": current})
+    except Exception as error:
+      print(f"Could not write Claude scan index {index_file}: {error}", file=sys.stderr)
+
+  return summarize_records(current, order)
+
+
+def summarize_records(indexed: dict[str, Any], order: list[str]) -> dict[str, Any]:
+  today = local_date_string()
+  recent_dates = recent_date_strings()
+  recent = {day: {"date": day, "messageCount": 0} for day in recent_dates}
+
+  seen: dict[str, list[Any]] = {}
+  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
+
+  for key in order:
+    for record in indexed[key].get("records") or []:
+      held = seen.get(record[0])
+      if held is not None and record[4] < held[4]:
+        continue
+      seen[record[0]] = record
+
+  for _, model, day, input_tokens, output_tokens, cache_read, cache_write, session_key in seen.values():
+    if day is None:
+      day = today
+    total = input_tokens + output_tokens + cache_read + cache_write
+
+    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:
+      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
+
+  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),
+  }
+
+
+def cached_scan(projects_path: Path, max_age_seconds: float, use_index: bool) -> dict[str, Any]:
+  digest = hashlib.sha1(str(projects_path).encode("utf-8")).hexdigest()[:16]
+  root = cache_root()
+  cache_file = root / f"claude-scan-{digest}.json"
+  lock_file = root / f"claude-scan-{digest}.lock"
+  index_file = root / f"claude-index-{digest}.json"
+
+  cached = read_fresh_json(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_fresh_json(cache_file, max_age_seconds)
+    if cached is not None:
+      return cached
+    if not use_index:
+      index_file.unlink(missing_ok=True)
+    summary = scan_projects(projects_path, index_file)
+    write_json(cache_file, summary)
+    return summary
+
+
+def stats_cache_fallback(claude_dir: Path) -> dict[str, Any] | None:
+  data = read_json(claude_dir / "stats-cache.json")
+  if data is None:
+    return None
+
+  today = local_date_string()
+  today_tokens = {}
+  for entry in data.get("dailyModelTokens") or []:
+    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 = {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(value) for value 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),
+  }
+
+
+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)
+
+
+def oauth_login(claude_dir: Path) -> tuple[str, int, int, str]:
+  data = read_json(claude_dir / ".credentials.json") or {}
+  login = data.get("claudeAiOauth")
+  if not isinstance(login, dict):
+    return "", 0, 0, ""
+  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,
+  )
+
+
+def profile_tier(claude_dir: Path) -> str:
+  profile = read_json(profile_path(claude_dir)) or {}
+  account = profile.get("oauthAccount")
+  if not isinstance(account, dict):
+    return ""
+  for key in ("userRateLimitTier", "organizationRateLimitTier"):
+    tier = str(account.get(key) or "")
+    if tier:
+      return tier
+  return ""
+
+
+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"
+
+
+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")
+
+
+def normalize_utilization(value: Any, percent_scale: bool) -> float:
+  n = parse_utilization(value)
+  if not (n >= 0):
+    return -1.0
+  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():
+    timestamp = int(raw)
+    if timestamp < 1e12:
+      timestamp *= 1000
+    try:
+      return dt.datetime.fromtimestamp(timestamp / 1000, dt.timezone.utc).isoformat()
+    except Exception:
+      return raw
+  try:
+    return dt.datetime.fromisoformat(raw.replace("Z", "+00:00")).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
+
+
+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 ""
+
+
+def scoped_limits(payload: dict[str, Any], percent_scale: bool) -> list[dict[str, Any]]:
+  entries = payload.get("limits")
+  if not isinstance(entries, list):
+    return []
+  scoped: 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
+    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
+    scoped.append({
+      "label": title,
+      "title": title,
+      "percent": percent,
+      "resetsAt": normalize_reset_at(entry.get("resets_at")),
+    })
+  return scoped
+
+
+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:
+    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 ""
+      ) + "."
+    else:
+      help_text = f"Anthropic's usage endpoint returned status {error.code}."
+    return {"ok": False, "helpText": help_text}
+  except Exception:
+    return {
+      "ok": False,
+      "transport": True,
+      "helpText": "Couldn't reach Anthropic's usage endpoint. Retrying shortly.",
+    }
+
+  weekly = usage_bucket(payload, "seven_day_oauth_apps") or usage_bucket(payload, "seven_day")
+  session = usage_bucket(payload, "five_hour")
+  raw_values = [session.get("utilization") if session else None, weekly.get("utilization") if weekly else None]
+  entries = payload.get("limits")
+  if isinstance(entries, list):
+    raw_values += [entry.get("percent") for entry in entries if isinstance(entry, dict)]
+  percent_scale = any(parse_utilization(value) >= 1 for value in raw_values)
+
+  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."}
+  return {"ok": True, "limits": limits}
+
+
+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, refresh_expires_at_ms: int, force: bool) -> dict[str, Any]:
+  result = {"limits": [], "usageStatusText": "", "authHelpText": AUTH_HELP, "live": False, "fetchedAtMs": 0}
+
+  probe_cache = cache_root() / "claude-limits.json"
+  cached = read_json(probe_cache) or {}
+  fallback = usable_cached_limits(cached)
+  result["fetchedAtMs"] = number(cached.get("fetchedAtMs"))
+  now_ms = time.time() * 1000
+
+  if access_token == "":
+    result["limits"] = fallback
+    result["usageStatusText"] = "Waiting for sign-in"
+    return result
+
+  if expires_at_ms > 0 and expires_at_ms <= now_ms:
+    result["limits"] = fallback
+    shown = " Showing the last known limits." if fallback else ""
+    if refresh_expires_at_ms > now_ms:
+      result["usageStatusText"] = "Limits paused"
+      result["authHelpText"] = "Claude Code's access token has lapsed." + shown + " Start Claude Code to refresh it."
+    else:
+      result["usageStatusText"] = "Sign-in expired"
+      result["authHelpText"] = "Claude Code's saved sign-in expired." + shown + " Run `claude auth login` to sign in again."
+    return result
+
+  fetched_age = time.time() - number(cached.get("fetchedAtMs")) / 1000
+  if fallback and not force and 0 <= fetched_age < PROBE_MIN_INTERVAL_SECONDS:
+    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"] = round(time.time() * 1000)
+    write_json(probe_cache, {"fetchedAtMs": result["fetchedAtMs"], "limits": probe["limits"]})
+    return result
+
+  if probe.get("transport"):
+    result["retryAdvised"] = True
+  if fallback:
+    result["limits"] = fallback
+    fetched_age = time.time() - result["fetchedAtMs"] / 1000
+    result["live"] = 0 <= fetched_age < CURRENT_ENOUGH_SECONDS
+  else:
+    result["usageStatusText"] = "Limits unavailable"
+    result["authHelpText"] = probe["helpText"]
+  return result
+
+
+def local_stats(claude_dir: Path, scan_age: float, force: bool) -> dict[str, Any]:
+  stats = cached_scan(claude_dir / "projects", scan_age, use_index=not force)
+  if number(stats.get("totalPrompts")) > 0:
+    return stats
+
+  fallback = stats_cache_fallback(claude_dir)
+  if fallback is not None:
+    return fallback
+
+  today_prompts, today_sessions = today_prompts_from_history(claude_dir)
+  if today_prompts or today_sessions:
+    return dict(stats, todayPrompts=today_prompts, todaySessions=today_sessions)
+  return stats
+
+
+def main() -> int:
+  parser = argparse.ArgumentParser()
+  parser.add_argument("--force", action="store_true")
+  parser.add_argument("--limits-only", action="store_true")
+  parser.add_argument("--cache-seconds", type=float, default=20)
+  args = parser.parse_args()
+
+  claude_dir = config_dir()
+  scan_age = 0 if args.force else (900 if args.limits_only else args.cache_seconds)
+  stats = local_stats(claude_dir, scan_age, args.force)
+
+  access_token, expires_at_ms, refresh_expires_at_ms, plan = oauth_login(claude_dir)
+  limits = collect_limits(access_token, expires_at_ms, refresh_expires_at_ms, args.force)
+
+  record = {
+    "schemaVersion": 1,
+    "id": AGENT_ID,
+    "name": AGENT_NAME,
+    "updatedAt": dt.datetime.now(dt.timezone.utc).isoformat(),
+    "tierLabel": plan,
+    "usageStatusText": limits["usageStatusText"],
+    "authHelpText": limits["authHelpText"],
+    "limits": limits["limits"],
+    "limitsStale": not limits["live"],
+    "limitsFetchedAt": limits["fetchedAtMs"],
+  }
+  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())

bin/blob-claude-usage

@@ -0,0 +1,32 @@
+#!/bin/bash
+
+# blob:summary=Print Claude Code session and weekly usage limits
+# blob:group=claude
+# blob:hidden=true
+
+set -euo pipefail
+
+credentials_file="$HOME/.claude/.credentials.json"
+usage_url="https://api.anthropic.com/api/oauth/usage"
+
+[[ -r $credentials_file ]] || { echo "Sign in to Claude Code first" >&2; exit 1; }
+
+access_token=$(jq -r '.claudeAiOauth.accessToken // empty' "$credentials_file")
+[[ -n $access_token ]] || { echo "Sign in to Claude Code first" >&2; exit 1; }
+
+response=$(printf 'Authorization: Bearer %s\nanthropic-beta: oauth-2025-04-20\n' "$access_token" |
+  curl -fsS --max-time 10 -H @- "$usage_url" 2>/dev/null) || {
+  echo "Could not load usage, open Claude Code to refresh sign-in" >&2
+  exit 1
+}
+
+jq -r '
+  def limit_rows(name; limit):
+    if limit then
+      "\(name)_percent\t\((limit.utilization // 0) | floor)",
+      "\(name)_resets\t\(limit.resets_at // "")"
+    else empty end;
+  limit_rows("session"; .five_hour),
+  limit_rows("weekly"; .seven_day),
+  limit_rows("weekly_opus"; .seven_day_opus)
+' <<<"$response"