Grok Bot

#!/usr/bin/env python3
"""Adom Grok Bot Bridge Server — localhost HTTP server.

Adom Bridge (ab) spawns this on the Windows PC. There is no public Grok HTTP API;
we talk TO Grok Bot the way kicad talks to KiCad: detect the host app,
drive it via ab desktop_* callbacks. Never modifies ab-core.

Usage:
    python server.py
    python server.py --port 8769
"""

from __future__ import annotations

import json
import time
import os
import sys
import traceback
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

import caller_identity
from grok_detect import detect_host, status_chip
from handlers.app import (
    handle_close,
    handle_login,
    handle_logout,
    handle_plugin_install,
    handle_record_start,
    handle_record_status,
    handle_record_stop,
    handle_resize,
    handle_settings,
)
from handlers.bots import handle_bot_create, handle_bot_list, handle_channel_create
from handlers.describe import handle_describe
from handlers.install import handle_install, handle_install_status
from handlers.readiness import handle_prewarm, handle_readiness, handle_status, handle_uninstall
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.compose import handle_chat_send, handle_wake
from handlers.windows import handle_list_windows, handle_open, handle_screenshot

DEFAULT_PORT = 8769
SERVICE = "grok-bridge"
REPORT_ISSUES = "adom-wiki issue create adom/grok-bridge"
DEFAULT_HINT = (
    "Call grok_describe for the verb catalog; grok_readiness for host-app state. "
    "CLI: adom-bridge grok_* . Always --ai-thread. Background by default. "
    "Host app is Grok Bot.exe. "
    f"Bugs: {REPORT_ISSUES}"
)

KNOWN_VERBS = [
    "grok_readiness",
    "grok_prewarm",
    "grok_describe",
    "grok_status",
    "grok_open",
    "grok_screenshot",
    "grok_list_windows",
    "grok_chat_status",
    "grok_chat_read",
    "grok_send",
    "grok_uninstall",
    "grok_close",
    "grok_resize",
    "grok_record_start",
    "grok_record_stop",
    "grok_record_status",
    "grok_login",
    "grok_logout",
    "grok_plugin_install",
    "grok_webhook_status",
    "grok_webhook_bind",
    "grok_webhook_ensure",
    "grok_webhook_fire",
    "grok_computer_screenshot",
    "grok_install",
    "grok_install_status",
    "grok_settings",
    "grok_run_status",
    "grok_bot_create",
    "grok_channel_create",
    "grok_bot_list",
]


def _read_bridge_version() -> str:
    try:
        p = Path(__file__).resolve().parent / "BRIDGE_VERSION"
        return p.read_text(encoding="utf-8").strip()
    except OSError:
        return ""


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
    # Filter is "thread", never "caller": ab injects args["caller"] = {aiThread, ...} into every
    # relayed call, and filtering on that dict matched nothing (1.3.13 always reported 0 calls).
    thread = args.get("thread")
    out = activity.read_recent(minutes, limit, str(args.get("verb") or ""),
                               thread if isinstance(thread, str) else "")
    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 / thread; "
        "limit sets how many recent rows come back."
    )
    return out

COMMAND_HANDLERS = {
    "readiness": handle_readiness,
    "prewarm": handle_prewarm,
    "describe": handle_describe,
    "status": handle_status,
    "open": handle_open,
    "screenshot": handle_screenshot,
    "list_windows": handle_list_windows,
    "chat_status": handle_chat_status,
    "chat_read": handle_chat_read,
    "send": handle_send,
    "uninstall": handle_uninstall,
    "close": handle_close,
    "resize": handle_resize,
    "record_start": handle_record_start,
    "record_stop": handle_record_stop,
    "record_status": handle_record_status,
    "login": handle_login,
    "logout": handle_logout,
    "plugin_install": handle_plugin_install,
    "webhook_status": handle_webhook_status,
    "webhook_bind": handle_webhook_bind,
    "webhook_ensure": handle_webhook_ensure,
    "webhook_fire": handle_webhook_fire,
    "computer_screenshot": handle_computer_screenshot,
    "install": handle_install,
    "install_status": handle_install_status,
    "settings": handle_settings,
    "run_status": handle_run_status,
    "bot_create": handle_bot_create,
    "channel_create": handle_channel_create,
    "bot_list": handle_bot_list,
    "activity": handle_activity,
    "chat_send": handle_chat_send,
    "wake": handle_wake,
}


def _strip_prefix(command: str) -> str:
    """Accept grok_open, grok_list_windows, or bare open / list_windows."""
    c = (command or "").strip()
    if c.startswith("grok_"):
        return c[len("grok_"):]
    return c


def dispatch_command(command: str, args: dict) -> dict:
    if not isinstance(args, dict):
        args = {}
    internal = _strip_prefix(command)
    handler = COMMAND_HANDLERS.get(internal)
    if handler is None:
        return {
            "success": False,
            "errorCode": "unknown_command",
            "error": f"Unknown command: {command}",
            "_hint": (
                f"Unknown command {command!r}. Call grok_describe for the catalog. "
                f"Known verbs: {', '.join(KNOWN_VERBS)}."
            ),
            "_next": "grok_describe {}",
        }
    return handler(args)


