Grok Bot

"""Grok Bridge webhook bind / status / fire / ensure.

Stores {url, key, header, boundAt} in ~/.adom/grok-webhook.json
(Windows: %USERPROFILE%\\.adom\\). The HTTP server runs ON the desktop, so this
is local FS — never pull_file.

Never log the key. Never put the key in _hint, logs, wiki, or git.
Auth on fire is Authorization: Bearer <key> — a JSON body field named key is WRONG.

grok_webhook_ensure reads UIA Edit controls on the Grok Bot routine pane
(names: "Webhook URL", "Webhook key", "Authorization header"). No OCR.
"""

from __future__ import annotations

import json
import secrets
import socket
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse

from handlers.windows import _cb, _callback_ok, grok_hwnd

WEBHOOK_FILENAME = "grok-webhook.json"
SOURCE = "grok-bridge"
DEFAULT_HEADER = "Authorization: Bearer"

# Proven UIA names on the Grok Bot routine pane. Do not invent others.
OPEN_ROUTINE_BUTTON = "Open routine Grok Bridge webhook"
TRIGGER_NAME = "When a webhook fires"
EDIT_URL = "Webhook URL"
EDIT_KEY = "Webhook key"
EDIT_AUTH = "Authorization header"
PROMPT_NAME = "Prompt"


def webhook_path() -> Path:
    return Path.home() / ".adom" / WEBHOOK_FILENAME


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def load_webhook() -> dict:
    """Return {url, key, header, boundAt} internally. Missing file → {}."""
    path = webhook_path()
    try:
        raw = path.read_text(encoding="utf-8")
    except OSError:
        return {}
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return {}
    if not isinstance(data, dict):
        return {}
    url = str(data.get("url") or "").strip()
    key = str(data.get("key") or "").strip()
    header = str(data.get("header") or "").strip() or DEFAULT_HEADER
    bound_at = str(data.get("boundAt") or "")
    out = {}
    if url:
        out["url"] = url
    if key:
        out["key"] = key
    if header:
        out["header"] = header
    if bound_at:
        out["boundAt"] = bound_at
    return out


def url_host(url: str) -> str | None:
    try:
        host = urlparse(url).hostname
        return host or None
    except Exception:
        return None


def _public_bind(cfg: dict) -> dict:
    url = cfg.get("url") or ""
    return {
        "bound": bool(url),
        "urlHost": url_host(url) if url else None,
        "path": str(webhook_path()),
        "boundAt": cfg.get("boundAt"),
        "authHeader": "Authorization",
    }


def write_webhook(url: str, key: str, header: str = DEFAULT_HEADER) -> dict:
    """Write the bind file. Returns public bind dict or error dict. Never returns the key."""
    path = webhook_path()
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        payload = {
            "url": url,
            "key": key,
            "header": header or DEFAULT_HEADER,
            "boundAt": _now_iso(),
        }
        tmp = path.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
        tmp.replace(path)
    except OSError as e:
        return {
            "success": False,
            "errorCode": "bind_write_failed",
            "bound": False,
            "error": f"Could not write {path}: {e}",
            "_hint": f"Failed to write {path}. Check that ~/.adom is writable. The sender key was not logged.",
            "_next": "Retry grok_webhook_bind or grok_webhook_ensure",
        }
    pub = _public_bind(payload)
    return {
        "success": True,
        "bound": True,
        **pub,
    }


def handle_webhook_status(_args: dict) -> dict:
    cfg = load_webhook()
    pub = _public_bind(cfg)
    return {
        "success": True,
        **pub,
        "_hint": (
            f"Webhook bound to host {pub['urlHost']} (key not shown). Fire with grok_webhook_fire / grok_bot_list."
            if pub["bound"] else
            "No webhook bound. Prefer grok_webhook_ensure {} (reads the routine pane via UIA). "
            "Or grok_webhook_bind {url, key} with values from the Grok Bridge webhook routine panel."
        ),
    }


