diff --git a/daemon/codex-usage-daemon.py b/daemon/codex-usage-daemon.py index c26be1a..57cb090 100755 --- a/daemon/codex-usage-daemon.py +++ b/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 diff --git a/install.sh b/install.sh index 9b16cf6..40f5814 100755 --- a/install.sh +++ b/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" < + + + + Labelcom.justin.codexmeter.claude-plan-scrape + ProgramArguments + + $PYTHON + $SCRAPE_SCRIPT + + StartInterval60 + RunAtLoad + StandardOutPath$HOME/Library/Logs/codexmeter-claude-scrape.log + StandardErrorPath$HOME/Library/Logs/codexmeter-claude-scrape.err.log + + +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" diff --git a/scripts/claude-plan-usage-cache.py b/scripts/claude-plan-usage-cache.py index 08c5d1f..159d4cd 100755 --- a/scripts/claude-plan-usage-cache.py +++ b/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 diff --git a/scripts/claude-statusline-limits.py b/scripts/claude-statusline-limits.py new file mode 100755 index 0000000..cd3485b --- /dev/null +++ b/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())