javis_bot/src/jarvis/tools/builtin/control_browser.py

"""Operate the on-screen (Go-Live) Chrome like a person would.

Only meaningful in screen-share mode (``STREAM_BROWSER`` true): it drives the
on-screen Chrome via the Node CDP helper (``chrome-control.mjs``) so every
action is visible on the broadcast. Pointer moves, clicks and typing are real
xdotool input (visible cursor, char-by-char typing), never synthetic DOM
events.

This is the general browser-control tool (navigate to any site, manage
windows/tabs, click, type, scroll, screenshot). ``browseAndPlay`` remains the
YouTube-playback shortcut.
"""

from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path
from typing import Dict, Any, Optional

from ..base import Tool, ToolContext
from ..types import ToolExecutionResult
from ...debug import debug_log

# .../owner/src/jarvis/tools/builtin/control_browser.py -> parents[4] == .../owner
_REPO_ROOT = Path(__file__).resolve().parents[4]
_NODE_SCRIPT = _REPO_ROOT / "bot" / "scripts" / "stream-test" / "chrome-control.mjs"

# Actions that don't change anything (safe, fast, no human-input latency).
_READ_ACTIONS = {"status", "listTabs"}


class ControlBrowserTool(Tool):
    """Drive the on-screen Chrome: navigate, manage tabs, click, type, scroll."""

    @property
    def name(self) -> str:
        return "controlBrowser"

    @property
    def description(self) -> str:
        return (
            "Drive the on-screen Chrome that is shown on the broadcast, like a person would. "
            "Use this (NOT webSearch) whenever the user wants something done or shown IN the "
            "browser on screen: open a website or URL, search on a specific site (action "
            "'search' with site=naver/google/daum/youtube/bing), go back/forward, refresh, "
            "manage tabs (list/new/close/switch), close popups, move the mouse onto an element, "
            "click, type, scroll, or screenshot. webSearch only returns text and shows nothing "
            "on screen; this tool actually navigates the visible browser. "
            "Cursor behaviour matters: 'search' and 'type' (with a selector) move the REAL mouse "
            "cursor to the on-page box, click it, then type one character at a time — use these "
            "when the user wants to search or type on a page. 'navigate' only types the URL into "
            "the address bar (no mouse movement). 'moveMouse' moves/hovers the visible cursor "
            "onto an element (selector, or site=naver/... for that site's search box) WITHOUT "
            "clicking — use it when the user just asks to move the mouse somewhere. "
            "Only available in screen-share mode. "
            "Never claim you did any of this unless this tool returns success."
        )

    @property
    def inputSchema(self) -> Dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "action": {
                    "type": "string",
                    "enum": [
                        "status", "listTabs", "navigate", "search", "back",
                        "forward", "refresh", "newTab", "closeTab", "activateTab",
                        "closePopups", "moveMouse", "click", "type", "scroll", "pressKey",
                        "screenshot",
                    ],
                    "description": "What to do in the browser.",
                },
                "url": {"type": "string", "description": "Target URL/site for navigate/newTab (e.g. 'naver.com')."},
                "query": {"type": "string", "description": "Search text for action 'search'."},
                "site": {"type": "string", "description": "Site for action 'search' or 'moveMouse': naver, google, daum, youtube, bing. For moveMouse it targets that site's search box."},
                "index": {"type": "integer", "description": "Tab index for closeTab/activateTab (from listTabs)."},
                "selector": {"type": "string", "description": "CSS selector for click/type/moveMouse."},
                "text": {"type": "string", "description": "Text to type."},
                "key": {"type": "string", "description": "Key to press, e.g. 'Return', 'Escape'."},
                "dir": {"type": "string", "description": "Scroll direction: 'down' or 'up'."},
            },
            "required": ["action"],
        }

    def run(self, args: Optional[Dict[str, Any]], context: ToolContext) -> ToolExecutionResult:
        cfg = context.cfg
        if not getattr(cfg, "stream_browser", True):
            return ToolExecutionResult(
                success=False,
                reply_text="화면 공유 모드(STREAM_BROWSER=true)에서만 브라우저를 제어할 수 있습니다.",
            )
        if not _NODE_SCRIPT.exists():
            return ToolExecutionResult(success=False, reply_text="브라우저 제어 도구를 찾을 수 없습니다.")
        if not args or not isinstance(args, dict):
            return ToolExecutionResult(success=False, reply_text="수행할 동작(action)을 알려주세요.")

        action = str(args.get("action", "")).strip()
        if not action:
            return ToolExecutionResult(success=False, reply_text="수행할 동작(action)을 알려주세요.")

        # Pass the whole arg dict through as the command (the Node side reads the
        # fields it needs per action).
        command = json.dumps(args, ensure_ascii=False)
        context.user_print(f"🖱️ 브라우저 제어: {action}")
        debug_log(f"    🖱️ controlBrowser {command[:120]}", "tools")
        # Human-input actions need time for the visible cursor move + char typing.
        timeout = 25 if action in _READ_ACTIONS else 90
        # Split deployment: when the browser (Chrome + X + xdotool) lives on a
        # different machine, send the command to its control-server so the REAL
        # input lands on that host's screen. Otherwise run chrome-control.mjs
        # locally (all-in-one / browser-host layout).
        control_url = os.environ.get("BROWSER_CONTROL_URL", "").strip()
        try:
            if control_url:
                import urllib.request
                req = urllib.request.Request(
                    control_url.rstrip("/") + "/control",
                    data=command.encode("utf-8"),
                    headers={"Content-Type": "application/json"},
                )
                with urllib.request.urlopen(req, timeout=timeout) as resp:
                    data = json.loads((resp.read().decode("utf-8") or "").strip() or "{}")
            else:
                proc = subprocess.run(
                    ["node", str(_NODE_SCRIPT), command],
                    capture_output=True,
                    text=True,
                    timeout=timeout,
                    env={**os.environ, "CDP_PORT": os.environ.get("CDP_PORT", "9222")},
                )
                data = json.loads((proc.stdout or "").strip() or "{}")
        except Exception as e:
            return ToolExecutionResult(success=False, reply_text=f"브라우저 제어에 실패했습니다: {e}")

        if not data.get("ok"):
            return ToolExecutionResult(
                success=False, reply_text=f"브라우저 제어에 실패했습니다: {data.get('error', 'unknown')}"
            )

        # Return a factual summary of what ACTUALLY happened so the reply engine
        # describes the real outcome and doesn't confabulate.
        summary = self._summarise(action, args, data)
        if action == "status":
            # The broadcast (Go-Live) state lives in the bot runtime, surfaced
            # to the tool via the turn context — report it alongside the tabs.
            bc = getattr(context, "broadcasting", None)
            state = "켜짐" if bc else ("꺼짐" if bc is not None else "알 수 없음")
            summary = f"📡 방송: {state}\n{summary}"
        return ToolExecutionResult(success=True, reply_text=summary)

    @staticmethod
    def _summarise(action: str, args: Dict[str, Any], data: Dict[str, Any]) -> str:
        if action == "navigate":
            return f"브라우저에서 {data.get('url', args.get('url'))} 로 이동했습니다."
        if action == "search":
            base = f"{data.get('site', '')}에서 '{data.get('query', args.get('query'))}'를 검색해 화면에 띄웠습니다."
            # Flag when the real cursor path didn't run, so a silent fallback to
            # cursor-less DOM input is visible rather than reported as "human".
            if data.get("input") in ("api", "api-fallback"):
                base += " (참고: 실제 마우스 커서 이동 없이 처리됨)"
            return base
        if action in ("back", "forward", "refresh"):
            return f"브라우저: {action} 완료 ({data.get('url', '')})."
        if action in ("status", "listTabs"):
            tabs = data.get("tabs", [])
            lines = "\n".join(f"  [{t['index']}] {t.get('title') or t['url']}" for t in tabs) or "  (탭 없음)"
            return f"브라우저 탭 {len(tabs)}개:\n{lines}"
        if action == "newTab":
            return f"새 탭을 열었습니다 (index {data.get('index')})."
        if action == "closeTab":
            return f"탭 {data.get('closed')}번을 닫았습니다 (남은 탭 {data.get('remaining')}개)."
        if action == "activateTab":
            return f"탭 {data.get('active')}번으로 전환했습니다."
        if action == "closePopups":
            return f"팝업/빈 탭 {data.get('closed')}개를 닫았습니다."
        if action == "moveMouse":
            target = data.get("target") or args.get("selector") or args.get("site") or "대상"
            return f"마우스 커서를 {target} 위치로 옮겼습니다."
        if action == "screenshot":
            return f"화면을 캡처했습니다: {data.get('path')}"
        return "완료했습니다."