"""
Neko plugin: one-click install from the dashboard, then room link and controls.

Copy this folder to ~/.mugi/plugins/neko and approve it in Settings > Plugins.
Uses Colima + Mugi Homebrew (same stack as SearXNG); no Docker Desktop required.
All setup lives under ~/.mugi/neko/. Install shows live status and a progress bar; if a container or compose file is already present, "Remove Neko" appears on the same panel. After a successful install, the dashboard shows Open in browser, Copy link, Remove Neko, and Audio; tools get_neko_room_link and open_url_in_neko are available.
"""

from __future__ import annotations

import base64
import json
import os
import re
import subprocess
import threading
import time
import uuid
from collections.abc import Callable
from typing import Any
from urllib.parse import urlparse

from langchain_core.tools import tool

# Directory under ~/.mugi for compose file and state (same pattern as SearXNG under ~/.mugi/searxng)
NEKO_DIR = os.path.join(os.path.expanduser("~"), ".mugi", "neko")
NEKO_COMPOSE_PATH = os.path.join(NEKO_DIR, "docker-compose.yml")
# Chromium is default so user and agent can share the same browser; CDP allows agent control.
NEKO_IMAGE_CHROMIUM = "ghcr.io/m1k1o/neko/chromium:3.0.8"
NEKO_IMAGE_CHROMIUM_CDP = "mugi-neko-chromium-cdp:latest"  # Built at install from stock + supervisord with --remote-debugging-port
NEKO_IMAGE_FIREFOX = "ghcr.io/m1k1o/neko/firefox:latest"
NEKO_DEFAULT_PORT = 8080
NEKO_CDP_PORT = 9222
# Single dashboard stream id so repeated open_neko_dashboard refreshes one panel (Swift keys streams by stream_id).
NEKO_DASHBOARD_STREAM_ID = "neko_dashboard"
# Appended to tool results so the model does not try to "join" the room by loading the room URL in Chromium.
_NEKO_AGENT_ALREADY_IN_SESSION = (
    "Agent: CDP already attaches to the same Chromium the user sees after they open the room link and log in on their Mac—"
    "you are in that session; there is no separate connect step. "
    "Never call neko_navigate with the room URL or "
    f"http://127.0.0.1:{NEKO_DEFAULT_PORT} / localhost:{NEKO_DEFAULT_PORT} (that nests Neko inside itself). "
    "Use neko_navigate only for external https:// URLs or about:blank. "
    "If the user asks whether you can see the Neko session or what page is showing, call neko_browser_snapshot in that same turn "
    "before answering—do not rely on older chat where Neko was not installed yet. "
    "To summarize or quote what is on screen, use neko_page_visible_text (CDP), not read_webpage on the tab URL (separate fetch; can differ). "
    "Interact with pages using neko_browser_snapshot, neko_page_visible_text, neko_click_selector, neko_fill_field, neko_insert_text, neko_press_key."
)
# WebRTC: use UDP+TCP mux on one port. Large EPR ranges (52000-52100/udp) often fail to forward on
# Docker Desktop / Colima macOS; mux matches Neko docs and is much more reliable locally.
_NEKO_WEBRTC_MUX_PORT = 59000

# Only treat docker ps -aq lines as IDs (hex); ignore blank lines or any stray stdout.
_DOCKER_CONTAINER_ID_RE = re.compile(r"^[a-fA-F0-9]{12,64}$")
_HOST_IPV4_RE = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}$")
# Navigating the in-container browser to the Neko HTTP port on loopback (or host.docker.internal)
# loads the Neko UI inside Chromium → nested "Neko in Neko" / login recursion.
_LOOPBACK_NEKO_NAV_HOSTS = frozenset({"localhost", "127.0.0.1", "0.0.0.0", "::1"})
# compose `container_name` — same Docker network DNS can point Chromium back at Neko's HTTP port.
_NEKO_CONTAINER_DNS_NAMES = frozenset({"mugi-neko"})
# Host port in neko_base_url can differ (e.g. 9999:8080 publish), but inside the container the Neko web UI
# is still on NEKO_DEFAULT_PORT — loopback:8080 in Chromium always risks nested Neko.
_SCHEMELESS_HOST_PORT_RE = re.compile(
    r"^([\w.-]+|\d{1,3}(?:\.\d{1,3}){3}):\d+(?:/|$)",
    re.I,
)


def _with_http_scheme_if_missing(url: str) -> str:
    u = (url or "").strip()
    if not u or "://" in u:
        return u
    if _SCHEMELESS_HOST_PORT_RE.match(u):
        return "http://" + u
    return u


def _http_url_host_port(url: str) -> tuple[str | None, int | None]:
    """Return (host lowercased, port) for http(s) URLs, else (None, None). Port is explicit or scheme default."""
    try:
        p = urlparse(_with_http_scheme_if_missing(url))
    except Exception:
        return None, None
    if p.scheme not in ("http", "https"):
        return None, None
    host = p.hostname
    if not host:
        return None, None
    port = p.port
    if port is None:
        port = 443 if p.scheme == "https" else 80
    return host.lower(), port


def neko_navigate_would_self_embed(nav_url: str, neko_base_url: str) -> bool:
    """True if loading nav_url in the shared browser would hit the Neko room HTTP service (recursive UI)."""
    nh, np = _http_url_host_port(nav_url)
    if nh is None or np is None:
        return False
    # Inside the shared browser, loopback is the container; Neko's HTTP UI listens on NEKO_DEFAULT_PORT.
    if nh in _LOOPBACK_NEKO_NAV_HOSTS and np == NEKO_DEFAULT_PORT:
        return True
    if nh in _NEKO_CONTAINER_DNS_NAMES and np == NEKO_DEFAULT_PORT:
        return True
    bh, bp = _http_url_host_port(neko_base_url or "")
    # From the container, host.docker.internal reaches the Mac; :8080 or the published room port can still be Neko.
    if nh == "host.docker.internal":
        if np == NEKO_DEFAULT_PORT:
            return True
        if bh is not None and bp is not None and np == bp:
            return True
    if bh is not None and bp is not None and nh == bh and np == bp:
        return True
    return False


# Rough install progress for the plugin dashboard (maps _do_install phases → 0–100).
_NEKO_INSTALL_PHASE_PCT: dict[str, int] = {
    "start": 5,
    "colima": 12,
    "docker_wait": 22,
    "cdp_files": 28,
    "pull_chromium": 40,
    "build_cdp": 58,
    "compose": 68,
    "pull_firefox": 72,
    "up": 88,
    "done": 100,
}


def _neko_install_stream_payload(phase: str, message: str) -> dict:
    pl: dict = {"status": "installing", "message": message, "metrics": {}}
    pct = _NEKO_INSTALL_PHASE_PCT.get(phase)
    if pct is not None:
        pl["progress_pct"] = float(pct)
    return pl


# Embedded Dockerfile and supervisord for CDP-enabled Chromium (so install works even if plugin dir lacks the files)
_DOCKERFILE_CDP = """# CDP-enabled Chromium: stock Neko image + socat to expose the debug port on 0.0.0.0
FROM ghcr.io/m1k1o/neko/chromium:3.0.8
RUN apt-get update && apt-get install -y --no-install-recommends socat python3 && rm -rf /var/lib/apt/lists/* \\
    && mkdir -p /usr/local/lib/mugi
COPY supervisord-chromium-cdp.conf /etc/neko/supervisord/chromium.conf
COPY mugi_cdp_container_runner.py /usr/local/lib/mugi/mugi_cdp_container_runner.py
"""
_SUPERVISORD_CHROMIUM_CDP = """# Chromium with remote debugging (CDP) for agent control. Based on m1k1o/neko apps/chromium.
# Chromium in headed mode ignores --remote-debugging-address; it always binds 127.0.0.1.
# socat_cdp forwards 0.0.0.0:9222 -> 127.0.0.1:9223 so the host can reach CDP.
[program:chromium]
environment=HOME="/home/%(ENV_USER)s",USER="%(ENV_USER)s",DISPLAY="%(ENV_DISPLAY)s"
command=/usr/bin/chromium
 --remote-debugging-port=9223
 --remote-allow-origins=*
 --window-position=0,0
 --display=%(ENV_DISPLAY)s
 --user-data-dir=/home/neko/.config/chromium
 --no-first-run
 --start-maximized
 --force-dark-mode
 --disable-file-system
 --disable-gpu
 --disable-software-rasterizer
 --disable-dev-shm-usage
stopsignal=INT
autorestart=true
priority=800
user=%(ENV_USER)s
stdout_logfile=/var/log/neko/chromium.log
stdout_logfile_maxbytes=100MB
stdout_logfile_backups=10
redirect_stderr=true

[program:socat_cdp]
command=socat TCP-LISTEN:9222,fork,reuseaddr,bind=0.0.0.0 TCP:127.0.0.1:9223
autorestart=true
priority=850
user=root
stdout_logfile=/var/log/neko/socat_cdp.log
stdout_logfile_maxbytes=10MB
stdout_logfile_backups=2
redirect_stderr=true

[program:openbox]
environment=HOME="/home/%(ENV_USER)s",USER="%(ENV_USER)s",DISPLAY="%(ENV_DISPLAY)s"
command=/usr/bin/openbox --config-file /etc/neko/openbox.xml
autorestart=true
priority=300
user=%(ENV_USER)s
stdout_logfile=/var/log/neko/openbox.log
stdout_logfile_maxbytes=100MB
stdout_logfile_backups=10
redirect_stderr=true
"""


def _mugi_cdp_runner_script_text() -> str:
    """Load the in-container CDP helper (sibling file in plugin dir, or last install copy under NEKO_DIR)."""
    here = os.path.dirname(os.path.abspath(__file__))
    for d in (here, NEKO_DIR):
        p = os.path.join(d, "mugi_cdp_container_runner.py")
        try:
            with open(p, encoding="utf-8") as f:
                t = f.read()
                if t.strip():
                    return t
        except OSError:
            continue
    return ""


