hermes-agent/tools/browser_lightpanda.py

"""Lightpanda local engine.

With ``browser.engine: lightpanda``, each browser session spawns one ``lightpanda serve`` and
``browser_exec`` drives its CDP endpoint (``BU_CDP_URL``). ``tools.browser_session`` owns the
session cache, inactivity reaper and atexit sweep.
"""

import functools
import json
import logging
import os
import shutil
import socket
import subprocess
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Optional, Tuple

from utils import atomic_json_write

logger = logging.getLogger(__name__)

LIGHTPANDA_INSTALL_URL = "https://lightpanda.io/docs/run-locally/installation/one-liner"
LIGHTPANDA_INSTALL_HINT = f"Install Lightpanda from {LIGHTPANDA_INSTALL_URL} and make sure " "`lightpanda` is on PATH"

_READY_TIMEOUT_S = 10.0
_POLL_INTERVAL_S = 0.1
_STDERR_TAIL_LIMIT = 2000

_servers: "dict[str, LightpandaServer]" = {}
_servers_lock = threading.Lock()


@dataclass
class LightpandaServer:
    session_name: str
    port: int
    proc: subprocess.Popen
    log_path: str
    start_time: Optional[int] = None

    @property
    def cdp_url(self) -> str:
        return f"http://127.0.0.1:{self.port}"  # HTTP discovery URL; the harness resolves /json/version itself

    def is_alive(self) -> bool:
        return self.proc.poll() is None


def _home_candidates() -> list:
    home = Path.home()
    candidates = [home / ".lightpanda" / "lightpanda", home / ".local" / "bin" / "lightpanda"]
    try:
        from hermes_constants import get_hermes_home
        candidates.append(Path(get_hermes_home()) / "bin" / "lightpanda")
    except Exception as e:  # pragma: no cover - defensive
        logger.debug("hermes home unavailable for lightpanda lookup: %s", e)
    return candidates


# System PATH fallbacks for a minimal service PATH (systemd units, launchd), Homebrew included.
_SANE_PATH_DIRS = ("/opt/homebrew/bin", "/opt/homebrew/sbin", "/usr/local/sbin", "/usr/local/bin",
                   "/usr/sbin", "/usr/bin", "/sbin", "/bin")


def merge_sane_path(existing_path: str = "") -> str:
    """``existing_path`` with the standard system dirs that exist appended (existing order kept)."""
    parts = [p for p in (existing_path or "").split(os.pathsep) if p]
    return os.pathsep.join(parts + [d for d in _SANE_PATH_DIRS if d not in parts and os.path.isdir(d)])


def find_lightpanda_binary() -> Optional[str]:
    """Return the lightpanda executable, or None. Order: PATH (plus the standard system dirs), then the
    installer's locations, then ``$HERMES_HOME/bin``. No Windows build."""
    if os.name == "nt":
        logger.debug("Lightpanda has no Windows build")
        return None
    return shutil.which("lightpanda", path=merge_sane_path(os.environ.get("PATH", ""))) or next(
        (str(c) for c in _home_candidates() if c.is_file() and os.access(c, os.X_OK)), None)


def _pick_free_loopback_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind(("127.0.0.1", 0))
        return int(sock.getsockname()[1])


def _state_dir() -> Path:
    from hermes_constants import get_hermes_home
    path = Path(get_hermes_home()) / "cache" / "browser-use" / "lightpanda"
    path.mkdir(parents=True, exist_ok=True)
    return path


def _http_cache_dir() -> Path:
    """Filesystem HTTP cache shared by every Lightpanda this Hermes spawns.

    Shared rather than per-session so a cached asset survives session churn.
    Lightpanda holds it in sqlite (WAL) with a best-effort write path, and
    ``--http-cache-entry-limit`` (upstream default 1000, not passed here)
    bounds it without Hermes managing eviction.
    """
    path = _state_dir() / "http-cache"
    path.mkdir(parents=True, exist_ok=True)
    return path


_HTTP_CACHE_FLAG = "--http-cache-dir"


@functools.lru_cache(maxsize=1)
def _binary_supports_http_cache(binary: str) -> bool:
    """True if ``lightpanda serve`` accepts ``--http-cache-dir``.

    The flag landed upstream in 0.3.x; older binaries fatally reject it
    ("unknown argument"), which would break every launch. Probing ``help``
    output keeps working across future flag additions without parsing
    versions, and the lru_cache keeps it once per binary per process.
    """
    try:
        proc = subprocess.run(
            [binary, "help"],
            capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=3.0,
            stdin=subprocess.DEVNULL,
            check=False,
        )
        return _HTTP_CACHE_FLAG in ((proc.stdout or "") + (proc.stderr or ""))
    except Exception as e:
        logger.debug("lightpanda http-cache probe failed (%s); assuming no", e)
        return False


def _record_path(session_name: str) -> Path:
    return _state_dir() / f"{session_name}.json"


def _browser_env() -> dict:
    from tools.browser_session import build_browser_env
    return build_browser_env()