def handle_webhook_bind(args: dict) -> dict:
    url = str(args.get("url") or "").strip()
    key = str(args.get("key") or args.get("senderKey") or "").strip()
    header = str(args.get("header") or DEFAULT_HEADER).strip() or DEFAULT_HEADER
    if not url:
        return {
            "success": False,
            "errorCode": "url_required",
            "bound": False,
            "_hint": (
                "grok_webhook_bind needs {url, key}. Prefer grok_webhook_ensure {} to read them "
                "from the routine pane. Or copy both from the panel named \"Grok Bridge webhook\"."
            ),
            "_next": "grok_webhook_ensure {} or grok_webhook_bind {\"url\":\"https://...\",\"key\":\"...\"}",
        }
    parsed = urlparse(url)
    if parsed.scheme != "https" or not parsed.netloc:
        return {
            "success": False,
            "errorCode": "url_must_be_https",
            "bound": False,
            "_hint": "url must be https (https host required).",
            "_next": "grok_webhook_ensure {} or grok_webhook_bind {\"url\":\"https://...\",\"key\":\"...\"}",
        }
    if not key:
        return {
            "success": False,
            "errorCode": "key_required",
            "bound": False,
            "_hint": (
                "key is required. Prefer grok_webhook_ensure {}. The key is stored locally "
                "and is never logged or returned."
            ),
            "_next": "grok_webhook_ensure {}",
        }

    written = write_webhook(url, key, header)
    if not written.get("success"):
        return written
    written["_hint"] = (
        f"Bound Grok Bridge webhook to host {written.get('urlHost')} in {written.get('path')}. "
        "The sender key is stored locally and is never logged or returned. "
        "POSTs use Authorization: Bearer. Next: grok_bot_create / grok_channel_create / grok_bot_list."
    )
    return written


def webhook_not_bound_error(verb: str) -> dict:
    return {
        "success": False,
        "errorCode": "webhook_not_bound",
        "bound": False,
        "error": "Grok Bridge webhook is not bound.",
        "_hint": (
            f"{verb} needs a bound webhook. Call grok_webhook_ensure {{}} to read the "
            "Grok Bridge webhook routine pane (UIA Edit names Webhook URL / Webhook key), "
            "or grok_webhook_bind {url, key} from that panel. The binding is per PC "
            "(%USERPROFILE%\\.adom\\grok-webhook.json): a PC that just got Grok Bot via grok_install "
            "has none even when another PC on the same account is bound. Never print the key."
        ),
        "_next": "grok_webhook_ensure {}",
    }



def _normalize_payload_agent(args: dict, payload):
    """Copy agent (name or id) from args or payload into payload.agent.

    The Grok Bridge webhook routine is the only mailbox. payload.agent tells
    grok-bridge to relay into that teammate's thread and paste the reply back.
    """
    pl = dict(payload) if isinstance(payload, dict) else {}
    raw = args.get("agent")
    if raw is None:
        raw = pl.get("agent") or pl.get("agentId") or pl.get("agent_id")
    if raw is None:
        return pl, None
    s = str(raw).strip()
    if not s:
        return pl, None
    pl["agent"] = s
    return pl, s



_LAST_FIRE: dict = {}


def _extract_run_uuid(parsed, raw: str | None = None) -> str | None:
    if isinstance(parsed, str):
        s = parsed.strip().strip('"').strip("'")
        if 8 <= len(s) <= 80 and " " not in s and "/" not in s:
            return s
    if isinstance(parsed, list) and parsed:
        got = _extract_run_uuid(parsed[0], raw)
        if got:
            return got
    if not isinstance(parsed, dict):
        if raw:
            for marker in ("runUuid", "run_uuid", "runId"):
                key = f'"{marker}"'
                i = raw.find(key)
                if i < 0:
                    i = raw.find(marker)
                if i < 0:
                    continue
                chunk = raw[i:i + 120]
                for sep in (":", "="):
                    if sep in chunk:
                        rest = chunk.split(sep, 1)[1].strip().strip(",} ").strip('"').strip("'")
                        tok = rest.split()[0].strip(",} ") if rest else ""
                        tok = tok.strip('"').strip("'")
                        if 8 <= len(tok) <= 80:
                            return tok
        return None
    for k in ("runUuid", "run_uuid", "runId", "run_id", "uuid", "id"):
        v = parsed.get(k)
        if v is None or v is False:
            continue
        s = str(v).strip()
        if s and s.lower() not in ("none", "null"):
            # skip obvious non-uuids like event names for generic "id"
            if k == "id" and len(s) < 8:
                continue
            return s
    for nest in ("data", "result", "response", "run"):
        inner = parsed.get(nest)
        if isinstance(inner, dict):
            got = _extract_run_uuid(inner, raw)
            if got:
                return got
    if raw:
        got = _extract_run_uuid(None, raw)
        if got:
            return got
    return None