def _cdp_run_page_via_container(
    target_id: str,
    session_steps: list[tuple[str, dict]],
    timeout: float,
) -> tuple[bool, list[dict] | str]:
    """Run page-level CDP inside mugi-neko against 127.0.0.1:9223 (bypasses host:socat for flattened sessions)."""
    try:
        from tools.searxng_docker import get_docker_path, _run_docker
    except ImportError:
        return False, "Docker helpers not available (tools.searxng_docker)."

    if not get_docker_path():
        return False, "Docker not found."

    ok_ps, out_ps = _run_docker(
        ["ps", "-q", "-f", "name=mugi-neko", "-f", "status=running"],
        timeout=10,
    )
    if not ok_ps or not (out_ps or "").strip():
        return False, "mugi-neko container is not running."

    runner_text = _mugi_cdp_runner_script_text()
    if not runner_text.strip():
        return False, (
            "mugi_cdp_container_runner.py is missing next to plugin.py. "
            "Reinstall Neko from the dashboard or copy the full neko plugin folder."
        )

    os.makedirs(NEKO_DIR, exist_ok=True)
    runner_host = os.path.join(NEKO_DIR, "mugi_cdp_container_runner.py")
    try:
        with open(runner_host, "w", encoding="utf-8") as f:
            f.write(runner_text)
    except OSError as e:
        return False, str(e)

    job = {
        "target_id": target_id,
        "steps": [[a, dict(b)] for a, b in session_steps],
        "timeout": float(timeout),
    }
    job_host = os.path.join(NEKO_DIR, ".mugi_cdp_job.json")
    try:
        with open(job_host, "w", encoding="utf-8") as f:
            json.dump(job, f)
    except OSError as e:
        return False, str(e)

    ok_cp1, err1 = _run_docker(["cp", runner_host, "mugi-neko:/tmp/mugi_cdp_runner.py"], timeout=30)
    if not ok_cp1:
        return False, f"docker cp runner: {err1}"
    ok_cp2, err2 = _run_docker(["cp", job_host, "mugi-neko:/tmp/mugi_cdp_job.json"], timeout=30)
    if not ok_cp2:
        return False, f"docker cp job: {err2}"

    tout = int(max(timeout, 5.0)) + 90
    ok_ex, stdout = _run_docker(
        ["exec", "mugi-neko", "python3", "/tmp/mugi_cdp_runner.py", "/tmp/mugi_cdp_job.json"],
        timeout=tout,
    )
    if not ok_ex:
        return False, (stdout or "docker exec failed")[:800]
    raw = (stdout or "").strip()
    if not raw:
        return False, "empty runner output"
    last = raw.splitlines()[-1].strip()
    try:
        data = json.loads(last)
    except json.JSONDecodeError:
        return False, f"bad runner output: {raw[:400]}"
    if data.get("ok"):
        return True, list(data.get("result") or [])
    return False, str(data.get("error") or "runner error")


def _cdp_ws_origin_header_line(ws_url: str) -> str:
    """Return ``Origin: …\\r\\n`` for WebSocket upgrade (Chromium checks remote-allow-origins)."""
    p = urlparse(ws_url)
    scheme = "https" if (p.scheme or "").lower() == "wss" else "http"
    host = p.hostname or "localhost"
    port = p.port
    origin = f"{scheme}://{host}" + (f":{port}" if port else "")
    return f"Origin: {origin}\r\n"


def _cdp_ws_send_json_masked(sock, obj: dict) -> None:
    import struct as _struct

    payload = json.dumps(obj).encode("utf-8")
    mask_key = os.urandom(4)
    length = len(payload)
    if length <= 125:
        header = _struct.pack("!BB", 0x81, 0x80 | length) + mask_key
    elif length <= 65535:
        header = _struct.pack("!BBH", 0x81, 0x80 | 126, length) + mask_key
    else:
        header = _struct.pack("!BBQ", 0x81, 0x80 | 127, length) + mask_key
    masked = bytes(b ^ mask_key[i % 4] for i, b in enumerate(payload))
    sock.sendall(header + masked)


def _cdp_ws_read_frame(sock, deadline: float) -> tuple[int, int, bytes] | None:
    import struct as _struct

    if time.monotonic() >= deadline:
        return None
    sock.settimeout(max(0.05, deadline - time.monotonic()))
    try:
        hdr = sock.recv(2)
    except Exception:
        return None
    if len(hdr) < 2:
        return None
    b1, b2 = hdr[0], hdr[1]
    fin = (b1 >> 7) & 1
    opcode = b1 & 0x0F
    masked = (b2 >> 7) & 1
    plen = b2 & 0x7F
    if plen == 126:
        sock.settimeout(max(0.05, deadline - time.monotonic()))
        ext = sock.recv(2)
        if len(ext) < 2:
            return None
        plen = _struct.unpack("!H", ext)[0]
    elif plen == 127:
        sock.settimeout(max(0.05, deadline - time.monotonic()))
        ext = sock.recv(8)
        if len(ext) < 8:
            return None
        plen = _struct.unpack("!Q", ext)[0]
    mask_key = b""
    if masked:
        sock.settimeout(max(0.05, deadline - time.monotonic()))
        mask_key = sock.recv(4)
        if len(mask_key) < 4:
            return None
    payload = b""
    while len(payload) < plen:
        sock.settimeout(max(0.05, deadline - time.monotonic()))
        chunk = sock.recv(plen - len(payload))
        if not chunk:
            return None
        payload += chunk
    if masked:
        payload = bytes(payload[i] ^ mask_key[i % 4] for i in range(len(payload)))
    return fin, opcode, payload


def _cdp_ws_send_pong_masked(sock, payload: bytes) -> None:
    import struct as _struct

    mask_key = os.urandom(4)
    length = len(payload)
    opcode = 0x8A
    if length <= 125:
        header = _struct.pack("!BB", opcode, 0x80 | length) + mask_key
    elif length <= 65535:
        header = _struct.pack("!BBH", opcode, 0x80 | 126, length) + mask_key
    else:
        header = _struct.pack("!BBQ", opcode, 0x80 | 127, length) + mask_key
    if length == 0:
        sock.sendall(header)
        return
    masked = bytes(payload[i] ^ mask_key[i % 4] for i in range(len(payload)))
    sock.sendall(header + masked)


def _cdp_ws_recv_json(sock, deadline: float) -> dict | None:
    """Read one JSON text frame; answers WebSocket ping so Chromium keepalives do not desync the stream."""
    while time.monotonic() < deadline:
        fr = _cdp_ws_read_frame(sock, deadline)
        if fr is None:
            return None
        _fin, opcode, payload = fr
        if opcode == 0x9:
            _cdp_ws_send_pong_masked(sock, payload)
            continue
        if opcode == 0x8:
            return None
        if opcode == 0xA:
            continue
        if opcode == 0x1:
            try:
                return json.loads(payload.decode("utf-8", errors="replace"))
            except json.JSONDecodeError:
                continue
    return None


def _cdp_ws_recv_until_id(sock, msg_id: int, deadline: float) -> dict | None:
    while time.monotonic() < deadline:
        msg = _cdp_ws_recv_json(sock, deadline)
        if msg is None:
            return None
        if msg.get("id") == msg_id:
            return msg
    return None


def _cdp_ws_command(ws_url: str, method: str, params: dict | None = None, timeout: int = 15) -> tuple[bool, dict]:
    """Send a single CDP command over a raw WebSocket and return (ok, result_or_error).
    Works with browser-level WS endpoints through the socat proxy."""
    import socket as _socket

    parsed = urlparse(ws_url)
    host = parsed.hostname or "localhost"
    port = parsed.port or 9222
    path = parsed.path or "/"

    try:
        sock = _socket.create_connection((host, port), timeout=timeout)
    except Exception as e:
        return False, {"error": f"TCP connect failed ({host}:{port}): {e}"}

    key = base64.b64encode(os.urandom(16)).decode()
    req = (
        f"GET {path} HTTP/1.1\r\n"
        f"Host: {host}:{port}\r\n"
        f"{_cdp_ws_origin_header_line(ws_url)}"
        f"Upgrade: websocket\r\n"
        f"Connection: Upgrade\r\n"
        f"Sec-WebSocket-Key: {key}\r\n"
        f"Sec-WebSocket-Version: 13\r\n"
        f"\r\n"
    )

    try:
        sock.sendall(req.encode())
        buf = b""
        while b"\r\n\r\n" not in buf:
            chunk = sock.recv(4096)
            if not chunk:
                sock.close()
                return False, {"error": "Connection closed during WebSocket upgrade"}
            buf += chunk
        if b"101" not in buf.split(b"\r\n")[0]:
            sock.close()
            return False, {"error": f"WebSocket upgrade failed: {buf[:200].decode(errors='replace')}"}

        _cdp_ws_send_json_masked(sock, {"id": 1, "method": method, **({"params": params} if params else {})})

        deadline = time.monotonic() + timeout
        result = _cdp_ws_recv_until_id(sock, 1, deadline)
        sock.close()
        if result is None:
            return True, {}
        if "error" in result:
            return False, result["error"]
        return True, result.get("result", {})
    except Exception as e:
        try:
            sock.close()
        except Exception:
            pass
        return False, {"error": str(e)}


