#!/usr/bin/env python3 """ oauth_usage_collect.py — OpenClaw OAuth/Codex consumption telemetry collector. Calls `openclaw status`, parses output, appends one JSONL line to the history file. Scheduled by ai.openclaw.oauth-telemetry LaunchAgent (5-min interval). Output schema: { timestamp, day_pct, week_pct, hours_until_reset, credits_remaining, raw_output, error?, error_detail? } credits_remaining is always null — openclaw status does not report absolute credit counts. The 200-credit alert threshold falls back to 80% day_pct proxy. """ import subprocess import json import re import sys import os import tempfile from datetime import datetime, timezone JSONL_PATH = "/Users/openclaw/.openclaw/workspace/state/oauth-usage-history.jsonl" OPENCLAW_BIN = "/opt/homebrew/bin/openclaw" TIMEOUT_SECS = 15 MAX_LINES = 10000 ARCHIVE_KEEP = 3 # --------------------------------------------------------------------------- # openclaw status --json --usage # --------------------------------------------------------------------------- def get_openclaw_status(): """Run `openclaw status --json --usage` and return (stdout, returncode).""" try: result = subprocess.run( [OPENCLAW_BIN, "status", "--json", "--usage"], capture_output=True, text=True, timeout=TIMEOUT_SECS ) return result.stdout.strip(), result.returncode except subprocess.TimeoutExpired: return "", -1 except FileNotFoundError: return "", -2 def parse_status(raw): """ Parse openclaw status --json --usage output. Extracts day_pct, week_pct, hours_until_reset from the usage.providers[] entry for openai-codex. Falls back to regex parsing if JSON fails. Returns (day_pct, week_pct, hours_until_reset) as floats, or None per field if not found. """ day_pct = None week_pct = None hours_until_reset = None # The JSON output may be preceded by stderr lines (e.g. token refresh # warnings) printed to stdout. Find the first '{' and parse from there. json_start = raw.find('{') if json_start == -1: return day_pct, week_pct, hours_until_reset try: data = json.loads(raw[json_start:]) except json.JSONDecodeError: return day_pct, week_pct, hours_until_reset usage = data.get("usage", {}) providers = usage.get("providers", []) for provider in providers: if provider.get("provider") != "openai-codex": continue windows = provider.get("windows", []) now_epoch = datetime.now(timezone.utc).timestamp() for w in windows: label = (w.get("label") or "").lower() pct = w.get("usedPercent") reset_at = w.get("resetAt") if "5h" in label or "day" in label or "daily" in label: day_pct = float(pct) if pct is not None else None if reset_at and day_pct is not None: hours_until_reset = max(0, (reset_at / 1000 - now_epoch) / 3600) elif "week" in label: week_pct = float(pct) if pct is not None else None break return day_pct, week_pct, hours_until_reset # --------------------------------------------------------------------------- # Log rotation # --------------------------------------------------------------------------- def rotate_if_needed(path): """Rotate JSONL if it exceeds MAX_LINES. Retain ARCHIVE_KEEP archives.""" if not os.path.exists(path): return with open(path, 'r') as f: lines = f.readlines() if len(lines) < MAX_LINES: return date_str = datetime.now(timezone.utc).strftime('%Y-%m-%d') base = path.replace('.jsonl', '') archive_path = f"{base}-{date_str}.jsonl" # Avoid clobbering an existing archive for the same date counter = 0 while os.path.exists(archive_path): counter += 1 archive_path = f"{base}-{date_str}-{counter}.jsonl" os.rename(path, archive_path) # Prune oldest archives beyond ARCHIVE_KEEP dirpath = os.path.dirname(path) basename = os.path.basename(base) archives = sorted([ f for f in os.listdir(dirpath) if f.startswith(basename + '-') and f.endswith('.jsonl') ]) while len(archives) > ARCHIVE_KEEP: os.remove(os.path.join(dirpath, archives.pop(0))) # --------------------------------------------------------------------------- # Atomic append # --------------------------------------------------------------------------- def append_line(path, line): """Append a single JSONL line atomically (write to temp, then append).""" os.makedirs(os.path.dirname(path), exist_ok=True) dirpath = os.path.dirname(path) with tempfile.NamedTemporaryFile('w', dir=dirpath, delete=False, suffix='.tmp') as tmp: tmp.write(line + '\n') tmp_path = tmp.name with open(tmp_path, 'r') as src, open(path, 'a') as dst: dst.write(src.read()) os.unlink(tmp_path) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): ts = datetime.now(timezone.utc).isoformat() raw, returncode = get_openclaw_status() # Hard failure: command didn't run or timed out if returncode not in (0,) or not raw: record = { "timestamp": ts, "day_pct": None, "week_pct": None, "hours_until_reset": None, "credits_remaining": None, "raw_output": raw, "error": True, "error_detail": ( "timeout" if returncode == -1 else "openclaw binary not found" if returncode == -2 else f"non-zero exit ({returncode})" if returncode != 0 else "empty output" ) } append_line(JSONL_PATH, json.dumps(record)) sys.exit(1) day_pct, week_pct, hours_until_reset = parse_status(raw) # Parse failure: command ran but output was unrecognisable if day_pct is None and week_pct is None and hours_until_reset is None: record = { "timestamp": ts, "day_pct": None, "week_pct": None, "hours_until_reset": None, "credits_remaining": None, "raw_output": raw, "error": True, "error_detail": "could not parse any fields from openclaw status output" } append_line(JSONL_PATH, json.dumps(record)) sys.exit(1) # Success path rotate_if_needed(JSONL_PATH) record = { "timestamp": ts, "day_pct": day_pct, "week_pct": week_pct, "hours_until_reset": hours_until_reset, "credits_remaining": None, "raw_output": raw } append_line(JSONL_PATH, json.dumps(record)) sys.exit(0) if __name__ == "__main__": main()