Grok Bot

"""Activity log + webhook governor (1.3.13).

Why: John (2026-10-15) saw Claude Desktop's Grok requests fail with throttling while other AI
threads were hammering the same Grok Bridge webhook, and grok.log could not say who, when or
what came back (no timestamps, no results). Now every verb call is one JSONL line in
~/.adom/grok-bridge-activity.jsonl, and grok_webhook_fire goes through a governor:

  - a per-PC budget (default 6 fires a minute) and a per-thread budget (default 3 a minute),
    so one noisy thread cannot use up the shared mailbox;
  - a cooldown after Grok answers 429 / "rate limit": honour Retry-After (else 60 s, doubling
    to 300 s), refuse locally until it passes, so callers do not extend the throttle.

Never logs the webhook key, payload messages or transcript text. Limits are overridable in
~/.adom/grok-bridge-settings.json: {"webhookPerMinute": 6, "webhookPerThreadPerMinute": 3}.
"""

from __future__ import annotations

import collections
import json
import os
import threading
import time
from datetime import datetime

_DIR = os.path.join(os.path.expanduser("~"), ".adom")
LOG_PATH = os.path.join(_DIR, "grok-bridge-activity.jsonl")
COOLDOWN_PATH = os.path.join(_DIR, "grok-webhook-cooldown.json")
SETTINGS_PATH = os.path.join(_DIR, "grok-bridge-settings.json")
MAX_BYTES = 5 * 1024 * 1024

DEFAULT_PER_MINUTE = 6
DEFAULT_PER_THREAD_PER_MINUTE = 3
COOLDOWN_DEFAULT = 60
COOLDOWN_MAX = 300

_lock = threading.Lock()
_fires: collections.deque = collections.deque()          # (t, caller) of accepted fires

# Result fields worth keeping (never message text, keys or transcripts).
_KEEP = ("errorCode", "event", "httpStatus", "runUuid", "nonce", "agent", "retryAfterSeconds",
         "throttledBy", "warning", "computerState", "launched", "restarted", "mainWindowHidden")


def now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def _write(entry: dict) -> None:
    try:
        os.makedirs(_DIR, exist_ok=True)
        with _lock:
            if os.path.exists(LOG_PATH) and os.path.getsize(LOG_PATH) > MAX_BYTES:
                os.replace(LOG_PATH, LOG_PATH + ".1")
            with open(LOG_PATH, "a", encoding="utf-8") as f:
                f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    except Exception:
        pass   # logging must never break a verb


def log_call(verb: str, caller: str, ms: int, result) -> None:
    entry = {"ts": now_iso(), "verb": verb, "caller": caller or "anonymous", "ms": ms}
    if isinstance(result, dict):
        entry["ok"] = result.get("success") is not False
        for k in _KEEP:
            if result.get(k) is not None:
                entry[k] = result.get(k)
    _write(entry)


def _settings() -> dict:
    try:
        with open(SETTINGS_PATH, encoding="utf-8") as f:
            d = json.load(f)
            return d if isinstance(d, dict) else {}
    except Exception:
        return {}


def limits() -> tuple[int, int]:
    s = _settings()
    try:
        per_min = max(1, int(s.get("webhookPerMinute") or DEFAULT_PER_MINUTE))
    except (TypeError, ValueError):
        per_min = DEFAULT_PER_MINUTE
    try:
        per_thread = max(1, int(s.get("webhookPerThreadPerMinute") or DEFAULT_PER_THREAD_PER_MINUTE))
    except (TypeError, ValueError):
        per_thread = DEFAULT_PER_THREAD_PER_MINUTE
    return per_min, min(per_thread, per_min)


def _read_cooldown() -> dict:
    try:
        with open(COOLDOWN_PATH, encoding="utf-8") as f:
            return json.load(f) or {}
    except Exception:
        return {}


def cooldown_state() -> dict:
    c = _read_cooldown()
    left = int(float(c.get("until") or 0) - time.time())
    return {"active": left > 0, "secondsLeft": max(0, left), "since": c.get("since"),
            "httpStatus": c.get("httpStatus"), "strikes": c.get("strikes", 0)}