def _cdp_page_session_run(
    browser_ws: str,
    target_id: str,
    session_steps: list[tuple[str, dict]],
    timeout: float = 45.0,
) -> tuple[bool, list[dict] | str]:
    """Single WebSocket: Target.attachToTarget, then CDP calls with sessionId (flattened protocol).

    Uses ``sessionId`` from the attach command result first, then ``Target.attachedToTarget`` if needed.
    Sends ``Runtime.runIfWaitingForDebugger`` before page commands so
    paused targets resume. Handles WebSocket ping frames.
    """
    import socket as _socket

    parsed = urlparse(browser_ws)
    host = parsed.hostname or "localhost"
    port = parsed.port or 9222
    path = parsed.path or "/"

    sock: _socket.socket | None = None
    try:
        sock = _socket.create_connection((host, port), timeout=15)
        key = base64.b64encode(os.urandom(16)).decode()
        req = (
            f"GET {path} HTTP/1.1\r\n"
            f"Host: {host}:{port}\r\n"
            f"{_cdp_ws_origin_header_line(browser_ws)}"
            f"Upgrade: websocket\r\n"
            f"Connection: Upgrade\r\n"
            f"Sec-WebSocket-Key: {key}\r\n"
            f"Sec-WebSocket-Version: 13\r\n"
            f"\r\n"
        )
        sock.sendall(req.encode())
        buf = b""
        while b"\r\n\r\n" not in buf:
            chunk = sock.recv(4096)
            if not chunk:
                return False, "Connection closed during WebSocket upgrade"
            buf += chunk
        if b"101" not in buf.split(b"\r\n")[0]:
            return False, f"WebSocket upgrade failed: {buf[:200].decode(errors='replace')}"

        deadline = time.monotonic() + timeout
        _cdp_ws_send_json_masked(
            sock,
            {"id": 1, "method": "Target.attachToTarget", "params": {"targetId": target_id, "flatten": True}},
        )
        event_sid: str | None = None
        r1: dict | None = None
        while time.monotonic() < deadline:
            msg = _cdp_ws_recv_json(sock, deadline)
            if msg is None:
                return False, "Target.attachToTarget closed or timed out"
            if msg.get("method") == "Target.attachedToTarget":
                p = msg.get("params") or {}
                es = p.get("sessionId")
                if isinstance(es, str) and es:
                    event_sid = es
            if msg.get("id") == 1:
                r1 = msg
                break
        if not r1:
            return False, "Target.attachToTarget timed out"
        if "error" in r1:
            return False, str(r1.get("error"))
        resp_sid = (r1.get("result") or {}).get("sessionId")
        # Prefer attach *response* sessionId; event-based id can race or disagree on some stacks.
        sid = resp_sid if (isinstance(resp_sid, str) and resp_sid) else event_sid
        if not sid:
            return False, "No sessionId from Target.attachToTarget"

        _cdp_ws_send_json_masked(
            sock,
            {"id": 2, "method": "Runtime.runIfWaitingForDebugger", "params": {}, "sessionId": sid},
        )
        _cdp_ws_recv_until_id(sock, 2, deadline)

        out: list[dict] = []
        nid = 3
        for method, params in session_steps:
            _cdp_ws_send_json_masked(sock, {"id": nid, "method": method, "params": params, "sessionId": sid})
            rn = _cdp_ws_recv_until_id(sock, nid, deadline)
            if not rn:
                return False, f"CDP timeout: {method}"
            if "error" in rn:
                return False, f"{method}: {rn.get('error')}"
            out.append(rn.get("result") or {})
            nid += 1
        return True, out
    except Exception as e:
        return False, str(e)
    finally:
        if sock is not None:
            try:
                sock.close()
            except Exception:
                pass


def _cdp_http_get_json(cdp_http_base: str, path: str) -> tuple[bool, Any]:
    """GET JSON from Chromium devtools HTTP (e.g. /json/list). path must start with /."""
    import json as _json

    base = (cdp_http_base or "").rstrip("/")
    if not base:
        return False, "empty cdp http base"
    try:
        import urllib.request

        with urllib.request.urlopen(f"{base}{path}", timeout=10) as resp:
            return True, _json.loads(resp.read().decode("utf-8", errors="replace"))
    except Exception as e:
        return False, str(e)


def _cdp_chromium_major_version(cdp_http_base: str) -> int | None:
    ok, data = _cdp_http_get_json((cdp_http_base or "").rstrip("/"), "/json/version")
    if not ok or not isinstance(data, dict):
        return None
    browser = str(data.get("Browser") or "")
    m = re.match(r"(?:Chrome|Chromium)/(\d+)", browser)
    return int(m.group(1)) if m else None


def _neko_cdp_broken_chromium_hint(cdp_http_base: str) -> str:
    """Extra user-facing text when page-level CDP fails on known-bad Chromium builds."""
    maj = _cdp_chromium_major_version(cdp_http_base)
    if maj is None or maj < 145:
        return ""
    return (
        " [Neko CDP] Chromium 145+ in the rolling Neko image breaks page-level DevTools (clicks, DOM text). "
        "Remove Neko, run: docker rmi mugi-neko-chromium-cdp:latest, then reinstall so the plugin rebuilds "
        "from ghcr.io/m1k1o/neko/chromium:3.0.8 (Chromium ~139). Install now uses docker build --pull."
    )


def _cdp_direct_page_run(
    page_ws: str,
    session_steps: list[tuple[str, dict]],
    timeout: float = 45.0,
) -> tuple[bool, list[dict] | str]:
    """Run CDP on a page's webSocketDebuggerUrl (no Target.attachToTarget / sessionId).

    Sends ``Runtime.runIfWaitingForDebugger`` first (not included in *out*). Handles WebSocket pings.
    """
    import socket as _socket

    parsed = urlparse(page_ws)
    host = parsed.hostname or "localhost"
    port = parsed.port or 9222
    path = parsed.path or "/"

    sock: _socket.socket | None = None
    try:
        sock = _socket.create_connection((host, port), timeout=15)
        key = base64.b64encode(os.urandom(16)).decode()
        req = (
            f"GET {path} HTTP/1.1\r\n"
            f"Host: {host}:{port}\r\n"
            f"{_cdp_ws_origin_header_line(page_ws)}"
            f"Upgrade: websocket\r\n"
            f"Connection: Upgrade\r\n"
            f"Sec-WebSocket-Key: {key}\r\n"
            f"Sec-WebSocket-Version: 13\r\n"
            f"\r\n"
        )
        sock.sendall(req.encode())
        buf = b""
        while b"\r\n\r\n" not in buf:
            chunk = sock.recv(4096)
            if not chunk:
                return False, "Connection closed during WebSocket upgrade"
            buf += chunk
        if b"101" not in buf.split(b"\r\n")[0]:
            return False, f"WebSocket upgrade failed: {buf[:200].decode(errors='replace')}"

        deadline = time.monotonic() + timeout
        _cdp_ws_send_json_masked(sock, {"id": 1, "method": "Runtime.runIfWaitingForDebugger", "params": {}})
        _cdp_ws_recv_until_id(sock, 1, deadline)

        out: list[dict] = []
        nid = 2
        for method, params in session_steps:
            _cdp_ws_send_json_masked(sock, {"id": nid, "method": method, "params": params})
            rn = _cdp_ws_recv_until_id(sock, nid, deadline)
            if not rn:
                return False, f"CDP timeout: {method}"
            if "error" in rn:
                return False, f"{method}: {rn.get('error')}"
            out.append(rn.get("result") or {})
            nid += 1
        return True, out
    except Exception as e:
        return False, str(e)
    finally:
        if sock is not None:
            try:
                sock.close()
            except Exception:
                pass


def _page_ws_from_json_list(cdp_http_base: str, target_id: str) -> str:
    ok_j, data = _cdp_http_get_json(cdp_http_base, "/json/list")
    if not ok_j or not isinstance(data, list):
        return ""
    tid = str(target_id)
    for e in data:
        if isinstance(e, dict) and str(e.get("id") or "") == tid:
            raw = (e.get("webSocketDebuggerUrl") or "").strip()
            if not raw:
                return ""
            return _rewrite_ws_url_to_cdp_base(raw, cdp_http_base)
    return ""


def _rewrite_ws_url_to_cdp_base(ws_url: str, cdp_http_base: str) -> str:
    """Rewrite a ``ws://`` URL so its host:port matches *cdp_http_base*.

    Chromium reports ``webSocketDebuggerUrl`` relative to its own listening
    address (e.g. ``127.0.0.1:9223``).  Behind socat, only the proxy port
    (``9222``) is reachable from the host, so we must substitute.
    """
    from urllib.parse import urlparse as _up

    base_p = _up((cdp_http_base or "").rstrip("/"))
    ws_p = _up(ws_url)
    new_host = base_p.hostname or ws_p.hostname or "localhost"
    new_port = base_p.port or ws_p.port or 9222
    scheme = ws_p.scheme or "ws"
    return f"{scheme}://{new_host}:{new_port}{ws_p.path}"


def _neko_cdp_stale_session_error(msg: object) -> bool:
    """True when CDP indicates the page target or DevTools session is no longer valid."""
    s = str(msg)
    return (
        "-32001" in s
        or "Session with given id not found" in s
        or "No such target id" in s
    )


def _cdp_run_page_or_session(
    cdp_http_base: str,
    browser_ws: str,
    target_id: str,
    session_steps: list[tuple[str, dict]],
    timeout: float = 45.0,
) -> tuple[bool, list[dict] | str]:
    """Try direct page WebSocket first (host CDP). Then in-container session attach (bypasses socat).
    Falls back to Target.attachToTarget on the host browser WS.

    Retries once with a freshly picked page ``targetId`` when CDP reports a stale
    session or missing target — including failures on the **direct** page WebSocket
    path (otherwise we could burn a full timeout on container/attach with a dead
    ``targetId`` after ``neko_navigate`` replaces the tab).
    """
    tid = target_id
    final_ok = False
    final_out: list[dict] | str = ""

    for attempt in range(2):
        page_ws = _page_ws_from_json_list(cdp_http_base, tid)
        if page_ws:
            ok_d, out_d = _cdp_direct_page_run(page_ws, session_steps, timeout=timeout)
            if ok_d:
                return True, out_d
            final_out = out_d
            if attempt == 0 and _neko_cdp_stale_session_error(out_d):
                time.sleep(0.25)
                try:
                    ok_pick, picked = _neko_pick_page_target_id(browser_ws)
                    if ok_pick and isinstance(picked, str) and picked:
                        tid = picked
                        continue
                except Exception:
                    pass

        ok_in, out_in = _cdp_run_page_via_container(tid, session_steps, timeout)
        if ok_in:
            return True, out_in
        final_out = out_in

        ok_a, out_a = _cdp_page_session_run(browser_ws, tid, session_steps, timeout=timeout)
        if ok_a:
            return True, out_a
        final_ok = ok_a
        final_out = out_a

        err_a = str(out_a) if not ok_a else ""
        if attempt == 0 and (
            _neko_cdp_stale_session_error(err_a) or _neko_cdp_stale_session_error(out_in)
        ):
            time.sleep(0.5)
            try:
                ok_pick, picked = _neko_pick_page_target_id(browser_ws)
                if ok_pick and isinstance(picked, str) and picked:
                    tid = picked
                    continue
            except Exception:
                pass
        break

    if not final_ok and isinstance(final_out, str):
        hint = _neko_cdp_broken_chromium_hint(cdp_http_base)
        if hint:
            return False, final_out + hint
    return final_ok, final_out


def _page_target_sort_key(t: dict) -> tuple[int, str]:
    """Prefer real web pages for CDP attach; deprioritize extension, chrome://, about:, devtools."""
    u = (t.get("url") or "").lower()
    if u.startswith("about:"):
        return (3, u)
    if u.startswith("chrome-extension:") or u.startswith("chrome://") or u.startswith("devtools:"):
        return (2, u)
    if u.startswith("http://") or u.startswith("https://"):
        return (0, u)
    return (1, u)


def _neko_page_targets_sorted(targets_payload: dict) -> list[dict]:
    pages = [t for t in targets_payload.get("targetInfos", []) if t.get("type") == "page"]
    pages.sort(key=_page_target_sort_key)
    return pages


