Grok Bot

@@ -0,0 +1,354 @@+"""Chat verbs: grok_chat_status / grok_chat_read via UIA on the Grok Bot hwnd.++Never steal focus. Never return webhook keys. UIA first, not OCR. Never CDP 9222.+Uses grok_hwnd / _cb / _callback_ok from handlers.windows after the hwnd unwrap fix.+"""++from __future__ import annotations++import re++from grok_detect import HOST_APP, detect_host+from handlers.windows import (+    _cb,+    _callback_ok,+    _title_of,+    flatten_callback,+    grok_hwnd,+)++# Proven UIA on Grok Bot.exe hwnd (AdomLapper): Document named "Grok Bot",+# Edit "Prompt", heading / sand-conversation-heading ("New Bot"),+# Text/button "<agent> is working" while thinking.+DOCUMENT_NAME = "Grok Bot"+PROMPT_NAME = "Prompt"+HEADING_AID = "sand-conversation-heading"+VIEW_DETAILS = "View conversation details"++SIDEBAR_NOISE = {+    "bot list",+    "new bot or channel",+    "search",+    "prompt",+    "view conversation details",+    "conversation",+    "grok bot",+}++_WORKING_SUFFIXES = (" is working", " is thinking")++# Never return webhook keys / bearer tokens in transcript.+_SECRET_RE = re.compile(+    r"(?:crsr_[A-Za-z0-9_-]+)"+    r"|(?:Bearer\s+[A-Za-z0-9._\-+=/]+)"+    r"|(?:Webhook\s+key\s*[:=]\s*\S+)",+    re.IGNORECASE,+)+++def _as_controls(blob) -> list[dict]:+    merged = flatten_callback(blob)+    if isinstance(merged, list):+        return [c for c in merged if isinstance(c, dict)]+    if not isinstance(merged, dict):+        return []+    out: list[dict] = []+    for key in ("controls", "matches", "items", "result", "nodes", "windows"):+        val = merged.get(key)+        if isinstance(val, list):+            out.extend(c for c in val if isinstance(c, dict))+    data = merged.get("data")+    if isinstance(data, list):+        out.extend(c for c in data if isinstance(c, dict))+    best = merged.get("best")+    if isinstance(best, dict):+        out.append(best)+    if not out and (merged.get("name") or merged.get("hwnd") or merged.get("role")):+        out.append(merged)+    return out+++def _ctrl_name(c: dict) -> str:+    return str(c.get("name") or c.get("Name") or c.get("title") or "")+++def _ctrl_value(c: dict) -> str:+    for k in ("value", "Value", "currentValue", "text", "content"):+        v = c.get(k)+        if v is not None and str(v).strip():+            return str(v).strip()+    return ""+++def _extract_text(blob) -> str:+    merged = flatten_callback(blob)+    if isinstance(merged, str):+        return merged+    if isinstance(merged, list):+        parts = []+        for item in merged:+            if isinstance(item, str) and item.strip():+                parts.append(item)+            elif isinstance(item, dict):+                t = _extract_text(item)+                if t:+                    parts.append(t)+        return "\n".join(parts)+    if not isinstance(merged, dict):+        return ""+    for k in ("text", "value", "content", "all", "visible"):+        v = merged.get(k)+        if isinstance(v, str) and v.strip():+            return v+    data = merged.get("data")+    if isinstance(data, str) and data.strip():+        return data+    if isinstance(data, dict):+        for k in ("text", "value", "content"):+            v = data.get(k)+            if isinstance(v, str) and v.strip():+                return v+    best = merged.get("best")+    if isinstance(best, dict):+        t = _extract_text(best)+        if t:+            return t+    return ""+++def _looks_working(name: str) -> bool:+    nl = (name or "").strip().lower()+    if not nl:+        return False+    if nl.endswith(" is working") or nl.endswith(" is thinking"):+        return True+    if nl in ("working", "thinking"):+        return True+    return False+++def _strip_working(name: str) -> str:+    n = (name or "").strip()+    nl = n.lower()+    for suf in _WORKING_SUFFIXES:+        if nl.endswith(suf):+            return n[: len(n) - len(suf)].strip()+    return ""+++def _redact_secrets(text: str) -> str:+    if not text:+        return text+    return _SECRET_RE.sub("[redacted]", text)+++def _window_missing(verb: str) -> dict:+    info = detect_host()+    thinking = False+    return {+        "success": False,+        "errorCode": "window_not_found",+        "error": "No Grok Bot window.",+        "thinking": thinking,+        "idle": False,+        "agentName": None,+        "hwnd": None,+        "title": None,+        "hostApp": HOST_APP,+        "hostAppFallback": info.get("hostAppFallback"),+        "windowEtiquette": "background",+        "_hint": (+            "No Grok Bot hwnd (desktop_find_window titleContains exact \"Grok Bot\"). "+            f"Call grok_open {{}} then grok_{verb} {{}}. Never steal focus. Never kill Cursor. "+            "Do not treat Google Chrome / ai-grok-bot as the host."+        ),+        "_next": f"grok_open {{}} then grok_{verb} {{}}",+    }+++def _find_working(hwnd: int) -> tuple[bool, str]:+    """Return (thinking, working_control_name). Read-only; never focuses."""+    queries = (+        {"hwnd": hwnd, "contains": "is working"},+        {"hwnd": hwnd, "nameContains": "is working"},+        {"hwnd": hwnd, "contains": "working"},+    )+    for args in queries:+        blob = _cb("desktop_find_controls", args)+        ctrls = _as_controls(blob)+        if not ctrls:+            ctrls = _as_controls(_cb("desktop_find_control", args))+        for c in ctrls:+            n = _ctrl_name(c) or _ctrl_value(c)+            if _looks_working(n):+                return True, n+    return False, ""+++def _read_agent_name(hwnd: int, working_name: str = "") -> str:+    stripped = _strip_working(working_name)+    if stripped:+        return stripped+    queries = (+        {"hwnd": hwnd, "automationId": HEADING_AID},+        {"hwnd": hwnd, "name": HEADING_AID},+        {"hwnd": hwnd, "role": "Heading"},+        {"hwnd": hwnd, "name": VIEW_DETAILS},+    )+    for args in queries:+        blob = _cb("desktop_find_control", args)+        for c in _as_controls(blob):+            n = (_ctrl_name(c) or _ctrl_value(c)).strip()+            if n and n.lower() not in SIDEBAR_NOISE:+                return n+        blob2 = _cb("desktop_find_controls", args)+        for c in _as_controls(blob2):+            n = (_ctrl_name(c) or _ctrl_value(c)).strip()+            if n and n.lower() not in SIDEBAR_NOISE:+                return n+    return HOST_APP+++def _prompt_value(hwnd: int) -> str:+    blob = _cb("desktop_ui_get", {"hwnd": hwnd, "name": PROMPT_NAME, "role": "Edit"})+    text = _extract_text(blob)+    if text:+        return text.strip()+    blob = _cb("desktop_find_control", {"hwnd": hwnd, "name": PROMPT_NAME, "role": "Edit"})+    for c in _as_controls(blob):+        v = _ctrl_value(c)+        if v:+            return v+    return ""+++def _strip_prompt_garbage(text: str, prompt_val: str) -> str:+    """Drop the Prompt compose box if it is tacked onto the Document transcript."""+    out = text or ""+    pv = (prompt_val or "").strip()+    if pv and out.rstrip().endswith(pv):+        out = out.rstrip()[: -len(pv)].rstrip()+    # Trailing isolated "Prompt" label (compose chrome, not a message).+    lines = out.splitlines()+    while lines and lines[-1].strip().lower() in ("prompt", ""):+        if lines[-1].strip() == "":+            lines.pop()+            continue+        if lines[-1].strip().lower() == "prompt":+            lines.pop()+            continue+        break+    return "\n".join(lines).rstrip()+++def _read_document_text(hwnd: int) -> tuple[str, dict | None]:+    """desktop_ui_text {hwnd, role:Document, range:all} — proven Document 'Grok Bot'."""+    last = None+    for args in (+        {"hwnd": hwnd, "role": "Document", "range": "all"},+        {"hwnd": hwnd, "name": DOCUMENT_NAME, "role": "Document", "range": "all"},+        {"hwnd": hwnd, "name": DOCUMENT_NAME, "range": "all"},+    ):+        last = _cb("desktop_ui_text", args)+        if not _callback_ok(last):+            continue+        text = _extract_text(last)+        if text.strip():+            return text, last+    return "", last+++def handle_chat_status(_args: dict) -> dict:+    """thinking=true when UIA exposes '<agent> is working' / working chip. Never steals focus."""+    hwnd, win = grok_hwnd(allow_cursor=False)+    if not hwnd:+        hwnd, win = grok_hwnd(allow_cursor=True)+    if not hwnd:+        return _window_missing("chat_status")++    thinking, working_name = _find_working(hwnd)+    agent = _read_agent_name(hwnd, working_name)+    title = _title_of(win) or HOST_APP+    return {+        "success": True,+        "thinking": bool(thinking),+        "idle": (not thinking),+        "agentName": agent,+        "hwnd": hwnd,+        "title": title,+        "windowEtiquette": "background",+        "_hint": (+            f"Grok Bot hwnd={hwnd} agent={agent!r} "+            + ("thinking (UIA name like '<agent> is working')." if thinking else "idle (no working chip).")+            + " Read-only UIA — focus not stolen. grok_chat_read for the visible transcript."+        ),+    }+++def handle_chat_read(args: dict) -> dict:+    """Visible transcript via desktop_ui_text Document range=all. limit truncates from the END."""+    hwnd, win = grok_hwnd(allow_cursor=False)+    if not hwnd:+        hwnd, win = grok_hwnd(allow_cursor=True)+    if not hwnd:+        miss = _window_missing("chat_read")+        miss["text"] = ""+        miss["length"] = 0+        return miss++    thinking, working_name = _find_working(hwnd)+    agent = _read_agent_name(hwnd, working_name)+    raw, last = _read_document_text(hwnd)+    prompt_val = _prompt_value(hwnd)+    text = _strip_prompt_garbage(raw, prompt_val)+    text = _redact_secrets(text)++    limit = args.get("limit")+    try:+        limit_n = int(limit) if limit is not None and str(limit).strip() != "" else None+    except (TypeError, ValueError):+        limit_n = None+    if limit_n is not None and limit_n < 0:+        limit_n = None+    if limit_n:+        text = text[-limit_n:]++    ok = bool(text) or _callback_ok(last)+    if not ok and not text:+        return {+            "success": False,+            "errorCode": "transcript_unavailable",+            "agentName": agent,+            "thinking": bool(thinking),+            "idle": (not thinking),+            "text": "",+            "length": 0,+            "hwnd": hwnd,+            "title": _title_of(win),+            "windowEtiquette": "background",+            "_hint": (+                "desktop_ui_text {hwnd, role:Document, range:all} returned no text. "+                "Needs a live Grok Bot window with the chat Document (name \"Grok Bot\"). "+                "Never OCR-first. Never CDP 9222. grok_chat_status {} then retry grok_chat_read."+            ),+            "_next": "grok_chat_status {} then grok_chat_read {}",+        }++    return {+        "success": True,+        "agentName": agent,+        "thinking": bool(thinking),+        "idle": (not thinking),+        "text": text,+        "length": len(text),+        "hwnd": hwnd,+        "title": _title_of(win),+        "windowEtiquette": "background",+        "_hint": (+            f"Read UIA Document range=all on Grok Bot hwnd={hwnd} "+            f"({len(text)} chars, agent={agent!r}, "+            + ("thinking" if thinking else "idle")+            + "). Latest messages are at the end. Webhook keys redacted. "+            "Prompt compose box stripped when distinguishable. Never stole focus."+        ),+    }