def handle_run_status(args: dict) -> dict:
    """Cheap last-fire echo. Optional GET of bound url is not required for runUuid surfacing."""
    want = str(args.get("runUuid") or args.get("run_uuid") or "").strip() or None
    last = dict(_LAST_FIRE) if _LAST_FIRE else {}
    run = want or last.get("runUuid")
    out = {
        "success": True if (run or last) else False,
        "runUuid": run,
        "event": last.get("event"),
        "httpStatus": last.get("httpStatus"),
        "firedAt": last.get("firedAt"),
        "urlHost": last.get("urlHost"),
    }
    if not run and not last:
        out["errorCode"] = "no_run"
        out["_hint"] = (
            "No runUuid yet. Fire with grok_webhook_fire {event} first; HTTP 200 now "
            "surfaces runUuid at the top level (and in response)."
        )
        out["_next"] = "grok_webhook_fire {\"event\":\"ping\"}"
        return out
    if want and last.get("runUuid") and want != last.get("runUuid"):
        out["success"] = True
        out["matchedLastFire"] = False
        out["_hint"] = (
            f"Requested runUuid={want}. Last fire in this process was "
            f"{last.get('runUuid')} event={last.get('event')}. "
            "There is no public run-status HTTP API; poll grok_chat_read for the nonce."
        )
        return out
    out["matchedLastFire"] = bool(last.get("runUuid") and run == last.get("runUuid"))
    out["_hint"] = (
        f"Last grok_webhook_fire in this process: runUuid={run} event={last.get('event')} "
        f"HTTP {last.get('httpStatus')}. Poll grok_chat_read for payload.nonce — "
        "the HTTP 200 is only the run id, not the answer."
    )
    return out


APPROVAL_WAIT_DEFAULT = 180
APPROVAL_WAIT_MIN = 30
APPROVAL_WAIT_MAX = 270
APPROVE_LABEL = "Allow"
DENY_LABEL = "Deny"


def _ask_user_approval(event: str, args: dict, agent, hostname) -> dict:
    """Raise a buttoned Adom Bridge toast on the user's machine and wait for the click.

    Returns {approved:True, stamp} or {approved:False, refusal}. The click is the human's;
    the caller's own claims of approval are never trusted (stripped in handle_webhook_fire).
    """
    import caller_identity
    summary = str(args.get("approvalSummary") or "").strip()
    thread = caller_identity.thread_name() or "an unnamed AI thread"
    if len(summary) < 10:
        return {"approved": False, "refusal": {
            "success": False, "fired": False, "errorCode": "approval_summary_required",
            "_hint": "requireApproval needs approvalSummary: one plain sentence (>= 10 chars) saying "
                     "what Grok Bot's agent will do, shown to the user on the Allow / Deny prompt.",
        }}
    try:
        wait = int(args.get("approvalTimeoutSeconds") or APPROVAL_WAIT_DEFAULT)
    except (TypeError, ValueError):
        wait = APPROVAL_WAIT_DEFAULT
    wait = max(APPROVAL_WAIT_MIN, min(APPROVAL_WAIT_MAX, wait))
    body = (f'AI thread "{thread}" wants Grok Bot\'s agent'
            + (f" ({agent})" if agent else "") + f" to: {summary}")
    res = _cb("notify_user", {
        "title": "Allow this Grok Bot request?",
        "body": body[:400],
        "scenario": "reminder",
        "buttons": [{"label": APPROVE_LABEL}, {"label": DENY_LABEL}],
        "wait": True,
        "timeoutSec": wait,
    }, timeout=wait + 20)
    from handlers.windows import flatten_callback
    merged = flatten_callback(res)
    merged = merged if isinstance(merged, dict) else {}
    action = str(merged.get("action") or (res or {}).get("action") or "").strip()
    if action != APPROVE_LABEL:
        why = ("the user pressed Deny" if action == DENY_LABEL else
               "no answer before the prompt timed out" if not action else f"the user chose {action!r}")
        return {"approved": False, "refusal": {
            "success": False, "fired": False, "approved": False,
            "errorCode": "approval_denied" if action == DENY_LABEL else "approval_not_given",
            "approvalAction": action or None,
            "_hint": f"Not fired: {why}. Do not retry the same request without asking the user in chat first.",
        }}
    now = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
    return {"approved": True, "stamp": {
        "approvedBy": "user",
        "approvedVia": "Adom Bridge Allow/Deny prompt on " + (hostname or "the user's PC"),
        "approvedAt": now,
        "approvalSummary": summary,
        "requestedByThread": thread,
    }}


