Grok Bot

@@ -317,6 +317,66 @@ def handle_run_status(args: dict) -> dict:     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()@@ -356,6 +416,17 @@ def handle_webhook_fire(args: dict) -> dict:         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"]+     body = {         "event": event,         "payload": payload,@@ -459,6 +530,8 @@ def handle_webhook_fire(args: dict) -> dict:     }     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)

@@ -7,6 +7,7 @@ HTTP callback. All driving goes through caller_identity.ad_callback. from __future__ import annotations  import json+import os import subprocess import sys import time@@ -34,6 +35,13 @@ HOST_FLAGS = [ ] _RESTART_REASON_MIN = 10 +# Grok Bot owns small secondary top-level windows titled "Grok Bot" too (a 240x30 strip+# was captured instead of the main window once its main window was hidden: wiki #5).+# Anything smaller than this is never "the" window.+STRAY_MIN_W = 400+STRAY_MIN_H = 300+_MAIN_FILE = os.path.join(os.path.expanduser("~"), ".adom", "grok-main-window.json")+  def _cb(command: str, args: dict | None = None, timeout: int = 30) -> dict:     """Best-effort ab callback. Never raises."""@@ -287,21 +295,72 @@ def collect_grok_windows(title_contains: str | None = None) -> tuple[list[dict],     return _dedupe_windows(filtered), ab_up  +def _wh(w: dict | None) -> tuple[int, int]:+    r = (w or {}).get("rect") or {}+    try:+        return int(r.get("width") or 0), int(r.get("height") or 0)+    except (TypeError, ValueError):+        return 0, 0+++def _is_stray(w: dict | None) -> bool:+    """True for a window too small to be the main Grok Bot window. Unknown size counts+    as not stray (some ab shapes carry no rect)."""+    wd, ht = _wh(w)+    if not wd and not ht:+        return False+    return wd < STRAY_MIN_W or ht < STRAY_MIN_H+++def _remember_main(hwnd) -> None:+    try:+        os.makedirs(os.path.dirname(_MAIN_FILE), exist_ok=True)+        with open(_MAIN_FILE, "w", encoding="utf-8") as f:+            json.dump({"hwnd": int(hwnd), "at": time.time()}, f)+    except Exception:+        pass+++def remembered_main_hwnd() -> int | None:+    try:+        with open(_MAIN_FILE, encoding="utf-8") as f:+            return int(json.load(f).get("hwnd")) or None+    except Exception:+        return None+++# What the last pick saw, for verbs that want to say "main window hidden".+LAST_PICK: dict = {}++ def pick_grok_window(windows: list[dict] | None = None, *, allow_cursor: bool = True) -> dict | None:+    """Best title rank, never a stray, then the LARGEST visible window."""     if windows is None:         windows, _ = collect_grok_windows()     ranked = []+    strays = []     for w in windows or []:         rank = _title_rank(_title_of(w))         if rank == 2 and not allow_cursor:             continue         if rank >= 9:             continue-        ranked.append((rank, w))+        if _is_stray(w):+            strays.append(w)+            continue+        wd, ht = _wh(w)+        ranked.append((rank, -(wd * ht), w))+    LAST_PICK.clear()+    LAST_PICK["strays"] = [{"hwnd": _hwnd_of(w), "size": "%dx%d" % _wh(w)} for w in strays]     if not ranked:+        LAST_PICK["mainWindowHidden"] = bool(strays)         return None-    ranked.sort(key=lambda x: x[0])-    return ranked[0][1]+    ranked.sort(key=lambda x: (x[0], x[1]))+    best = ranked[0][2]+    LAST_PICK["mainWindowHidden"] = False+    if _hwnd_of(best):+        _remember_main(_hwnd_of(best))+    return best   def grok_hwnd(*, allow_cursor: bool = True) -> tuple[int | None, dict | None]:@@ -309,6 +368,27 @@ def grok_hwnd(*, allow_cursor: bool = True) -> tuple[int | None, dict | None]:     return _hwnd_of(win), win  +def main_window_state() -> dict:+    """Fields for callers when no main window is visible."""+    return {"mainWindowHidden": bool(LAST_PICK.get("mainWindowHidden")),+            "strayWindows": LAST_PICK.get("strays") or [],+            "rememberedMainHwnd": remembered_main_hwnd()}+++def show_hidden_main() -> tuple[int | None, dict | None, dict | None]:+    """Re-show the remembered main window BEHIND the user's work (show, then bottom)."""+    h = remembered_main_hwnd()+    if not h:+        return None, None, None+    res = _cb("desktop_set_window_state", {"hwnd": h, "state": "show"})+    if not _callback_ok(res) or (isinstance(res, dict) and res.get("error")):+        return None, None, res+    _cb("desktop_set_window_state", {"hwnd": h, "state": "bottom"})+    time.sleep(0.5)+    hwnd, win = grok_hwnd(allow_cursor=False)+    return hwnd, win, res++ def window_rect(win: dict | None) -> dict:     """Physical px {x,y,w,h} from a desktop_list_windows entry (ab physical px)."""     if not isinstance(win, dict):@@ -526,7 +606,9 @@ def handle_list_windows(args: dict) -> dict: def handle_screenshot(args: dict) -> dict:     title = (args.get("titleContains") or "").strip()     info = detect_host()-    needles = [title] if title else window_title_needles(info)+    # Explicit titleContains only: ab's titleContains picks the TOPMOST match, which can be a+    # 240x30 secondary strip (wiki #5). By default capture the picked main window by hwnd.+    needles = [title] if title else []     last = None     for needle in needles:         last = _cb("desktop_screenshot_window", {"titleContains": needle})@@ -557,6 +639,15 @@ def handle_screenshot(args: dict) -> dict:                 "Window was NOT raised. PNG is on the desktop; pull it with the CLI."             )             return shot+    ms = main_window_state()+    if ms["mainWindowHidden"]:+        out = {"success": False, "errorCode": "main_window_hidden",+               "error": "Grok Bot's main window is hidden; only small secondary windows are visible.",+               "_hint": ("Not capturing a stray strip. grok_open {} shows the main window again, behind "+                         "the user's other windows (no focus steal), then retry grok_screenshot."),+               "_next": "grok_open {} then grok_screenshot {}"}+        out.update(ms)+        return out     return {         "success": False,         "errorCode": "window_not_found",@@ -629,6 +720,13 @@ def handle_open(args: dict) -> dict:         time.sleep(3)      hwnd0, win0, proc0 = _host_alive(info)+    hidden_fix = None+    if proc0 and not hwnd0 and not fallback and restarted is None:+        hidden_state = main_window_state()+        h2, w2, res = show_hidden_main()+        hidden_fix = dict(hidden_state, shown=bool(h2), showCallback=res)+        if h2:+            hwnd0, win0 = h2, w2     listed = handle_list_windows(args) if (hwnd0 or proc0) else None      def _envelope(*, success, launched, already_open, opened, hwnd, win, via, callback, host_running, extra_hint=""):@@ -687,6 +785,14 @@ def handle_open(args: dict) -> dict:             ),         )         env["launchedWithAccessibility"] = flags_ok+        if hidden_fix is not None:+            env.update({k: hidden_fix[k] for k in ("mainWindowHidden", "strayWindows", "rememberedMainHwnd")})+            env["shownHiddenMain"] = hidden_fix["shown"]+            env["_hint"] = (("Grok Bot's main window was hidden; showed it again BEHIND the user's other "+                             "windows (show + bottom, no focus steal). ") if hidden_fix["shown"] else+                            ("Grok Bot is running but its main window is hidden and its hwnd is not "+                             "remembered on this PC. grok_open {\"restart\":true,\"reason\":\"...\"} "+                             "relaunches it in the background. ")) + env["_hint"]         if restarted is not None:             env["restartKill"] = restarted             env["_hint"] = "Restart asked, but Grok Bot was still running after the kill. " + env["_hint"]