9a8cdf96bf
- tools/browser_tool_install.py: keep pm-clean's frozen old-updater stub; main's UTF-8 decode fix touched only the npx prefetch body it replaces. - tests/hermes_cli/test_update_scoped_reconciliation.py: keep pm-clean's test subset (catch-up rides the PM completion owner) and take main's gateway-less host evidence (#120740): the updated seed that holds the host at a running gateway, and the two gateway-less matrices for the source change that merged cleanly into update_cmd_fleet.py.
341 lines
13 KiB
Python
341 lines
13 KiB
Python
"""Lightpanda local engine for Browser Use mode.
|
|
|
|
With ``browser.engine: lightpanda``, Browser Use mode spawns one ``lightpanda serve`` per
|
|
browser session and points ``browser_exec`` at its CDP endpoint (``BU_CDP_URL``); the
|
|
built-in ``browser_*`` tools keep going through ``agent-browser --engine lightpanda``.
|
|
``tools.browser_tool`` 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
|
|
|
|
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
|
|
|
|
|
|
def find_lightpanda_binary() -> Optional[str]:
|
|
"""Return the lightpanda executable, or None. Order: PATH (with agent-browser's Homebrew/managed-node
|
|
fallbacks), then installer/agent-browser locations, then ``$HERMES_HOME/bin``. No Windows build."""
|
|
if os.name == "nt":
|
|
logger.debug("Lightpanda has no Windows build")
|
|
return None
|
|
path_env = os.environ.get("PATH", "")
|
|
try:
|
|
from tools.browser_tool_install import _merge_browser_path
|
|
path_env = _merge_browser_path(path_env)
|
|
except Exception as e:
|
|
logger.debug("browser PATH merge unavailable: %s", e)
|
|
return shutil.which("lightpanda", path=path_env) 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,
|
|
)
|
|
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:
|
|
try:
|
|
from tools.browser_tool import _build_browser_env
|
|
return _build_browser_env()
|
|
except Exception as e:
|
|
logger.debug("credential-scrubbed browser env unavailable: %s", e)
|
|
return os.environ.copy()
|
|
|
|
|
|
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; shared with the real-profile Chrome cleanup."""
|
|
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:
|
|
(_state_dir() / f"{server.session_name}.json").write_text(json.dumps(record), encoding="utf-8")
|
|
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,
|
|
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
|