Grok Bot

@@ -0,0 +1,212 @@+"""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:],+    }

@@ -13,6 +13,7 @@ Usage: from __future__ import annotations  import json+import time import os import sys import traceback@@ -41,6 +42,7 @@ from handlers.readiness import handle_prewarm, handle_readiness, handle_status, from handlers.send import handle_send from handlers.webhook import handle_computer_screenshot, handle_run_status, handle_webhook_bind, handle_webhook_ensure, handle_webhook_fire, handle_webhook_status from handlers.chat import handle_chat_read, handle_chat_status+from handlers import activity from handlers.windows import handle_list_windows, handle_open, handle_screenshot  DEFAULT_PORT = 8769@@ -98,6 +100,28 @@ def _read_bridge_version() -> str:  BRIDGE_VERSION = _read_bridge_version() ++def handle_activity(args: dict) -> dict:+    """Who called what, when, and what came back (from ~/.adom/grok-bridge-activity.jsonl)."""+    try:+        minutes = max(1, min(24 * 60, int(args.get("minutes") or 60)))+    except (TypeError, ValueError):+        minutes = 60+    try:+        limit = max(0, min(500, int(args.get("limit") or 40)))+    except (TypeError, ValueError):+        limit = 40+    out = activity.read_recent(minutes, limit, str(args.get("verb") or ""), str(args.get("caller") or ""))+    out.update({"success": True, "logPath": activity.LOG_PATH, "webhookBudget": activity.budget_state()})+    w = out["webhook"]+    out["_hint"] = (+        f"Last {minutes} min on this PC: {out['calls']} calls; webhook fires {w['fires']} "+        f"(sent {w['sent']}, throttled by Grok {w['throttledByGrok']}, refused by the local budget "+        f"{w['refusedByBudget']}). byCallerVerb names the threads. Filter with verb / caller; "+        "limit sets how many recent rows come back."+    )+    return out+ COMMAND_HANDLERS = {     "readiness": handle_readiness,     "prewarm": handle_prewarm,@@ -130,6 +154,7 @@ COMMAND_HANDLERS = {     "bot_create": handle_bot_create,     "channel_create": handle_channel_create,     "bot_list": handle_bot_list,+    "activity": handle_activity, }  @@ -267,23 +292,27 @@ class GrokBridgeHandler(BaseHTTPRequestHandler):         args = request.get("args") if isinstance(request.get("args"), dict) else {}         caller_identity.set_from_request(self.headers, args)         full = command if str(command).startswith("grok_") else f"grok_{_strip_prefix(str(command))}"-        print(f"[verb] {full}{caller_identity.log_suffix()}", flush=True)+        print(f"{activity.now_iso()} [verb] {full}{caller_identity.log_suffix()}", flush=True)+        t0 = time.monotonic()+        result = None         try:             result = dispatch_command(str(command), args)             self.sendJSON(200, result)         except Exception as e:             traceback.print_exc()-            self.sendJSON(500, {+            result = {                 "success": False,                 "errorCode": "internal_error",                 "error": f"Internal error: {e}",                 "_hint": f"Bridge bug, not your call. {REPORT_ISSUES} with the verb and args.",-            })+            }+            self.sendJSON(500, result)         finally:+            activity.log_call(full, caller_identity.thread_name(), int((time.monotonic() - t0) * 1000), result)             caller_identity.clear()      def log_message(self, fmt, *args):-        print(f"[grok-bridge] {args[0]} {args[1]} {args[2]}", flush=True)+        print(f"{activity.now_iso()} [grok-bridge] {args[0]} {args[1]} {args[2]}", flush=True)   def main() -> None: