Fix Claude 0/0 usage with multi-source limits + auto-refresh

daemon/codex-usage-daemon.py

@@ -150,6 +150,14 @@ def load_env_file() -> None:
         str(CACHE_DIR / "claude-plan-usage.json"),
     )
 )
+CLAUDE_LIMITS_FILE = Path(
+    os.getenv(
+        "CODEXMETER_CLAUDE_LIMITS_FILE",
+        str(CLAUDE_HOME / "plan-limits.json"),
+    )
+)
+CLAUDE_HOOK_TTL_SECONDS = int(os.getenv("CODEXMETER_CLAUDE_HOOK_TTL", str(15 * 60)))
+CLAUDE_PLAN_TTL_SECONDS = int(os.getenv("CODEXMETER_CLAUDE_PLAN_TTL", str(60 * 60)))
 SERIAL_PATTERNS = (
     "/dev/cu.usbmodemDC5475CBBC601",
     "/dev/cu.usbmodem*",
@@ -769,6 +777,54 @@ def claude_usage_tokens(usage: dict) -> int:
     return total
 
 
+def _number_or_none(value: Any) -> float | int | None:
+    if value is None:
+        return None
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    return int(number) if number.is_integer() else number
+
+
+def claude_hook_limits_snapshot() -> UsageSnapshot | None:
+    """Read normalized rate limits written by scripts/claude-statusline-limits.py."""
+    try:
+        stat = CLAUDE_LIMITS_FILE.stat()
+    except OSError:
+        return None
+    if time.time() - stat.st_mtime > CLAUDE_HOOK_TTL_SECONDS:
+        return None
+    try:
+        data = json.loads(CLAUDE_LIMITS_FILE.read_text(errors="replace"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    if not isinstance(data, dict):
+        return None
+
+    five_hour = data.get("five_hour") if isinstance(data.get("five_hour"), dict) else None
+    seven_day = data.get("seven_day") if isinstance(data.get("seven_day"), dict) else None
+    five_used = _number_or_none(five_hour.get("used_percentage")) if five_hour else None
+    week_used = _number_or_none(seven_day.get("used_percentage")) if seven_day else None
+
+    # Require both windows so a partial payload can't override a complete cached
+    # source downstream (a missing window would otherwise display as 100% remaining).
+    if five_used is None or week_used is None:
+        return None
+
+    five_reset = minutes_until_epoch(five_hour.get("resets_at"))
+    week_reset = minutes_until_epoch(seven_day.get("resets_at"))
+
+    return UsageSnapshot(
+        session_pct=remaining_pct_from_used(float(five_used)),
+        session_reset_mins=five_reset,
+        weekly_pct=remaining_pct_from_used(float(week_used)),
+        weekly_reset_mins=week_reset,
+        status=f"5h {round(float(five_used))}% used · 7d {round(float(week_used))}% used",
+        ok=True,
+    )
+
+
 def claude_rate_limit_snapshot() -> UsageSnapshot | None:
     """Read Claude Code's official status-line rate limit fields, if present."""
     candidates = [CLAUDE_RATE_LIMITS_FILE]
@@ -779,7 +835,7 @@ def claude_rate_limit_snapshot() -> UsageSnapshot | None:
             stat = candidate.stat()
         except OSError:
             continue
-        if time.time() - stat.st_mtime > 15 * 60:
+        if time.time() - stat.st_mtime > CLAUDE_HOOK_TTL_SECONDS:
             continue
         try:
             data = json.loads(candidate.read_text(errors="replace"))
@@ -845,7 +901,7 @@ def claude_plan_usage_snapshot() -> UsageSnapshot | None:
     except OSError:
         return None
 
-    if time.time() - stat.st_mtime > 10 * 60:
+    if time.time() - stat.st_mtime > CLAUDE_PLAN_TTL_SECONDS:
         return None
 
     try:
@@ -924,6 +980,10 @@ def claude_limit_message_snapshot(message: str) -> UsageSnapshot | None:
 
 
 def local_claude_activity_snapshot() -> UsageSnapshot:
+    hook_snapshot = claude_hook_limits_snapshot()
+    if hook_snapshot is not None:
+        return hook_snapshot
+
     plan_usage_snapshot = claude_plan_usage_snapshot()
     if plan_usage_snapshot is not None:
         return plan_usage_snapshot

install.sh

@@ -63,6 +63,33 @@ PLIST
     launchctl load "$PLIST_FILE"
     echo "Installed LaunchAgent: $PLIST_FILE"
     echo "Logs: $HOME/Library/Logs/codexmeter.log"
+
+    SCRAPE_SCRIPT="$SCRIPT_DIR/scripts/claude-plan-usage-cache.py"
+    SCRAPE_PLIST="$PLIST_DIR/com.justin.codexmeter.claude-plan-scrape.plist"
+    if [ -x "$SCRAPE_SCRIPT" ]; then
+      cat > "$SCRAPE_PLIST" <<PLIST
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
+  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
+<plist version="1.0">
+<dict>
+  <key>Label</key><string>com.justin.codexmeter.claude-plan-scrape</string>
+  <key>ProgramArguments</key>
+  <array>
+    <string>$PYTHON</string>
+    <string>$SCRAPE_SCRIPT</string>
+  </array>
+  <key>StartInterval</key><integer>60</integer>
+  <key>RunAtLoad</key><true/>
+  <key>StandardOutPath</key><string>$HOME/Library/Logs/codexmeter-claude-scrape.log</string>
+  <key>StandardErrorPath</key><string>$HOME/Library/Logs/codexmeter-claude-scrape.err.log</string>
+</dict>
+</plist>
+PLIST
+      launchctl unload "$SCRAPE_PLIST" >/dev/null 2>&1 || true
+      launchctl load "$SCRAPE_PLIST"
+      echo "Installed LaunchAgent: $SCRAPE_PLIST (every 60s)"
+    fi
     ;;
   Linux)
     USER_SERVICE_DIR="$HOME/.config/systemd/user"

scripts/claude-plan-usage-cache.py

@@ -7,7 +7,6 @@
 import re
 import subprocess
 import sys
-import time
 from datetime import datetime, timedelta
 from pathlib import Path
 

scripts/claude-statusline-limits.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""Persist Claude Code statusLine rate limits for CodexMeter."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+OUT_PATH = Path(
+    os.getenv("CODEXMETER_CLAUDE_LIMITS_OUT", str(Path.home() / ".claude" / "plan-limits.json"))
+)
+DEBUG_PATH_ENV = os.getenv("CODEXMETER_CLAUDE_LIMITS_DEBUG")
+DEBUG_PATH = Path(DEBUG_PATH_ENV) if DEBUG_PATH_ENV else None
+
+
+def number_or_none(value: Any) -> float | int | None:
+    if value is None:
+        return None
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    return int(number) if number.is_integer() else number
+
+
+def limit_window(raw: Any) -> dict | None:
+    if not isinstance(raw, dict):
+        return None
+    used = number_or_none(raw.get("used_percentage"))
+    resets_at = number_or_none(raw.get("resets_at"))
+    if used is None and resets_at is None:
+        return None
+    return {
+        "used_percentage": used,
+        "resets_at": resets_at,
+    }
+
+
+def normalized_limits(statusline: dict) -> dict | None:
+    rate_limits = statusline.get("rate_limits")
+    if not isinstance(rate_limits, dict):
+        return None
+
+    five_hour = limit_window(rate_limits.get("five_hour"))
+    seven_day = limit_window(rate_limits.get("seven_day"))
+    if five_hour is None and seven_day is None:
+        return None
+
+    payload: dict[str, Any] = {
+        "provider": "claude",
+        "source": "claude_code_statusline",
+        "updated_at": datetime.now(timezone.utc).isoformat(timespec="seconds").replace(
+            "+00:00", "Z"
+        ),
+    }
+    if five_hour is not None:
+        payload["five_hour"] = five_hour
+    if seven_day is not None:
+        payload["seven_day"] = seven_day
+    return payload
+
+
+def write_payload(path: Path, payload: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".json.tmp")
+    tmp.write_text(json.dumps(payload, separators=(",", ":")))
+    tmp.replace(path)
+
+
+def main() -> int:
+    raw = sys.stdin.read()
+    try:
+        statusline = json.loads(raw or "{}")
+    except json.JSONDecodeError:
+        statusline = {}
+
+    if DEBUG_PATH is not None:
+        debug = {
+            "captured_at": datetime.now(timezone.utc)
+            .isoformat(timespec="seconds")
+            .replace("+00:00", "Z"),
+            "keys": sorted(list(statusline.keys())) if isinstance(statusline, dict) else None,
+            "has_rate_limits": isinstance(statusline, dict)
+            and isinstance(statusline.get("rate_limits"), dict),
+            "raw": statusline,
+        }
+        try:
+            write_payload(DEBUG_PATH, debug)
+        except OSError:
+            pass
+
+    payload = normalized_limits(statusline)
+    if payload is None:
+        print("Claude limits pending")
+        return 0
+
+    write_payload(OUT_PATH, payload)
+    five = payload.get("five_hour", {}).get("used_percentage", "--")
+    seven = payload.get("seven_day", {}).get("used_percentage", "--")
+    print(f"Claude 5h: {five}% 7d: {seven}%")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())