@@ -6,6 +6,8 @@ HTTP callback. All driving goes through caller_identity.ad_callback.  from __future__ import annotations +import json+ import caller_identity from grok_detect import (     HOST_APP,@@ -31,36 +33,144 @@ def _callback_ok(res: dict | None) -> bool:     return True  +def _parse_output_field(blob: dict):+    """ad_callback may wrap the real payload in blob['output'] (JSON object/array)."""+    raw = blob.get("output")+    if isinstance(raw, (dict, list)):+        return raw+    if not isinstance(raw, str):+        return None+    s = raw.strip()+    if not s or s[0] not in "{[":+        return None+    try:+        parsed = json.loads(s)+    except json.JSONDecodeError:+        return None+    if isinstance(parsed, (dict, list)):+        return parsed+    return None+++def flatten_callback(blob):+    """Unwrap ad_callback wrappers: output JSON string, then data dict.++    1. json.loads(blob['output']) when it is a JSON object/array and merge+    2. unwrap blob['data'] if dict+    """+    if isinstance(blob, list):+        return blob+    if not isinstance(blob, dict):+        return blob+    merged = dict(blob)+    parsed = _parse_output_field(blob)+    if isinstance(parsed, list):+        return parsed+    if isinstance(parsed, dict):+        merged.update(parsed)+    data = merged.get("data")+    if isinstance(data, dict):+        inner = dict(merged)+        inner.update(data)+        merged = inner+    elif isinstance(data, list):+        return data+    return merged+++def _add_unique_dicts(out: list[dict], seen: set, items) -> None:+    for item in items or []:+        if not isinstance(item, dict):+            continue+        h = item.get("hwnd")+        if h is None:+            h = item.get("handle")+        pid = item.get("pid") if item.get("pid") is not None else item.get("id")+        try:+            if h is not None:+                key = ("hwnd", int(h))+            elif pid is not None:+                key = ("pid", int(pid) if str(pid).lstrip("-").isdigit() else pid)+            else:+                key = ("obj", id(item), str(item.get("title") or item.get("name") or ""))+        except (TypeError, ValueError):+            key = ("obj", id(item), str(item.get("title") or item.get("name") or ""))+        if key in seen:+            continue+        seen.add(key)+        out.append(item)++ def _as_windows(blob) -> list[dict]:+    """Collect window dicts from a desktop_list_windows / desktop_find_window blob.++    Unwraps output JSON + data dict. Collects windows/items/result lists AND+    blob['best'] when it has hwnd/title (find_window shape).+    """+    if isinstance(blob, list):+        return [w for w in blob if isinstance(w, dict)]     if not isinstance(blob, dict):-        if isinstance(blob, list):-            return [w for w in blob if isinstance(w, dict)]         return []+    parsed = _parse_output_field(blob)+    if isinstance(parsed, list):+        windows = [w for w in parsed if isinstance(w, dict)]+        best = blob.get("best")+        if isinstance(best, dict) and (best.get("hwnd") or best.get("title") or best.get("handle")):+            windows.append(best)+        return windows++    merged = flatten_callback(blob)+    if isinstance(merged, list):+        return [w for w in merged if isinstance(w, dict)]+    if not isinstance(merged, dict):+        return []++    out: list[dict] = []+    seen: set = set()     for key in ("windows", "items", "result"):-        val = blob.get(key)+        val = merged.get(key)         if isinstance(val, list):-            return [w for w in val if isinstance(w, dict)]-    data = blob.get("data")-    if isinstance(data, dict):-        return _as_windows(data)+            _add_unique_dicts(out, seen, val)+    data = merged.get("data")     if isinstance(data, list):-        return [w for w in data if isinstance(w, dict)]-    if blob.get("hwnd") or blob.get("title") or blob.get("handle"):-        return [blob]-    return []+        _add_unique_dicts(out, seen, data)+    best = merged.get("best")+    if isinstance(best, dict) and (best.get("hwnd") or best.get("title") or best.get("handle")):+        _add_unique_dicts(out, seen, [best])+    if not out and (merged.get("hwnd") or merged.get("title") or merged.get("handle")):+        _add_unique_dicts(out, seen, [merged])+    return out   def _as_processes(blob) -> list[dict]:+    """Same unwrap as _as_windows: output JSON, data dict, processes/items/result, best."""+    if isinstance(blob, list):+        return [p for p in blob if isinstance(p, dict)]     if not isinstance(blob, dict):         return []+    parsed = _parse_output_field(blob)+    if isinstance(parsed, list):+        return [p for p in parsed if isinstance(p, dict)]++    merged = flatten_callback(blob)+    if isinstance(merged, list):+        return [p for p in merged if isinstance(p, dict)]+    if not isinstance(merged, dict):+        return []++    out: list[dict] = []+    seen: set = set()     for key in ("processes", "items", "result"):-        val = blob.get(key)+        val = merged.get(key)         if isinstance(val, list):-            return [p for p in val if isinstance(p, dict)]-    data = blob.get("data")-    if isinstance(data, dict):-        return _as_processes(data)-    return []+            _add_unique_dicts(out, seen, val)+    data = merged.get("data")+    if isinstance(data, list):+        _add_unique_dicts(out, seen, data)+    best = merged.get("best")+    if isinstance(best, dict) and (best.get("pid") or best.get("name") or best.get("imageName")):+        _add_unique_dicts(out, seen, [best])+    return out   def _hwnd_of(win: dict | None):@@ -81,12 +191,18 @@ def _title_of(win: dict | None) -> str:     return str(win.get("title") or win.get("name") or "")  +def _is_host_impostor(title: str) -> bool:+    """Google Chrome / ai-grok-bot are NOT the Grok Bot host."""+    tl = (title or "").lower()+    return "google chrome" in tl or "ai-grok-bot" in tl++ def _title_rank(title: str) -> int:     tl = (title or "").lower()+    if _is_host_impostor(title):+        return 9     if "grok bot" in tl:         return 0-    if "grok" in tl:-        return 1     if "cursor" in tl:         return 2     return 9@@ -106,23 +222,41 @@ def _dedupe_windows(windows: list[dict]) -> list[dict]:   def collect_grok_windows(title_contains: str | None = None) -> tuple[list[dict], bool]:-    """List Grok Bot windows. titleContains: Grok Bot first, then Grok (then Cursor if fallback)."""+    """List Grok Bot windows. Prefer desktop_find_window (best.hwnd) over list_windows.++    titleContains: exact "Grok Bot" first. Bare "Grok" is too greedy (Chrome "ai-grok-bot").+    """     info = detect_host()     needles = [title_contains] if (title_contains and str(title_contains).strip()) else window_title_needles(info)     windows: list[dict] = []     ab_up = False++    def _ingest(blob) -> None:+        nonlocal ab_up+        ab_up = ab_up or _callback_ok(blob)+        windows.extend(_as_windows(blob))++    def _hosts() -> list[dict]:+        return [w for w in windows if not _is_host_impostor(_title_of(w))]+     for needle in needles:         needle = (needle or "").strip()         if not needle:             continue-        blob = _cb("desktop_list_windows", {"titleContains": needle})-        ab_up = ab_up or _callback_ok(blob)-        windows.extend(_as_windows(blob))-        found = _cb("desktop_find_window", {"titleContains": needle})-        ab_up = ab_up or _callback_ok(found)-        windows.extend(_as_windows(found))-    windows.sort(key=lambda w: _title_rank(_title_of(w)))-    return _dedupe_windows(windows), ab_up+        _ingest(_cb("desktop_find_window", {"titleContains": needle}))+        if _hosts():+            break+    if not _hosts():+        for needle in needles:+            needle = (needle or "").strip()+            if not needle:+                continue+            _ingest(_cb("desktop_list_windows", {"titleContains": needle}))+            if _hosts():+                break+    filtered = _hosts()+    filtered.sort(key=lambda w: _title_rank(_title_of(w)))+    return _dedupe_windows(filtered), ab_up   def pick_grok_window(windows: list[dict] | None = None, *, allow_cursor: bool = True) -> dict | None:@@ -248,7 +382,8 @@ def handle_list_windows(args: dict) -> dict:         "abCallbackReached": ab_up,         "windowEtiquette": "background",         "_hint": (-            "Listed via ab desktop_list_windows (titleContains Grok Bot first, then Grok) and "+            "Listed via ab desktop_find_window (preferred; best.hwnd) then desktop_list_windows, "+            "titleContains exact Grok Bot (not bare Grok / Chrome ai-grok-bot), and "             f"process_list nameFilter {filt!r}. Read-only — no focus steal. "             + empty_hint             + fallback_hint