def handle_webhook_fire(args: dict) -> dict:

    event = str(args.get("event") or "").strip()
    if not event:
        return {
            "success": False,
            "errorCode": "event_required",
            "error": "grok_webhook_fire needs {event}.",
            "_hint": (
                "Pass event (and optional payload). Optional agent (name or id) rides in "
                "payload.agent so grok-bridge relays into that teammate and pastes the reply "
                "back. Bound events: ping, create_bot, create_channel, list_bots."
            ),
            "_next": "grok_webhook_fire {\"event\":\"list_bots\"}",
        }
    cfg = load_webhook()
    url = cfg.get("url") or ""
    if not url:
        return webhook_not_bound_error("grok_webhook_fire")

    payload = args.get("payload")
    if payload is None:
        payload = {}
    payload, agent = _normalize_payload_agent(args, payload)
    # Top-level args.nonce / args.message merge into payload. payload wins on conflict.
    if not isinstance(payload, dict):
        payload = {}
    else:
        payload = dict(payload)
    for field in ("nonce", "message"):
        if field in payload and str(payload.get(field) or "").strip():
            continue
        raw = args.get(field)
        if raw is not None and str(raw).strip():
            payload[field] = str(raw).strip()
    try:
        hostname = socket.gethostname()
    except Exception:
        hostname = None

    # Wiki #5 item 6: a caller can never stamp approval itself; only a human click below can.
    for k in ("approval", "approvedBy", "approvedAt"):
        payload.pop(k, None)
    approval = None
    if args.get("requireApproval"):
        approval = _ask_user_approval(event, args, agent, hostname)
        if not approval.get("approved"):
            return approval["refusal"]
        payload["approval"] = approval["stamp"]

    # Governor (1.3.13): Grok's webhook is ONE mailbox shared by every AI thread. Refuse
    # locally while Grok's throttle cools down, and cap each thread so others still get through.
    import caller_identity
    from handlers import activity
    refusal = activity.admit(caller_identity.thread_name())
    if refusal:
        return {
            "success": False, "fired": False, "event": event,
            "errorCode": "webhook_throttled",
            "throttledBy": refusal["throttledBy"],
            "retryAfterSeconds": refusal["retryAfterSeconds"],
            "_hint": ("Not sent: " + refusal["why"] + " Every AI thread shares this one Grok mailbox. "
                      f"Wait at least {refusal['retryAfterSeconds']} s; do not loop or retry sooner, and "
                      "do not fire again while waiting for a reply you already asked for (poll "
                      "grok_chat_read for its nonce instead). grok_activity shows who is using it."),
            "_next": "grok_activity {\"minutes\":10}",
        }

    body = {
        "event": event,
        "payload": payload,
        "source": SOURCE,
    }
    nonce = str(payload.get("nonce") or "").strip()
    if nonce:
        body["nonce"] = nonce
    if agent:
        body["agent"] = agent
    if hostname:
        body["hostname"] = hostname
        body["desktop"] = hostname

    data = json.dumps(body).encode("utf-8")
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "User-Agent": f"{SOURCE}/1.2.4",
    }
    key = cfg.get("key") or ""
    if key:
        # Proven card: Authorization: Bearer <key>. Never a JSON field named key.
        headers["Authorization"] = f"Bearer {key}"

    status = None
    resp_text = ""
    try:
        req = urllib.request.Request(url, data=data, headers=headers, method="POST")
        with urllib.request.urlopen(req, timeout=30) as resp:
            status = getattr(resp, "status", None) or resp.getcode()
            resp_text = (resp.read() or b"").decode("utf-8", errors="replace")
    except urllib.error.HTTPError as e:
        status = e.code
        try:
            resp_text = (e.read() or b"").decode("utf-8", errors="replace")
        except Exception:
            resp_text = str(e)
        low = resp_text.lower()
        if status == 429 or "rate limit" in low or "too many requests" in low or "throttl" in low:
            from handlers import activity
            import re as _re
            wait = activity.note_throttled(status, e.headers.get("Retry-After") if e.headers else None)
            m_lim = _re.search(r"limit\D{0,20}?(\d+)\s*per\s*minute", low)
            m_cnt = _re.search(r"current_count\D{0,3}(\d+)", low)
            out = {
                "success": False, "fired": False, "event": event,
                "errorCode": "webhook_throttled", "throttledBy": "grok",
                "httpStatus": status, "retryAfterSeconds": wait,
                "urlHost": url_host(url), "bodyPreview": resp_text[:300],
                "_hint": (f"Grok refused the webhook for its rate limit (HTTP {status}). Wait or stop; do NOT "
                          f"re-run grok_webhook_ensure (the URL and key are fine). This PC holds every fire for "
                          f"{wait} s so the limit is not pushed further; all AI threads share this mailbox. "
                          "On 2026-10-15 Grok's automation counter sat at its cap for over an hour while no "
                          "grok-bridge PC was firing, so the load can be outside grok-bridge. Urgent and cannot "
                          "wait: grok_chat_send {message, reason} reaches the agent through the chat instead."),
                "_next": "grok_chat_send {\"message\":\"...\",\"reason\":\"...\"} if urgent; else wait retryAfterSeconds",
            }
            if m_lim:
                out["rateLimitPerMinute"] = int(m_lim.group(1))
            if m_cnt:
                out["currentCount"] = int(m_cnt.group(1))
            return out
        if "bot is unavailable" in low:
            return {
                "success": False,
                "errorCode": "bot_unavailable",
                "event": event,
                "httpStatus": status,
                "urlHost": url_host(url),
                "bodyPreview": resp_text[:500],
                "_hint": (
                    "The webhook is reachable and the key is accepted, but Grok says the bot that owns the "
                    "Grok Bridge webhook routine is unavailable. Usually its cloud computer is not up yet "
                    "(the app shows 'Couldn't load Bots' or 'having trouble connecting'; wait 10 to 20 min), or "
                    "the routine belongs to a bot that was removed, or this PC's binding was copied from another "
                    "PC whose routine changed. Check the sidebar in grok_screenshot, then run grok_webhook_ensure "
                    "on this PC to re-read the routine. Retrying the same fire immediately will not help."
                ),
                "_next": "grok_screenshot {} then grok_webhook_ensure {}",
            }
        return {
            "success": False,
            "errorCode": "webhook_http_error",
            "event": event,
            "httpStatus": status,
            "urlHost": url_host(url),
            "bodyPreview": resp_text[:500],
            "_hint": (
                f"Webhook POST to {url_host(url)} returned HTTP {status} for event={event}. "
                "Auth was header Authorization: Bearer (never a JSON body field named key). "
                "Re-run grok_webhook_ensure if the routine URL/key rotated."
            ),
            "_next": "grok_webhook_status {} then retry, or grok_webhook_ensure {}",
        }
    except Exception as e:
        return {
            "success": False,
            "errorCode": "webhook_fire_failed",
            "event": event,
            "urlHost": url_host(url),
            "error": str(e),
            "_hint": (
                f"Could not POST event={event} to the bound webhook host {url_host(url)}. "
                "Check grok_webhook_status. The sender key was not logged."
            ),
            "_next": "grok_webhook_status {} then grok_webhook_fire {\"event\":\"...\"}",
        }

    parsed = None
    if resp_text:
        try:
            parsed = json.loads(resp_text)
        except json.JSONDecodeError:
            parsed = None

    run_uuid = _extract_run_uuid(parsed, resp_text)
    from handlers import activity
    activity.note_ok()
    out = {
        "success": True,
        "fired": True,
        "event": event,
        "httpStatus": status,
        "urlHost": url_host(url),
        "hostname": hostname,
        "runUuid": run_uuid,
    }
    if agent:
        out["agent"] = agent
    if approval is not None:
        out["approval"] = approval["stamp"]
    from handlers.chat import UNREACHABLE_HINT, probe_computer
    comp = probe_computer()
    out.update(comp)
    if comp.get("computerReachable") is False:
        # Wiki #5: HTTP 200 + runUuid even though no reply can ever come back.
        out["warning"] = "computer_unreachable"
        out["_warningHint"] = UNREACHABLE_HINT
    if isinstance(parsed, dict):
        safe = {k: v for k, v in parsed.items() if k.lower() not in ("key", "senderkey", "token", "secret")}
        out["response"] = safe
        if run_uuid is None:
            run_uuid = _extract_run_uuid(safe)
            out["runUuid"] = run_uuid
    elif resp_text:
        out["bodyPreview"] = resp_text[:500]
    _LAST_FIRE.clear()
    _LAST_FIRE.update({
        "runUuid": run_uuid,
        "event": event,
        "httpStatus": status,
        "urlHost": url_host(url),
        "firedAt": _now_iso(),
    })
    out["_hint"] = (
        f"POSTed event={event} to webhook host {url_host(url)} (HTTP {status})"
        + (f" for agent={agent!r}." if agent else ".")
        + (f" runUuid={run_uuid}." if run_uuid else " runUuid not present in HTTP body.")
        + " Auth was header Authorization: Bearer, never a body field named key. "
        "The HTTP response is only a runUuid; the answer lands in the grok-bridge "
        "thread — poll grok_chat_read for a nonce in payload. grok_run_status {runUuid} echoes the last fire."
    )
    return out



