browser-cli/browser_cli/client.py

"""
Unix socket client — sends commands to the native host relay socket.
Used by both the CLI and the public Python API.

Profile selection order:
  1. Explicit `profile` argument to send_command()
  2. BROWSER_CLI_PROFILE environment variable
  3. First entry in /tmp/.browser_cli/registry.json
  4. Otherwise, no browser can be resolved automatically
"""
import json
import os
import socket
import struct
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any

SOCKET_DIR = Path("/tmp/.browser_cli")
REGISTRY_PATH = SOCKET_DIR / "registry.json"


class BrowserNotConnected(Exception):
    """Raised when the native host socket is not available."""


@dataclass(frozen=True)
class BrowserTarget:
    profile: str
    display_name: str
    socket_path: str


def _active_sockets(reg: dict) -> dict:
    """Return only entries whose socket file exists on disk."""
    return {k: v for k, v in reg.items() if Path(v).exists()}


def display_browser_name(profile_name: str, sock_path: str) -> str:
    if profile_name != "default":
        return profile_name
    return Path(sock_path).stem or profile_name


def active_browser_targets() -> list[BrowserTarget]:
    if not REGISTRY_PATH.exists():
        return []
    try:
        reg = json.loads(REGISTRY_PATH.read_text())
    except Exception:
        return []
    return [
        BrowserTarget(profile=profile, display_name=display_browser_name(profile, sock_path), socket_path=sock_path)
        for profile, sock_path in _active_sockets(reg).items()
    ]


def _resolve_socket(profile: str | None = None) -> str:
    """Return the socket path for the given profile (or auto-detect)."""
    target = profile or os.environ.get("BROWSER_CLI_PROFILE")

    if target:
        if REGISTRY_PATH.exists():
            try:
                reg = json.loads(REGISTRY_PATH.read_text())
                if target in reg:
                    return reg[target]
            except Exception:
                pass
        safe = target.replace(" ", "_").replace("/", "_")
        return str(SOCKET_DIR / f"{safe}.sock")

    # Auto-detect: error when multiple browser instances are active
    try:
        active = active_browser_targets()
        if len(active) > 1:
            aliases = [target.profile for target in active]
            examples = "\n".join(f"  browser-cli --browser {a} ..." for a in aliases)
            raise BrowserNotConnected(
                f"Multiple browser instances are active: {', '.join(aliases)}\n"
                f"Use --browser <alias> to select one:\n{examples}"
            )
        if active:
            return active[0].socket_path
    except BrowserNotConnected:
        raise
    except Exception:
        pass

    raise BrowserNotConnected(
        "Cannot resolve a browser socket automatically.\n"
        "Make sure the browser is running with the browser-cli extension enabled,\n"
        "or pass --browser <alias> / set BROWSER_CLI_PROFILE to a known alias."
    )


def send_command(command: str, args: dict | None = None, profile: str | None = None) -> Any:
    """Send a command to the browser and return the response data."""
    sock_path = _resolve_socket(profile)
    msg = {
        "id": str(uuid.uuid4()),
        "command": command,
        "args": args or {},
    }
    payload = json.dumps(msg).encode("utf-8")
    framed = struct.pack("<I", len(payload)) + payload

    try:
        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
            sock.connect(sock_path)
            sock.sendall(framed)
            response = _recv_all(sock)
    except (FileNotFoundError, ConnectionRefusedError, OSError):
        profile_hint = f" (profile: {profile})" if profile else ""
        raise BrowserNotConnected(
            f"Cannot connect to browser{profile_hint}.\n"
            "Make sure:\n"
            "  1. The browser-cli extension is installed and enabled\n"
            "  2. The native host is registered: uv run browser-cli install chrome\n"
            "  3. Your browser is running\n"
            "  Tip: use BROWSER_CLI_PROFILE=<name> to select a specific profile"
        )

    result = json.loads(response)
    if not result.get("success", True):
        raise RuntimeError(result.get("error", "unknown error from browser"))
    return result.get("data")


def _recv_all(sock: socket.socket) -> bytes:
    raw_len = _recv_exact(sock, 4)
    msg_len = struct.unpack("<I", raw_len)[0]
    return _recv_exact(sock, msg_len)


def _recv_exact(sock: socket.socket, n: int) -> bytes:
    buf = b""
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:
            raise ConnectionError("Socket closed before full message received")
        buf += chunk
    return buf