Grok Bot

"""grok_chat_send + grok_wake (1.3.15, wiki #8 / #9 / #10).

grok_chat_send: a second route into the grok-bridge agent when the webhook is refused (#8).
Sets the grok-bridge chat's Prompt box through UIA ValuePattern and invokes its "Send message"
button: programmatic, no keyboard, no focus steal (measured on AdomLapper 2026-10-15). The
message lands in the user's own chat, so it shows as the user. To keep the agent's approval
rule honest, every message carries a fixed prefix naming the AI thread; it cannot be turned off.
It refuses when the Prompt box already holds text (the user's draft is never overwritten).

grok_wake: Grok's agent computer pauses a few minutes after an agent run ends and resumes when
the next run starts (#9). There is no Grok Bot setting to keep it up (Settings > General /
Computer / Usage checked 2026-10-15). An agent run that is still working keeps it running, so
grok_wake starts a run that only sleeps for N minutes. It costs Grok usage like any agent run.
"""

from __future__ import annotations

import secrets
import time

import caller_identity
from handlers import activity
from handlers.chat import _select_sidebar_thread
from handlers.windows import _callback_ok, flatten_callback, grok_hwnd

PROMPT = {"name": "Prompt", "role": "Edit"}
SEND = {"name": "Send message", "role": "Button"}
MIN_REASON = 10
WAKE_MAX_MIN = 30


def _ui(cmd: str, args: dict, timeout: int = 10) -> dict:
    from handlers.windows import _cb
    return _cb(cmd, args, timeout=timeout)


def _prompt_text(hwnd: int) -> str | None:
    res = _ui("desktop_find_control", dict(PROMPT, hwnd=hwnd))
    m = flatten_callback(res)
    best = (m.get("best") or m.get("control")) if isinstance(m, dict) else None
    if not isinstance(best, dict):
        return None
    got = _ui("desktop_ui_get", dict(PROMPT, hwnd=hwnd))
    g = flatten_callback(got)
    if isinstance(g, dict):
        for k in ("value", "text", "Value"):
            if isinstance(g.get(k), str):
                return g[k]
    return ""


def chat_send(message: str, *, reason: str, thread: str = "grok-bridge", count_budget: bool = True) -> dict:
    who = caller_identity.thread_name() or "an unnamed AI thread"
    message = (message or "").strip()
    if not message:
        return {"success": False, "errorCode": "message_required", "_hint": "Pass message."}
    if len((reason or "").strip()) < MIN_REASON:
        return {"success": False, "errorCode": "reason_required",
                "_hint": f"grok_chat_send types into the user's own Grok Bot chat; pass reason (>= {MIN_REASON} chars)."}
    if count_budget:
        refusal = activity.admit(who)
        if refusal and refusal["throttledBy"] != "grok":   # Grok's webhook throttle does not apply to chat
            return {"success": False, "sent": False, "errorCode": "webhook_throttled",
                    "throttledBy": refusal["throttledBy"], "retryAfterSeconds": refusal["retryAfterSeconds"],
                    "_hint": "Not sent: " + refusal["why"] + " The chat and the webhook share one budget."}
    hwnd, _win = grok_hwnd(allow_cursor=False)
    if not hwnd:
        return {"success": False, "errorCode": "window_not_found",
                "_hint": "Grok Bot's main window is not visible. grok_open {} first (a hidden window cannot be typed into)."}
    _select_sidebar_thread(hwnd, thread)
    current = _prompt_text(hwnd)
    if current is None:
        return {"success": False, "errorCode": "prompt_not_found",
                "_hint": "The Prompt box is not exposed to UIA. If grok_chat_status says launchedWithAccessibility:false, "
                         "grok_open {\"restart\":true,...} first."}
    if current.strip():
        return {"success": False, "errorCode": "prompt_busy",
                "_hint": "The Prompt box already holds text (likely the user's draft). Not overwritten, not sent."}
    text = f'[from AI thread "{who}" via grok-bridge, not typed by John] {message}'
    st = _ui("desktop_ui_set", dict(PROMPT, hwnd=hwnd, text=text))
    if not _callback_ok(st) or (isinstance(st, dict) and st.get("error")):
        return {"success": False, "errorCode": "prompt_set_failed", "callback": st}
    time.sleep(0.6)
    clicked = _ui("desktop_ui_click", dict(SEND, hwnd=hwnd))
    ok = _callback_ok(clicked) and not (isinstance(clicked, dict) and clicked.get("error"))
    after = _prompt_text(hwnd)
    sent = ok and (after is not None and not after.strip())
    if not sent:
        _ui("desktop_ui_set", dict(PROMPT, hwnd=hwnd, text=""))   # never leave our text in the user's box
    return {
        "success": sent, "sent": sent, "via": "chat", "thread": thread, "hwnd": hwnd,
        **({} if sent else {"errorCode": "send_not_confirmed", "clickCallback": clicked}),
        "_hint": ("Typed into the user's grok-bridge chat through UIA (no focus steal) and pressed Send message. "
                  "It shows as the user's message with the 'from AI thread' prefix, so the agent treats it as an "
                  "AI request, not the user's approval. Read the reply with grok_chat_read."
                  if sent else "Send was not confirmed; the Prompt box was cleared again."),
    }


def handle_chat_send(args: dict) -> dict:
    return chat_send(str(args.get("message") or ""), reason=str(args.get("reason") or ""),
                     thread=str(args.get("thread") or "grok-bridge"))


def handle_wake(args: dict) -> dict:
    """Start an agent run that only sleeps, so the agent computer stays up for N minutes."""
    from handlers.webhook import handle_webhook_fire
    reason = str(args.get("reason") or "").strip()
    if len(reason) < MIN_REASON:
        return {"success": False, "errorCode": "reason_required",
                "_hint": f"grok_wake starts a Grok agent run (it costs Grok usage); pass reason (>= {MIN_REASON} chars)."}
    try:
        minutes = int(args.get("minutes") or 10)
    except (TypeError, ValueError):
        minutes = 10
    minutes = max(1, min(WAKE_MAX_MIN, minutes))
    nonce = "WAKE-" + secrets.token_hex(4)
    msg = (f"Keep your computer awake for {minutes} minutes for {caller_identity.thread_name() or 'an AI thread'} "
           f"({reason}). Run exactly `sleep {minutes * 60}` in your terminal and nothing else, then reply "
           f"with just: {nonce} awake-done.")
    fire = handle_webhook_fire({"event": "ping", "payload": {"nonce": nonce, "message": msg}})
    via = "webhook"
    if not fire.get("success") and fire.get("errorCode") in ("webhook_throttled", "webhook_http_error", "bot_unavailable"):
        fire = chat_send(f"{nonce}: {msg}", reason="wake the agent computer: " + reason)
        via = "chat"
    out = {
        "success": bool(fire.get("success")), "via": via, "nonce": nonce, "minutes": minutes,
        "awakeUntilApprox": time.strftime("%H:%M", time.localtime(time.time() + minutes * 60 + 60)),
        "result": {k: fire.get(k) for k in ("errorCode", "httpStatus", "runUuid", "throttledBy", "retryAfterSeconds")},
        "_hint": (f"Asked the agent (via {via}) to sleep {minutes} min, which keeps its computer running. It resumes "
                  "within about a minute: check `adom-bridge targets` for grok-bot-vm-155244935. It pauses again a few "
                  f"minutes after the run ends; the reply '{nonce} awake-done' marks the end. Each wake costs Grok usage."),
    }
    if not out["success"]:
        out["errorCode"] = fire.get("errorCode") or "wake_failed"
    return out