def _neko_pick_page_target_id(browser_ws: str) -> tuple[bool, str | dict]:
    ok, targets = _cdp_ws_command(browser_ws, "Target.getTargets")
    if not ok:
        return False, targets
    pages = _neko_page_targets_sorted(targets)
    if not pages:
        return False, "No Chromium page targets."
    return True, pages[0]["targetId"]


def _neko_cdp_eval_object_value(eval_step_result: dict) -> dict | list | str | float | bool | None:
    """Extract JavaScript object value from Runtime.evaluate CDP result dict."""
    inner = eval_step_result.get("result")
    if not isinstance(inner, dict):
        return None
    if inner.get("subtype") == "error":
        return None
    val = inner.get("value")
    if isinstance(val, dict):
        return val
    if inner.get("type") == "object" and isinstance(val, dict):
        return val
    return val


def _cdp_get_browser_ws(cdp_url: str) -> tuple[bool, str]:
    """Query /json/version to get the browser-level webSocketDebuggerUrl."""
    import json as _json
    base = (cdp_url or "http://localhost:9222").rstrip("/")
    try:
        import urllib.request
        resp = urllib.request.urlopen(f"{base}/json/version", timeout=5)
        ver = _json.loads(resp.read())
        ws = ver.get("webSocketDebuggerUrl", "")
        if not ws:
            return False, "No browser WebSocket URL in /json/version."
        return True, _rewrite_ws_url_to_cdp_base(ws, base)
    except Exception as e:
        return False, (
            f"Could not reach CDP at {base}/json/version. "
            "Check: (1) container running (docker ps | grep mugi-neko), "
            "(2) neko_cdp_url in plugin settings (default http://localhost:9222). "
            f"Error: {e}"
        )


def _neko_cdp_page_snapshot_lines(browser_ws: str) -> tuple[bool, str]:
    """Read-only: list each Chromium page tab's url and title (Target.getTargets)."""
    ok, targets = _cdp_ws_command(browser_ws, "Target.getTargets")
    if not ok:
        return False, f"Target.getTargets failed: {targets}"
    pages = _neko_page_targets_sorted(targets)
    if not pages:
        return True, "No open page tabs reported by Chromium."
    lines: list[str] = []
    for i, p in enumerate(pages, 1):
        url = (p.get("url") or "").strip() or "(unknown url)"
        title = (p.get("title") or "").strip() or "(no title)"
        lines.append(f"Tab {i}: {url}\n        title: {title}")
    return True, "\n".join(lines)


# ---------------------------------------------------------------------------
# neko_wait_for primitives — pure helpers keep logic unit-testable without CDP.
# ---------------------------------------------------------------------------
# Hard bounds on polling so agents cannot burn CDP with a long/tight loop.
_WAIT_TIMEOUT_MIN_S = 1.0
_WAIT_TIMEOUT_MAX_S = 60.0
_WAIT_TIMEOUT_DEFAULT_S = 20.0
_WAIT_POLL_INITIAL_S = 0.25
_WAIT_POLL_MAX_S = 2.0
_WAIT_POLL_BACKOFF = 1.5
# Bound text-match probes so we never ship long needles to the page eval or
# echo large strings back to the agent in error messages.
_WAIT_TEXT_MAX_LEN = 200
_WAIT_SELECTOR_ECHO_LEN = 80


def _wait_clamp_timeout(raw: Any) -> tuple[bool, float | str]:
    """Return (True, seconds) or (False, err_code) after parsing/clamping."""
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return False, "invalid_timeout"
    if value < _WAIT_TIMEOUT_MIN_S or value > _WAIT_TIMEOUT_MAX_S:
        return False, "invalid_timeout"
    return True, value


def _wait_pick_condition(
    url_equals: str, url_prefix: str, selector: str, text: str
) -> tuple[str, str] | tuple[None, str]:
    """Choose which condition is active; return (kind, value) or (None, err_code)."""
    provided = [
        ("url_equals", (url_equals or "").strip()),
        ("url_prefix", (url_prefix or "").strip()),
        ("selector", (selector or "").strip()),
        ("text", (text or "")[: _WAIT_TEXT_MAX_LEN + 1]),
    ]
    active = [(k, v) for k, v in provided if v]
    if not active:
        return None, "no_condition"
    if len(active) > 1:
        return None, "too_many_conditions"
    kind, value = active[0]
    if kind == "text" and len(value) > _WAIT_TEXT_MAX_LEN:
        return None, "text_too_long"
    return kind, value


def _wait_url_current(browser_ws: str) -> tuple[bool, str]:
    """Return (True, url) for the top page target, or (False, err_code)."""
    ok, targets = _cdp_ws_command(browser_ws, "Target.getTargets")
    if not ok:
        return False, "cdp_unreachable"
    pages = _neko_page_targets_sorted(targets)
    if not pages:
        return False, "no_page_target"
    url = pages[0].get("url") or ""
    return True, str(url)


def _wait_probe_expr(kind: str, value: str) -> str:
    """Build a small, bounded JS probe for selector/text conditions.

    Returns ``{ok: true, found: bool}`` on success or ``{ok: false, err: <msg>}``
    when the probe itself raised. Kept intentionally short — agents never see
    page content, only the boolean plus stable error codes.
    """
    if kind == "selector":
        return (
            "(function(){try{return{ok:true,found:!!document.querySelector("
            + json.dumps(value)
            + ")};}catch(e){return{ok:false,err:String(e)};}})()"
        )
    if kind == "text":
        return (
            "(function(){try{var t=document.body?(document.body.innerText||''):'';"
            "var n=" + json.dumps(value) + ";"
            "return{ok:true,found:t.indexOf(n)!==-1};"
            "}catch(e){return{ok:false,err:String(e)};}})()"
        )
    return "(function(){return{ok:false,err:'unknown_probe'};})()"


def _wait_next_delay(prev: float, deadline: float, now: float) -> float:
    """Compute next poll sleep: jittered backoff, capped, bounded by deadline."""
    nxt = min(_WAIT_POLL_MAX_S, prev * _WAIT_POLL_BACKOFF)
    remaining = max(0.0, deadline - now)
    return max(0.0, min(nxt, remaining))


def _wait_timeout_tag(kind: str, value: str) -> str:
    """Stable failure code for the skip-reason catalog. Truncate selector echo;
    never echo user-provided text so page content cannot leak via an error string."""
    if kind == "selector":
        echo = value[:_WAIT_SELECTOR_ECHO_LEN]
        suffix = "…" if len(value) > _WAIT_SELECTOR_ECHO_LEN else ""
        return f"wait_timeout:selector={echo}{suffix}"
    if kind in ("url_equals", "url_prefix", "text"):
        return f"wait_timeout:{kind}"
    return "wait_timeout:unknown"


def _default_route_ipv4() -> str:
    """Best-effort primary IPv4 for WebRTC NAT1TO1 (Neko docs: use local IP, not always loopback).

    127.0.0.1 often fails with Colima/Lima UDP port forwarding; the browser still needs a routable
    host candidate for ICE. Falls back to loopback if detection fails.
    """
    import socket as _socket

    try:
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
        try:
            s.settimeout(0.5)
            s.connect(("8.8.8.8", 80))
            ip = s.getsockname()[0]
            if ip and not ip.startswith("127."):
                return ip
        finally:
            s.close()
    except Exception:
        pass
    return "127.0.0.1"


def _webrtc_nat1to1_for_room_url(room_url: str, user_override: str | None) -> str:
    """ICE host IP Neko advertises must match how the browser reaches this Mac (localhost vs LAN IP)."""
    o = (user_override or "").strip()
    if o:
        return o
    try:
        from urllib.parse import urlparse

        h = (urlparse(room_url).hostname or "").strip().lower()
    except Exception:
        h = ""
    if h in ("localhost", "127.0.0.1", ""):
        return "127.0.0.1"
    if h and _HOST_IPV4_RE.match(h):
        return h
    return _default_route_ipv4()


def _safe_url_for_open(url: str | None) -> str | None:
    """Return url only if it is http or https (safe to open in browser or copy). Otherwise None."""
    if not url or not isinstance(url, str):
        return None
    s = url.strip()
    if not s or len(s) > 2048:
        return None
    lower = s.lower()
    if not (lower.startswith("http://") or lower.startswith("https://")):
        return None
    try:
        from urllib.parse import urlparse
        parsed = urlparse(s)
        if parsed.scheme not in ("http", "https") or not parsed.netloc:
            return None
        return s
    except Exception:
        return None


def _compose_content(base_url: str, use_firefox: bool = False, *, nat1to1: str = "127.0.0.1") -> str:
    """Docker Compose: CDP Chromium by default; minimal env matching the official v3 install docs.

    IMPORTANT: Do NOT set NEKO_LEGACY=false. The bundled web client is still v2-compatible and connects
    via the legacy /ws endpoint. Disabling legacy mode removes that endpoint → "connection timeout".
    The official v3 docs explicitly warn against it unless using a v3-native client (demodesk/neko-client).
    """
    nat_ip = (nat1to1 or "127.0.0.1").strip() or "127.0.0.1"
    image = NEKO_IMAGE_FIREFOX if use_firefox else NEKO_IMAGE_CHROMIUM_CDP
    mux = _NEKO_WEBRTC_MUX_PORT
    ports = [
        f"8080:{NEKO_DEFAULT_PORT}",
        f"{mux}:{mux}/udp",
        f"{mux}:{mux}/tcp",
    ]
    if not use_firefox:
        ports.append(f"{NEKO_CDP_PORT}:{NEKO_CDP_PORT}/tcp")
    neko_svc = {
        "image": image,
        "container_name": "mugi-neko",
        "restart": "unless-stopped",
        "shm_size": "2gb",
        "ports": ports,
        "environment": {
            "NEKO_DESKTOP_SCREEN": "1280x720@30",
            "NEKO_MEMBER_MULTIUSER_USER_PASSWORD": "neko",
            "NEKO_MEMBER_MULTIUSER_ADMIN_PASSWORD": "neko",
            "NEKO_WEBRTC_UDPMUX": mux,
            "NEKO_WEBRTC_TCPMUX": mux,
            "NEKO_WEBRTC_ICELITE": True,
            "NEKO_WEBRTC_NAT1TO1": nat_ip,
        },
    }
    if not use_firefox:
        neko_svc["cap_add"] = ["SYS_ADMIN"]
    compose = {"version": "3", "services": {"neko": neko_svc}}
    try:
        import yaml
        body = yaml.dump(compose, default_flow_style=False, sort_keys=False, allow_unicode=True)
    except ImportError:
        lines = [
            "version: \"3\"",
            "services:",
            "  neko:",
            f"    image: {image}",
            "    container_name: mugi-neko",
            "    restart: unless-stopped",
            '    shm_size: "2gb"',
        ]
        if not use_firefox:
            lines.append("    cap_add:")
            lines.append("      - SYS_ADMIN")
        lines.append("    ports:")
        for p in ports:
            lines.append(f'      - "{p}"')
        lines.append("    environment:")
        lines.extend([
            '      NEKO_DESKTOP_SCREEN: "1280x720@30"',
            '      NEKO_MEMBER_MULTIUSER_USER_PASSWORD: "neko"',
            '      NEKO_MEMBER_MULTIUSER_ADMIN_PASSWORD: "neko"',
            f'      NEKO_WEBRTC_UDPMUX: "{mux}"',
            f'      NEKO_WEBRTC_TCPMUX: "{mux}"',
            '      NEKO_WEBRTC_ICELITE: true',
            f'      NEKO_WEBRTC_NAT1TO1: "{nat_ip}"',
        ])
        body = "\n".join(lines) + "\n"
    return "# Generated by Mugi Neko plugin. Do not edit manually.\n" + body