def _enrich(data: dict, ok_status: bool) -> dict:
    """Every JSON body gets _hint. Failures also get _next + errorCode + _reportIssues."""
    if not isinstance(data, dict):
        data = {"success": False, "error": str(data)}
    failed = (data.get("success") is False) or (not ok_status)
    data.setdefault("_hint", DEFAULT_HINT)
    if failed:
        data.setdefault("errorCode", data.get("errorCode") or "error")
        data.setdefault("_next", "Call grok_describe {} then grok_readiness {}. Surface _hint verbatim.")
        data.setdefault("_reportIssues", REPORT_ISSUES)
    return data


class GrokBridgeHandler(BaseHTTPRequestHandler):
    """POST {command, args} + GET /status + /health. Bind loopback only (see main)."""

    def sendJSON(self, status: int, data: dict) -> None:
        body = _enrich(data if isinstance(data, dict) else {"success": False, "error": str(data)},
                       ok_status=(status < 400))
        payload = json.dumps(body).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def _status_payload(self) -> dict:
        info = detect_host()
        chip = status_chip(info)
        return {
            "status": "ok",
            "led": chip["led"],
            "summary": chip["summary"],
            "tooltip": chip["tooltip"],
            "service": SERVICE,
            "version": BRIDGE_VERSION,
            "bridgeVersion": BRIDGE_VERSION,
            "hostAppInstalled": info.get("hostAppInstalled"),
            "hostAppRunning": info.get("running"),
            "paths": info.get("paths") or [],
            "windowsHint": info.get("windowsHint"),
            "_hint": chip["tooltip"],
        }

    def do_GET(self):
        path = self.path.split("?", 1)[0]
        if path in ("/status", "/health"):
            self.sendJSON(200, self._status_payload())
        elif path in ("/commands", "/verbs"):
            names = list(KNOWN_VERBS)
            self.sendJSON(200, {
                "success": True,
                "prefix": "grok_",
                "commands": names,
                "verbs": names,
                "commandNames": names,
                "_hint": (
                    "Grok verbs are grok_* (verbPrefixes). adom-bridge commands is JSON — "
                    "grep grok_ against this list or grok_describe.verbs[].name, not unprefixed internals."
                ),
            })
        else:
            self.sendJSON(404, {
                "success": False,
                "errorCode": "not_found",
                "error": "Not found",
                "_hint": "Health is GET /status or GET /health. Verbs are POST /command (or POST /) with {command, args}.",
                "_next": "GET /status",
            })

    def do_POST(self):
        path = self.path.split("?", 1)[0]
        if path not in ("/command", "/", "/command/"):
            self.sendJSON(404, {
                "success": False,
                "errorCode": "not_found",
                "error": "Not found",
                "_hint": "POST /command or POST / with body {command, args}. GET /status for the chip.",
                "_next": "POST /command {\"command\":\"grok_describe\",\"args\":{}}",
            })
            return

        length = int(self.headers.get("Content-Length", 0) or 0)
        raw = self.rfile.read(length) if length else b""
        try:
            request = json.loads(raw.decode("utf-8") or "{}")
        except json.JSONDecodeError as e:
            self.sendJSON(400, {
                "success": False,
                "errorCode": "invalid_json",
                "error": f"Invalid JSON: {e}",
                "_hint": "Body must be {\"command\":\"grok_describe\",\"args\":{}}.",
            })
            return

        if not isinstance(request, dict):
            self.sendJSON(400, {
                "success": False,
                "errorCode": "invalid_json",
                "error": "JSON body must be an object.",
            })
            return

        command = request.get("command") or ""
        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"{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()
            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"{activity.now_iso()} [grok-bridge] {args[0]} {args[1]} {args[2]}", flush=True)


def main() -> None:
    try:
        sys.stdout.reconfigure(line_buffering=True)
        sys.stderr.reconfigure(line_buffering=True)
    except Exception:
        pass

    port = DEFAULT_PORT
    if "--port" in sys.argv:
        idx = sys.argv.index("--port")
        if idx + 1 < len(sys.argv):
            port = int(sys.argv[idx + 1])

    # Bind loopback ONLY. ADOM_BIND_HOST is always 127.0.0.1 when ab spawns us
    # (Bridge SDK Runtime contract, v1.9.63+) — never 0.0.0.0, no firewall prompt.
    bind_host = os.environ.get("ADOM_BIND_HOST", "127.0.0.1")
    if bind_host not in ("127.0.0.1", "::1", "localhost"):
        print(f"[grok-bridge] refusing non-loopback ADOM_BIND_HOST={bind_host!r}; using 127.0.0.1",
              flush=True)
        bind_host = "127.0.0.1"

    class _Server(ThreadingHTTPServer):
        allow_reuse_address = False
        daemon_threads = True

    print(f"[grok-bridge] starting v{BRIDGE_VERSION} on {bind_host}:{port} (pid {os.getpid()})",
          flush=True)
    try:
        server = _Server((bind_host, port), GrokBridgeHandler)
    except OSError as e:
        print(f"[grok-bridge] Port {port} already in use ({e}). "
              "Another instance owns it — exiting cleanly (single-instance guard).",
              flush=True)
        return

    info = detect_host()
    print(f"[grok-bridge] detect hostAppInstalled={info.get('hostAppInstalled')} "
          f"running={info.get('running')} paths={info.get('paths')}",
          flush=True)
    print(f"[grok-bridge] Listening on http://{bind_host}:{port} — GET /status  POST /command",
          flush=True)

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n[grok-bridge] Shutting down.", flush=True)
        server.server_close()


if __name__ == "__main__":
    main()