def _cdp_ready(url: str, timeout: float = 0.2) -> bool:
    try:
        from hermes_cli.browser_connect import is_browser_debug_ready
        return is_browser_debug_ready(url, timeout=timeout)
    except Exception as e:
        logger.debug("CDP readiness probe failed for %s: %s", url, e)
        return False


def _read_log_tail(path: str) -> str:
    """Last non-empty line of the child's stderr log ('' when unreadable)."""
    try:
        text = Path(path).read_bytes()[-_STDERR_TAIL_LIMIT:].decode("utf-8", errors="replace")
    except OSError:
        return ""
    return next((ln for ln in reversed(text.strip().splitlines()) if ln.strip()), "")


def _terminate(proc: subprocess.Popen, what: str = "lightpanda") -> None:
    """poll -> terminate -> wait(5) -> kill (fallback when the tree-kill failed)."""
    try:
        if proc.poll() is None:
            proc.terminate()
            try:
                proc.wait(timeout=5)
            except Exception:
                proc.kill()
    except Exception as e:
        logger.debug("%s terminate failed: %s", what, e)


def _safe_start_time(pid: int) -> Optional[int]:
    try:
        from tools.process_registry import ProcessRegistry
        return ProcessRegistry._safe_host_start_time(pid)
    except Exception:
        return None


def _tree_kill(pid: int, expected_start) -> None:
    """Tree-kill ``pid`` via ProcessRegistry, verifying its start time first."""
    from tools.process_registry import ProcessRegistry
    ProcessRegistry._terminate_host_pid(pid, expected_start=expected_start)


def _write_record(server: LightpandaServer) -> None:
    record = {"pid": server.proc.pid, "port": server.port, "owner_pid": os.getpid(),
              "start_time": server.start_time, "started_at": time.time()}
    try:
        # Temp + rename: an orphan sweep (another Hermes, or a cleanup thread) unlinks a record it
        # cannot parse, so a torn in-place write would drop a live server's record.
        atomic_json_write(_state_dir() / f"{server.session_name}.json", record, indent=None)
    except OSError as e:
        logger.debug("could not write lightpanda record for %s: %s", server.session_name, e)


def launch_lightpanda(session_name: str, *, block_private_networks: bool = False) -> tuple[Optional[LightpandaServer], Optional[str]]:
    """Start ``lightpanda serve`` on a free loopback port; ``(server, None)`` once ``/json/version`` answers,
    else ``(None, error)``. stderr goes to ``<state_dir>/<session>.log`` so a chatty child never blocks on a pipe."""
    binary = find_lightpanda_binary()
    if not binary:
        if os.name == "nt":
            return None, ("browser.engine is 'lightpanda' but Lightpanda has no Windows "
                          "build. Set browser.engine to auto (or run Hermes under WSL2).")
        return None, ("browser.engine is 'lightpanda' but no lightpanda binary was found on PATH, ~/.lightpanda "
                      f"or ~/.local/bin. {LIGHTPANDA_INSTALL_HINT}, or set browser.engine to auto.")

    port = _pick_free_loopback_port()
    argv = [binary, "serve", "--host", "127.0.0.1", "--port", str(port)]
    if _binary_supports_http_cache(binary):
        argv += [_HTTP_CACHE_FLAG, str(_http_cache_dir())]
    if block_private_networks:
        argv.append("--block-private-networks")
    log_path = str(_state_dir() / f"{session_name}.log")
    try:
        with open(log_path, "wb") as log_file:
            proc = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                                    # detached-spawn: ok — a crashed owner's server is reaped from its state record by the next Hermes (reap_orphaned_lightpanda); owning it is a follow-up with the launcher's Chromium
                                    stderr=log_file, env=_browser_env(), start_new_session=True)
    except (OSError, subprocess.SubprocessError) as e:
        return None, f"Failed to launch lightpanda serve ({binary}): {e}"

    url = f"http://127.0.0.1:{port}"
    deadline = time.monotonic() + _READY_TIMEOUT_S
    while True:
        rc = proc.poll()
        if rc is not None:
            tail = _read_log_tail(log_path)
            return None, f"lightpanda serve exited with code {rc} before {url}/json/version answered{f': {tail}' if tail else ''}"
        if _cdp_ready(url):
            break
        if time.monotonic() >= deadline:
            _terminate(proc)
            tail = _read_log_tail(log_path)
            detail = f" (last stderr line: {tail})" if tail else ""
            return None, f"lightpanda serve did not expose {url}/json/version within {int(_READY_TIMEOUT_S)}s{detail}"
        time.sleep(_POLL_INTERVAL_S)

    server = LightpandaServer(session_name, port, proc, log_path, _safe_start_time(proc.pid))
    _write_record(server)
    with _servers_lock:
        _servers[session_name] = server
    logger.info("Started lightpanda serve (pid %s, port %s) for session %s", proc.pid, port, session_name)
    return server, None


def get_server(session_name: str) -> Optional[LightpandaServer]:
    with _servers_lock:
        return _servers.get(session_name)


