Use Anthropic ratelimit headers as primary Claude scrape (Clawdmeter method)

CLAUDE.md

@@ -12,21 +12,31 @@ together four sources in `local_claude_activity_snapshot()`
 (`daemon/codex-usage-daemon.py`). They are tried in order; the first to
 return a real snapshot wins.
 
-1. **`~/.claude/plan-limits.json`** — written by
+1. **`~/.config/codexmeter/claude-plan-usage.json`** — preferred. Written
+   every 15 s by the `com.justin.codexmeter.claude-plan-scrape` LaunchAgent.
+   Scrape sources are tried in this order inside the script chain:
+   1. **`scripts/claude-plan-usage-oauth-headers.py`** (primary) — reads
+      the Claude Code OAuth bearer from the macOS Keychain
+      (`Claude Code-credentials`), POSTs a 1-token Haiku request to
+      `https://api.anthropic.com/v1/messages`, ignores the body, and
+      parses the `anthropic-ratelimit-unified-{5h,7d}-{utilization,reset}`
+      response headers. No browser, no Cloudflare — documented API.
+      Approach adapted from
+      [HermannBjorgvin/Clawdmeter](https://github.com/HermannBjorgvin/Clawdmeter).
+   2. **`scripts/claude-plan-usage-desktop.py`** (fallback) — decrypts
+      Claude Desktop's cookie SQLite via the `Claude Safe Storage`
+      Keychain entry and calls `claude.ai/api/organizations/{org}/usage`.
+   3. **`scripts/claude-plan-usage-firefox.py`** (fallback to the fallback)
+      — same endpoint, Firefox cookies.
+   TTL: `CLAUDE_PLAN_TTL_SECONDS` (default 60 min).
+2. **`~/.claude/plan-limits.json`** — written by
    `scripts/claude-statusline-limits.py`, which Claude Code itself runs
-   from its `statusLine` hook (`~/.claude/settings.json`). Only contains
-   `rate_limits` when Claude Code is in an interactive Pro/Max session
-   and Anthropic chooses to populate them; can be silent for hours.
+   from its `statusLine` hook. **Lags** the truth: only updates when
+   Claude Code makes its own API call, which can be minutes behind
+   actual utilization. Used only when the plan-usage cache is missing.
    TTL: `CLAUDE_HOOK_TTL_SECONDS` (default 15 min).
-2. **`~/.config/codexmeter/claude-plan-usage.json`** — written by
-   `scripts/claude-plan-usage-firefox.py` (primary) every 60 s via the
-   `com.justin.codexmeter.claude-plan-scrape` LaunchAgent. Reads
-   `sessionKey` from Firefox's unencrypted `cookies.sqlite` and calls
-   `https://claude.ai/api/organizations/{org}/usage` directly. Structured
-   JSON — immune to claude.ai page-layout changes.
-   TTL: `CLAUDE_PLAN_TTL_SECONDS` (default 60 min).
 3. **`~/.config/codexmeter/claude-statusline.json`** — legacy raw
-   statusline capture; same data as #1 but read from a different cache.
+   statusline capture; same data as #2 but read from a different cache.
 4. **Stale cached `claude-plan-usage.json`** — past TTL, surfaced with
    `ok=False` and `status="stale Nh"`. Intentional: better to show the
    user the *last known good* percentages with a stale flag than a
@@ -126,9 +136,11 @@ GET http://localhost:9596/health   # diagnostic — source freshness, last scrap
 | Path / process | What it is |
 |---|---|
 | `daemon/codex-usage-daemon.py` | HTTP daemon, polls every 30 s |
-| `scripts/claude-plan-usage-firefox.py` | Primary scrape (Firefox cookies → claude.ai API) |
+| `scripts/claude-plan-usage-oauth-headers.py` | **Primary** scrape (OAuth bearer → /v1/messages rate-limit headers) |
+| `scripts/claude-plan-usage-desktop.py` | Fallback (Claude Desktop cookies → claude.ai API) |
+| `scripts/claude-plan-usage-firefox.py` | Second fallback (Firefox cookies → claude.ai API) |
 | `scripts/claude-plan-usage-cache.py` | Legacy Chrome AppleScript scrape — manual fallback only |
-| `scripts/claude-statusline-limits.py` | Persists Claude Code statusLine `rate_limits` |
+| `scripts/claude-statusline-limits.py` | Persists Claude Code statusLine `rate_limits` (lagging) |
 | `~/Library/LaunchAgents/com.justin.codexmeter.claude.plist` | Daemon (Claude provider, port 9596) |
 | `~/Library/LaunchAgents/com.justin.codexmeter.claude-plan-scrape.plist` | Scrape every 60 s |
 | `~/.config/codexmeter/claude-plan-usage.json` | Scrape output |
@@ -148,7 +160,7 @@ silently overriding a complete one is the recurring bug pattern
 
 ```bash
 # Force-refresh scrape and see what it'd write
-.venv/bin/python scripts/claude-plan-usage-firefox.py print
+.venv/bin/python scripts/claude-plan-usage-oauth-headers.py print
 
 # Watch the daemon's view in real time
 watch -n 5 'curl -s http://localhost:9596/usage'

daemon/codex-usage-daemon.py

@@ -1025,14 +1025,19 @@ def claude_stale_plan_usage_snapshot() -> UsageSnapshot | None:
 
 
 def local_claude_activity_snapshot() -> UsageSnapshot:
-    hook_snapshot = claude_hook_limits_snapshot()
-    if hook_snapshot is not None:
-        return hook_snapshot
-
+    # Prefer the plan-usage cache first: the new oauth-headers scraper writes
+    # to it on every tick, so it reflects the *current* utilization. The hook
+    # source lags behind because it only updates when Claude Code itself
+    # makes an inference call (could be minutes ago) — so it should only fill
+    # in when the cache is missing.
     plan_usage_snapshot = claude_plan_usage_snapshot()
     if plan_usage_snapshot is not None:
         return plan_usage_snapshot
 
+    hook_snapshot = claude_hook_limits_snapshot()
+    if hook_snapshot is not None:
+        return hook_snapshot
+
     rate_limit_snapshot = claude_rate_limit_snapshot()
     if rate_limit_snapshot is not None:
         return rate_limit_snapshot

install.sh

@@ -64,7 +64,7 @@ PLIST
     echo "Installed LaunchAgent: $PLIST_FILE"
     echo "Logs: $HOME/Library/Logs/codexmeter.log"
 
-    SCRAPE_SCRIPT="$SCRIPT_DIR/scripts/claude-plan-usage-firefox.py"
+    SCRAPE_SCRIPT="$SCRIPT_DIR/scripts/claude-plan-usage-oauth-headers.py"
     SCRAPE_PLIST="$PLIST_DIR/com.justin.codexmeter.claude-plan-scrape.plist"
     if [ -x "$SCRAPE_SCRIPT" ]; then
       cat > "$SCRAPE_PLIST" <<PLIST

scripts/claude-plan-usage-oauth-headers.py

@@ -0,0 +1,283 @@
+#!/usr/bin/env python3
+"""Read Claude plan usage from anthropic-ratelimit-unified-* response headers.
+
+Approach (adapted from Hermann Bjorgvin's Clawdmeter
+https://github.com/HermannBjorgvin/Clawdmeter):
+
+  1. Read the Claude Code OAuth access token from the macOS Keychain
+     (service "Claude Code-credentials") or from ~/.claude/.credentials.json
+     on Linux.
+  2. POST a minimal /v1/messages request to api.anthropic.com (1 token of
+     Haiku, ~$0.0001 per scrape).
+  3. Ignore the body. Parse:
+       anthropic-ratelimit-unified-5h-utilization     (0.0-1.0 float)
+       anthropic-ratelimit-unified-5h-reset           (epoch seconds)
+       anthropic-ratelimit-unified-7d-utilization
+       anthropic-ratelimit-unified-7d-reset
+       anthropic-ratelimit-unified-5h-status          (string)
+  4. Write the normalized JSON to claude-plan-usage.json — same shape the
+     daemon already reads.
+
+This is more durable than the cookie scrapes because:
+  * No browser or Cloudflare in the loop.
+  * Uses the documented /v1/messages endpoint.
+  * Claude Code rotates the OAuth token in Keychain on its own.
+
+If anything fails, the script falls back to claude-plan-usage-desktop.py
+(which itself falls through to the Firefox cookie scraper).
+"""
+
+from __future__ import annotations
+
+import argparse
+import getpass
+import json
+import os
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+from datetime import datetime
+from pathlib import Path
+
+CONFIG_DIR = Path.home() / ".config" / "codexmeter"
+CACHE_PATH = Path(
+    os.getenv(
+        "CODEXMETER_CLAUDE_PLAN_USAGE_FILE", str(CONFIG_DIR / "claude-plan-usage.json")
+    )
+)
+HEALTH_PATH = Path(
+    os.getenv(
+        "CODEXMETER_CLAUDE_SCRAPE_HEALTH", str(CONFIG_DIR / "claude-scrape-health.json")
+    )
+)
+KEYCHAIN_SERVICE = os.getenv(
+    "CODEXMETER_CLAUDE_CODE_KEYCHAIN_SERVICE", "Claude Code-credentials"
+)
+CREDENTIALS_PATH = Path(
+    os.getenv(
+        "CODEXMETER_CLAUDE_CODE_CREDENTIALS",
+        str(Path.home() / ".claude" / ".credentials.json"),
+    )
+)
+API_URL = "https://api.anthropic.com/v1/messages"
+API_HEADERS_TEMPLATE = {
+    "anthropic-version": "2023-06-01",
+    "anthropic-beta": "oauth-2025-04-20",
+    "content-type": "application/json",
+    "user-agent": os.getenv("CODEXMETER_CLAUDE_UA", "claude-code/2.1.5"),
+    "accept": "application/json",
+}
+API_BODY = {
+    "model": os.getenv("CODEXMETER_CLAUDE_PROBE_MODEL", "claude-haiku-4-5-20251001"),
+    "max_tokens": 1,
+    "messages": [{"role": "user", "content": "hi"}],
+}
+
+
+def _read_token_keychain() -> str | None:
+    try:
+        out = subprocess.run(
+            [
+                "security",
+                "find-generic-password",
+                "-w",
+                "-s",
+                KEYCHAIN_SERVICE,
+                "-a",
+                getpass.getuser(),
+            ],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        return None
+    if out.returncode != 0 or not out.stdout.strip():
+        return None
+    return _extract_access_token(out.stdout)
+
+
+def _read_token_file() -> str | None:
+    try:
+        raw = CREDENTIALS_PATH.read_text()
+    except OSError:
+        return None
+    return _extract_access_token(raw)
+
+
+def _extract_access_token(blob: str) -> str | None:
+    blob = blob.strip()
+    if not blob:
+        return None
+    try:
+        data = json.loads(blob)
+    except json.JSONDecodeError:
+        return None
+    if isinstance(data, dict):
+        # nested: {"claudeAiOauth": {"accessToken": "..."}}
+        for value in data.values():
+            if isinstance(value, dict) and isinstance(value.get("accessToken"), str):
+                return value["accessToken"]
+        if isinstance(data.get("accessToken"), str):
+            return data["accessToken"]
+    return None
+
+
+def read_oauth_token() -> str:
+    token = _read_token_keychain() if sys.platform == "darwin" else _read_token_file()
+    if not token:
+        # Try the other source as a last-ditch.
+        token = _read_token_file() if sys.platform == "darwin" else _read_token_keychain()
+    if not token:
+        raise SystemExit(
+            "Could not read Claude Code OAuth token (Keychain service "
+            f"'{KEYCHAIN_SERVICE}' or {CREDENTIALS_PATH}). Sign in to "
+            "Claude Code so it stores a fresh token."
+        )
+    return token
+
+
+def probe_headers(token: str) -> dict[str, str]:
+    body = json.dumps(API_BODY).encode()
+    headers = dict(API_HEADERS_TEMPLATE)
+    headers["authorization"] = f"Bearer {token}"
+    req = urllib.request.Request(API_URL, data=body, method="POST", headers=headers)
+    try:
+        with urllib.request.urlopen(req, timeout=20) as resp:
+            return {k.lower(): v for k, v in resp.getheaders()}
+    except urllib.error.HTTPError as exc:
+        # Some 4xx responses still return ratelimit headers; keep them in case
+        # we still get a useful snapshot. But error out on auth failures so the
+        # fallback chain runs.
+        body_text = exc.read().decode("utf-8", errors="replace")[:200]
+        if exc.code == 401:
+            raise SystemExit(
+                f"OAuth bearer rejected by /v1/messages (HTTP 401). Token may "
+                f"have expired. Body: {body_text}"
+            ) from exc
+        raise SystemExit(
+            f"/v1/messages returned HTTP {exc.code}: {body_text}"
+        ) from exc
+
+
+def pct(util: str | None) -> int:
+    if util is None:
+        return 0
+    try:
+        return int(round(float(util) * 100))
+    except ValueError:
+        return 0
+
+
+def reset_minutes(reset_ts: str | None) -> int:
+    if reset_ts is None:
+        return -1
+    try:
+        when = float(reset_ts)
+    except ValueError:
+        return -1
+    delta = (when - time.time()) / 60.0
+    return max(0, int(round(delta)))
+
+
+def normalize(headers: dict[str, str]) -> dict:
+    five_util = headers.get("anthropic-ratelimit-unified-5h-utilization")
+    five_reset = headers.get("anthropic-ratelimit-unified-5h-reset")
+    week_util = headers.get("anthropic-ratelimit-unified-7d-utilization")
+    week_reset = headers.get("anthropic-ratelimit-unified-7d-reset")
+    if five_util is None or week_util is None:
+        raise SystemExit(
+            f"Expected anthropic-ratelimit-unified-* headers; got: "
+            f"{sorted(k for k in headers if 'ratelimit' in k)}"
+        )
+    return {
+        "source": "claude_api_oauth_headers",
+        "captured_at": datetime.now().astimezone().isoformat(timespec="seconds"),
+        "session": {
+            "used_pct": pct(five_util),
+            "reset_mins": reset_minutes(five_reset),
+        },
+        "weekly": {
+            "used_pct": pct(week_util),
+            "reset_mins": reset_minutes(week_reset),
+        },
+    }
+
+
+def write_atomic(path: Path, payload: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    tmp.write_text(json.dumps(payload, separators=(",", ":")))
+    tmp.replace(path)
+
+
+def write_health(ok: bool, message: str, payload: dict | None = None) -> None:
+    health: dict = {
+        "ok": ok,
+        "source": "oauth_headers",
+        "checked_at": datetime.now().astimezone().isoformat(timespec="seconds"),
+        "message": message,
+    }
+    if payload:
+        health["last_payload"] = payload
+    try:
+        write_atomic(HEALTH_PATH, health)
+    except OSError:
+        pass
+
+
+def try_fallback(command: str) -> int:
+    fallback = Path(__file__).parent / "claude-plan-usage-desktop.py"
+    if not fallback.exists():
+        return 1
+    print(
+        f"oauth-headers scrape failed; falling back to {fallback.name} {command}",
+        file=sys.stderr,
+    )
+    proc = subprocess.run([sys.executable, str(fallback), command], check=False)
+    return proc.returncode
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument(
+        "command", nargs="?", default="scrape", choices=["scrape", "print"]
+    )
+    ap.add_argument(
+        "--no-fallback",
+        action="store_true",
+        help="Don't try the desktop-cookie script if this one fails.",
+    )
+    args = ap.parse_args()
+
+    try:
+        token = read_oauth_token()
+        headers = probe_headers(token)
+        payload = normalize(headers)
+    except SystemExit as exc:
+        msg = f"oauth-headers scrape failed: {exc}"
+        write_health(False, str(exc))
+        if args.no_fallback:
+            raise
+        print(msg, file=sys.stderr)
+        return try_fallback(args.command)
+    except Exception as exc:  # noqa: BLE001
+        msg = f"oauth-headers scrape failed: {type(exc).__name__}: {exc}"
+        write_health(False, msg)
+        if args.no_fallback:
+            raise SystemExit(msg) from exc
+        print(msg, file=sys.stderr)
+        return try_fallback(args.command)
+
+    if args.command == "scrape":
+        write_atomic(CACHE_PATH, payload)
+    write_health(True, "ok", payload)
+    print(json.dumps(payload, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())