AI Usage

{
  "id": "ai-usage",
  "name": "AI Usage",
  "icon": "icon.png",
  "description": "How much of your coding agents' plan limits you've used (Claude Code and Codex, whichever you have), with reset times. Pops up as a limit passes 50%, 75% and 90%.",
  "watch": [
    {
      "script": "usage.sh",
      "every": 60,
      "as": "usage"
    }
  ],
  "initial": "glance",
  "anytime": true,
  "states": {
    "glance": {
      "compact": {
        "leading": [
          {
            "image": "{{usage.top.logo | logos/claude.png}}"
          },
          {
            "ring": {
              "value": "{{usage.top.pct | 0}}",
              "max": "100",
              "tint": "{{usage.top.tint | gray}}"
            }
          }
        ]
      },
      "expanded": {
        "size": "compact",
        "header": {
          "symbol": "gauge.with.dots.needle.50percent",
          "tint": "{{usage.top.tint | gray}}",
          "title": "AI usage"
        },
        "blocks": [
          {
            "blocks_from": "usage.panel"
          }
        ]
      },
      "keys": [
        {
          "key": "cmd+r",
          "label": "Refresh",
          "do": [
            {
              "refresh": "usage"
            }
          ]
        }
      ]
    },
    "alert": {
      "compact": {
        "leading": [
          {
            "image": "{{usage.top.logo | logos/claude.png}}"
          },
          {
            "ring": {
              "value": "{{usage.top.pct | 0}}",
              "max": "100",
              "tint": "{{usage.top.tint | gray}}"
            }
          }
        ]
      },
      "expanded": {
        "size": "compact",
        "header": {
          "symbol": "gauge.with.dots.needle.50percent",
          "tint": "{{usage.top.tint | gray}}",
          "title": "{{usage.top.name}} at {{usage.top.pct}}%"
        },
        "blocks": [
          {
            "blocks_from": "usage.panel"
          }
        ]
      },
      "keys": [
        {
          "key": "cmd+r",
          "label": "Refresh",
          "do": [
            {
              "refresh": "usage"
            }
          ]
        }
      ],
      "open": true,
      "dismiss_after": 10
    }
  },
  "commands": {
    "refresh": {
      "description": "Fetch the latest usage now.",
      "phrase": "Refresh AI usage",
      "do": [
        {
          "refresh": "usage"
        }
      ]
    },
    "show": {
      "description": "Show plan usage for Claude Code and Codex.",
      "phrase": "Show AI usage", "preview": { "state": "glance" },
      "do": [
        {
          "go": "glance"
        }
      ]
    }
  },
  "preview": {
    "states": [
      "glance",
      "alert"
    ],
    "data": {
      "usage": {
        "ok": true,
        "top": {
          "pct": 82,
          "logo": "logos/claude.png",
          "tint": "#D97757",
          "name": "Claude Code"
        },
        "panel": {
          "data": {},
          "blocks": [
            {
              "text": "Claude Code",
              "style": "caption"
            },
            {
              "progress": {
                "value": "82",
                "max": "100",
                "label": "5-hour session",
                "tint": "#D97757"
              }
            },
            {
              "note": "Resets in 1h 12m (at 13:30)"
            },
            {
              "progress": {
                "value": "21",
                "max": "100",
                "label": "Week",
                "tint": "#D97757"
              }
            },
            {
              "note": "Resets in 3d 22h (at Fri 01:00)"
            },
            {
              "text": "Codex",
              "style": "caption"
            },
            {
              "progress": {
                "value": "35",
                "max": "100",
                "label": "5-hour session",
                "tint": "#10A37F"
              }
            },
            {
              "note": "Resets in 3h 40m (at 16:00)"
            }
          ]
        }
      }
    }
  },
  "triggers": [
    {
      "when": {
        "usage.pop": true
      },
      "open": "alert"
    }
  ]
}

# Plan usage for the coding agents you use: Claude Code (the limits its /usage shows, asked with
# its stored login) and Codex (the limits it last reported, from its local session logs). Agents
# you don't have are left out. Prints the fields and a `panel` of bars for the card.
import glob, json, os, subprocess, time, urllib.error, urllib.request
from datetime import datetime, timezone

STORE = os.environ.get("CHOPSTICK_STORAGE", ".")
CACHE = os.path.join(STORE, "claude.json")
FRESH = 300      # seconds a good Claude reply is reused before asking again
BACKOFF = 900    # wait after a 429 when the server gives no Retry-After


def left(t):
    """Time until `t` (a datetime) as "2h 5m" or "3d 4h", and the clock time it happens."""
    if t is None:
        return "—", "—"
    s = max(0, int((t - datetime.now(timezone.utc)).total_seconds()))
    d, h, m = s // 86400, s % 86400 // 3600, s % 3600 // 60
    return (f"{d}d {h}h" if d else f"{h}h {m}m"), t.astimezone().strftime("%a %H:%M" if d else "%H:%M")


def window_name(minutes):
    return {300: "5-hour session", 10080: "Week", 43200: "Month"}.get(minutes, f"{round(minutes / 60)}-hour window")


