opencode_usage_monitor.py

#!/usr/bin/env python3
"""opencode-go / Zen API usage watchdog.

Polls https://opencode.ai/zen/go/v1/usage with OPENCODE_GO_API_KEY (fallback
name: OPENCODE_ZEN_API_KEY). Silent unless something matters:

  - any window moved >= 2 percentage points since the last report
  - any window >= 85% (hard alert) or its quota status != ok
  - any window sits in the 70-84% band (warning)
  - first check of a new UTC day (daily digest)
  - very first run ever (proves the pipeline works)

API key resolution order (first hit wins):

  1. environment variables OPENCODE_GO_API_KEY / OPENCODE_ZEN_API_KEY
  2. $HERMES_HOME/.env
  3. ~/.hermes/.env

State/history files default to the Hermes home ($HERMES_HOME, else ~/.hermes)
and can be overridden with OPENCODE_WATCHDOG_STATE / OPENCODE_WATCHDOG_HISTORY.
All filesystem paths go through expanduser.

Modes:

  (default)   one silent-first tick; empty stdout means "all fine"
  --once      force-print the current report (debug; ignores and does not
              touch state/history)
  --selftest  offline fixture run over every report branch; exit code != 0
              on failure
"""

import argparse
import json
import os
import re
import shutil
import sys
import tempfile
import urllib.error
import urllib.request
from datetime import datetime, timedelta, timezone

URL = "https://opencode.ai/zen/go/v1/usage"
KEY_NAMES = ("OPENCODE_GO_API_KEY", "OPENCODE_ZEN_API_KEY")

ALERT_PCT = 85      # hard alert threshold
WARN_PCT = 70       # warning threshold
DELTA_PP = 2        # report when any window moved >= this many percentage points
def _dollar_limits():
    """Go plan cash caps; OPENCODE_WATCHDOG_DOLLAR_LIMITS ('12,30,60') overrides."""
    default = {"rolling": 12.0, "weekly": 30.0, "monthly": 60.0}
    raw = os.environ.get("OPENCODE_WATCHDOG_DOLLAR_LIMITS")
    if not raw:
        return default
    try:
        vals = [float(x.strip()) for x in raw.split(",")]
    except ValueError:
        return default
    if len(vals) != len(default):
        return default
    return dict(zip(default, vals))


DOLLAR_LIMITS = _dollar_limits()  # Go plan cash caps
WINDOWS = ("rolling", "weekly", "monthly")
FAIL_COOLDOWN_SEC = 3600  # rate-limit fetch-failure reports to one per hour

# The endpoint sits behind Cloudflare and 403s non-browser User-Agents.
# Version tokens are shortened so leak scans don't flag them as IP addresses.
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/126 Safari/537.36")


# ------------------------------------------------------------- paths and key

def hermes_home():
    """Portable Hermes home: $HERMES_HOME when set, else ~/.hermes."""
    hh = os.environ.get("HERMES_HOME")
    if hh:
        return os.path.expanduser(hh)
    return os.path.join(os.path.expanduser("~"), ".hermes")


def resolve_paths():
    """Return (state_path, history_path); env override beats the defaults."""
    home = hermes_home()
    state = os.environ.get("OPENCODE_WATCHDOG_STATE") or os.path.join(
        home, "opencode-usage-state.json")
    history = os.environ.get("OPENCODE_WATCHDOG_HISTORY") or os.path.join(
        home, "opencode-usage-history.jsonl")
    return os.path.expanduser(state), os.path.expanduser(history)


def env_file_candidates():
    """.env files to consult, in priority order."""
    paths = []
    hh = os.environ.get("HERMES_HOME")
    if hh:
        paths.append(os.path.join(os.path.expanduser(hh), ".env"))
    paths.append(os.path.join(os.path.expanduser("~"), ".hermes", ".env"))
    return paths


def _key_from_file(path):
    """First key found in one .env file; tolerates 'export' and quotes."""
    try:
        with open(path, "r", encoding="utf-8") as f:
            lines = f.readlines()
    except OSError:
        return None
    for name in KEY_NAMES:
        pat = re.compile(r'\s*(?:export\s+)?' + name + r'\s*=\s*(["\']?)([^"\'\n]+)')
        for line in lines:
            m = pat.match(line)
            if m:
                quoted, val = m.group(1), m.group(2).strip()
                if not quoted:
                    val = re.split(r"\s+#", val, maxsplit=1)[0].rstrip()
                return val
    return None