def stop_lightpanda(session_name: str) -> None:
    """Stop the server for ``session_name`` (tree-kill, plain terminate on failure) and drop its record."""
    with _servers_lock:
        server = _servers.pop(session_name, None)
    if server is not None and server.is_alive():
        try:
            _tree_kill(server.proc.pid, server.start_time)
        except Exception as e:
            logger.debug("lightpanda tree-kill failed for %s: %s", server.session_name, e)
            _terminate(server.proc)
        try:
            server.proc.wait(timeout=5)
        except Exception:
            _terminate(server.proc)
    try:
        (_state_dir() / f"{session_name}.json").unlink(missing_ok=True)
    except OSError as e:
        logger.debug("could not remove lightpanda record for %s: %s", session_name, e)
    if server is not None:
        logger.debug("Stopped lightpanda serve for session %s", session_name)


def stop_all_lightpanda() -> None:
    """Stop every server this process started. Idempotent; safe from atexit."""
    with _servers_lock:
        names = list(_servers)
    for name in names:
        try:
            stop_lightpanda(name)
        except Exception as e:
            logger.debug("lightpanda stop failed for %s: %s", name, e)


def _is_lightpanda_process(pid: int, port, start_time) -> bool:
    """True only when ``pid`` is verifiably the ``lightpanda serve`` we recorded."""
    try:
        import psutil
        proc = psutil.Process(pid)
        if "lightpanda" not in proc.name().lower():
            return False
        cmdline = proc.cmdline()
        if "serve" not in cmdline or str(port) not in cmdline:
            return False
        if start_time:
            from gateway.status import get_process_start_time
            return get_process_start_time(pid) == start_time
    except Exception:
        return False
    return True


def reap_orphaned_lightpanda() -> int:
    """Kill ``lightpanda serve`` processes whose owning Hermes is gone; return the count. A live owner is
    never touched; a PID is only signalled after psutil confirms it is still ``lightpanda serve`` on the recorded port."""
    try:
        state_dir = _state_dir()
    except Exception as e:
        logger.debug("lightpanda state dir unavailable: %s", e)
        return 0
    try:
        from gateway.status import _pid_exists
    except Exception:  # pragma: no cover - defensive
        return 0

    reaped = 0
    for record_path in sorted(state_dir.glob("*.json")):
        session_name = record_path.stem
        try:
            record = json.loads(record_path.read_text(encoding="utf-8-sig"))
            if not isinstance(record, dict):
                raise ValueError(f"expected a JSON object, got {type(record).__name__}")
        except (OSError, ValueError):
            record_path.unlink(missing_ok=True)
            continue
        owner_pid = record.get("owner_pid")
        if owner_pid == os.getpid():
            with _servers_lock:
                if session_name in _servers:
                    continue
        elif owner_pid and _pid_exists(int(owner_pid)):
            continue
        pid = record.get("pid")
        if pid and _is_lightpanda_process(int(pid), record.get("port"), record.get("start_time")):
            try:
                _tree_kill(int(pid), record.get("start_time"))
                reaped += 1
                logger.info("Reaped orphaned lightpanda serve pid %s (session %s)", pid, session_name)
            except Exception as e:
                logger.debug("orphan lightpanda kill failed for pid %s: %s", pid, e)
        record_path.unlink(missing_ok=True)
    return reaped


def using_lightpanda_engine() -> bool:
    """True when ``browser.engine`` is ``lightpanda``."""
    from tools.browser_session_cloud import get_browser_engine
    return get_browser_engine() == "lightpanda"


def lightpanda_engine_status() -> tuple[bool, str]:
    """Whether ``browser.engine: lightpanda`` is actually in effect, and why.

    ``(False, "")`` when the engine isn't lightpanda; else the reason names the setting shadowing it, in
    session-creation precedence, with config-only gates (no network I/O) for ``/browser status`` and
    ``hermes doctor``."""
    if not using_lightpanda_engine():
        return False, ""
    from tools.browser_session_cdp import get_cdp_override_raw
    from tools.browser_session_cloud import get_cloud_provider, use_real_profile
    if get_cdp_override_raw():
        return False, "a CDP override is active (/browser connect or browser.cdp_url)"
    # Real profile before the cloud provider: browser_exec resolves it first, so it claims the session.
    if use_real_profile():
        return False, "browser.use_real_profile is on (Lightpanda cannot load a Chromium profile)"
    try:
        provider = get_cloud_provider()
    except ValueError:  # an unknown explicit provider name: browser_exec reports it on use
        provider = None
    if provider is not None:
        name = getattr(provider, "display_name", "") or type(provider).__name__
        return False, f"cloud provider {name} is selected (browser.cloud_provider, or auto-detected from credentials)"
    from tools.browser_use_cli import _read_browser_cfg, is_legacy_browser_use_cloud_config
    if is_legacy_browser_use_cloud_config(_read_browser_cfg()):
        return False, "Browser Use cloud (BROWSER_USE_API_KEY) is selected"
    return True, "Hermes spawns `lightpanda serve` per browser session"