SHOT_NONCE_PREFIX = "BOX-SHOT-"
SHOT_POLL_SECONDS = 4
SHOT_TIMEOUT_DEFAULT = 180
SHOT_TIMEOUT_MIN = 30
SHOT_TIMEOUT_MAX = 300
SHOT_READ_LIMIT = 8000
SHOT_SNIPPET = 600
SHOT_MESSAGE = (
    "Screenshot grok-bridge computer. Reply in grok-bridge with this nonce and attach the PNG."
)
SHOT_HINT = (
    "This screenshots grok-bridge's computer (the agent box), NOT Grok Bot.exe. "
    "HTTP 200 has no image. This verb waits and returns when the nonce is in grok-bridge. "
    "Do not bash-poll grok_chat_read. Do not reuse a nonce. "
    "Full PNG is the in-thread attachment; shotId is the Grok Bot.exe thumbnail."
)


def _mint_shot_nonce() -> str:
    """BOX-SHOT- + 4 uppercase hex + '-' + last 4 of unix time. Never BOX-SHOT-4821."""
    hex4 = secrets.token_hex(2).upper()
    last4 = str(int(time.time()))[-4:]
    return f"{SHOT_NONCE_PREFIX}{hex4}-{last4}"


def _snippet_around(text: str, needle: str, width: int = SHOT_SNIPPET) -> str:
    if not text:
        return ""
    i = text.find(needle)
    if i < 0:
        return text[-width:]
    half = max(width - len(needle), 0) // 2
    start = max(0, i - half)
    end = min(len(text), start + width)
    if end - start < width:
        start = max(0, end - width)
    return text[start:end]