def _do_install(
    ctx,
    progress: Callable[[str, str], None] | None = None,
) -> tuple[bool, str]:
    """Pull image, write compose, run docker compose up -d via Colima + Mugi Homebrew. Return (success, message).

    *progress* is optional; called as progress(phase_id, user_visible_message) for Live View / stream updates.
    """
    try:
        from tools.searxng_docker import (
            ensure_colima_running,
            get_docker_path,
            _run_docker,
        )
    except ImportError:
        return False, "Colima/Docker helpers not available (tools.searxng_docker)."

    def p(phase: str, msg: str) -> None:
        if progress:
            progress(phase, msg)

    p("start", "Starting Neko install…")

    docker_bin = get_docker_path()
    if not docker_bin:
        return False, (
            "Docker not found. Set up Colima via 'Set up SearXNG (Docker)' in Settings → Network, "
            "or install Colima so Neko can use the same container runtime."
        )

    p("colima", "Checking Colima / Docker…")
    ok, err = ensure_colima_running()
    if not ok:
        return False, f"Colima could not start: {err[:400]}"

    # Wait for the Docker socket to be ready (Colima may have just started)
    daemon_hint = (
        "Colima is not running. In Terminal run: colima start. Wait for it to finish, then try again. "
        "If Colima is not installed, use Settings → Network → Set up SearXNG (Docker) first."
    )
    p("docker_wait", "Waiting for Docker daemon…")
    for attempt in range(10):
        ok, msg = _run_docker(["info"], timeout=15)
        if ok:
            break
        if attempt < 9:
            time.sleep(2)
    if not ok:
        if msg and ("Cannot connect to the Docker daemon" in msg or "Is the docker daemon running" in msg):
            return False, daemon_hint
        return False, (msg or "Docker daemon not reachable.")[:280] + " " + daemon_hint

    os.makedirs(NEKO_DIR, exist_ok=True)
    base_url = ctx.get_setting("neko_base_url", "http://localhost:8080") or "http://localhost:8080"
    use_firefox = bool(ctx.get_setting("neko_use_firefox", False))

    if not use_firefox:
        # Build CDP-enabled Chromium image so the agent can control the browser.
        p("cdp_files", "Writing CDP Chromium build files…")
        dockerfile_path = os.path.join(NEKO_DIR, "Dockerfile.cdp")
        supervisord_path = os.path.join(NEKO_DIR, "supervisord-chromium-cdp.conf")
        try:
            runner_text = _mugi_cdp_runner_script_text()
            if not runner_text.strip():
                return False, (
                    "mugi_cdp_container_runner.py is missing next to the Neko plugin. "
                    "Copy the full templates/plugin_examples/neko folder into ~/.mugi/plugins/neko/."
                )
            runner_neko = os.path.join(NEKO_DIR, "mugi_cdp_container_runner.py")
            with open(runner_neko, "w", encoding="utf-8") as rf:
                rf.write(runner_text)
            with open(dockerfile_path, "w", encoding="utf-8") as f:
                f.write(_DOCKERFILE_CDP)
            with open(supervisord_path, "w", encoding="utf-8") as f:
                f.write(_SUPERVISORD_CHROMIUM_CDP)
        except OSError as e:
            return False, f"Cannot write CDP build files: {e}"
        p("pull_chromium", "Pulling base Chromium image (may take several minutes)…")
        ok, msg = _run_docker(["pull", NEKO_IMAGE_CHROMIUM], timeout=300)
        if not ok:
            return False, f"Pull base Chromium image failed: {msg[:300]}"
        p("build_cdp", "Building CDP-enabled Chromium image (may take a few minutes)…")
        ok, msg = _run_docker(
            ["build", "--pull", "-t", NEKO_IMAGE_CHROMIUM_CDP, "-f", dockerfile_path, NEKO_DIR],
            timeout=300,
        )
        if not ok:
            return False, f"Build CDP Chromium image failed: {msg[:400]}"

    nat_ip = _webrtc_nat1to1_for_room_url(base_url, ctx.get_setting("neko_webrtc_nat1to1", "") or "")
    p(
        "compose",
        f"Writing docker-compose.yml (WebRTC mux UDP+TCP {_NEKO_WEBRTC_MUX_PORT}, NAT 1:1 → {nat_ip})…",
    )
    content = _compose_content(base_url, use_firefox=use_firefox, nat1to1=nat_ip)
    try:
        with open(NEKO_COMPOSE_PATH, "w", encoding="utf-8") as f:
            f.write(content)
    except OSError as e:
        return False, f"Cannot write compose file: {e}"

    if use_firefox:
        p("pull_firefox", "Pulling Firefox image…")
        ok, msg = _run_docker(
            ["compose", "-f", NEKO_COMPOSE_PATH, "pull"],
            timeout=300,
        )
        if not ok:
            if msg and ("Cannot connect to the Docker daemon" in msg or "Is the docker daemon running" in msg):
                return False, daemon_hint
            return False, ((msg or "docker compose pull failed.")[:280] + " " + daemon_hint)[:400]

    p("up", "Starting container (docker compose up -d)…")
    ok, msg = _run_docker(
        ["compose", "-f", NEKO_COMPOSE_PATH, "up", "-d"],
        timeout=60,
    )
    if not ok:
        return False, (msg or "docker compose up failed.")[:300]
    p(
        "done",
        "Neko is running (WebRTC mux). If video still stalls, set NAT 1:1 override to your LAN IP when "
        "using http://192.168.x.x in the browser.",
    )
    return True, "Neko is running. Open the dashboard to get the room link."


def _list_neko_container_ids() -> tuple[list[str] | None, str]:
    """IDs for this plugin's Neko stack: fixed container name plus known images (covers compose renames / drift).

    Returns (None, err) if a docker ps invocation failed; ([], "") if none match.
    """
    try:
        from tools.searxng_docker import _run_docker

        filter_argsets: list[list[str]] = [
            ["ps", "-aq", "-f", "name=mugi-neko"],
            ["ps", "-aq", "-f", f"ancestor={NEKO_IMAGE_CHROMIUM}"],
            ["ps", "-aq", "-f", f"ancestor={NEKO_IMAGE_CHROMIUM_CDP}"],
            ["ps", "-aq", "-f", f"ancestor={NEKO_IMAGE_FIREFOX}"],
        ]
        seen: set[str] = set()
        ordered: list[str] = []
        for args in filter_argsets:
            ok, text = _run_docker(args, timeout=15)
            if not ok:
                return None, (text or "docker ps failed.")[:300]
            for ln in (text or "").splitlines():
                s = ln.strip()
                if s and _DOCKER_CONTAINER_ID_RE.fullmatch(s) and s not in seen:
                    seen.add(s)
                    ordered.append(s)
        return ordered, ""
    except Exception as e:
        return None, str(e)[:200]


def _container_exists() -> bool:
    """True if the Neko container exists (running or stopped). Used so dashboard state matches reality even when neko_installed was never set."""
    ids, _ = _list_neko_container_ids()
    return bool(ids)


def _neko_install_panel_show_remove() -> bool:
    """True if Remove should appear on the pre-install dashboard (container or compose present)."""
    return _container_exists() or os.path.isfile(NEKO_COMPOSE_PATH)


def _do_remove() -> tuple[bool, str]:
    """Stop and remove the Neko container. Compose file is left in ~/.mugi/neko when present so Install can reuse it.

    Always force-removes by container ID after a best-effort `compose down`, so a failing or mismatched
    compose project cannot leave an orphaned `mugi-neko` running.
    """
    try:
        from tools.searxng_docker import get_docker_path, _run_docker
    except ImportError:
        return False, "Docker helpers not available (tools.searxng_docker)."
    if not get_docker_path():
        return False, "Docker not found."

    had_compose = os.path.isfile(NEKO_COMPOSE_PATH)
    ids_start, list_err = _list_neko_container_ids()
    if ids_start is None:
        return False, list_err or "Could not list Docker containers."

    compose_down_err = ""
    if had_compose:
        ok_cd, msg_cd = _run_docker(
            ["compose", "-f", NEKO_COMPOSE_PATH, "down", "--remove-orphans"],
            timeout=120,
        )
        if not ok_cd:
            compose_down_err = (msg_cd or "docker compose down failed.").strip()[:400]
        ids, list_err2 = _list_neko_container_ids()
        if ids is None:
            return False, list_err2 or "Could not list Docker containers."
    else:
        ids = ids_start

    if ids:
        for cid in ids:
            ok, msg = _run_docker(["rm", "-f", cid], timeout=90)
            if not ok:
                tail = f" (compose down: {compose_down_err[:180]})" if compose_down_err else ""
                return False, ((msg or f"docker rm failed for {cid[:12]}…") + tail)[:450]

    ids_left, _ = _list_neko_container_ids()
    if ids_left:
        return False, (
            "Neko container(s) still present: "
            + ", ".join(c[:12] for c in ids_left[:4])
            + ". Start Colima if needed and ensure `docker` uses the same context as when Neko was started."
        )[:450]

    if not ids_start and not had_compose:
        return (
            True,
            "Nothing to remove — there is no Neko container and no compose file yet. "
            "Use Install Neko when you want to set up.",
        )

    if not ids_start and had_compose:
        note = f" {compose_down_err[:200]}" if compose_down_err else ""
        return (
            True,
            "Neko compose stack is down; no mugi-neko container was running."
            + (f" Note:{note}" if note else "")
            + " Click Install Neko to start again.",
        )[:450]

    if not had_compose:
        return (
            True,
            "Removed the leftover Neko container (docker-compose.yml was missing under ~/.mugi/neko). "
            "Click Install Neko to create a fresh setup.",
        )

    base = "Neko container stopped and removed. Click Install Neko if you want to run it again."
    if compose_down_err:
        base += f" (docker compose down reported an issue: {compose_down_err[:180]})"
    return True, base[:450]