def get_key():
    """Env var first, then $HERMES_HOME/.env, then ~/.hermes/.env."""
    for name in KEY_NAMES:
        val = os.environ.get(name)
        if val and val.strip():
            return val.strip()
    for path in env_file_candidates():
        key = _key_from_file(path)
        if key:
            return key
    raise SystemExit("error: OPENCODE_GO_API_KEY / OPENCODE_ZEN_API_KEY not found "
                     "(looked in: environment, $HERMES_HOME/.env, ~/.hermes/.env)")


# --------------------------------------------------------------------- fetch

def fetch():
    req = urllib.request.Request(URL)
    req.add_header("Authorization", "Bearer " + get_key())
    req.add_header("User-Agent", UA)
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        raw = ""
        try:
            raw = e.read().decode("utf-8", "replace")
        except Exception:
            raw = ""
        detail = ""
        if raw:
            try:
                parsed = json.loads(raw)
                if isinstance(parsed, dict):
                    bits = [f"{k}={parsed[k]}" for k in
                            ("message", "error", "reset", "resetsAt", "resetAt")
                            if k in parsed]
                    if bits:
                        detail = "; ".join(bits)
            except Exception:
                pass
            if not detail:
                detail = raw[:200]
        raise RuntimeError(
            f"HTTP {e.code}{': ' + detail if detail else ''}"
        ) from e


# --------------------------------------------------------------------- state

def load_state(path):
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {}


def save_state(state, path):
    """Atomic write: serialize to a tmp file, then os.replace into place."""
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(state, f)
    os.replace(tmp, path)


# -------------------------------------------------------------------- report