def handle_computer_screenshot(args: dict) -> dict:
    """Fire screenshot_computer, wait for nonce in grok-bridge, then thumbnail Grok Bot.exe.

    Runs on the laptop (AdomLapper) via Adom Bridge. Does not screenshot this Linux box.
    Never steals focus. Never types into chat. Never logs webhook keys.
    """
    from handlers.chat import handle_chat_read
    from handlers.windows import flatten_callback, handle_screenshot

    nonce = str((args or {}).get("nonce") or "").strip()
    if not nonce:
        nonce = _mint_shot_nonce()

    raw_timeout = (args or {}).get("timeoutSeconds")
    try:
        timeout = int(raw_timeout) if raw_timeout is not None and str(raw_timeout).strip() != "" else SHOT_TIMEOUT_DEFAULT
    except (TypeError, ValueError):
        timeout = SHOT_TIMEOUT_DEFAULT
    timeout = max(SHOT_TIMEOUT_MIN, min(SHOT_TIMEOUT_MAX, timeout))

    from handlers.chat import UNREACHABLE_HINT, probe_computer
    comp = probe_computer()
    if comp.get("computerReachable") is False:
        # Wiki #5: do not fire and then wait out the whole timeout for a reply that cannot come.
        out = {"success": False, "found": False, "fired": False, "nonce": nonce,
               "errorCode": "computer_unreachable", "_hint": UNREACHABLE_HINT,
               "_next": "grok_chat_status {} after the user presses Retry, then grok_computer_screenshot {}"}
        out.update(comp)
        return out

    fire = handle_webhook_fire({
        "event": "screenshot_computer",
        "nonce": nonce,
        "message": SHOT_MESSAGE,
        "payload": {
            "nonce": nonce,
            "message": SHOT_MESSAGE,
        },
    })
    if not fire.get("success"):
        if "nonce" not in fire:
            fire = dict(fire)
            fire["nonce"] = nonce
        return fire

    run_uuid = fire.get("runUuid")
    t0 = time.monotonic()
    polls = 0
    last_read: dict = {}
    found = False
    snippet = ""
    deadline = t0 + timeout

    while True:
        polls += 1
        last_read = handle_chat_read({"thread": "grok-bridge", "limit": SHOT_READ_LIMIT})
        text = str(last_read.get("text") or "")
        if nonce in text:
            found = True
            snippet = _snippet_around(text, nonce)
            break
        if last_read.get("errorCode") == "computer_unreachable":
            break    # the box went down mid-wait; say so now instead of at the timeout
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            snippet = _snippet_around(text, nonce)
            break
        time.sleep(min(SHOT_POLL_SECONDS, remaining))

    wait_ms = int((time.monotonic() - t0) * 1000)
    out = {
        "success": found,
        "found": found,
        "nonce": nonce,
        "runUuid": run_uuid,
        "waitMs": wait_ms,
        "polls": polls,
        "snippet": snippet,
        "thinking": last_read.get("thinking"),
        "idle": last_read.get("idle"),
        "_hint": SHOT_HINT,
    }
    if found:
        try:
            shot = handle_screenshot({})
        except Exception:
            shot = None
        if isinstance(shot, dict) and shot.get("success") is not False:
            merged = flatten_callback(shot)
            blob = merged if isinstance(merged, dict) else shot
            shot_id = (
                blob.get("shotId") or shot.get("shotId")
                or blob.get("screenshotId") or shot.get("screenshotId")
                or blob.get("id") or shot.get("id")
            )
            if shot_id and str(shot_id).lower() not in ("none", "null"):
                out["shotId"] = shot_id
        return out

    if last_read.get("errorCode") == "computer_unreachable":
        out["errorCode"] = "computer_unreachable"
        out["_hint"] = UNREACHABLE_HINT
        out.update({k: last_read.get(k) for k in ("computerState", "computerReachable", "computerBanner")})
        return out
    out["errorCode"] = "nonce_not_in_transcript"
    out["_next"] = "retry grok_computer_screenshot {} with a FRESH nonce; FIFO may be busy."
    return out