def _install_controls(*, show_remove: bool = False) -> list[dict]:
    """Install when not in 'ready' dashboard state; optionally Remove if a partial/stuck deploy exists."""
    ctrls: list[dict] = [
        {
            "id": "install_neko",
            "type": "button",
            "label": "Install Neko",
            "action_id": "install_neko",
            "style": "primary",
            "description": "Download and start the Neko Chromium container (uses Colima; setup under ~/.mugi/neko). Progress appears above while this runs.",
        },
    ]
    if show_remove:
        ctrls.append(
            {
                "id": "remove_neko",
                "type": "button",
                "label": "Remove Neko",
                "action_id": "remove_neko",
                "style": "default",
                "description": "Stop the container and clear install state. Use if install failed, WebRTC is stuck, or you need a clean slate.",
            },
        )
    return ctrls


def _session_controls(room_url: str) -> list[dict]:
    """Controls shown after install for the current room. Remove is placed early so it isn't missed below the fold."""
    return [
        {
            "id": "open_browser",
            "type": "button",
            "label": "Open in browser",
            "action_id": "open_browser",
            "style": "primary",
        },
        {
            "id": "copy_link",
            "type": "button",
            "label": "Copy link",
            "action_id": "copy_link",
        },
        {
            "id": "remove_neko",
            "type": "button",
            "label": "Remove Neko",
            "action_id": "remove_neko",
            "style": "default",
            "description": "Stop and remove the container (compose files stay under ~/.mugi/neko). Use before re-installing.",
        },
        {
            "id": "audio_enabled",
            "type": "toggle",
            "label": "Audio",
            "default": True,
        },
    ]


def _make_tools(ctx, state: dict, state_lock: threading.Lock):
    def _is_installed() -> bool:
        """Installed only if a matching container exists; clear stale neko_installed after remove or manual docker rm."""
        has_container = _container_exists()
        flagged = ctx.get_setting("neko_installed", False)
        if has_container:
            if not flagged:
                ctx.set_setting("neko_installed", True)
            return True
        if flagged:
            ctx.set_setting("neko_installed", False)
        return False

    def _base_url() -> str:
        return (ctx.get_setting("neko_base_url", "") or "http://localhost:8080").rstrip("/")

    @tool
    def open_neko_dashboard() -> str:
        """Open or refresh the Neko plugin panel in Mugi (Install, room link, Open in browser). Idempotent: calling again updates the same panel instead of stacking duplicates. The room link is for the human on the Mac; the agent is already on the shared Chromium via CDP once Neko is installed—do not neko_navigate to the room URL."""
        stream_id = ctx.open_stream(
            NEKO_DASHBOARD_STREAM_ID,
            stream_type="timeseries",
            title="Neko",
            metadata={"plugin": "neko"},
        )
        with state_lock:
            state["active_stream_id"] = stream_id
        if not _is_installed():
            show_rm = _neko_install_panel_show_remove()
            ctx.register_stream_controls(stream_id, _install_controls(show_remove=show_rm))
            pending_msg = (
                "Neko is not installed. Click 'Install Neko' to download and start the container (uses Colima, same as SearXNG). "
                "Status and progress appear here while install runs."
            )
            if show_rm:
                pending_msg += " If a previous attempt left things stuck, use 'Remove Neko' first, then install again."
            ctx.push_stream_update(
                stream_id,
                {
                    "status": "pending_install",
                    "message": pending_msg,
                    "metrics": {},
                },
            )
            return "Neko dashboard opened. Neko is not installed yet — click 'Install Neko' in the dashboard to install."
        room_url = _base_url()
        ctx.register_stream_controls(stream_id, _session_controls(room_url))
        ctx.push_stream_update(
            stream_id,
            {
                "status": "ready",
                "message": f"Room link: {room_url}",
                "metrics": {"room_url": room_url},
            },
        )
        return (
            "Neko dashboard opened (panel refreshed). Do not call open_neko_dashboard again in the same flow—"
            "reply to the user or use neko_browser_snapshot / neko_page_visible_text / neko_click_selector unless they explicitly ask to open the dashboard again. "
            f"Room link for the user (Mac browser only): {room_url}. "
            f"{_NEKO_AGENT_ALREADY_IN_SESSION}"
        )

    @tool
    def install_neko() -> str:
        """Install/start the Neko container (Chromium + CDP). The user opens the room link on their Mac to log in and see the desktop; CDP then lets you drive that same Chromium—no neko_navigate to the room URL. If Colima is missing, use Settings → Network → Set up SearXNG (Docker) first."""
        if _is_installed():
            return (
                "Neko is already installed. "
                f"{_NEKO_AGENT_ALREADY_IN_SESSION} "
                "Use get_neko_room_link if the user needs the URL for their browser."
            )
        with state_lock:
            if state.get("installing"):
                return (
                    "Install already in progress. Open the Neko dashboard to watch status, or wait for it to finish."
                )
            state["installing"] = True
        try:
            sid = state.get("active_stream_id")

            def _prog(phase: str, message: str) -> None:
                if not sid:
                    return
                ctx.push_stream_update(sid, _neko_install_stream_payload(phase, message))

            ok, msg = _do_install(ctx, progress=_prog if sid else None)
        finally:
            with state_lock:
                state["installing"] = False
        if not ok:
            return f"Install failed: {msg}"
        ctx.set_setting("neko_installed", True)
        return (
            f"Install succeeded. {msg} "
            "Give the user the room URL (open_neko_dashboard or this tool's sibling get_neko_room_link) so they can open it in Safari/Chrome on the Mac and log in. "
            f"{_NEKO_AGENT_ALREADY_IN_SESSION}"
        )

    @tool
    def get_neko_room_link(room_name: str | None = None) -> str:
        """Return the room URL for the human to paste into their Mac browser (login / view the shared desktop). This is NOT for neko_navigate—the agent is already on that Chromium via CDP. room_name is ignored in single-room setups."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko or open the Neko dashboard and click 'Install Neko'."
        base = _base_url()
        default_room = (ctx.get_setting("default_room_name", "") or "").strip()
        if room_name and room_name.strip():
            # If we had neko-rooms API we could resolve room slug to URL
            url = f"{base}/room/{room_name.strip()}" if "/room/" not in base else base
        else:
            url = base
        stream_id = state.get("active_stream_id")
        if stream_id:
            ctx.push_stream_update(
                stream_id,
                {"status": "ready", "message": f"Room: {url}", "metrics": {"room_url": url}},
            )
        return f"Room URL for the user's Mac browser: {url}. {_NEKO_AGENT_ALREADY_IN_SESSION}"

    @tool
    def open_url_in_neko(url: str, room_name: str | None = None) -> str:
        """Ask the user to open `url` in the shared browser (they can type it in the address bar), or use neko_navigate(url) yourself with a normal https URL. Never use neko_navigate with the room link—the user already opened that on the Mac; CDP is your connection."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."
        room_link = _base_url()
        return (
            f"User: with the room open ({room_link}), go to: {url}. "
            f"Agent: if that is a normal https URL, use neko_navigate with it—not the room URL. {_NEKO_AGENT_ALREADY_IN_SESSION}"
        )

    @tool
    def neko_browser_snapshot() -> str:
        """Read-only: list each open Chromium tab's URL and title (CDP). Use when the user asks what website is showing, what's loaded, or to inspect the current page—do NOT use neko_navigate or about:blank for that; neko_navigate closes/replaces tabs and will destroy what they had open."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."
        cdp_url = (ctx.get_setting("neko_cdp_url", "") or "http://localhost:9222").strip()
        ok, browser_ws = _cdp_get_browser_ws(cdp_url)
        if not ok:
            return f"Failed: {browser_ws}"
        ok2, text = _neko_cdp_page_snapshot_lines(browser_ws)
        if not ok2:
            return text
        return f"Current Chromium tabs (read-only, session unchanged):\n{text}"

    @tool
    def neko_page_visible_text(max_chars: int = 40000) -> str:
        """Read visible text from the active Chromium tab (document.body.innerText via CDP). Use to summarize or quote what the user sees in the shared Neko browser. Do not use read_webpage on the tab URL as a substitute—that is a separate HTTP fetch and can differ (cookies, paywall, SSR vs hydrated DOM). Call neko_browser_snapshot first if you need the exact URL. max_chars is capped 1000–120000 (default 40000)."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."
        cap = 40000
        try:
            cap = int(max_chars)
        except (TypeError, ValueError):
            cap = 40000
        cap = max(1000, min(cap, 120000))
        okp, packed = _neko_cdp_triple()
        if not okp:
            return f"Failed: {packed}"
        cdp_url, browser_ws, tid = packed
        expr = (
            "(function(){try{var cap="
            + json.dumps(cap)
            + ";var raw=document.body?(document.body.innerText||''):'';"
            "var truncated=raw.length>cap;var t=truncated?raw.slice(0,cap):raw;"
            "return{ok:true,text:t,truncated:truncated,approx_len:raw.length};"
            "}catch(e){return{ok:false,err:String(e)};}})()"
        )
        steps: list[tuple[str, dict]] = [
            ("Runtime.enable", {}),
            ("Runtime.evaluate", {"expression": expr, "returnByValue": True, "awaitPromise": True}),
        ]
        ok3, parts = _cdp_run_page_or_session(cdp_url, browser_ws, tid, steps, timeout=45.0)
        if not ok3:
            return f"Failed: {parts}"
        val = _neko_cdp_eval_object_value(parts[1])
        if not isinstance(val, dict) or not val.get("ok"):
            err = (val or {}).get("err") if isinstance(val, dict) else "evaluate failed"
            return f"Failed: {err}"
        text = val.get("text")
        if not isinstance(text, str):
            return "Failed: no text from page."
        truncated = bool(val.get("truncated"))
        approx = val.get("approx_len")
        note = ""
        if truncated:
            note = f"\n\n[Truncated to {cap} characters; page innerText length ≈ {approx}.]"
        return f"Visible text from the active Neko Chromium tab:\n\n{text.strip()}{note}"

    @tool
    def neko_navigate(url: str) -> str:
        """Open a URL by creating a new tab and closing an existing page tab if one exists—this REPLACES what is on screen. Do not use this to learn what page is open (use neko_browser_snapshot). Do not use about:blank to \"check\" state; it wipes the user's tab.

You already control the same Chromium the user sees (CDP). Do not open the room link or localhost:8080 here (nested Neko; matches NEKO_DEFAULT_PORT). Use external https:// URLs or about:blank only when intentionally clearing/navigating. Requires Chromium + CDP."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."
        base = _base_url()
        if neko_navigate_would_self_embed(url, base):
            return (
                f"Refused: {url!r} is the Neko room web UI. The user already uses that URL on their Mac to see this desktop; "
                "you are already driving this Chromium via CDP—there is nothing to 'connect' by navigating here. "
                "Use an external site (https://…) or about:blank."
            )
        nav_url = _with_http_scheme_if_missing(url)
        cdp_url = (ctx.get_setting("neko_cdp_url", "") or "http://localhost:9222").strip()
        ok, browser_ws = _cdp_get_browser_ws(cdp_url)
        if not ok:
            return f"Failed: {browser_ws}"
        # Close existing page and open a new tab at the URL via browser-level CDP
        ok_t, targets = _cdp_ws_command(browser_ws, "Target.getTargets")
        if ok_t:
            pages = _neko_page_targets_sorted(targets)
            if pages:
                _cdp_ws_command(browser_ws, "Target.closeTarget", {"targetId": pages[0]["targetId"]})
        ok2, result = _cdp_ws_command(browser_ws, "Target.createTarget", {"url": nav_url})
        if not ok2:
            return f"Failed: {result}"
        return f"Navigated to {nav_url}"

    def _neko_cdp_triple() -> tuple[bool, tuple[str, str, str] | str]:
        """Return (True, (cdp_url, browser_ws, page_target_id)) or (False, error)."""
        cdp_url = (ctx.get_setting("neko_cdp_url", "") or "http://localhost:9222").strip()
        ok, browser_ws = _cdp_get_browser_ws(cdp_url)
        if not ok:
            return False, browser_ws
        ok2, tid = _neko_pick_page_target_id(browser_ws)
        if not ok2:
            return False, tid if isinstance(tid, str) else str(tid)
        return True, (cdp_url, browser_ws, tid)

    @tool
    def neko_click_selector(css_selector: str = "", selector: str = "") -> str:
        """Click the center of the first element matching a CSS selector via CDP (preferred for buttons, links, inputs). Use argument **css_selector** (alias **selector**). Examples: `main a`, `button[type='submit']`, `#search`. Use neko_browser_snapshot first if you need the current URL; do not use neko_navigate to \"inspect\" the page."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."
        sel = (css_selector or "").strip() or (selector or "").strip()
        if not sel:
            return "Provide a non-empty css_selector (or alias selector)."
        okp, packed = _neko_cdp_triple()
        if not okp:
            return f"Failed: {packed}"
        cdp_url, browser_ws, tid = packed
        expr = (
            "(function(){const sel="
            + json.dumps(sel)
            + """;const el=document.querySelector(sel);
if(!el)return{ok:false,err:'no element matches selector'};
el.scrollIntoView({block:'center',inline:'center',behavior:'instant'});
const r=el.getBoundingClientRect();
if(r.width<1&&r.height<1)return{ok:false,err:'element has no visible box'};
return{ok:true,x:r.left+r.width/2,y:r.top+r.height/2};})()"""
        )
        steps_a: list[tuple[str, dict]] = [
            ("Runtime.enable", {}),
            ("Input.enable", {}),
            ("Runtime.evaluate", {"expression": expr, "returnByValue": True, "awaitPromise": True}),
        ]
        ok3, parts = _cdp_run_page_or_session(cdp_url, browser_ws, tid, steps_a, timeout=30.0)
        if not ok3:
            return f"Failed: {parts}"
        coords = _neko_cdp_eval_object_value(parts[2])
        if not isinstance(coords, dict) or not coords.get("ok"):
            err = (coords or {}).get("err") if isinstance(coords, dict) else "bad evaluate result"
            return f"Failed: {err}"
        try:
            x = float(coords["x"])
            y = float(coords["y"])
        except (KeyError, TypeError, ValueError):
            return "Failed: could not parse click coordinates from page."
        steps_b: list[tuple[str, dict]] = [
            ("Runtime.enable", {}),
            ("Input.enable", {}),
            ("Input.dispatchMouseEvent", {"type": "mouseMoved", "x": x, "y": y, "buttons": 0}),
            (
                "Input.dispatchMouseEvent",
                {
                    "type": "mousePressed",
                    "x": x,
                    "y": y,
                    "button": "left",
                    "buttons": 1,
                    "clickCount": 1,
                },
            ),
            (
                "Input.dispatchMouseEvent",
                {
                    "type": "mouseReleased",
                    "x": x,
                    "y": y,
                    "button": "left",
                    "buttons": 0,
                    "clickCount": 1,
                },
            ),
        ]
        ok4, _parts2 = _cdp_run_page_or_session(cdp_url, browser_ws, tid, steps_b, timeout=25.0)
        if not ok4:
            return f"Failed: {_parts2}"
        return f"Clicked element matching {sel!r} at ({x:.1f}, {y:.1f})."

    @tool
    def neko_fill_field(css_selector: str = "", text: str = "", clear_first: bool = True, selector: str = "") -> str:
        """Focus an input/textarea (or contenteditable), optionally clear it, then type text via CDP Input.insertText—use for forms. Use **css_selector** (alias **selector**). After filling, use neko_press_key with Enter if you need to submit."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."
        sel = (css_selector or "").strip() or (selector or "").strip()
        if not sel:
            return "Provide a non-empty css_selector (or alias selector)."
        payload = text if text is not None else ""
        okp, packed = _neko_cdp_triple()
        if not okp:
            return f"Failed: {packed}"
        cdp_url, browser_ws, tid = packed
        expr = (
            "(function(){const sel="
            + json.dumps(sel)
            + ";const clear="
            + json.dumps(bool(clear_first))
            + """;const el=document.querySelector(sel);