def fmt_until(iso):
    """Human-friendly countdown from an ISO reset timestamp."""
    try:
        dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
        secs = (dt - datetime.now(timezone.utc)).total_seconds()
        if secs <= 0:
            return "resets now"
        h = int(secs // 3600)
        m = int((secs % 3600) // 60)
        return f"~{h}h{m}m" if h < 24 else f"~{h // 24}d{h % 24}h"
    except Exception:
        return "?"


def build_report(data, change=None, alert=None, warn=None,
                 digest=False, first=False, forced=False):
    """Render one report. Exactly one title header, then one line per window."""
    u = data["usage"]
    lines = []
    if alert:
        lines.append("⚠️ OPENCODE-GO USAGE ALERT")
        lines.append(f"  (>= {ALERT_PCT}% or quota error: {', '.join(alert)})")
    if warn:
        lines.append(f"⚡ approaching limit (>= {WARN_PCT}%): {', '.join(warn)}")
    if not alert and not warn:
        if digest:
            lines.append("📊 opencode-go usage digest")
        elif change:
            lines.append("🔄 opencode-go usage change")
        elif first:
            lines.append("📡 opencode-go usage — first report")
        elif forced:
            lines.append("📡 opencode-go usage — current status")

    for w in WINDOWS:
        x = u[w]
        mark = ""
        if change and w in change:
            mark = f"  ({'+' if change[w] > 0 else ''}{change[w]}pp)"
        st = "⚠️" if x["status"] != "ok" else ""
        buck = f"${x['percent'] * DOLLAR_LIMITS[w] / 100:.2f}/{DOLLAR_LIMITS[w]:.0f}"
        lines.append(f"  {w:<8} {x['percent']:>3}% {buck:>10}  {st}  "
                     f"{fmt_until(x['resetsAt'])}{mark}")
    if "channel" in u:
        lines.append(f"  channel: {u['channel']}")
    return "\n".join(lines)


# ---------------------------------------------------------------------- tick

def tick(data, now=None, state_path=None, history_path=None):
    """Process one successful fetch. Returns report text, or None when silent."""
    now = now or datetime.now(timezone.utc)
    default_state, default_history = resolve_paths()
    state_path = state_path or default_state
    history_path = history_path or default_history
    u = data["usage"]

    # history: one JSON line per successful check
    rec = {"ts": now.isoformat()}
    for w in WINDOWS:
        rec[w] = u[w]["percent"]
    rec["status"] = {w: u[w]["status"] for w in WINDOWS}
    with open(history_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(rec) + "\n")

    prev = load_state(state_path)
    first = "last" not in prev  # no previous successful snapshot -> first run
    last = prev.get("last", {})

    change = {}
    for w in WINDOWS:
        if w in last:
            d = u[w]["percent"] - last[w]
            if abs(d) >= DELTA_PP:
                change[w] = d

    alerts = [w for w in WINDOWS
              if u[w]["percent"] >= ALERT_PCT or u[w]["status"] != "ok"]
    warns = [w for w in WINDOWS
             if WARN_PCT <= u[w]["percent"] < ALERT_PCT]
    digest = prev.get("day") != now.strftime("%Y-%m-%d")

    out = None
    if first or change or alerts or warns or digest:
        out = build_report(data, change=change, alert=alerts, warn=warns,
                           digest=digest and not first, first=first)

    save_state({"last": {w: u[w]["percent"] for w in WINDOWS},
                "day": now.strftime("%Y-%m-%d"),
                "init": True}, state_path)
    return out


def tick_failure(err, now=None, state_path=None):
    """Rate-limited failure reporting: at most one line per hour."""
    now = now or datetime.now(timezone.utc)
    default_state, _ = resolve_paths()
    state_path = state_path or default_state
    st = load_state(state_path)
    if st.get("last_failed", 0) >= now.timestamp() - FAIL_COOLDOWN_SEC:
        return None
    save_state({**st, "last_failed": now.timestamp()}, state_path)
    return f"⚠️ opencode-go usage check FAILED: {err}"


# ------------------------------------------------------------------ selftest

def _fixture(rolling=9, weekly=3, monthly=1):
    """Built-in sample payload shaped like the real /usage response."""
    now = datetime.now(timezone.utc)

    def resets(**kw):
        return (now + timedelta(**kw)).isoformat().replace("+00:00", "Z")

    return {"usage": {
        "rolling": {"percent": rolling, "status": "ok",
                    "resetsAt": resets(hours=3, minutes=11)},
        "weekly": {"percent": weekly, "status": "ok",
                   "resetsAt": resets(days=2, hours=14)},
        "monthly": {"percent": monthly, "status": "ok",
                    "resetsAt": resets(days=30, hours=19)},
        "channel": "go",
    }}


def selftest():
    """Offline check of every branch. No network, no real state files.

    Returns 0 when every check passes, 1 otherwise.
    """
    checks = []

    def check(name, ok, detail=""):
        checks.append(bool(ok))
        line = f"[{'PASS' if ok else 'FAIL'}] {name}"
        if detail:
            line += f" — {detail}"
        print(line)

    def show(text):
        for ln in (text or "(silent — empty stdout)").splitlines():
            print(f"    | {ln}")

    saved_env = dict(os.environ)
    tmp = tempfile.mkdtemp(prefix="ocw-selftest-")
    try:
        home = os.path.join(tmp, "home")
        hh = os.path.join(tmp, "hermes-home")
        os.makedirs(os.path.join(home, ".hermes"))
        os.makedirs(hh)
        state = os.path.join(tmp, "state.json")
        history = os.path.join(tmp, "history.jsonl")

        # fully isolated sandbox: no real keys, no real $HOME, no network
        for k in ("OPENCODE_GO_API_KEY", "OPENCODE_ZEN_API_KEY", "HERMES_HOME"):
            os.environ.pop(k, None)
        os.environ["HOME"] = home
        os.environ["OPENCODE_WATCHDOG_STATE"] = state
        os.environ["OPENCODE_WATCHDOG_HISTORY"] = history

        with open(os.path.join(hh, ".env"), "w", encoding="utf-8") as f:
            f.write("OPENCODE_GO_API_KEY=file-key-hermes-home\n")
        with open(os.path.join(home, ".hermes", ".env"), "w", encoding="utf-8") as f:
            f.write('export OPENCODE_ZEN_API_KEY="file-key-home"\n')

        today = datetime.now(timezone.utc).strftime("%Y-%m-%d")

        print("== key resolution ==")
        os.environ["HERMES_HOME"] = hh
        os.environ["OPENCODE_GO_API_KEY"] = "env-key-go"
        check("env OPENCODE_GO_API_KEY beats $HERMES_HOME/.env",
              get_key() == "env-key-go", get_key())
        del os.environ["OPENCODE_GO_API_KEY"]
        os.environ["OPENCODE_ZEN_API_KEY"] = "env-key-zen"
        check("env OPENCODE_ZEN_API_KEY accepted as fallback name",
              get_key() == "env-key-zen", get_key())
        del os.environ["OPENCODE_ZEN_API_KEY"]
        check("$HERMES_HOME/.env used when no env key is set",
              get_key() == "file-key-hermes-home", get_key())
        del os.environ["HERMES_HOME"]
        check("~/.hermes/.env used as last resort (export/quotes parsed)",
              get_key() == "file-key-home", get_key())

        print("\n== report branches ==")
        out = tick(_fixture())
        check("first run prints a first report",
              out is not None
              and out.splitlines()[0] == "📡 opencode-go usage — first report")
        show(out)

        out = tick(_fixture())
        check("identical second run is silent", out is None)
        show(out)

        out = tick(_fixture(rolling=14))  # 9 -> 14 = +5pp
        check(">=2pp move prints the change header",
              out is not None
              and out.splitlines()[0] == "🔄 opencode-go usage change"
              and "+5pp" in out)
        show(out)

        out = tick(_fixture(rolling=90))  # crosses ALERT_PCT
        check(">=85% prints the alert header (single title, no change header)",
              out is not None
              and out.splitlines()[0] == "⚠️ OPENCODE-GO USAGE ALERT"
              and "usage change" not in out)
        show(out)

        save_state({"last": {"rolling": 10, "weekly": 3, "monthly": 1},
                    "day": today, "init": True}, state)
        fx = _fixture(rolling=10)
        fx["usage"]["weekly"]["status"] = "degraded"
        out = tick(fx)
        check("status != ok alerts even below the percent threshold",
              out is not None
              and out.splitlines()[0] == "⚠️ OPENCODE-GO USAGE ALERT"
              and "weekly" in out.splitlines()[1])
        show(out)

        save_state({"last": {"rolling": 90, "weekly": 3, "monthly": 1},
                    "day": today, "init": True}, state)
        out = tick(_fixture(rolling=75))  # 70-84 band
        check("70-84% prints the warning line as header",
              out is not None
              and out.splitlines()[0].startswith("⚡ approaching limit"))
        show(out)

        save_state({"last": {"rolling": 9, "weekly": 3, "monthly": 1},
                    "day": "2000-01-01", "init": True}, state)
        out = tick(_fixture())
        check("first check of a new UTC day prints the digest header",
              out is not None
              and out.splitlines()[0] == "📊 opencode-go usage digest")
        show(out)

        print("\n== failure & storage ==")
        err1 = tick_failure("simulated fetch error")
        err2 = tick_failure("simulated fetch error")
        check("first failure reports once", err1 is not None and "FAILED" in err1)
        check("second failure within the hour stays silent", err2 is None)

        tmp_leftover = os.path.exists(state + ".tmp")
        try:
            with open(state, encoding="utf-8") as f:
                json.load(f)
            valid = True
        except Exception:
            valid = False
        check("state file is valid JSON with no tmp residue (atomic write)",
              valid and not tmp_leftover)

        with open(history, encoding="utf-8") as f:
            n_history = sum(1 for _ in f)
        check("history gains one line per successful tick",
              n_history == 7, f"{n_history} lines")

        forced = build_report(_fixture(), forced=True)
        check("--once forced report renders with a header",
              forced.splitlines()[0] == "📡 opencode-go usage — current status")
        show(forced)

        passed = sum(checks)
        print(f"\nselftest: {passed}/{len(checks)} checks passed")
        return 0 if passed == len(checks) else 1
    finally:
        os.environ.clear()
        os.environ.update(saved_env)
        shutil.rmtree(tmp, ignore_errors=True)


# ---------------------------------------------------------------------- main

def parse_args(argv=None):
    p = argparse.ArgumentParser(
        description="opencode-go usage watchdog — silent unless something matters.")
    p.add_argument("--once", action="store_true",
                   help="force-print the current report (debug; ignores and "
                        "does not touch state/history)")
    p.add_argument("--selftest", action="store_true",
                   help="offline self-check of every report branch; "
                        "exit code != 0 on failure")
    return p.parse_args(argv)


def main(argv=None):
    args = parse_args(argv)

    if args.selftest:
        return selftest()

    if args.once:
        try:
            data = fetch()
        except SystemExit:
            raise  # missing API key: keep the clear message and exit 1
        except Exception as e:
            print(f"⚠️ opencode-go usage check FAILED: {e}", file=sys.stderr)
            return 2
        u = data["usage"]
        alerts = [w for w in WINDOWS
                  if u[w]["percent"] >= ALERT_PCT or u[w]["status"] != "ok"]
        warns = [w for w in WINDOWS
                 if WARN_PCT <= u[w]["percent"] < ALERT_PCT]
        print(build_report(data, alert=alerts, warn=warns, forced=True))
        return 0

    now = datetime.now(timezone.utc)
    try:
        data = fetch()
    except Exception as e:
        out = tick_failure(e, now=now)
    else:
        try:
            out = tick(data, now=now)
        except Exception as e:
            out = tick_failure(e, now=now)
    if out:
        print(out)
    return 0


if __name__ == "__main__":
    sys.exit(main())