# ── grok_webhook_ensure: UIA Edit read of the routine pane (no OCR) ──────────


def _as_controls(blob) -> list[dict]:
    if isinstance(blob, list):
        return [c for c in blob if isinstance(c, dict)]
    if not isinstance(blob, dict):
        return []
    for key in ("controls", "matches", "items", "result", "nodes"):
        val = blob.get(key)
        if isinstance(val, list):
            return [c for c in val if isinstance(c, dict)]
    data = blob.get("data")
    if isinstance(data, dict):
        return _as_controls(data)
    if isinstance(data, list):
        return [c for c in data if isinstance(c, dict)]
    best = blob.get("best")
    if isinstance(best, dict):
        return [best]
    return []


def _ctrl_name(c: dict) -> str:
    return str(c.get("name") or c.get("Name") or "")


def _ctrl_value(c: dict) -> str:
    for k in ("value", "Value", "currentValue", "text"):
        v = c.get(k)
        if v is not None and str(v).strip():
            return str(v).strip()
    return ""


def _ctrl_center(c: dict):
    rect = c.get("rect") or {}
    if not isinstance(rect, dict):
        rect = {}
    cx = rect.get("centerX", c.get("centerX"))
    cy = rect.get("centerY", c.get("centerY"))
    if cx is not None and cy is not None:
        try:
            return int(cx), int(cy)
        except (TypeError, ValueError):
            return None
    x = rect.get("x", rect.get("left"))
    y = rect.get("y", rect.get("top"))
    w = rect.get("w", rect.get("width"))
    h = rect.get("h", rect.get("height"))
    try:
        if x is None or y is None or w is None or h is None:
            return None
        return int(x + w / 2), int(y + h / 2)
    except (TypeError, ValueError):
        return None


def _ui_get_value(hwnd: int, name: str) -> str:
    blob = _cb("desktop_ui_get", {"hwnd": hwnd, "name": name})
    if not isinstance(blob, dict):
        return ""
    inner = blob.get("data") if isinstance(blob.get("data"), dict) else blob
    if not isinstance(inner, dict):
        return ""
    for k in ("value", "Value", "text", "currentValue"):
        v = inner.get(k)
        if v is not None and str(v).strip():
            return str(v).strip()
    best = inner.get("best")
    if isinstance(best, dict):
        return str(best.get("value") or "").strip()
    return ""


def _read_webhook_edits(hwnd: int) -> dict:
    """Read Webhook URL / Webhook key / Authorization header UIA Edits. Never logs values."""
    url = key = header_val = ""
    blob = _cb("desktop_find_controls", {"hwnd": hwnd, "role": "Edit"})
    for c in _as_controls(blob):
        n = _ctrl_name(c)
        v = _ctrl_value(c)
        if n == EDIT_URL and v:
            url = v
        elif n == EDIT_KEY and v:
            key = v
        elif n == EDIT_AUTH and v:
            header_val = v
    if not url:
        url = _ui_get_value(hwnd, EDIT_URL)
    if not key:
        key = _ui_get_value(hwnd, EDIT_KEY)
    if not header_val:
        header_val = _ui_get_value(hwnd, EDIT_AUTH)
    if header_val and not key:
        parts = header_val.split()
        if parts and parts[-1].startswith("crsr_"):
            key = parts[-1]
    if url.startswith("https://") and key:
        return {"url": url, "key": key, "header": DEFAULT_HEADER}
    return {}


def _clear_prompt_if_needed(hwnd: int, touched: bool) -> None:
    if not touched:
        return
    _cb("desktop_ui_set", {"hwnd": hwnd, "name": PROMPT_NAME, "text": ""})


def _open_routine_pane(hwnd: int) -> bool:
    res = _cb("desktop_ui_click", {
        "hwnd": hwnd,
        "name": OPEN_ROUTINE_BUTTON,
        "role": "Button",
    })
    return _callback_ok(res)