if(!el)return{ok:false,err:'no element matches selector'};
el.scrollIntoView({block:'center',inline:'center',behavior:'instant'});
el.focus();
if(clear&&(el.tagName==='INPUT'||el.tagName==='TEXTAREA'))el.value='';
else if(clear&&el.isContentEditable)el.textContent='';
return{ok:true};})()"""
        )
        steps: list[tuple[str, dict]] = [
            ("Runtime.enable", {}),
            ("Input.enable", {}),
            ("Runtime.evaluate", {"expression": expr, "returnByValue": True, "awaitPromise": True}),
            ("Input.insertText", {"text": payload}),
        ]
        ok3, parts = _cdp_run_page_or_session(cdp_url, browser_ws, tid, steps, timeout=45.0)
        if not ok3:
            return f"Failed: {parts}"
        focus_res = _neko_cdp_eval_object_value(parts[2])
        if not isinstance(focus_res, dict) or not focus_res.get("ok"):
            err = (focus_res or {}).get("err") if isinstance(focus_res, dict) else "focus failed"
            return f"Failed: {err}"
        return f"Filled {sel!r} ({len(payload)} characters)."

    @tool
    def neko_insert_text(text: str) -> str:
        """Insert Unicode text into whatever is already focused in Chromium (CDP). Use after neko_click_selector on an input or contenteditable; do not use for opening URLs."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."
        payload = text if text is not None else ""
        okp, packed = _neko_cdp_triple()
        if not okp:
            return f"Failed: {packed}"
        cdp_url, browser_ws, tid = packed
        steps: list[tuple[str, dict]] = [
            ("Runtime.enable", {}),
            ("Input.enable", {}),
            ("Input.insertText", {"text": payload}),
        ]
        ok3, _parts = _cdp_run_page_or_session(cdp_url, browser_ws, tid, steps, timeout=30.0)
        if not ok3:
            return f"Failed: {_parts}"
        return f"Inserted {len(payload)} characters at focus."

    @tool
    def neko_press_key(key: str) -> str:
        """Send a key to the focused element (CDP). Allowed key names: Enter, Tab, Escape, Backspace—e.g. submit a form after neko_fill_field with Enter."""
        allowed = {
            "Enter": (13, "Enter", "Enter"),
            "Tab": (9, "Tab", "Tab"),
            "Escape": (27, "Escape", "Escape"),
            "Backspace": (8, "Backspace", "Backspace"),
        }
        name = (key or "").strip()
        if name not in allowed:
            return f"Unsupported key {name!r}. Use one of: {', '.join(sorted(allowed))}."
        vk, k, code = allowed[name]
        okp, packed = _neko_cdp_triple()
        if not okp:
            return f"Failed: {packed}"
        cdp_url, browser_ws, tid = packed
        down = {
            "type": "keyDown",
            "windowsVirtualKeyCode": vk,
            "nativeVirtualKeyCode": vk,
            "key": k,
            "code": code,
        }
        up = {
            "type": "keyUp",
            "windowsVirtualKeyCode": vk,
            "nativeVirtualKeyCode": vk,
            "key": k,
            "code": code,
        }
        steps: list[tuple[str, dict]] = [
            ("Runtime.enable", {}),
            ("Input.enable", {}),
            ("Input.dispatchKeyEvent", down),
            ("Input.dispatchKeyEvent", up),
        ]
        ok3, _parts = _cdp_run_page_or_session(cdp_url, browser_ws, tid, steps, timeout=20.0)
        if not ok3:
            return f"Failed: {_parts}"
        return f"Pressed {name}."

    @tool
    def neko_wait_for(
        url_equals: str = "",
        url_prefix: str = "",
        selector: str = "",
        text: str = "",
        timeout_seconds: float = _WAIT_TIMEOUT_DEFAULT_S,
    ) -> str:
        """Wait until the shared Chromium tab reaches a target state, then return. Provide EXACTLY ONE of: url_equals (full URL match), url_prefix (starts_with), selector (CSS selector must appear in the DOM), or text (visible-text substring, max 200 chars). timeout_seconds is clamped to [1, 60]; default 20. Returns 'ok:<kind>' on success or a stable failure like 'wait_timeout:selector=<sel>' / 'wait_timeout:url_prefix'. Use after neko_navigate instead of polling neko_browser_snapshot in a loop."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."

        kind, cond = _wait_pick_condition(url_equals, url_prefix, selector, text)
        if kind is None:
            return f"Failed: {cond}. Provide exactly one of url_equals, url_prefix, selector, text."

        ok_t, timeout = _wait_clamp_timeout(timeout_seconds)
        if not ok_t:
            return f"Failed: invalid_timeout. timeout_seconds must be a number between {_WAIT_TIMEOUT_MIN_S:g} and {_WAIT_TIMEOUT_MAX_S:g}."

        cdp_url = (ctx.get_setting("neko_cdp_url", "") or "http://localhost:9222").strip()
        ok_ws, browser_ws = _cdp_get_browser_ws(cdp_url)
        if not ok_ws:
            return f"Failed: cdp_unreachable ({browser_ws})"

        deadline = time.monotonic() + float(timeout)
        delay = _WAIT_POLL_INITIAL_S
        last_err: str = ""
        while True:
            if kind in ("url_equals", "url_prefix"):
                ok_u, current = _wait_url_current(browser_ws)
                if not ok_u:
                    last_err = current
                else:
                    if kind == "url_equals" and current == cond:
                        return "ok:url_equals"
                    if kind == "url_prefix" and current.startswith(cond):
                        return "ok:url_prefix"
            else:
                ok_t2, target_id = _neko_pick_page_target_id(browser_ws)
                if not ok_t2 or not isinstance(target_id, str):
                    last_err = "no_page_target"
                else:
                    expr = _wait_probe_expr(kind, cond)
                    steps: list[tuple[str, dict]] = [
                        ("Runtime.enable", {}),
                        ("Runtime.evaluate", {"expression": expr, "returnByValue": True, "awaitPromise": True}),
                    ]
                    ok_r, parts = _cdp_run_page_or_session(cdp_url, browser_ws, target_id, steps, timeout=10.0)
                    if not ok_r:
                        last_err = "cdp_unreachable"
                    else:
                        val = _neko_cdp_eval_object_value(parts[1])
                        if isinstance(val, dict) and val.get("ok") and val.get("found"):
                            return f"ok:{kind}"

            now = time.monotonic()
            if now >= deadline:
                return _wait_timeout_tag(kind, cond)
            sleep_for = _wait_next_delay(delay, deadline, now)
            if sleep_for <= 0:
                return _wait_timeout_tag(kind, cond)
            time.sleep(sleep_for)
            delay = min(_WAIT_POLL_MAX_S, delay * _WAIT_POLL_BACKOFF)

    @tool
    def neko_click(x: int, y: int) -> str:
        """Low-level: click pixel (x,y) via xdotool on the Neko display (1280x720). Prefer neko_click_selector(css) for real pages so you hit buttons and fields reliably."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."
        try:
            from tools.searxng_docker import _run_docker
        except ImportError:
            return "Failed: Docker helpers not available."
        ok, msg = _run_docker(["exec", "-e", "DISPLAY=:99.0", "mugi-neko", "xdotool", "mousemove", "--", str(x), str(y)], timeout=5)
        if not ok:
            return f"Failed to move mouse: {msg}"
        ok2, msg = _run_docker(["exec", "-e", "DISPLAY=:99.0", "mugi-neko", "xdotool", "click", "1"], timeout=5)
        if not ok2:
            return f"Failed to click: {msg}"
        return f"Clicked at ({x}, {y})"

    @tool
    def neko_type(text: str) -> str:
        """Legacy: type via xdotool into whatever has X11 focus. Prefer neko_fill_field(selector, text) or neko_click_selector then neko_insert_text for real forms. Do not type the Neko room URL into the address bar."""
        if not _is_installed():
            return "Neko is not installed. Run install_neko first."
        try:
            from tools.searxng_docker import _run_docker
        except ImportError:
            return "Failed: Docker helpers not available."
        ok, msg = _run_docker(["exec", "-e", "DISPLAY=:99.0", "mugi-neko", "xdotool", "type", "--clearmodifiers", "--delay", "50", "--", text], timeout=max(10, len(text)))
        if not ok:
            return f"Failed to type: {msg}"
        return f"Typed: {text}"

    return [
        open_neko_dashboard,
        install_neko,
        get_neko_room_link,
        open_url_in_neko,
        neko_browser_snapshot,
        neko_page_visible_text,
        neko_navigate,
        neko_click_selector,
        neko_fill_field,
        neko_insert_text,
        neko_press_key,
        neko_wait_for,
        neko_click,
        neko_type,
    ]