def claude():
    try:
        cache = json.load(open(CACHE))
    except Exception:
        cache = {}
    now, error = time.time(), None
    if now - cache.get("fetched", 0) > FRESH and now > cache.get("retry_after", 0):
        raw = subprocess.run(["security", "find-generic-password", "-s", "Claude Code-credentials", "-w"],
                             capture_output=True, text=True, timeout=5).stdout.strip()
        if not raw and "data" not in cache:
            return None  # no Claude Code login here
        try:
            tok = json.loads(raw)["claudeAiOauth"]["accessToken"]
            req = urllib.request.Request("https://api.anthropic.com/api/oauth/usage", headers={
                "Authorization": "Bearer " + tok, "anthropic-beta": "oauth-2025-04-20", "User-Agent": "claude-code"})
            cache["data"] = json.loads(urllib.request.urlopen(req, timeout=10).read())
            cache["fetched"] = now
        except urllib.error.HTTPError as e:
            if e.code == 429:
                wait = e.headers.get("Retry-After")
                cache["retry_after"] = now + (int(wait) if wait and wait.isdigit() else BACKOFF)
                error = "Rate limited, trying again at " + datetime.fromtimestamp(cache["retry_after"]).strftime("%H:%M")
            else:
                error = f"Couldn't ask (HTTP {e.code})"
        except Exception as e:
            error = str(e)[:100]
        os.makedirs(STORE, exist_ok=True)
        with open(CACHE, "w") as f:
            json.dump(cache, f)
    d = cache.get("data")
    if not d:
        return {"name": "Claude Code", "logo": "logos/claude.png", "tint": "#D97757", "limits": [], "error": error or "No usage yet"}
    limits = []
    for key, name in (("five_hour", "5-hour session"), ("seven_day", "Week")):
        part = d.get(key) or {}
        if part.get("utilization") is None:
            continue
        at = datetime.fromisoformat(part["resets_at"]) if part.get("resets_at") else None
        # The reset time jitters by a fraction of a second between replies, so the period is named
        # by its nearest minute; otherwise every fetch looks like a new period and pops up again.
        limits.append({"name": name, "pct": round(part["utilization"]), "resets": left(at),
                       "window": at and round(at.timestamp() / 60) * 60})
    return {"name": "Claude Code", "logo": "logos/claude.png", "tint": "#D97757", "limits": limits,
            "error": error and error + " · showing the last numbers"}


def codex():
    files = sorted(glob.glob(os.path.expanduser("~/.codex/sessions/*/*/*/*.jsonl")), key=os.path.getmtime, reverse=True)
    if not files:
        return None
    for path in files[:5]:
        last = None
        with open(path, errors="ignore") as f:
            for line in f:
                if '"rate_limits"' in line:
                    last = line
        if not last:
            continue
        found = None

        def walk(x):
            nonlocal found
            if isinstance(x, dict):
                if "rate_limits" in x and isinstance(x["rate_limits"], dict):
                    found = x["rate_limits"]
                for v in x.values():
                    walk(v)
            elif isinstance(x, list):
                for v in x:
                    walk(v)
        walk(json.loads(last))
        if not found:
            continue
        limits = []
        for key in ("primary", "secondary"):
            part = found.get(key)
            if not part:
                continue
            at = datetime.fromtimestamp(part["resets_at"], timezone.utc) if part.get("resets_at") else None
            limits.append({"name": window_name(part.get("window_minutes") or 0), "pct": round(part.get("used_percent") or 0),
                           "resets": left(at), "window": part.get("resets_at")})
        return {"name": "Codex", "logo": "logos/codex.svg", "tint": "#10A37F", "limits": limits,
                "note": "As of " + datetime.fromtimestamp(os.path.getmtime(path)).strftime("%a %H:%M")}
    return {"name": "Codex", "logo": "logos/codex.svg", "tint": "#10A37F", "limits": [], "error": "No usage reported yet"}


agents = [a for a in (claude(), codex()) if a]
blocks = []
for a in agents:
    blocks.append({"text": a["name"], "style": "caption"})
    for lim in a["limits"]:
        blocks.append({"progress": {"value": str(lim["pct"]), "max": "100", "label": lim["name"], "tint": a["tint"]}})
        blocks.append({"note": f"Resets in {lim['resets'][0]} (at {lim['resets'][1]})"})
    for extra in (a.get("error"), a.get("note")):
        if extra:
            blocks.append({"note": extra})
if not agents:
    blocks.append({"note": "No Claude Code or Codex found on this Mac."})

# The fullest limit across agents picks the ring.
top = max(((lim["pct"], a) for a in agents for lim in a["limits"][:1]), default=(0, None), key=lambda x: x[0])

# Pop up once as a limit passes 50, 75 and 90%: `pop` is true for one run per threshold, per limit
# period (remembered, so a restart doesn't announce it again).
NOTIFIED = os.path.join(STORE, "notified.json")
try:
    notified = json.load(open(NOTIFIED))
except Exception:
    notified = {}
pop = False
for a in agents:
    for lim in a["limits"][:1]:
        key = f"{a['name']}|{lim.get('window')}"
        level = max([t for t in (50, 75, 90) if lim["pct"] >= t], default=0)
        if level > notified.get(key, 0):
            notified[key], pop = level, True
if pop:
    os.makedirs(STORE, exist_ok=True)
    with open(NOTIFIED, "w") as f:
        json.dump(notified, f)
print(json.dumps({
    "ok": bool(agents),
    "pop": pop,
    "top": {"pct": top[0], "logo": top[1]["logo"] if top[1] else "logos/claude.png",
            "tint": top[1]["tint"] if top[1] else "gray", "name": top[1]["name"] if top[1] else ""},
    "panel": {"blocks": blocks, "data": {}},
}))

cd "$(dirname "$0")" 2>/dev/null
exec /usr/bin/env python3 -I usage.py