def _expand_webhook_trigger(hwnd: int) -> str | None:
    """Prefer invokable Button; ListItem Invoke is false — click its center as fallback."""
    btn = _cb("desktop_ui_click", {
        "hwnd": hwnd,
        "name": TRIGGER_NAME,
        "role": "Button",
    })
    if _callback_ok(btn):
        return "ui_click Button"
    found = _cb("desktop_find_control", {
        "hwnd": hwnd,
        "name": TRIGGER_NAME,
        "role": "listitem",
    })
    ctrls = _as_controls(found)
    if not ctrls:
        listed = _cb("desktop_find_controls", {
            "hwnd": hwnd,
            "role": "ListItem",
            "name": TRIGGER_NAME,
        })
        ctrls = _as_controls(listed)
    for c in ctrls:
        if _ctrl_name(c) != TRIGGER_NAME:
            continue
        center = _ctrl_center(c)
        if not center:
            continue
        click = _cb("desktop_click", {
            "x": center[0],
            "y": center[1],
            "hwnd": hwnd,
            "reason": "expand webhook trigger (ListItem Invoke is false)",
        })
        if _callback_ok(click):
            return "desktop_click ListItem center"
    return None


def handle_webhook_ensure(args: dict) -> dict:
    """Drive Grok Bot UI and bind from UIA Edit 'Webhook URL' / 'Webhook key'. No OCR."""
    force = bool(args.get("force"))
    existing = load_webhook()
    if existing.get("url") and existing.get("key") and not force:
        pub = _public_bind(existing)
        return {
            "success": True,
            "bound": True,
            "alreadyBound": True,
            **pub,
            "_hint": (
                f"Webhook already bound to host {pub['urlHost']} (key not shown). "
                "Pass force:true to re-read the routine pane. Next: grok_bot_list / grok_bot_create."
            ),
        }

    hwnd, win = grok_hwnd(allow_cursor=False)
    if not hwnd:
        hwnd, win = grok_hwnd(allow_cursor=True)
    if not hwnd:
        try:
            from handlers.windows import handle_open
            handle_open({})
            time.sleep(1.5)
        except Exception:
            pass
        hwnd, win = grok_hwnd(allow_cursor=False)
        if not hwnd:
            hwnd, win = grok_hwnd(allow_cursor=True)
    if not hwnd:
        return {
            "success": False,
            "errorCode": "window_not_found",
            "bound": False,
            "_hint": (
                "No Grok Bot window. grok_open {} then grok_webhook_ensure {}. "
                "The in-chat button is named exactly \"Open routine Grok Bridge webhook\". "
                "Settings gear is not the Routines list. Never kill Cursor."
            ),
            "_next": "grok_open {} then grok_webhook_ensure {}",
        }

    steps = []
    prompt_touched = False

    fields = _read_webhook_edits(hwnd)
    if not fields:
        if _open_routine_pane(hwnd):
            steps.append("ui_click Open routine Grok Bridge webhook")
            time.sleep(1.2)
            fields = _read_webhook_edits(hwnd)
        else:
            steps.append("open-routine button not found")

    if not fields:
        how = _expand_webhook_trigger(hwnd)
        if how:
            steps.append(f"expand When a webhook fires via {how}")
            time.sleep(0.8)
            fields = _read_webhook_edits(hwnd)
        else:
            steps.append("webhook trigger not expanded")

    _clear_prompt_if_needed(hwnd, prompt_touched)

    if not fields:
        return {
            "success": False,
            "errorCode": "routine_fields_not_found",
            "bound": False,
            "hwnd": hwnd,
            "steps": steps,
            "needsHuman": True,
            "_hint": (
                "Could not read UIA Edit controls named \"Webhook URL\" and \"Webhook key\". "
                "Click the in-chat button exactly named \"Open routine Grok Bridge webhook\", "
                "expand \"When a webhook fires\" if needed, then retry grok_webhook_ensure. "
                "Settings gear is not the Routines list. Do not type into Prompt "
                "(if Prompt has leftover garbage, clear it — do not submit). "
                "Or grok_webhook_bind {url, key} from that pane."
            ),
            "_next": "Human clicks Open routine Grok Bridge webhook, then grok_webhook_ensure {}",
        }

    written = write_webhook(fields["url"], fields["key"], fields.get("header") or DEFAULT_HEADER)
    if not written.get("success"):
        written["steps"] = steps
        written["hwnd"] = hwnd
        return written
    written["hwnd"] = hwnd
    written["steps"] = steps
    written["via"] = "uia-edit"
    written["_hint"] = (
        f"Read UIA Edits Webhook URL / Webhook key from the Grok Bot routine pane "
        f"and bound host {written.get('urlHost')} (key not shown). "
        "POSTs use Authorization: Bearer. Next: grok_bot_list / grok_bot_create / grok_channel_create."
    )
    return written