def register(ctx):
    ctx.log("Neko plugin loading...")
    state_lock = threading.Lock()
    state = {"active_stream_id": None, "room_url": None, "installing": False}

    def _on_control(event, plugin_ctx):
        stream_id = str(event.get("stream_id") or "")
        control_id = str(event.get("control_id") or "")
        value = event.get("value")
        action_id = str(event.get("action_id") or "")

        if control_id == "install_neko" and action_id == "install_neko":
            with state_lock:
                if state.get("installing"):
                    return {
                        "status": "rejected",
                        "message": "Install already in progress — watch the status above.",
                    }
                state["installing"] = True
            try:

                def _prog(phase: str, message: str) -> None:
                    plugin_ctx.push_stream_update(
                        stream_id, _neko_install_stream_payload(phase, message)
                    )

                ok, msg = _do_install(plugin_ctx, progress=_prog)
            finally:
                with state_lock:
                    state["installing"] = False
            if not ok:
                plugin_ctx.push_stream_update(
                    stream_id,
                    {"status": "error", "message": msg, "metrics": {}},
                )
                plugin_ctx.emit_stream_error(stream_id, msg)
                show_rm = _neko_install_panel_show_remove()
                plugin_ctx.register_stream_controls(
                    stream_id, _install_controls(show_remove=show_rm)
                )
                return {"status": "rejected", "message": msg}
            plugin_ctx.set_setting("neko_installed", True)
            base_url = plugin_ctx.get_setting("neko_base_url", "http://localhost:8080") or "http://localhost:8080"
            plugin_ctx.push_stream_update(
                stream_id,
                {
                    "status": "ready",
                    "message": f"Installed. Room link: {base_url}",
                    "metrics": {"room_url": base_url},
                },
            )
            plugin_ctx.register_stream_controls(stream_id, _session_controls(base_url))
            return {"status": "applied", "message": "Neko installed. Use 'Open in browser' or copy the link."}

        if control_id == "open_browser" and action_id == "open_browser":
            raw = plugin_ctx.get_setting("neko_base_url", "http://localhost:8080") or "http://localhost:8080"
            base_url = _safe_url_for_open(raw) or "http://localhost:8080"
            plugin_ctx.push_stream_update(
                stream_id,
                {
                    "status": "ready",
                    "message": f"Open this URL in your browser: {base_url}",
                    "metrics": {"room_url": base_url},
                },
            )
            # open_url in the response tells the app to open this URL in the system browser (sanitized)
            return {"status": "applied", "message": "Opening in browser.", "open_url": base_url}

        if control_id == "copy_link" and action_id == "copy_link":
            raw = plugin_ctx.get_setting("neko_base_url", "http://localhost:8080") or "http://localhost:8080"
            base_url = _safe_url_for_open(raw) or "http://localhost:8080"
            plugin_ctx.push_stream_update(
                stream_id,
                {
                    "status": "ready",
                    "message": f"Copy this link: {base_url}",
                    "metrics": {"room_url": base_url},
                },
            )
            # copy_text in the response tells the app to copy this string to the clipboard (sanitized)
            return {"status": "applied", "message": "Link copied to clipboard.", "copy_text": base_url}

        if control_id == "audio_enabled":
            # Store for future use (e.g. URL params); no-op for now
            return {"status": "applied", "message": "Audio preference updated."}

        if control_id == "remove_neko" and action_id == "remove_neko":
            ok, msg = _do_remove()
            if not ok:
                plugin_ctx.push_stream_update(
                    stream_id,
                    {"status": "error", "message": msg, "metrics": {}},
                )
                plugin_ctx.emit_stream_error(stream_id, msg)
                return {"status": "rejected", "message": msg}
            plugin_ctx.set_setting("neko_installed", False)
            show_rm = _neko_install_panel_show_remove()
            plugin_ctx.register_stream_controls(
                stream_id, _install_controls(show_remove=show_rm)
            )
            # Same text for stream + control_ack — Swift shows [applied] on the ack line and it must not contradict the card.
            panel_msg = msg
            if show_rm and "Nothing to remove" not in msg:
                panel_msg += " You can tap Remove again if something is still running in Docker."
            plugin_ctx.push_stream_update(
                stream_id,
                {
                    "status": "pending_install",
                    "message": panel_msg,
                    "metrics": {},
                },
            )
            return {"status": "applied", "message": panel_msg}

        return {"status": "rejected", "message": f"Unknown control: {control_id}"}

    ctx.register_control_handler(_on_control)
    ctx.register_tools(_make_tools(ctx, state, state_lock))

    ctx.register_settings({
        "name": "Neko",
        "fields": [
            {
                "key": "neko_base_url",
                "type": "string",
                "default": "http://localhost:8080",
                "label": "Neko room URL (use the same host you type in the browser — localhost vs 192.168.x.x changes WebRTC)",
            },
            {"key": "neko_use_firefox", "type": "boolean", "default": False, "label": "Use Firefox instead of Chromium (remove and re-install to apply)"},
            {"key": "neko_cdp_url", "type": "string", "default": "http://localhost:9222", "label": "Chrome DevTools Protocol URL (for agent control; requires CDP-enabled Chromium)"},
            {
                "key": "neko_webrtc_nat1to1",
                "type": "string",
                "default": "",
                "label": "WebRTC NAT 1:1 override (empty: auto from room URL — localhost→127.0.0.1, else LAN; set manually if green button stalls)",
            },
            {"key": "default_room_name", "type": "string", "default": "", "label": "Default room name (for neko-rooms)"},
        ],
    })

    def _route_handler(req: dict) -> dict:
        if req.get("path") == "/open" and req.get("method", "GET").upper() == "GET":
            raw = ctx.get_setting("neko_base_url", "http://localhost:8080") or "http://localhost:8080"
            base = _safe_url_for_open(raw) or "http://localhost:8080"
            return {
                "status": 302,
                "headers": {"Location": base},
                "body": "",
            }
        return {"status": 404, "headers": {"content-type": "text/plain"}, "body": "Not found"}

    ctx.register_route_handler(_route_handler)

    ctx.log("Neko plugin loaded.")