def note_throttled(http_status, retry_after) -> int:
    """Grok throttled us: start/extend the shared cooldown. Returns seconds to wait."""
    c = _read_cooldown()
    strikes = int(c.get("strikes") or 0) + 1 if float(c.get("until") or 0) > time.time() - 600 else 1
    try:
        wait = int(float(retry_after)) if retry_after not in (None, "") else 0
    except (TypeError, ValueError):
        wait = 0
    if wait <= 0:
        wait = min(COOLDOWN_MAX, COOLDOWN_DEFAULT * (2 ** (strikes - 1)))
    wait = min(max(wait, 5), COOLDOWN_MAX)
    try:
        os.makedirs(_DIR, exist_ok=True)
        with open(COOLDOWN_PATH, "w", encoding="utf-8") as f:
            json.dump({"until": time.time() + wait, "since": now_iso(), "httpStatus": http_status,
                       "strikes": strikes}, f)
    except Exception:
        pass
    return wait


def note_ok() -> None:
    """A fire went through: clear strikes once any cooldown has passed."""
    c = _read_cooldown()
    if c and float(c.get("until") or 0) <= time.time():
        try:
            os.remove(COOLDOWN_PATH)
        except Exception:
            pass


def admit(caller: str) -> dict | None:
    """None = may fire now (and the fire is counted). Otherwise a refusal dict."""
    cd = cooldown_state()
    if cd["active"]:
        return {"throttledBy": "grok", "retryAfterSeconds": cd["secondsLeft"],
                "why": f"Grok throttled this PC's webhook (HTTP {cd['httpStatus']}) at {cd['since']}; "
                       f"cooling down {cd['secondsLeft']} s more. Not sent, so the throttle is not extended."}
    per_min, per_thread = limits()
    caller = caller or "anonymous"
    now = time.time()
    with _lock:
        while _fires and now - _fires[0][0] > 60:
            _fires.popleft()
        mine = [t for t, c in _fires if c == caller]
        if len(mine) >= per_thread:
            wait = int(60 - (now - mine[0])) + 1
            return {"throttledBy": "thread-budget", "retryAfterSeconds": wait,
                    "why": f"thread {caller!r} already fired {len(mine)} times in the last minute "
                           f"(limit {per_thread} per thread, so other threads can still reach Grok)."}
        if len(_fires) >= per_min:
            wait = int(60 - (now - _fires[0][0])) + 1
            return {"throttledBy": "pc-budget", "retryAfterSeconds": wait,
                    "why": f"{len(_fires)} webhook fires from this PC in the last minute (limit {per_min})."}
        _fires.append((now, caller))
    return None


def budget_state() -> dict:
    per_min, per_thread = limits()
    now = time.time()
    with _lock:
        recent = [(t, c) for t, c in _fires if now - t <= 60]
    by = collections.Counter(c for _, c in recent)
    return {"perMinute": per_min, "perThreadPerMinute": per_thread, "firesLastMinute": len(recent),
            "byThreadLastMinute": dict(by), "cooldown": cooldown_state()}


def read_recent(minutes: int = 60, limit: int = 40, verb: str = "", caller: str = "") -> dict:
    cutoff = time.time() - minutes * 60
    rows = []
    for path in (LOG_PATH + ".1", LOG_PATH):
        try:
            with open(path, encoding="utf-8") as f:
                for line in f:
                    try:
                        e = json.loads(line)
                        t = datetime.fromisoformat(e["ts"]).timestamp()
                    except Exception:
                        continue
                    if t < cutoff:
                        continue
                    if verb and e.get("verb") != verb:
                        continue
                    if caller and e.get("caller") != caller:
                        continue
                    rows.append(e)
        except FileNotFoundError:
            continue
    counts = collections.Counter((e.get("caller"), e.get("verb")) for e in rows)
    fires = [e for e in rows if e.get("verb") == "grok_webhook_fire"]
    return {
        "windowMinutes": minutes,
        "calls": len(rows),
        "byCallerVerb": [{"caller": c, "verb": v, "count": n} for (c, v), n in counts.most_common(30)],
        "webhook": {
            "fires": len(fires),
            "sent": sum(1 for e in fires if e.get("httpStatus") == 200),
            "throttledByGrok": sum(1 for e in fires if e.get("httpStatus") == 429
                                   or e.get("throttledBy") == "grok"),
            "refusedByBudget": sum(1 for e in fires if e.get("throttledBy") in ("thread-budget", "pc-budget")),
            "byCaller": dict(collections.Counter(e.get("caller") for e in fires)),
        },
        "recent": rows[-limit:],
    }