diff --git a/tools/browser_tool.py b/tools/browser_tool.py index ef8a199070..98bc2cd8df 100644 --- a/tools/browser_tool.py +++ b/tools/browser_tool.py @@ -1,21 +1,15 @@ #!/usr/bin/env python3 """Browser automation tools driven by the agent-browser CLI. -Backends — local headless Chromium (default; ``agent-browser install -[--with-deps]`` one-time setup), Browser Use / Browserbase / Firecrawl cloud -(auto-detected from config + credentials), a user-supplied CDP endpoint, or -Camofox — share one agent-facing behaviour: per-task sessions, text snapshots -of the accessibility tree with ``@eN`` element refs, and automatic cleanup. - -Env: BROWSERBASE_API_KEY / BROWSERBASE_PROJECT_ID / BROWSER_USE_API_KEY select -direct cloud credentials; BROWSERBASE_PROXIES (default "true"), -BROWSERBASE_ADVANCED_STEALTH ("false", Scale plan), BROWSERBASE_KEEP_ALIVE -("true", paid plan) and BROWSERBASE_SESSION_TIMEOUT (seconds, max 21600) tune -Browserbase sessions. Behavioural settings live under ``browser.*`` in config.yaml. - -Sibling modules hold extracted clusters (eval policy, lightpanda fallback, -real-profile CDP, snapshot store); their names are re-imported here so -``patch("tools.browser_tool.X")`` keeps working. +Backends — local headless Chromium (``agent-browser install [--with-deps]``), +Browser Use / Browserbase / Firecrawl cloud (auto-detected from config + +credentials), a user-supplied CDP endpoint, or Camofox — share one agent-facing +behaviour: per-task sessions, accessibility-tree snapshots with ``@eN`` refs, +automatic cleanup. Cloud credentials come from BROWSERBASE_API_KEY / +BROWSERBASE_PROJECT_ID / BROWSER_USE_API_KEY; behavioural settings live under +``browser.*`` in config.yaml. Sibling ``browser_tool_*`` modules hold extracted +clusters; their names are re-imported here so ``patch("tools.browser_tool.X")`` +keeps working. """ import atexit @@ -44,11 +38,8 @@ from hermes_cli._subprocess_compat import windows_hide_flags # noqa: F401 (tes def __getattr__(name: str): - """Lazy module attributes (PEP 562): ``requests`` and ``call_llm`` load on first use. - - First access binds the real object into module globals so the test-patch - surface (``patch("tools.browser_tool.requests.get")`` / ``.call_llm``) works. - """ + """PEP 562 lazy attributes: ``requests`` / ``call_llm`` load on first use and are + bound into module globals so ``patch("tools.browser_tool.requests.get")`` works.""" if name == "requests": import requests as _requests @@ -62,19 +53,9 @@ def __getattr__(name: str): raise AttributeError(f"module {__name__!r} has no attribute {name!r}") -def _lazy_call_llm(*args, **kwargs): - """Invoke ``call_llm`` through module globals so test patches of - ``tools.browser_tool.call_llm`` are honored, importing lazily otherwise.""" - fn = globals().get("call_llm") - if fn is None: - fn = __getattr__("call_llm") - return fn(*args, **kwargs) - - -# Keys re-added to the agent-browser subprocess env AFTER credential stripping. +# Env keys re-added to the agent-browser subprocess AFTER credential stripping. # agent-browser is a Node process loading npm deps: a compromised transitive -# dependency could read every Hermes secret from process.env, so only the -# browser-backend keys the worker legitimately needs pass through. +# dependency could read every Hermes secret from process.env. _BROWSER_PASSTHROUGH_KEYS: tuple[str, ...] = ( "BROWSERBASE_API_KEY", "BROWSERBASE_PROJECT_ID", @@ -86,11 +67,8 @@ _BROWSER_PASSTHROUGH_KEYS: tuple[str, ...] = ( def _build_browser_env() -> dict: - """Credential-scrubbed env for an agent-browser subprocess (only browser-backend keys re-added). - - The ``hermes_subprocess_env`` import is deferred so the module imports under - test harnesses that stub the ``tools`` package. - """ + """Credential-scrubbed env for an agent-browser subprocess (deferred import: test + harnesses stub the ``tools`` package).""" from tools.environments.local import hermes_subprocess_env env = hermes_subprocess_env(inherit_credentials=False) @@ -117,9 +95,8 @@ except Exception: _is_always_blocked_url = lambda url: True # noqa: E731 — fail-closed on the floor too _normalize_url_for_request = lambda url: url # noqa: E731 — best-effort fallback _sensitive_query_param_name = lambda url: None # noqa: E731 — best-effort fallback -# Browser-provider ABC + registry. Per-vendor providers live under -# ``plugins/browser//``; the legacy class names are re-exported below as -# backward-compat shims for callers that import them from this module. +# Browser-provider ABC + registry; per-vendor providers live under +# ``plugins/browser//``. Legacy class names are re-exported as shims. from agent.browser_provider import BrowserProvider as CloudBrowserProvider # noqa: F401 (legacy alias) from agent.browser_registry import ( # noqa: F401 (test-patchable surface) get_provider as _registry_get_browser_provider, @@ -129,9 +106,8 @@ try: registry_generation as _browser_registry_generation, ) except ImportError: - # A few isolated compatibility tests intentionally install a minimal - # ``agent.browser_registry`` stub exposing only ``get_provider``. Those - # harnesses have no mutable registry, so a constant generation is exact. + # Isolated compat tests install a minimal ``agent.browser_registry`` stub + # with only ``get_provider``; no mutable registry → constant generation. def _browser_registry_generation(*, scope=None): return (0, 0) from plugins.browser.browserbase.provider import ( # noqa: F401 (legacy import surface) @@ -144,14 +120,12 @@ from plugins.browser.firecrawl.provider import ( # noqa: F401 FirecrawlBrowserProvider as FirecrawlProvider, ) from tools.tool_backend_helpers import normalize_browser_cloud_provider # noqa: F401 (read via origin) -# Camofox local anti-detection browser backend (optional). -# When CAMOFOX_URL is set, all browser operations route through the -# camofox REST API instead of the agent-browser CLI. +# Optional backends: Camofox (CAMOFOX_URL routes everything through its REST API) +# and the Browser Use CLI. try: from tools.browser_camofox import is_camofox_mode as _is_camofox_mode except ImportError: _is_camofox_mode = lambda: False # noqa: E731 -# Browser Use CLI (optional) try: from tools.browser_use_cli import is_browser_use_cli_mode as _is_browser_use_cli_mode except ImportError: @@ -159,9 +133,8 @@ except ImportError: logger = logging.getLogger(__name__) -# Standard PATH entries for environments with minimal PATH (e.g. systemd services). -# Includes Android/Termux and macOS Homebrew locations needed for agent-browser, -# npx, node, and Android's glibc runner (grun). +# PATH fallbacks for minimal-PATH environments (systemd services): Termux, +# macOS Homebrew, and the usual system dirs — needed for agent-browser/npx/node. _SANE_PATH_DIRS = ( "/data/data/com.termux/files/usr/bin", "/data/data/com.termux/files/usr/sbin", @@ -201,37 +174,29 @@ from tools.browser_tool_install import ( # noqa: F401 (re-exported; tests patc # Throttle screenshot cleanup to avoid repeated full directory scans. _last_screenshot_cleanup_by_dir: dict[str, float] = {} -# ============================================================================ +# ---------------------------------------------------------------------------- # Configuration -# ============================================================================ +# ---------------------------------------------------------------------------- -# Default timeout for browser commands (seconds) -DEFAULT_COMMAND_TIMEOUT = 30 +DEFAULT_COMMAND_TIMEOUT = 30 # seconds -# Floor for ``open`` (navigate) — cold daemon + first Chromium launch can exceed -# the generic command_timeout on slow or library-starved Linux hosts. +# Floors for ``open``: cold daemon + first Chromium launch can exceed the +# generic command_timeout on slow or library-starved Linux hosts. MIN_OPEN_TIMEOUT = 60 MIN_FIRST_OPEN_TIMEOUT = 120 -# Default max chars for snapshot content before truncation. Aligned with -# web_tools.DEFAULT_EXTRACT_CHAR_LIMIT (15000) — the snapshot and -# web_extract paths share the same truncate-and-store pattern, so the model -# gets the same per-page budget from both. Configurable via -# ``browser.snapshot_threshold`` in config.yaml. +# Snapshot truncation budget — aligned with web_tools.DEFAULT_EXTRACT_CHAR_LIMIT +# so the model gets the same per-page budget from both paths. Configurable via +# ``browser.snapshot_threshold``. DEFAULT_SNAPSHOT_THRESHOLD = 15000 MIN_SNAPSHOT_THRESHOLD = 1000 +SNAPSHOT_SUMMARIZE_THRESHOLD = DEFAULT_SNAPSHOT_THRESHOLD # legacy import surface -# Backwards-compatible import surface. Runtime call sites use -# ``get_browser_snapshot_threshold()`` so config overrides take effect. -SNAPSHOT_SUMMARIZE_THRESHOLD = DEFAULT_SNAPSHOT_THRESHOLD - -# Hard ceiling on the full-snapshot file written to cache/web when a snapshot -# is truncated. Mirrors web_tools.MAX_STORED_TEXT_CHARS — -# the model only ever sees the truncated view; the stored copy exists for -# read_file paging and must not write unbounded bytes to disk. +# Ceiling on the stored full-snapshot file (mirrors web_tools.MAX_STORED_TEXT_CHARS): +# the stored copy exists for read_file paging and must not be unbounded. MAX_STORED_SNAPSHOT_CHARS = 2_000_000 -# Commands that legitimately return empty stdout (e.g. close, record). +# Commands that legitimately return empty stdout. _EMPTY_OK_COMMANDS: frozenset = frozenset({"close", "record"}) _cached_command_timeout: Optional[int] = None @@ -239,19 +204,13 @@ _command_timeout_resolved = False _cached_snapshot_threshold: Optional[int] = None _snapshot_threshold_resolved = False - -def _sanitize_url_for_logs(value: object) -> str: - """Mask secrets in logged CDP URLs; :func:`agent.redact.redact_cdp_url` is the single policy.""" - return redact_cdp_url(value) +# Mask secrets in logged CDP URLs; agent.redact.redact_cdp_url is the single policy. +_sanitize_url_for_logs = redact_cdp_url def _browser_cfg(key: str, default, parse, log_label: str): - """Read ``browser.`` from the raw profile config and ``parse`` it. - - Returns ``default`` when the key is absent, the section is not a mapping, - or reading/parsing raises (logged at debug as "Could not read "). - Raw config is used so tool JSON output is not affected by loader warnings. - """ + """``parse(browser.)`` from the RAW profile config (loader warnings must not + leak into tool JSON); ``default`` when absent, not a mapping, or on any error.""" try: from hermes_cli.config import read_raw_config browser_cfg = read_raw_config().get("browser", {}) @@ -262,61 +221,51 @@ def _browser_cfg(key: str, default, parse, log_label: str): return default -def _get_command_timeout() -> int: - """Return ``browser.command_timeout`` (floored at 5s; default 30s). +def _cached_browser_cfg(cache_name: str, flag_name: str, key: str, default, parse, log_label: str): + """Process-cached ``_browser_cfg`` read (cache cleared by ``cleanup_all_browsers``). - Cached after the first call and cleared by ``cleanup_all_browsers()``. + The value is stored BEFORE the resolved flag flips so a concurrent reader can + never observe ``resolved=True`` with a ``None`` cache. """ - global _cached_command_timeout, _command_timeout_resolved - if _command_timeout_resolved and _cached_command_timeout is not None: - return _cached_command_timeout + g = globals() + if g[flag_name] and g[cache_name] is not None: + return g[cache_name] + result = _browser_cfg(key, default, parse, log_label) + g[cache_name] = result + g[flag_name] = True + return result - result = _browser_cfg( + +def _get_command_timeout() -> int: + """``browser.command_timeout`` (floored at 5s; default 30s).""" + return _cached_browser_cfg( + "_cached_command_timeout", "_command_timeout_resolved", "command_timeout", DEFAULT_COMMAND_TIMEOUT, lambda v: DEFAULT_COMMAND_TIMEOUT if v is None else max(int(v), 5), "command_timeout from config", ) - # Assign the cached value BEFORE flipping the resolved flag so a - # concurrent reader cannot observe ``resolved=True`` with a ``None`` cache. - _cached_command_timeout = result - _command_timeout_resolved = True - return result def _safe_command_timeout() -> int: - """``_get_command_timeout`` guaranteed non-None (cache reset mid-flight). - - Uses ``is not None`` rather than ``or`` so a configured ``0`` is preserved. - """ + """``_get_command_timeout`` guaranteed non-None (cache reset mid-flight); ``is not + None`` rather than ``or`` so a configured ``0`` is preserved.""" val = _get_command_timeout() return val if val is not None else DEFAULT_COMMAND_TIMEOUT def get_browser_snapshot_threshold() -> int: - """Return ``browser.snapshot_threshold`` (floored at MIN_SNAPSHOT_THRESHOLD). - - Cached for the browser lifecycle and reset by :func:`cleanup_all_browsers`. - """ - global _cached_snapshot_threshold, _snapshot_threshold_resolved - if _snapshot_threshold_resolved and _cached_snapshot_threshold is not None: - return _cached_snapshot_threshold - - result = _browser_cfg( + """``browser.snapshot_threshold`` (floored at MIN_SNAPSHOT_THRESHOLD).""" + return _cached_browser_cfg( + "_cached_snapshot_threshold", "_snapshot_threshold_resolved", "snapshot_threshold", DEFAULT_SNAPSHOT_THRESHOLD, lambda v: DEFAULT_SNAPSHOT_THRESHOLD if v is None else max(int(v), MIN_SNAPSHOT_THRESHOLD), "browser.snapshot_threshold", ) - # Same race-safety invariant as the command-timeout cache. - _cached_snapshot_threshold = result - _snapshot_threshold_resolved = True - return result def _get_open_command_timeout(*, first_open: bool = False) -> int: """Timeout for agent-browser ``open`` (navigation / daemon cold start).""" - base = _safe_command_timeout() - floor = MIN_FIRST_OPEN_TIMEOUT if first_open else MIN_OPEN_TIMEOUT - return max(base, floor) + return max(_safe_command_timeout(), MIN_FIRST_OPEN_TIMEOUT if first_open else MIN_OPEN_TIMEOUT) from tools.browser_tool_session import ( # noqa: F401 (re-exported; tests patch tools.browser_tool.) @@ -359,24 +308,17 @@ from tools.browser_tool_cdp import ( # noqa: F401 (re-exported; tests patch to _stop_cdp_supervisor, ) -# ============================================================================ -# Cloud Provider Registry -# ============================================================================ -# -# Per-vendor providers live as plugins under ``plugins/browser//`` and -# self-register with :mod:`agent.browser_registry`, which is what -# ``_get_cloud_provider()`` consults. The legacy class-name dict below is a -# backward-compat shim: when a test monkeypatches it, it is honoured; -# otherwise the registry-backed path wins. +# ---------------------------------------------------------------------------- +# Cloud provider registry — legacy class-name dict is a backward-compat shim: +# honoured when a test monkeypatches it, otherwise agent.browser_registry wins. +# ---------------------------------------------------------------------------- _PROVIDER_REGISTRY: Dict[str, type] = { "browserbase": BrowserbaseProvider, "browser-use": BrowserUseProvider, "firecrawl": FirecrawlProvider, } -# Frozen copy of the import-time _PROVIDER_REGISTRY, used by -# ``_is_legacy_provider_registry_overridden`` to detect test-time -# monkeypatching. NEVER mutate this dict. +# Frozen import-time copy used to detect test-time monkeypatching. NEVER mutate. _DEFAULT_PROVIDER_REGISTRY: Dict[str, type] = dict(_PROVIDER_REGISTRY) _cached_cloud_provider: Optional[CloudBrowserProvider] = None @@ -390,9 +332,7 @@ _allow_private_urls_resolved = False _cached_allow_private_urls: Optional[bool] = None _cached_agent_browser: Optional[str] = None _agent_browser_resolved = False - -# Lightpanda engine support — cached like _get_cloud_provider(). -# agent-browser v0.25.3+ supports ``--engine lightpanda`` natively. +# Lightpanda engine (agent-browser v0.25.3+ ``--engine lightpanda``), cached like the provider. _cached_browser_engine: Optional[str] = None _browser_engine_resolved = False @@ -419,23 +359,16 @@ from hermes_constants import is_termux as _is_termux_environment # noqa: F401 # Sentinel _find_agent_browser returns/caches to mean "resolve via npx" rather -# than a concrete executable path. A named constant + predicate keep the six -# comparison sites (four here, plus hermes_cli/tools_config.py and -# hermes_cli/doctor.py) from drifting if the sentinel's exact spelling ever -# changes. +# than a concrete path (also compared in hermes_cli/tools_config.py and doctor.py). NPX_AGENT_BROWSER_SENTINEL = "npx agent-browser" -# Pinned to match scripts/install.sh / scripts/install.ps1's -# "agent-browser@^0.26.0" managed install so a git-clone install resolving -# agent-browser via bare npx gets the same version as a managed install, -# instead of floating latest with no integrity check. Update both together. +# Pinned to match scripts/install.sh / install.ps1's managed install so a bare-npx +# resolution gets the same version instead of floating latest. Update together. AGENT_BROWSER_NPX_SPEC = "agent-browser@^0.26.0" _auto_local_for_private_urls_resolved = False _cached_auto_local_for_private_urls: bool = True - - _cached_headed_mode: Optional[bool] = None _headed_mode_resolved = False @@ -452,10 +385,8 @@ from tools.browser_tool_lightpanda_fallback import ( # noqa: F401 ) -# Session name for the single shared real-profile copy-browser. All consented -# local browsing attaches to this one agent-browser session so concurrent -# tasks reuse the same copy-browser instead of each launching a rival Chromium -# on the same copied user-data-dir. +# Single shared real-profile copy-browser session: concurrent tasks reuse it +# instead of each launching a rival Chromium on the same copied user-data-dir. _REAL_PROFILE_SESSION = "hermes-real-profile" _real_profile_cdp_lock = threading.Lock() _real_profile_cdp_cache: dict = {} @@ -477,63 +408,46 @@ from tools.browser_tool_real_profile import ( # noqa: F401 ) -def _url_is_private(url: str) -> bool: - """Return True when the URL's host resolves to a private/LAN/loopback address. +_PRIVATE_HOST_SUFFIXES = (".localhost", ".local", ".lan", ".internal") - Reuses ``tools.url_safety.is_safe_url`` as the oracle — if the SSRF check - would reject the URL, we treat it as "private" for routing purposes. DNS - resolution failures are treated as NOT private (fall through to whatever - backend is configured, which will surface the DNS error naturally). + +def _url_is_private(url: str) -> bool: + """True when the URL's host is (or resolves to) a private/LAN/loopback/CGNAT address. + + Routing oracle only: DNS failures are NOT private (fall through to the + configured backend, which surfaces the DNS error). Obvious names short-circuit + the DNS hop; bare ``localhost`` resolves via /etc/hosts otherwise. """ + import ipaddress + import socket + from urllib.parse import urlparse + + def private(ip) -> bool: + return ( + ip.is_private or ip.is_loopback or ip.is_link_local + or ip in ipaddress.ip_network("100.64.0.0/10") + ) + try: - # is_safe_url returns False for private/loopback/link-local/CGNAT AND - # for DNS failures. We only want the private-network case here, so - # we parse + check the host shape as a DNS-failure sieve first. - from urllib.parse import urlparse - import ipaddress - import socket - parsed = urlparse(url) - hostname = (parsed.hostname or "").strip().lower().rstrip(".") + hostname = (urlparse(url).hostname or "").strip().lower().rstrip(".") if not hostname: return False - # Literal IP → check directly try: - ip = ipaddress.ip_address(hostname) - return ( - ip.is_private - or ip.is_loopback - or ip.is_link_local - # 172.16.0.0/12: only covered by ip.is_private on Python - # ≥3.11 (bpo-40791). Explicit check keeps 3.10 runtimes - # routing these to the local sidecar correctly. - or ip in ipaddress.ip_network("172.16.0.0/12") - or ip in ipaddress.ip_network("100.64.0.0/10") - ) + return private(ipaddress.ip_address(hostname)) except ValueError: pass - # Hostname — must resolve to confirm it's private (bare "localhost" - # resolves to 127.0.0.1 via /etc/hosts). Short-circuit on obvious - # names to avoid a DNS hop. - if hostname in {"localhost",} or hostname.endswith(".localhost"): - return True - if hostname.endswith(".local") or hostname.endswith(".lan") or hostname.endswith(".internal"): + if hostname == "localhost" or hostname.endswith(_PRIVATE_HOST_SUFFIXES): return True try: addr_info = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) except socket.gaierror: - return False # DNS fail → not private, let the normal path fail - for _, _, _, _, sockaddr in addr_info: + return False + for *_, sockaddr in addr_info: try: - ip = ipaddress.ip_address(sockaddr[0]) + if private(ipaddress.ip_address(sockaddr[0])): + return True except ValueError: continue - if ( - ip.is_private - or ip.is_loopback - or ip.is_link_local - or ip in ipaddress.ip_network("100.64.0.0/10") - ): - return True return False except Exception as exc: logger.debug("URL-privacy check failed for %s: %s", url, exc) @@ -541,50 +455,38 @@ def _url_is_private(url: str) -> bool: def _navigation_session_key(task_id: str, url: str) -> str: - """Pick the session key that should handle ``url`` for ``task_id``. + """Session key that should handle ``url`` for ``task_id``. - Returns ``f"{task_id}::local"`` (hybrid routing: a local Chromium sidecar - while the cloud session keeps serving public URLs) only when ALL hold: a - cloud provider is configured, ``browser.auto_local_for_private_urls`` is - on (default), the URL resolves to a private/LAN/loopback address, no CDP - override is active (it owns the whole session), and Camofox is off (already - local-only). Otherwise the bare task_id. + ``f"{task_id}::local"`` (hybrid routing: local Chromium sidecar while the cloud + session keeps serving public URLs) only when ALL hold: cloud provider + configured, ``browser.auto_local_for_private_urls`` on, private URL, no CDP + override (it owns the whole session), Camofox off (already local-only). """ if task_id is None: task_id = "default" - if _get_cdp_override_raw(): - return task_id - if _is_camofox_mode(): - return task_id - if _get_cloud_provider() is None: - return task_id - if not _auto_local_for_private_urls(): - return task_id - if not _url_is_private(url): - return task_id - return f"{task_id}{_LOCAL_SUFFIX}" + hybrid = ( + not _get_cdp_override_raw() + and not _is_camofox_mode() + and _get_cloud_provider() is not None + and _auto_local_for_private_urls() + and _url_is_private(url) + ) + return f"{task_id}{_LOCAL_SUFFIX}" if hybrid else task_id def _is_local_sidecar_key(session_key: str) -> bool: - """Return True when ``session_key`` is a hybrid-routing local sidecar.""" + """True when ``session_key`` is a hybrid-routing local sidecar.""" return session_key.endswith(_LOCAL_SUFFIX) def _bare_task_id_for_session_key(session_key: str) -> str: - """Return the owning bare task id for an opaque browser session key.""" - if _is_local_sidecar_key(session_key): - return session_key[: -len(_LOCAL_SUFFIX)] - return session_key + """Owning bare task id for an opaque browser session key.""" + return session_key[: -len(_LOCAL_SUFFIX)] if _is_local_sidecar_key(session_key) else session_key def _session_info_owned_by_task(session_info: Dict[str, Any], task_id: str, session_key: str) -> bool: - """Return whether ``session_info`` still belongs to ``task_id``/``session_key``. - - Sessions created by current code carry explicit ownership metadata. Treat - older in-memory entries without those fields as valid for hot-reload/test - compatibility, but reject any explicit mismatch before a non-navigation - tool can act on the wrong tab/session. - """ + """Ownership check; entries without metadata (older in-memory / hot-reload) pass, + any explicit mismatch fails before a non-nav tool can act on the wrong session.""" owner = session_info.get("owner_task_id") key = session_info.get("session_key") return (owner is None or owner == task_id) and (key is None or key == session_key) @@ -593,9 +495,8 @@ def _session_info_owned_by_task(session_info: Dict[str, Any], task_id: str, sess def _last_session_key(task_id: str) -> str: """Session key a non-nav tool must use: the one that served the task's last navigation. - If that session was cleaned up or its ownership metadata no longer matches, - fail closed by dropping the stale binding rather than recreating or mutating - the wrong browser. + If that session was cleaned up or its ownership no longer matches, fail closed by + dropping the stale binding rather than recreating or mutating the wrong browser. """ if task_id is None: task_id = "default" @@ -616,39 +517,26 @@ def _last_session_key(task_id: str) -> str: def _socket_safe_tmpdir() -> str: - """Short temp dir for Unix domain sockets. - - macOS ``TMPDIR`` (``/var/folders/.../T/``) plus ``agent-browser-hermes_…`` - exceeds the 104-byte ``AF_UNIX`` path limit ("Failed to create socket - directory", silent screenshot failures), so ``/tmp`` is used there. - """ - if sys.platform == "darwin": - return "/tmp" - return tempfile.gettempdir() + """Short temp dir for Unix sockets: macOS ``TMPDIR`` + ``agent-browser-hermes_…`` + exceeds the 104-byte AF_UNIX limit (silent screenshot failures), so use /tmp there.""" + return "/tmp" if sys.platform == "darwin" else tempfile.gettempdir() # Active sessions keyed by "session key": the bare task_id, or f"{task_id}::local" -# for a hybrid-routing local sidecar. The key is opaque to _run_browser_command / -# cleanup_browser. Values: session_name (always), bb_session_id + cdp_url (cloud). +# for a hybrid-routing local sidecar (opaque to _run_browser_command / cleanup_browser). +# Values: session_name (always), bb_session_id + cdp_url (cloud). _active_sessions: Dict[str, Dict[str, Any]] = {} _recording_sessions: set = set() # session_keys with active recordings -# Most recent session_key per task_id, set by browser_navigate() and read by every -# non-nav tool so click/snapshot land in the session that served the last -# navigation (otherwise a localhost sidecar task would fall back to the cloud session). +# Most recent session_key per task_id (set by browser_navigate, read by every non-nav +# tool) so click/snapshot land in the session that served the last navigation. _last_active_session_key: Dict[str, str] = {} _LOCAL_SUFFIX = "::local" -# Flag to track if cleanup has been done _cleanup_done = False -# ============================================================================= -# Inactivity Timeout Configuration -# ============================================================================= - -# Session inactivity timeout (seconds) - cleanup if no activity for this long. -# config.yaml is authoritative; BROWSER_INACTIVITY_TIMEOUT remains a legacy -# fallback so old deployments keep working if they have not migrated yet. +# Inactivity timeout: config.yaml is authoritative; BROWSER_INACTIVITY_TIMEOUT +# remains a legacy env fallback for unmigrated deployments. DEFAULT_SESSION_INACTIVITY_TIMEOUT = int( DEFAULT_CONFIG.get("browser", {}).get("inactivity_timeout", 120) ) @@ -665,39 +553,35 @@ def _get_session_inactivity_timeout() -> int: BROWSER_SESSION_INACTIVITY_TIMEOUT = _get_session_inactivity_timeout() -# How often the cleanup thread re-runs the orphan reaper (a startup-only reap -# can never recover from a leak that appears after boot in a long-lived process). +# Orphan reaper cadence: a startup-only reap can never recover from a leak that +# appears after boot in a long-lived process. BROWSER_ORPHAN_REAP_INTERVAL = 300 # seconds -# Idle ceiling for a daemon whose owner process is alive but which fell out of -# its in-memory tracking — owner-alive alone would make it immortal. A large -# multiple of the inactivity timeout so a legitimately busy session is never touched. +# Idle ceiling for a daemon whose owner is alive but which fell out of in-memory +# tracking (owner-alive alone would make it immortal); large multiple so a busy +# session is never touched. BROWSER_ORPHAN_GRACE_SECONDS = max(3600, BROWSER_SESSION_INACTIVITY_TIMEOUT * 20) _session_last_activity: Dict[str, float] = {} -# Owner Hermes home per session: the janitor is one process-global thread with -# no profile scope of its own, so each teardown must re-enter the OWNING -# profile's scope (copy_context at spawn would pin the first profile's secrets -# onto every other profile's teardown). +# Owner Hermes home per session: the janitor is one process-global thread, so each +# teardown must re-enter the OWNING profile's scope (copy_context at spawn would +# pin the first profile's secrets onto every other profile's teardown). _session_owner_homes: Dict[str, str] = {} -# Consecutive janitor cleanup failures per session; force-reaped after MAX_INACTIVITY_CLEANUP_FAILURES. +# Consecutive janitor failures per session; force-reaped after MAX_INACTIVITY_CLEANUP_FAILURES. _cleanup_failures: Dict[str, int] = {} MAX_INACTIVITY_CLEANUP_FAILURES = 3 -# Session keys flagged suspect after a command timeout. Written by -# _BrowserSessionBackend.mark_suspect (a single GIL-atomic dict write — must stay -# cheap and lock-free per the SuspectableBackend contract); consumed by -# ensure_healthy() at next use, which recycles the session. +# Session keys flagged suspect after a command timeout (written lock-free by +# mark_suspect; consumed by ensure_healthy() at next use, which recycles). _suspect_browser_sessions: Dict[str, str] = {} class _BrowserSessionBackend: """``agent.deadline.SuspectableBackend`` adapter for one cached session key. - A thin stateless view over ``_active_sessions[key]`` + its daemon. The - timeout path calls ``mark_suspect`` inline; ``ensure_healthy`` runs at the - top of ``_get_session_info`` — the single choke point every command passes - through before reusing a cached session. + Stateless view over ``_active_sessions[key]``. The timeout path calls + ``mark_suspect`` inline; ``ensure_healthy`` runs at the top of + ``_get_session_info`` — the choke point every command passes through. """ __slots__ = ("_session_key",) @@ -706,20 +590,15 @@ class _BrowserSessionBackend: self._session_key = session_key def mark_suspect(self, reason: str) -> None: - """Flag the cached session as possibly poisoned. - - MUST stay cheap, non-blocking and lock-free (it runs inline on the - timed-out caller's thread); all recycle work is deferred to ``ensure_healthy``. - """ + """MUST stay cheap, non-blocking and lock-free (runs inline on the timed-out + caller's thread); all recycle work is deferred to ``ensure_healthy``.""" _suspect_browser_sessions[self._session_key] = reason def ensure_healthy(self) -> bool: - """Recycle the session when a prior timeout marked it suspect. + """Recycle the session when a prior timeout marked it suspect; False after teardown. - True when safe to reuse; False after tearing down a suspect session - (caller creates a fresh one). The flag is popped BEFORE teardown: the - ``close`` re-enters ``_get_session_info`` and must not recurse into - another recycle. + The flag is popped BEFORE teardown: the ``close`` re-enters + ``_get_session_info`` and must not recurse into another recycle. """ reason = _suspect_browser_sessions.pop(self._session_key, None) if reason is None: @@ -739,15 +618,12 @@ class _BrowserSessionBackend: def _browser_session_backend(session_key: str) -> _BrowserSessionBackend: - """Return the SuspectableBackend adapter for ``session_key``.""" return _BrowserSessionBackend(session_key) -# Background cleanup thread state _cleanup_thread = None _cleanup_running = False -# Protects _session_last_activity AND _active_sessions for thread safety -# (subagents run concurrently via ThreadPoolExecutor) +# Protects _session_last_activity AND _active_sessions (subagents run concurrently). _cleanup_lock = threading.Lock() @@ -781,24 +657,16 @@ from tools.browser_tool_lifecycle import ( # noqa: F401 (re-exported; tests pa cleanup_all_browsers, ) -# atexit only — NO SIGINT/SIGTERM handlers calling sys.exit(): a SystemExit -# raised inside a prompt_toolkit key-binding callback corrupts the coroutine -# state and makes the process unkillable. atexit runs on any normal exit. +# atexit only — NO SIGINT/SIGTERM handlers calling sys.exit(): a SystemExit raised +# inside a prompt_toolkit key-binding callback corrupts the coroutine state and +# makes the process unkillable. atexit.register(_emergency_cleanup_all_sessions) - - -# ============================================================================= -# Inactivity Cleanup Functions -# ============================================================================= - - -# Register cleanup thread stop on exit atexit.register(_stop_browser_cleanup_thread) -# ============================================================================ +# ---------------------------------------------------------------------------- # Tool Schemas -# ============================================================================ +# ---------------------------------------------------------------------------- BROWSER_TOOL_SCHEMAS = [ { @@ -943,11 +811,6 @@ BROWSER_TOOL_SCHEMAS = [ ] -# ============================================================================ -# Utility Functions -# ============================================================================ - - from tools.browser_tool_snapshot import ( # noqa: F401 _store_full_snapshot, _truncate_snapshot, @@ -956,64 +819,71 @@ from tools.browser_tool_snapshot import ( # noqa: F401 ) -# ============================================================================ +# ---------------------------------------------------------------------------- # Browser Tool Functions -# ============================================================================ +# ---------------------------------------------------------------------------- + +def _err(error: str, **extra) -> dict: + return {"success": False, "error": error, **extra} + + +def _dumps(payload: Dict[str, Any], **kw) -> str: + return json.dumps(payload, ensure_ascii=False, **kw) + def _secret_url_error(url: str) -> Optional[dict]: - """Refuse URLs that embed an API key/token (raw and URL-decoded, catching ``%2D`` tricks). - - A prompt injection could otherwise make the agent navigate to - ``https://evil.com/steal?key=sk-ant-...`` to exfiltrate secrets. - """ + """Refuse URLs embedding an API key/token (raw and URL-decoded, catching ``%2D`` + tricks) — a prompt injection could otherwise exfiltrate secrets via the URL.""" import urllib.parse from agent.redact import _PREFIX_RE if _PREFIX_RE.search(url) or _PREFIX_RE.search(urllib.parse.unquote(url)): - return {"success": False, "error": "Blocked: URL contains what appears to be an API key or token. Secrets must not be sent in URLs."} + return _err("Blocked: URL contains what appears to be an API key or token. Secrets must not be sent in URLs.") return None def _url_policy_error(url: str, *, auto_local: bool = False) -> Optional[dict]: """Backend-aware URL checks on an already-normalized URL; None if allowed. - Order matters and every step is a floor for the next: - 1. Credential-like query params are refused for cloud backends (third-party - readers) — allowed for local backends and for the hybrid local sidecar. - 2. Cloud metadata / IMDS endpoints are refused UNCONDITIONALLY, for every - backend including pure-local Chromium and off-host CDP (a local Chromium - on a cloud VM still reaches the host IMDS). - 3. Private/internal addresses are refused unless the backend is local, the - URL is being auto-routed to the local sidecar (``auto_local``), or - ``browser.allow_private_urls`` opts out. - 4. Website policy (config allow/deny lists). + Ordered floors: (1) credential-like query params refused for cloud backends + (third-party readers), allowed for local and the hybrid sidecar; (2) cloud + metadata / IMDS refused UNCONDITIONALLY for every backend (a local Chromium on + a cloud VM still reaches the host IMDS); (3) private addresses refused unless + local, auto-routed to the sidecar, or ``browser.allow_private_urls``; + (4) website policy allow/deny lists. """ local = _is_local_backend() sensitive_query_key = _sensitive_query_param_name(url) if sensitive_query_key and not local and not auto_local: - return {"success": False, "error": ( + return _err( "Blocked: URL contains a credential-like query parameter " f"({sensitive_query_key}). Cloud browser backends are third-party " "readers; use a local browser/CDP session or remove the sensitive " - "query parameter before navigating.")} + "query parameter before navigating.") if _is_always_blocked_url(url): - return {"success": False, "error": "Blocked: URL targets a cloud metadata endpoint"} + return _err("Blocked: URL targets a cloud metadata endpoint") if not local and not auto_local and not _allow_private_urls() and not _is_safe_url(url): - return {"success": False, "error": "Blocked: URL targets a private or internal address"} + return _err("Blocked: URL targets a private or internal address") blocked = check_website_access(url) if blocked: - return {"success": False, "error": blocked["message"], - "blocked_by_policy": {"host": blocked["host"], "rule": blocked["rule"], "source": blocked["source"]}} + return _err(blocked["message"], + blocked_by_policy={"host": blocked["host"], "rule": blocked["rule"], "source": blocked["source"]}) return None +def _secret_url_error_normalized(url: str) -> tuple[str, Optional[dict]]: + """Secret check on the raw URL, then again on the normalized one; returns ``(url, error)``.""" + err = _secret_url_error(url) + if err is None: + url = _normalize_url_for_request(url) + err = _secret_url_error(url) + return url, err + + def evaluate_url_safety(url: str) -> Optional[dict]: """Run URL safety checks; None if safe, else an error dict""" - err = _secret_url_error(url) - if err: - return err - url = _normalize_url_for_request(url) - return _secret_url_error(url) or _url_policy_error(url) + url, err = _secret_url_error_normalized(url) + return err or _url_policy_error(url) _BOT_DETECTION_TITLE_PATTERNS = ( @@ -1026,32 +896,49 @@ _BOT_DETECTION_TITLE_PATTERNS = ( def _post_redirect_block(nav_session_key: str, url: str, final_url: str, auto_local_this_nav: bool) -> Optional[str]: - """Post-redirect SSRF check; returns a blocked JSON payload or None. + """Post-redirect SSRF check; blocked JSON payload or None. - If the browser followed a redirect to a private/internal address the model - could read internal content via later snapshots, so the page is navigated to - about:blank first. The cloud-metadata floor fires for every backend (even the - local sidecar); the private-address check is skipped for local backends and - the hybrid sidecar, and when ``browser.allow_private_urls`` opts out. + A redirect onto a private/internal address would let later snapshots read + internal content, so the page is navigated to about:blank first. The metadata + floor fires for every backend; the private-address check is skipped for local + backends, the hybrid sidecar, and ``browser.allow_private_urls``. """ if not final_url or final_url == url: return None if _is_always_blocked_url(final_url): - _run_browser_command(nav_session_key, "open", ["about:blank"], timeout=10) - return json.dumps({ - "success": False, "error": "Blocked: redirect landed on a cloud metadata endpoint" - }) - if ( + what = "a cloud metadata endpoint" + elif ( not _is_local_backend() and not auto_local_this_nav and not _allow_private_urls() and not _is_safe_url(final_url) ): - _run_browser_command(nav_session_key, "open", ["about:blank"], timeout=10) - return json.dumps({ - "success": False, "error": "Blocked: redirect landed on a private/internal address" - }) - return None + what = "a private/internal address" + else: + return None + _run_browser_command(nav_session_key, "open", ["about:blank"], timeout=10) + return json.dumps(_err(f"Blocked: redirect landed on {what}")) + + +def _snapshot_fields(snap_result: Dict[str, Any]) -> Dict[str, Any]: + """``snapshot`` + ``element_count`` response fields from a successful snapshot result. + + Oversized snapshots truncate at line boundaries; the full tree is stored to + cache/web with a read_file paging note (same pattern as web_extract — no LLM). + """ + data = snap_result.get("data", {}) + snapshot_text = data.get("snapshot", "") + refs = data.get("refs", {}) + threshold = get_browser_snapshot_threshold() + if len(snapshot_text) > threshold: + snapshot_text = _truncate_snapshot(snapshot_text, max_chars=threshold) + return {"snapshot": _redact_browser_output(snapshot_text), "element_count": len(refs) if refs else 0} + + +def _merge_fallback_warning(response: Dict[str, Any], result: Dict[str, Any]) -> None: + """Copy a secondary result's fallback warning only if the response has none yet.""" + if result.get("fallback_warning") and not response.get("fallback_warning"): + _copy_fallback_warning(response, result) def _attach_auto_snapshot(response: Dict[str, Any], nav_session_key: str) -> None: @@ -1059,30 +946,20 @@ def _attach_auto_snapshot(response: Dict[str, Any], nav_session_key: str) -> Non try: snap_result = _run_browser_command(nav_session_key, "snapshot", ["-c"]) if snap_result.get("success"): - snap_data = snap_result.get("data", {}) - snapshot_text = snap_data.get("snapshot", "") - refs = snap_data.get("refs", {}) - threshold = get_browser_snapshot_threshold() - if len(snapshot_text) > threshold: - snapshot_text = _truncate_snapshot(snapshot_text, max_chars=threshold) - response["snapshot"] = _redact_browser_output(snapshot_text) - response["element_count"] = len(refs) if refs else 0 - if snap_result.get("fallback_warning") and not response.get("fallback_warning"): - _copy_fallback_warning(response, snap_result) + response.update(_snapshot_fields(snap_result)) + _merge_fallback_warning(response, snap_result) except Exception as e: logger.debug("Auto-snapshot after navigate failed: %s", e) def browser_navigate(url: str, task_id: Optional[str] = None) -> str: - """Navigate to ``url``; returns JSON with title, compact snapshot and, on first nav, stealth features.""" - # Hybrid routing decides BEFORE the safety checks whether this URL goes to a - # local Chromium sidecar (cloud provider configured + private URL + - # ``browser.auto_local_for_private_urls``); the cloud provider never sees - # the URL in that case, so the private-address checks are relaxed for it. - safety_error = _secret_url_error(url) - if safety_error is None: - url = _normalize_url_for_request(url) - safety_error = _secret_url_error(url) + """Navigate to ``url``; JSON with title, compact snapshot and, on first nav, stealth features. + + Hybrid routing decides BEFORE the safety checks whether this URL goes to a local + Chromium sidecar; the cloud provider never sees the URL then, so the + private-address checks are relaxed for it. + """ + url, safety_error = _secret_url_error_normalized(url) if safety_error is not None: return json.dumps(safety_error) @@ -1094,7 +971,6 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str: if safety_error is not None: return json.dumps(safety_error) - # Camofox backend — delegate after safety checks pass if _is_camofox_mode(): from tools.browser_camofox import camofox_navigate return camofox_navigate(url, task_id) @@ -1108,12 +984,8 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str: type(_get_cloud_provider()).__name__ if _get_cloud_provider() else "none", ) - # Get session info to check if this is a new session - # (will create one with features logged if not exists) session_info = _get_session_info(nav_session_key) is_first_nav = session_info.get("_first_nav", True) - - # Auto-start recording if configured and this is first navigation if is_first_nav: session_info["_first_nav"] = False _maybe_start_recording(nav_session_key) @@ -1121,11 +993,8 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str: result = _run_browser_command( nav_session_key, "open", [url], timeout=_get_open_command_timeout(first_open=is_first_nav) ) - if not result.get("success"): - return json.dumps({ - "success": False, "error": result.get("error", "Navigation failed") - }, ensure_ascii=False) + return _dumps(_err(result.get("error", "Navigation failed"))) data = result.get("data", {}) title = data.get("title", "") @@ -1135,19 +1004,15 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str: if blocked is not None: return blocked - response = { - "success": True, "url": final_url, "title": title - } - # Auditability: stamp navigations that ran on the user's real-profile - # copy-browser so usage is visible in the tool result. + response = {"success": True, "url": final_url, "title": title} + # Auditability: stamp navigations that ran on the user's real-profile copy-browser. try: if (session_info.get("features") or {}).get("real_profile"): response["used_real_profile"] = True except Exception: pass - # Remember only a successful, non-blocked navigation as the task owner. - # Failed opens and blocked redirects must not retarget follow-up clicks - # or snapshots to a newly-created but irrelevant session. + # Only a successful, non-blocked navigation becomes the task owner: failed opens + # and blocked redirects must not retarget follow-up clicks to an irrelevant session. _last_active_session_key[effective_task_id] = nav_session_key _copy_fallback_warning(response, result) @@ -1160,117 +1025,91 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str: "4) Some sites have very aggressive bot detection that may be unavoidable." ) - # Include feature info on first navigation so model knows what's active if is_first_nav and "features" in session_info: features = session_info["features"] - active_features = [k for k, v in features.items() if v] if not features.get("proxies"): response["stealth_warning"] = ( "Running WITHOUT residential proxies. Bot detection may be more aggressive. " "Consider upgrading Browserbase plan for proxy support." ) - response["stealth_features"] = active_features + response["stealth_features"] = [k for k, v in features.items() if v] _attach_auto_snapshot(response, nav_session_key) - return json.dumps(response, ensure_ascii=False) + return _dumps(response) def browser_snapshot( full: bool = False, task_id: Optional[str] = None, user_task: Optional[str] = None ) -> str: """Text snapshot of the page's accessibility tree (compact unless ``full``). - - ``user_task`` is deprecated and unused: oversized snapshots always - truncate-and-store (no LLM pass). - """ + ``user_task`` is deprecated and unused (oversized snapshots always truncate-and-store).""" if _is_camofox_mode(): from tools.browser_camofox import camofox_snapshot return camofox_snapshot(full, task_id) effective_task_id = _last_session_key(task_id or "default") + result = _run_browser_command(effective_task_id, "snapshot", [] if full else ["-c"]) + if not result.get("success"): + return _failed_response(result, "Failed to get snapshot") - # Build command args based on full flag - args = [] - if not full: - args.extend(["-c"]) # Compact mode + blocked = _blocked_private_page_content(effective_task_id) + if blocked is not None: + return blocked - result = _run_browser_command(effective_task_id, "snapshot", args) + response = {"success": True, **_snapshot_fields(result)} + _copy_fallback_warning(response, result) - if result.get("success"): - data = result.get("data", {}) - snapshot_text = data.get("snapshot", "") - refs = data.get("refs", {}) + # Merge supervisor state (pending dialogs + frame tree) when a CDP supervisor is + # attached. See website/docs/developer-guide/browser-supervisor.md. + try: + from tools.browser_supervisor import SUPERVISOR_REGISTRY # type: ignore[import-not-found] + _supervisor = SUPERVISOR_REGISTRY.get(effective_task_id) + if _supervisor is not None: + _sv_snap = _supervisor.snapshot() + if _sv_snap.active: + response.update(_redact_browser_output(_sv_snap.to_dict())) + except Exception as _sv_exc: + logger.debug("supervisor snapshot merge failed: %s", _sv_exc) - # ── Private-network guard: block snapshots from eval-navigated private pages ── - blocked = _blocked_private_page_content(effective_task_id) - if blocked is not None: - return blocked - - # Oversized snapshots truncate at line boundaries; the full - # accessibility tree is stored to cache/web and the appended note - # tells the agent how to page through it with read_file (same - # pattern as web_extract — no LLM summarization). Threshold is - # configurable via browser.snapshot_threshold. - threshold = get_browser_snapshot_threshold() - if len(snapshot_text) > threshold: - snapshot_text = _truncate_snapshot(snapshot_text, max_chars=threshold) - - response = { - "success": True, - "snapshot": _redact_browser_output(snapshot_text), - "element_count": len(refs) if refs else 0 - } - _copy_fallback_warning(response, result) - - # Merge supervisor state (pending dialogs + frame tree) when a CDP - # supervisor is attached to this task. No-op otherwise. See - # website/docs/developer-guide/browser-supervisor.md. - try: - from tools.browser_supervisor import SUPERVISOR_REGISTRY # type: ignore[import-not-found] - _supervisor = SUPERVISOR_REGISTRY.get(effective_task_id) - if _supervisor is not None: - _sv_snap = _supervisor.snapshot() - if _sv_snap.active: - response.update(_redact_browser_output(_sv_snap.to_dict())) - except Exception as _sv_exc: - logger.debug("supervisor snapshot merge failed: %s", _sv_exc) - - return json.dumps(response, ensure_ascii=False) - return _failed_response(result, "Failed to get snapshot") + return _dumps(response) def _json_with_fallback(response: Dict[str, Any], result: Dict[str, Any]) -> str: """``json.dumps`` of ``response`` with the Lightpanda fallback metadata copied from ``result``.""" - return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False) + return _dumps(_copy_fallback_warning(response, result)) def _failed_response(result: Dict[str, Any], default_error: str) -> str: - return _json_with_fallback({"success": False, "error": result.get("error", default_error)}, result) + return _json_with_fallback(_err(result.get("error", default_error)), result) def _tool_response(result: Dict[str, Any], ok: Dict[str, Any], default_error: str) -> str: - """Standard tool JSON for a ``_run_browser_command`` result. - - Success → ``{"success": True, **ok}``; failure → ``{"success": False, - "error": result.error or default_error}``. Lightpanda fallback metadata - is copied onto either shape. - """ + """Standard tool JSON: success → ``{"success": True, **ok}``, failure → + ``{"success": False, "error": result.error or default_error}``; fallback metadata on either.""" if not result.get("success"): return _failed_response(result, default_error) return _json_with_fallback({"success": True, **ok}, result) +def _camofox(func_name: str, *args): + """Call ``tools.browser_camofox.(*args)`` (Camofox mode delegation).""" + import importlib + return getattr(importlib.import_module("tools.browser_camofox"), func_name)(*args) + + +def _guarded_action_session(task_id: Optional[str], action: str) -> tuple[str, Optional[str]]: + """``(session_key, blocked_payload)`` for an input action on the task's current page.""" + effective_task_id = _last_session_key(task_id or "default") + return effective_task_id, _blocked_private_page_action(effective_task_id, action) + + def browser_click(ref: str, task_id: Optional[str] = None) -> str: """Click the element ``ref`` (e.g. "@e5").""" if _is_camofox_mode(): - from tools.browser_camofox import camofox_click - return camofox_click(ref, task_id) - - effective_task_id = _last_session_key(task_id or "default") - blocked = _blocked_private_page_action(effective_task_id, "click") + return _camofox("camofox_click", ref, task_id) + effective_task_id, blocked = _guarded_action_session(task_id, "click") if blocked is not None: return blocked - if not ref.startswith("@"): ref = f"@{ref}" result = _run_browser_command(effective_task_id, "click", [ref]) @@ -1280,55 +1119,39 @@ def browser_click(ref: str, task_id: Optional[str] = None) -> str: def browser_type(ref: str, text: str, task_id: Optional[str] = None) -> str: """Type ``text`` into the element ``ref`` (fill: clears, then types).""" if _is_camofox_mode(): - from tools.browser_camofox import camofox_type - return camofox_type(ref, text, task_id) - - effective_task_id = _last_session_key(task_id or "default") - blocked = _blocked_private_page_action(effective_task_id, "type") + return _camofox("camofox_type", ref, text, task_id) + effective_task_id, blocked = _guarded_action_session(task_id, "type") if blocked is not None: return blocked - if not ref.startswith("@"): ref = f"@{ref}" - # fill clears then types result = _run_browser_command(effective_task_id, "fill", [ref, text]) from agent.display import ( redact_browser_typed_text_for_display, redact_tool_args_for_display ) - - # Typed text goes through the secret-pattern redactor so API keys / tokens - # don't leak into tool progress or chat history (the raw value was already - # sent to the browser above); normal text passes through unchanged. + # Typed text goes through the secret-pattern redactor so API keys / tokens don't + # leak into tool progress or chat history (the raw value already went to the browser). display_text = (redact_tool_args_for_display("browser_type", {"text": text}) or {})["text"] if result.get("success"): response = {"success": True, "typed": display_text, "element": ref} else: - response = {"success": False, "error": result.get("error", f"Failed to type into {ref}")} + response = _err(result.get("error", f"Failed to type into {ref}")) response = _copy_fallback_warning(response, result) - response = redact_browser_typed_text_for_display(response, text) - return json.dumps(response, ensure_ascii=False) + return _dumps(redact_browser_typed_text_for_display(response, text)) def browser_scroll(direction: str, task_id: Optional[str] = None) -> str: """Scroll the page ``direction`` ("up"/"down") by about half a viewport.""" if direction not in {"up", "down"}: - return json.dumps({ - "success": False, "error": f"Invalid direction '{direction}'. Use 'up' or 'down'." - }, ensure_ascii=False) - - # Single scroll with a pixel amount (~half a viewport) instead of 5x subprocess calls. - _SCROLL_PIXELS = 500 - + return _dumps(_err(f"Invalid direction '{direction}'. Use 'up' or 'down'.")) + _SCROLL_PIXELS = 500 # ~half a viewport in one call instead of 5x subprocess calls if _is_camofox_mode(): - from tools.browser_camofox import camofox_scroll - # Camofox REST API doesn't support pixel args; use repeated calls - _SCROLL_REPEATS = 5 + # Camofox REST API has no pixel argument; use repeated calls. result = None - for _ in range(_SCROLL_REPEATS): - result = camofox_scroll(direction, task_id) + for _ in range(5): + result = _camofox("camofox_scroll", direction, task_id) return result - effective_task_id = _last_session_key(task_id or "default") result = _run_browser_command(effective_task_id, "scroll", [direction, str(_SCROLL_PIXELS)]) return _tool_response(result, {"scrolled": direction}, f"Failed to scroll {direction}") @@ -1337,17 +1160,12 @@ def browser_scroll(direction: str, task_id: Optional[str] = None) -> str: def browser_back(task_id: Optional[str] = None) -> str: """Navigate back in browser history.""" if _is_camofox_mode(): - from tools.browser_camofox import camofox_back - return camofox_back(task_id) - + return _camofox("camofox_back", task_id) effective_task_id = _last_session_key(task_id or "default") result = _run_browser_command(effective_task_id, "back", []) - if result.get("success"): - # History can land on a private/internal/cloud-metadata address the - # navigate preflight never saw (earlier redirect chain, manipulated - # client-side history). Re-check post-navigation like every other - # content-returning entry point — the floor fires for every backend. + # History can land on a private/internal/metadata address the navigate + # preflight never saw (earlier redirect chain, manipulated client-side history). blocked = _blocked_private_page( effective_task_id, "Browser history navigation (back) landed on this address.") if blocked is not None: @@ -1358,11 +1176,8 @@ def browser_back(task_id: Optional[str] = None) -> str: def browser_press(key: str, task_id: Optional[str] = None) -> str: """Press a keyboard key (e.g. "Enter", "Tab").""" if _is_camofox_mode(): - from tools.browser_camofox import camofox_press - return camofox_press(key, task_id) - - effective_task_id = _last_session_key(task_id or "default") - blocked = _blocked_private_page_action(effective_task_id, "press") + return _camofox("camofox_press", key, task_id) + effective_task_id, blocked = _guarded_action_session(task_id, "press") if blocked is not None: return blocked result = _run_browser_command(effective_task_id, "press", [key]) @@ -1371,17 +1186,12 @@ def browser_press(key: str, task_id: Optional[str] = None) -> str: def _blocked_private_page_json(blocked_url: str, why: str) -> str: """Refusal payload for a page whose URL targets a private/internal address.""" - return json.dumps({ - "success": False, - "error": f"Blocked: page URL targets a private or internal address ({blocked_url}). {why}", - }, ensure_ascii=False) + return _dumps(_err(f"Blocked: page URL targets a private or internal address ({blocked_url}). {why}")) def _blocked_private_page(effective_task_id: str, why: str) -> Optional[str]: - """Blocked payload when the SSRF guard is active and the current page is private, else None. - - Fail-open on probe failure (see ``_current_page_private_url``). - """ + """Blocked payload when the SSRF guard is active and the current page is private, else + None. Fail-open on probe failure (see ``_current_page_private_url``).""" if not _eval_ssrf_guard_active(effective_task_id): return None blocked_url = _current_page_private_url(effective_task_id) @@ -1389,7 +1199,7 @@ def _blocked_private_page(effective_task_id: str, why: str) -> Optional[str]: def _blocked_private_page_action(effective_task_id: str, action: str) -> Optional[str]: - """Return a blocked payload when an unsafe cloud page would receive input.""" + """Blocked payload when an unsafe cloud page would receive input.""" return _blocked_private_page( effective_task_id, f"Refusing to {action} on this page in this browser mode.") @@ -1398,38 +1208,31 @@ _EVAL_NAVIGATED_WHY = "This may have been caused by a JavaScript navigation via def _blocked_private_page_content(effective_task_id: str) -> Optional[str]: - """Content-returning tools (snapshot/vision/eval/get_images): after any eval that - may have changed ``location.href`` to a private address, returning page content - would expose it.""" + """Content-returning tools (snapshot/vision/eval/get_images): after an eval that may + have moved ``location.href`` to a private address, returning content would expose it.""" return _blocked_private_page(effective_task_id, _EVAL_NAVIGATED_WHY) def browser_console(clear: bool = False, expression: Optional[str] = None, task_id: Optional[str] = None) -> str: """Console messages + uncaught JS errors (optionally ``clear``ing the buffers), or — when ``expression`` is given — evaluate JS in the page like the DevTools console.""" - # --- JS evaluation mode --- if expression is not None: policy_error = _enforce_browser_eval_policy(expression) if policy_error: - return json.dumps({"success": False, "error": policy_error}, ensure_ascii=False) + return _dumps(_err(policy_error)) return _browser_eval(expression, task_id) - # --- Console output mode (original behaviour) --- if _is_camofox_mode(): - from tools.browser_camofox import camofox_console - return camofox_console(clear, task_id) + return _camofox("camofox_console", clear, task_id) effective_task_id = _last_session_key(task_id or "default") - blocked = _blocked_private_page_content(effective_task_id) if blocked is not None: return blocked - console_args = ["--clear"] if clear else [] - error_args = ["--clear"] if clear else [] - - console_result = _run_browser_command(effective_task_id, "console", console_args) - errors_result = _run_browser_command(effective_task_id, "errors", error_args) + clear_args = ["--clear"] if clear else [] + console_result = _run_browser_command(effective_task_id, "console", clear_args) + errors_result = _run_browser_command(effective_task_id, "errors", clear_args) messages = [] if console_result.get("success"): @@ -1439,7 +1242,6 @@ def browser_console(clear: bool = False, expression: Optional[str] = None, task_ "text": _redact_browser_output(msg.get("text", "")), "source": "console", }) - errors = [] if errors_result.get("success"): for err in errors_result.get("data", {}).get("errors", []): @@ -1455,9 +1257,8 @@ def browser_console(clear: bool = False, expression: Optional[str] = None, task_ "total_errors": len(errors), } _copy_fallback_warning(response, console_result) - if errors_result.get("fallback_warning") and not response.get("fallback_warning"): - _copy_fallback_warning(response, errors_result) - return json.dumps(response, ensure_ascii=False) + _merge_fallback_warning(response, errors_result) + return _dumps(response) from tools.browser_tool_eval_policy import ( # noqa: F401 @@ -1489,15 +1290,22 @@ def _parse_eval_value(raw_result: Any) -> Any: return raw_result -def _eval_supervisor_fast_path(effective_task_id: str, expression: str) -> Optional[str]: - """Run ``Runtime.evaluate`` on the CDP supervisor's persistent WebSocket. +def _eval_ok_response(parsed: Any, **extra) -> Dict[str, Any]: + return { + "success": True, + "result": _redact_browser_output(parsed), + "result_type": type(parsed).__name__, + **extra, + } - Zero subprocess startup cost vs spawning ``agent-browser eval``. Returns a - tool JSON string when the supervisor produced a definitive answer (a value, - a blocked private page, or a real JS-side exception — which is NOT retried - through the subprocess, that would just reproduce it slower), or None to - fall through to the subprocess path (no supervisor, supervisor-side failure, - import error), so behaviour is unchanged when no supervisor is running. + +def _eval_supervisor_fast_path(effective_task_id: str, expression: str) -> Optional[str]: + """``Runtime.evaluate`` on the CDP supervisor's persistent WebSocket (no subprocess cost). + + Returns tool JSON when the supervisor gave a definitive answer (a value, a + blocked private page, or a real JS-side exception — NOT retried through the + subprocess, that would just reproduce it slower), or None to fall through to the + subprocess path (no supervisor, supervisor-side failure, import error). """ try: from tools.browser_supervisor import SUPERVISOR_REGISTRY # type: ignore[import-not-found] @@ -1507,21 +1315,14 @@ def _eval_supervisor_fast_path(effective_task_id: str, expression: str) -> Optio sup_result = supervisor.evaluate_runtime(expression) if sup_result.get("ok"): parsed = _parse_eval_value(sup_result.get("result")) - # Post-eval page-URL recheck: if this (or a prior) eval navigated - # the page to a private address, withhold the result. + # Post-eval page-URL recheck: withhold the result if an eval navigated to a private address. blocked = _blocked_private_page_content(effective_task_id) if blocked is not None: return blocked - response = { - "success": True, - "result": _redact_browser_output(parsed), - "result_type": type(parsed).__name__, - "method": "cdp_supervisor", - } - return json.dumps(response, ensure_ascii=False, default=str) + return _dumps(_eval_ok_response(parsed, method="cdp_supervisor"), default=str) err = sup_result.get("error") or "evaluate_runtime failed" if "supervisor" not in err.lower(): - return json.dumps({"success": False, "error": err}, ensure_ascii=False) + return _dumps(_err(err)) logger.debug( "browser_eval: supervisor path unavailable (%s), falling back to subprocess", err ) @@ -1536,48 +1337,39 @@ def _eval_failure_response(result: Dict[str, Any]) -> str: """Tool JSON for a failed ``agent-browser eval``, with actionable rewrites of known errors.""" err = result.get("error", "eval failed") if any(hint in err.lower() for hint in ("unknown command", "not supported", "not found", "no such command")): - # Backend capability gap — give the model a clear signal. err = f"JavaScript evaluation is not supported by this browser backend. {err}" elif "reference chain is too long" in err.lower(): - # A live DOM node / NodeList / Window can't be JSON-serialized by CDP. - # The supervisor fast path retries with returnByValue=false; the CLI - # subprocess can't, so replace the cryptic protocol error with guidance. + # A live DOM node / NodeList / Window can't be JSON-serialized by CDP. The + # supervisor path retries with returnByValue=false; the CLI can't. err = ( "Expression returned a live DOM node / NodeList / Window, " "which can't be serialized. Extract a primitive value " "(e.g. .innerText, .href, .src, .value) or use " "JSON.stringify() / a snapshot tool instead." ) - return json.dumps(_copy_fallback_warning({"success": False, "error": err}, result)) + return json.dumps(_copy_fallback_warning(_err(err), result)) def _browser_eval(expression: str, task_id: Optional[str] = None) -> str: """Evaluate a JavaScript expression in the page context and return the result. - Private-network guard, both sub-paths gated on the same condition: the - literal pre-scan closes direct fetches (``fetch('http://127.0.0.1/...')``, - which never update ``location.href``); the post-eval page-URL recheck - closes navigate-then-read (``location.href = ...`` then read the DOM) — - eval returns arbitrary JS results directly, never via snapshot/vision. + Private-network guard, both halves gated on the same condition: the literal + pre-scan closes direct fetches (``fetch('http://127.0.0.1/...')`` never updates + ``location.href``); the post-eval page-URL recheck closes navigate-then-read. """ effective_task_id = _last_session_key(task_id or "default") if _eval_ssrf_guard_active(effective_task_id): blocked_literal = _expression_targets_private_url(expression) if blocked_literal: - return json.dumps({ - "success": False, - "error": ( - "Blocked: JavaScript expression targets a private or " - f"internal address ({blocked_literal}). Reading internal " - "endpoints via browser_console is not permitted in this " - "browser mode." - ), - }, ensure_ascii=False) + return _dumps(_err( + "Blocked: JavaScript expression targets a private or " + f"internal address ({blocked_literal}). Reading internal " + "endpoints via browser_console is not permitted in this " + "browser mode." + )) - # Camofox keeps its own raw-``task_id``-keyed session map, so pass the raw - # id (matching every other Camofox tool) rather than the resolved - # agent-browser session key. The literal pre-scan above already ran. + # Camofox keeps its own raw-task_id-keyed session map, so pass the raw id. if _is_camofox_mode(): return _camofox_eval(expression, task_id) @@ -1589,17 +1381,11 @@ def _browser_eval(expression: str, task_id: Optional[str] = None) -> str: if not result.get("success"): return _eval_failure_response(result) - parsed = _parse_eval_value(result.get("data", {}).get("result")) - response = { - "success": True, - "result": _redact_browser_output(parsed), - "result_type": type(parsed).__name__, - } - # Post-eval page-URL recheck (mirrors the supervisor path). + response = _eval_ok_response(_parse_eval_value(result.get("data", {}).get("result"))) blocked = _blocked_private_page_content(effective_task_id) if blocked is not None: return blocked - return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False, default=str) + return _dumps(_copy_fallback_warning(response, result), default=str) def _camofox_eval(expression: str, task_id: Optional[str] = None) -> str: @@ -1610,35 +1396,20 @@ def _camofox_eval(expression: str, task_id: Optional[str] = None) -> str: tab_id = tab_info.get("tab_id") or tab_info.get("id") user_id = tab_info["user_id"] resp = _post(f"/tabs/{tab_id}/evaluate", body={"expression": expression, "userId": user_id}) - - # Camofox returns the result in a JSON envelope - raw_result = resp.get("result") if isinstance(resp, dict) else resp - parsed = raw_result - if isinstance(raw_result, str): - try: - parsed = json.loads(raw_result) - except (json.JSONDecodeError, ValueError): - pass + parsed = _parse_eval_value(resp.get("result") if isinstance(resp, dict) else resp) if _eval_ssrf_guard_active(task_id or "default"): _blocked_url = _camofox_current_page_private_url(tab_id, user_id) if _blocked_url: return _blocked_private_page_json(_blocked_url, _EVAL_NAVIGATED_WHY) - return json.dumps({ - "success": True, - "result": _redact_browser_output(parsed), - "result_type": type(parsed).__name__, - }, ensure_ascii=False, default=str) + return _dumps(_eval_ok_response(parsed), default=str) except Exception as e: error_msg = str(e) - # Graceful degradation — server may not support eval - if any(code in error_msg for code in ("404", "405", "501")): - return json.dumps({ - "success": False, - "error": "JavaScript evaluation is not supported by this Camofox server. " - "Use browser_snapshot or browser_vision to inspect page state.", - }) + if any(code in error_msg for code in ("404", "405", "501")): # server without eval support + return json.dumps(_err( + "JavaScript evaluation is not supported by this Camofox server. " + "Use browser_snapshot or browser_vision to inspect page state.")) return tool_error(error_msg, success=False) @@ -1650,10 +1421,7 @@ def _maybe_start_recording(task_id: str): try: from hermes_cli.config import read_raw_config hermes_home = get_hermes_home() - cfg = read_raw_config() - record_enabled = cfg_get(cfg, "browser", "record_sessions", default=False) - - if not record_enabled: + if not cfg_get(read_raw_config(), "browser", "record_sessions", default=False): return recordings_dir = hermes_home / "browser_recordings" @@ -1691,16 +1459,7 @@ def _maybe_stop_recording(task_id: str): _recording_sessions.discard(task_id) -def browser_get_images(task_id: Optional[str] = None) -> str: - """List the page's images (src, alt, natural size), excluding data: URIs.""" - if _is_camofox_mode(): - from tools.browser_camofox import camofox_get_images - return camofox_get_images(task_id) - - effective_task_id = _last_session_key(task_id or "default") - - # Use eval to run JavaScript that extracts images - js_code = """JSON.stringify( +_GET_IMAGES_JS = """JSON.stringify( [...document.images].map(img => ({ src: img.src, alt: img.alt || '', @@ -1709,30 +1468,29 @@ def browser_get_images(task_id: Optional[str] = None) -> str: })).filter(img => img.src && !img.src.startsWith('data:')) )""" - result = _run_browser_command(effective_task_id, "eval", [js_code]) - if result.get("success"): - # ── Private-network guard (sibling of snapshot/vision/eval guards) ── - blocked = _blocked_private_page_content(effective_task_id) - if blocked is not None: - return blocked +def browser_get_images(task_id: Optional[str] = None) -> str: + """List the page's images (src, alt, natural size), excluding data: URIs.""" + if _is_camofox_mode(): + return _camofox("camofox_get_images", task_id) - data = result.get("data", {}) - raw_result = data.get("result", "[]") + effective_task_id = _last_session_key(task_id or "default") + result = _run_browser_command(effective_task_id, "eval", [_GET_IMAGES_JS]) + if not result.get("success"): + return _failed_response(result, "Failed to get images") - try: - # Parse the JSON string returned by JavaScript - if isinstance(raw_result, str): - images = json.loads(raw_result) - else: - images = raw_result + blocked = _blocked_private_page_content(effective_task_id) + if blocked is not None: + return blocked - return _json_with_fallback( - {"success": True, "images": _redact_browser_output(images), "count": len(images)}, result) - except json.JSONDecodeError: - return _json_with_fallback( - {"success": True, "images": [], "count": 0, "warning": "Could not parse image data"}, result) - return _failed_response(result, "Failed to get images") + raw_result = result.get("data", {}).get("result", "[]") + try: + images = json.loads(raw_result) if isinstance(raw_result, str) else raw_result + return _json_with_fallback( + {"success": True, "images": _redact_browser_output(images), "count": len(images)}, result) + except json.JSONDecodeError: + return _json_with_fallback( + {"success": True, "images": [], "count": 0, "warning": "Could not parse image data"}, result) _LP_VISION_FALLBACK_REASON = ( @@ -1751,15 +1509,13 @@ from tools.browser_tool_vision import ( # noqa: F401 (re-exported; tests patch def browser_vision(question: str, annotate: bool = False, task_id: Optional[str] = None) -> Union[str, Dict[str, Any]]: """Screenshot the current page for visual inspection (CAPTCHAs, images, layouts). - Native-vision models get the screenshot attached to the conversation (a - multimodal tool-result envelope); otherwise the auxiliary vision model - returns a text analysis as JSON. Either way the file is saved persistently - and its path returned so it can be shared via MEDIA:. + Native-vision models get the screenshot attached to the conversation (multimodal + tool-result envelope); otherwise the auxiliary vision model returns a text + analysis. The file is saved persistently and its path returned (MEDIA:). ``annotate`` overlays numbered [N] labels on interactive elements. """ if _is_camofox_mode(): - from tools.browser_camofox import camofox_vision - return camofox_vision(question, annotate, task_id) + return _camofox("camofox_vision", question, annotate, task_id) import uuid as uuid_mod from hermes_constants import get_hermes_dir @@ -1767,7 +1523,6 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str] screenshot_path = screenshots_dir / f"browser_screenshot_{uuid_mod.uuid4().hex}.png" effective_task_id = _last_session_key(task_id or "default") - # ── Private-network guard: block vision from eval-navigated private pages ── blocked = _blocked_private_page_content(effective_task_id) if blocked is not None: return blocked @@ -1775,10 +1530,9 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str] _lp_prerouted, _lp_fallback_warning, screenshot_path = _lightpanda_vision_preroute( effective_task_id, annotate, screenshot_path, ) - + result: Dict[str, Any] = {} try: screenshots_dir.mkdir(parents=True, exist_ok=True) - # Prune old screenshots (older than 24 hours) to prevent unbounded disk growth _cleanup_old_screenshots(screenshots_dir, max_age_hours=24) if _lp_prerouted and screenshot_path.exists(): @@ -1793,36 +1547,30 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str] effective_task_id, "screenshot", screenshot_args, - # If the Lightpanda pre-route already failed, force Chrome so - # _run_browser_command doesn't trigger a redundant LP fallback. + # A failed Lightpanda pre-route forces Chrome so _run_browser_command + # doesn't trigger a redundant LP fallback. _engine_override="auto" if _lp_prerouted else None, ) if not result.get("success"): error_detail = result.get("error", "Unknown error") - return _json_with_fallback({ - "success": False, - "error": f"Failed to take screenshot ({_vision_mode_label()} mode): {error_detail}", - }, result) + return _json_with_fallback( + _err(f"Failed to take screenshot ({_vision_mode_label()} mode): {error_detail}"), result) actual_screenshot_path = result.get("data", {}).get("path") if actual_screenshot_path: screenshot_path = Path(actual_screenshot_path) if not screenshot_path.exists(): - return json.dumps({ - "success": False, - "error": ( - f"Screenshot file was not created at {screenshot_path} ({_vision_mode_label()} mode). " - f"This may indicate a socket path issue (macOS /var/folders/), " - f"a missing Chromium install ('agent-browser install'), " - f"or a stale daemon process." - ), - }, ensure_ascii=False) + return _dumps(_err( + f"Screenshot file was not created at {screenshot_path} ({_vision_mode_label()} mode). " + f"This may indicate a socket path issue (macOS /var/folders/), " + f"a missing Chromium install ('agent-browser install'), " + f"or a stale daemon process." + )) - # Fast path: native image routing for the active main model — attach the - # screenshot directly instead of describing it through an aux vision LLM - # (no aux call, no information loss; consistent with vision_analyze). + # Native image routing for the active main model: attach the screenshot + # directly instead of describing it through an aux vision LLM (no information loss). from tools.vision_tools import _should_use_native_vision_fast_path if _should_use_native_vision_fast_path(): @@ -1837,37 +1585,23 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str] _copy_fallback_warning(response_data, result) if annotate and result.get("data", {}).get("annotations"): response_data["annotations"] = result["data"]["annotations"] - return json.dumps(response_data, ensure_ascii=False) + return _dumps(response_data) except Exception as e: - # Keep the screenshot if it was captured — the failure is in the vision - # analysis, not the capture, and deleting it loses evidence the user may - # need. The 24-hour cleanup bounds disk growth. + # Keep a captured screenshot — the failure is in the analysis, not the + # capture, and deleting it loses evidence. The 24-hour cleanup bounds disk growth. logger.warning("browser_vision failed: %s", e, exc_info=True) - error_info = {"success": False, "error": f"Error during vision analysis: {str(e)}"} + error_info = _err(f"Error during vision analysis: {str(e)}") if screenshot_path.exists(): error_info["screenshot_path"] = str(screenshot_path) error_info["note"] = "Screenshot was captured but vision analysis failed. You can still share it via MEDIA:." - _copy_fallback_warning(error_info, result if 'result' in locals() else {}) - return json.dumps(error_info, ensure_ascii=False) + _copy_fallback_warning(error_info, result) + return _dumps(error_info) -# ============================================================================ -# Cleanup and Management Functions -# ============================================================================ - - -# ============================================================================ -# Requirements Check -# ============================================================================ - - -# Cache for Chromium discovery. Invalidated by _reset_browser_caches. +# Chromium discovery cache / one-shot autoinstall flag (a failed 170MB download must +# not retry on every call). Both reset by cleanup_all_browsers(). _cached_chromium_installed: Optional[bool] = None - - -# One-shot per process: a 170MB download that fails (or is slow) must not be -# retried on every browser call. Reset by _reset_browser_caches() for tests. _chromium_autoinstall_attempted = False @@ -1882,21 +1616,14 @@ from tools.browser_extension_router import ( _BROWSER_SCHEMA_MAP = {s["name"]: s for s in BROWSER_TOOL_SCHEMAS} -def _browser_router_kw(kw: dict) -> dict: - """Identity kwargs forwarded to the extension router wrapper.""" - return { - "task_id": kw.get("task_id"), "session_id": kw.get("session_id") - } - - def check_browser_routed_requirements(action: str = "browser_snapshot") -> bool: """Availability gate for tools that can use either browser backend.""" return check_browser_requirements() or extension_controller_available(action) -# (tool name, emoji, availability gate, fallback call) — the routed-through- -# extension tools use the per-action gate; get_images/console/vision keep the -# plain requirement checks. ``_fallback`` receives (args, kw). +# (tool name, emoji, availability gate, fallback call) — routed-through-extension +# tools (gate None) use the per-action gate; get_images/console/vision keep the plain +# requirement checks. ``_fallback`` receives (args, kw). _BROWSER_TOOL_TABLE = ( ("browser_navigate", "🌐", None, lambda args, kw: browser_navigate(url=args.get("url", ""), task_id=kw.get("task_id"))), @@ -1934,25 +1661,23 @@ def _routed_check_fn(name: str): def _routed_handler(name: str, fallback): def handler(args, **kw): - return routed_browser_handler(name, args, fallback=lambda: fallback(args, kw), **_browser_router_kw(kw)) + return routed_browser_handler( + name, args, fallback=lambda: fallback(args, kw), + task_id=kw.get("task_id"), session_id=kw.get("session_id"), + ) return handler -# Legacy per-tool gate names (tests + external callers). -check_browser_navigate_requirements = _routed_check_fn("browser_navigate") -check_browser_snapshot_requirements = _routed_check_fn("browser_snapshot") -check_browser_click_requirements = _routed_check_fn("browser_click") -check_browser_type_requirements = _routed_check_fn("browser_type") -check_browser_scroll_requirements = _routed_check_fn("browser_scroll") -check_browser_back_requirements = _routed_check_fn("browser_back") -check_browser_press_requirements = _routed_check_fn("browser_press") - +# Legacy per-tool gate names (tests + external callers); also looked up by the +# registration loop below via globals(). for _name, _emoji, _check_fn, _fallback in _BROWSER_TOOL_TABLE: + if _check_fn is None: + _check_fn = globals()[f"check_{_name}_requirements"] = _routed_check_fn(_name) registry.register( name=_name, toolset="browser", schema=_BROWSER_SCHEMA_MAP[_name], handler=_routed_handler(_name, _fallback), - check_fn=_check_fn or globals()[f"check_{_name}_requirements"], + check_fn=_check_fn, emoji=_emoji, ) diff --git a/tools/browser_tool_lifecycle.py b/tools/browser_tool_lifecycle.py index 97a94472fd..e204bc8686 100644 --- a/tools/browser_tool_lifecycle.py +++ b/tools/browser_tool_lifecycle.py @@ -1,9 +1,8 @@ """Browser session lifecycle: inactivity janitor, orphan reaper, per-session teardown, atexit emergency cleanup. -Split out of ``tools/browser_tool.py``; every name is re-imported there so -``tools.browser_tool.`` keeps resolving (and monkeypatching). Origin +Split out of ``tools/browser_tool.py``; every name is re-imported there. Origin symbols and module state are read/written through ``_bt`` (the origin module, -resolved per call by :func:`tools.browser_tool_origin.origin_module`) so +the :data:`tools.browser_tool_origin.origin` proxy) so ``patch("tools.browser_tool.X")`` is honoured and no import cycle exists. """ @@ -19,17 +18,12 @@ from pathlib import Path from typing import Any, Dict, Optional, Tuple from hermes_constants import get_hermes_home, reset_hermes_home_override, set_hermes_home_override -from tools.browser_tool_origin import origin_module as _origin +from tools.browser_tool_origin import origin as _bt def _session_expiry_timestamp(session_info: Dict[str, Any]) -> Optional[float]: - """Return a provider-authoritative session expiry as epoch seconds. - - Cloud providers may omit ``expires_at``. Unknown or malformed values are - therefore treated as having no known expiry, preserving the existing - lifecycle for local browsers and providers without an expiry contract. - """ - _bt = _origin() + """Provider-authoritative session expiry as epoch seconds; None when absent or + malformed (cloud providers may omit ``expires_at``; local browsers never have one).""" value = session_info.get("expires_at") if isinstance(value, (int, float)) and not isinstance(value, bool): return float(value) @@ -52,8 +46,7 @@ def _session_expiry_timestamp(session_info: Dict[str, Any]) -> Optional[float]: def _session_has_expired( session_info: Dict[str, Any], *, now: Optional[float] = None ) -> bool: - """Return whether a cached browser session crossed its provider deadline.""" - _bt = _origin() + """Whether a cached browser session crossed its provider deadline.""" expires_at = _bt._session_expiry_timestamp(session_info) if expires_at is None: return False @@ -61,23 +54,16 @@ def _session_has_expired( def _emergency_cleanup_all_sessions(): - """ - Emergency cleanup of all active browser sessions. - Called on process exit or interrupt to prevent orphaned sessions. - - Also runs the orphan reaper to clean up daemons left behind by previously - crashed hermes processes — this way every clean hermes exit sweeps - accumulated orphans, not just ones that actively used the browser tool. - """ - _bt = _origin() + """atexit: close this process's sessions, then sweep orphans left by crashed + hermes processes — every clean exit reaps accumulated orphans, not only + processes that used the browser tool.""" if _bt._cleanup_done: return _bt._cleanup_done = True - # Clean up this process's own sessions first, so their owner_pid files - # are removed before the reaper scans. - # Real-profile Chrome processes are launched directly (not by - # agent-browser), so the session cleanup below never reaps them. + # Own sessions first so their owner_pid files are gone before the reaper scans. + # Real-profile Chrome is launched directly (not by agent-browser), so the + # session cleanup never reaps it. try: _bt._terminate_real_profile_chrome() except Exception as e: @@ -97,9 +83,7 @@ def _emergency_cleanup_all_sessions(): _bt._cleanup_failures.clear() _bt._recording_sessions.clear() - # Lightpanda servers (Browser Use mode) are processes we spawned; the - # session cleanup above stops the tracked ones, this catches any that - # fell out of ``_active_sessions``. + # Lightpanda servers we spawned that fell out of ``_active_sessions``. try: from tools.browser_lightpanda import stop_all_lightpanda @@ -107,8 +91,7 @@ def _emergency_cleanup_all_sessions(): except Exception as e: _bt.logger.debug("Lightpanda cleanup on exit failed: %s", e) - # Sweep orphans from other crashed hermes processes. Safe even if we - # never used the browser — uses owner_pid liveness to avoid reaping + # Safe even if we never used the browser — owner_pid liveness protects # daemons owned by other live hermes processes. try: _bt._reap_orphaned_browser_sessions() @@ -120,11 +103,10 @@ def _emergency_cleanup_all_sessions(): def _session_owner_scope(task_id: str): """Run under the Hermes home + secret scope owning ``task_id``'s session (no-op if unrecorded). - The janitor thread is process-global, so each teardown must re-enter its - OWN profile's scope rather than inherit the spawning profile's; never falls + The janitor thread is process-global, so each teardown must re-enter its OWN + profile's scope rather than inherit the spawning profile's; never falls through to ``os.environ``. """ - _bt = _origin() owner_home = _bt._session_owner_homes.get(task_id) if owner_home is None: yield @@ -147,22 +129,31 @@ def _session_owner_scope(task_id: str): reset_hermes_home_override(home_token) -def _cleanup_inactive_browser_sessions(): - """Close sessions inactive longer than the timeout (called by the cleanup thread). +def _forget_session_tracking(task_id: str, *, activity: bool = True, session: bool = False) -> None: + """Drop the janitor's bookkeeping (and optionally the session entry) for ``task_id``.""" + with _bt._cleanup_lock: + if session: + _bt._active_sessions.pop(task_id, None) + if activity: + _bt._session_last_activity.pop(task_id, None) + _bt._session_owner_homes.pop(task_id, None) + _bt._cleanup_failures.pop(task_id, None) - Each session is torn down under its owner profile's scope. A session whose - cleanup keeps failing is force-reaped after MAX_INACTIVITY_CLEANUP_FAILURES - attempts instead of retrying forever; only a successful cleanup clears its - failure count. + +def _cleanup_inactive_browser_sessions(): + """Close sessions inactive longer than the timeout (cleanup thread). + + Each teardown runs under its owner profile's scope. A session whose cleanup + keeps failing is force-reaped after MAX_INACTIVITY_CLEANUP_FAILURES attempts; + only a successful cleanup clears its failure count. """ - _bt = _origin() current_time = time.time() - sessions_to_cleanup = [] with _bt._cleanup_lock: - for task_id, last_time in list(_bt._session_last_activity.items()): - if current_time - last_time > _bt.BROWSER_SESSION_INACTIVITY_TIMEOUT: - sessions_to_cleanup.append(task_id) + sessions_to_cleanup = [ + task_id for task_id, last_time in list(_bt._session_last_activity.items()) + if current_time - last_time > _bt.BROWSER_SESSION_INACTIVITY_TIMEOUT + ] for task_id in sessions_to_cleanup: elapsed = int(current_time - _bt._session_last_activity.get(task_id, current_time)) @@ -170,10 +161,7 @@ def _cleanup_inactive_browser_sessions(): try: with _bt._session_owner_scope(task_id): _bt.cleanup_browser(task_id) - with _bt._cleanup_lock: - _bt._session_last_activity.pop(task_id, None) - _bt._session_owner_homes.pop(task_id, None) - _bt._cleanup_failures.pop(task_id, None) + _forget_session_tracking(task_id) except Exception as e: with _bt._cleanup_lock: failures = _bt._cleanup_failures[task_id] = _bt._cleanup_failures.get(task_id, 0) + 1 @@ -189,21 +177,13 @@ def _cleanup_inactive_browser_sessions(): except Exception as reap_exc: _bt.logger.error("Force-reap of browser session %s failed: %s", task_id, reap_exc) finally: - with _bt._cleanup_lock: - _bt._session_owner_homes.pop(task_id, None) - _bt._cleanup_failures.pop(task_id, None) + _forget_session_tracking(task_id, activity=False) def _write_owner_pid(socket_dir: str, session_name: str) -> None: - """Record the current hermes PID as the owner of a browser socket dir. - - Written atomically to ``/.owner_pid`` so the - orphan reaper can distinguish daemons owned by a live hermes process - (don't reap) from daemons whose owner crashed (reap). Best-effort — - an OSError here just falls back to the legacy ``tracked_names`` - heuristic in the reaper. - """ - _bt = _origin() + """Record this hermes PID in ``/.owner_pid`` so the orphan + reaper can tell live-owner daemons from crashed-owner ones. Best-effort: an + OSError falls back to the legacy ``tracked_names`` heuristic.""" try: path = os.path.join(socket_dir, f"{session_name}.owner_pid") with open(path, "w", encoding="utf-8") as f: @@ -217,18 +197,13 @@ def _verify_reapable_browser_daemon(daemon_pid: int, socket_dir: str, session_name: str) -> bool: """Confirm a live PID is genuinely *this* session's agent-browser daemon. - The ``.pid`` file lives in a world-writable, predictably-named temp dir and - is written by the daemon, not us: a same-user actor can plant one pointing - at a victim PID, or a recycled PID can land on an unrelated process — and - reaping is a *tree* kill, i.e. an arbitrary-process DoS. Two psutil checks - must both pass: (1) identity — ``agent-browser`` in the name or cmdline; - (2) binding — the socket dir path/basename in the cmdline, or - ``AGENT_BROWSER_SOCKET_DIR`` in its environ. (2) is the real spoof defense: - an attacker would need a real daemon embedding this exact path, which they - could already signal. Fail-closed on any ambiguity (unreadable cmdline, no - match): refuse to reap and leave process and socket dir alone. + The ``.pid`` file lives in a world-writable temp dir and is written by the + daemon: a same-user actor can plant one pointing at a victim PID, or a recycled + PID can land on an unrelated process — and reaping is a *tree* kill. Two + checks must pass: (1) identity — ``agent-browser`` in name or cmdline; + (2) binding — the socket dir in the cmdline or ``AGENT_BROWSER_SOCKET_DIR`` in + its environ (the real spoof defense). Fail-closed on any ambiguity. """ - _bt = _origin() try: import psutil except ImportError: # psutil is a hard dep; defensive only @@ -243,8 +218,7 @@ def _verify_reapable_browser_daemon(daemon_pid: int, socket_dir: str, name = (proc.name() or "").lower() cmdline = " ".join(proc.cmdline() or []).lower() except psutil.NoSuchProcess: - # Vanished between the liveness check and now — nothing to reap. - return False + return False # vanished between the liveness check and now except (psutil.AccessDenied, OSError) as exc: _bt.logger.warning( "Refusing to reap browser daemon PID %d (session %s): " @@ -252,14 +226,12 @@ def _verify_reapable_browser_daemon(daemon_pid: int, socket_dir: str, daemon_pid, session_name, exc) return False - looks_like_browser = "agent-browser" in name or "agent-browser" in cmdline - if not looks_like_browser: + if "agent-browser" not in name and "agent-browser" not in cmdline: _bt.logger.warning( "Refusing to reap PID %d (session %s): not an agent-browser " "process (name=%r)", daemon_pid, session_name, name) return False - # Binding check: the live process must reference *this* socket dir. socket_dir_l = socket_dir.lower() socket_base_l = os.path.basename(socket_dir).lower() bound = socket_dir_l in cmdline or ( @@ -271,9 +243,7 @@ def _verify_reapable_browser_daemon(daemon_pid: int, socket_dir: str, bound = bool(env_dir) and os.path.normpath(env_dir) == \ os.path.normpath(socket_dir) except (psutil.AccessDenied, psutil.NoSuchProcess, OSError): - # environ() can be denied even same-user on some platforms. - # cmdline already failed to bind — fail closed. - bound = False + bound = False # environ() can be denied even same-user; cmdline already failed — fail closed if not bound: _bt.logger.warning( @@ -288,10 +258,9 @@ def _verify_reapable_browser_daemon(daemon_pid: int, socket_dir: str, def _socket_dir_idle_seconds(socket_dir: str) -> Optional[float]: """Seconds since anything in ``socket_dir`` was last written; None if unknown (fail safe). - Every command writes ``_stdout_`` / ``_stderr_`` there, so the - newest mtime is a last-activity marker that survives hermes restarts and - lost in-memory bookkeeping. The dir's own mtime is not enough — rewriting - an existing ``_stdout_click`` doesn't touch it — so entries are scanned too. + Every command writes ``_stdout_`` there, so the newest mtime is a + last-activity marker surviving hermes restarts. The dir's own mtime is not + enough — rewriting an existing file doesn't touch it — so entries are scanned. """ try: latest = os.path.getmtime(socket_dir) @@ -311,34 +280,47 @@ def _socket_dir_idle_seconds(socket_dir: str) -> Optional[float]: return max(0.0, time.time() - latest) +def _read_pid_file(path: str) -> Optional[int]: + """Integer PID from ``path``; None when missing or corrupt.""" + try: + return int(Path(path).read_text(encoding="utf-8").strip()) + except (ValueError, OSError): + return None + + def _owner_pid_alive(socket_dir: str, session_name: str) -> Tuple[Optional[int], Optional[bool]]: """Read ``.owner_pid`` and report ``(pid, alive)``; ``(None, None)`` when missing/corrupt.""" - owner_pid_file = os.path.join(socket_dir, f"{session_name}.owner_pid") - if not os.path.isfile(owner_pid_file): + owner_pid = _read_pid_file(os.path.join(socket_dir, f"{session_name}.owner_pid")) + if owner_pid is None: return None, None - try: - owner_pid = int(Path(owner_pid_file).read_text(encoding="utf-8").strip()) - # ``os.kill(pid, 0)`` is NOT a no-op on Windows; use the cross-platform check. - from gateway.status import _pid_exists - return owner_pid, _pid_exists(owner_pid) - except (ValueError, OSError): - return None, None # corrupt file — fall through to legacy handling + # ``os.kill(pid, 0)`` is NOT a no-op on Windows; use the cross-platform check. + from gateway.status import _pid_exists + return owner_pid, _pid_exists(owner_pid) + + +def _terminate_verified_daemon(daemon_pid: int) -> bool: + """Tree-kill ``daemon_pid`` if it has a start-time fingerprint (so a PID swapped + between check and kill is refused); False when no fingerprint. Raises on OS errors.""" + from gateway.status import get_process_start_time + from tools.process_registry import ProcessRegistry + daemon_start = get_process_start_time(daemon_pid) + if daemon_start is None: + return False + ProcessRegistry._terminate_host_pid(daemon_pid, daemon_start) + return True def _reap_socket_dir(socket_dir: str, session_name: str, tracked_names: set) -> bool: - """Reap one ``agent-browser-`` dir if orphaned; return True when a daemon was killed. + """Reap one ``agent-browser-`` dir if orphaned; True when a daemon was killed. - Ownership priority: (1) a live ``owner_pid`` means another hermes process - owns it — leave it alone UNLESS it is untracked here and idle past + Ownership priority: (1) a live ``owner_pid`` means another hermes process owns + it — leave it alone UNLESS it is untracked here and idle past ``BROWSER_ORPHAN_GRACE_SECONDS`` (owner-alive alone made leaked daemons - immortal: in-memory tracking is lost on any exception path and the daemon's - own idle timeout doesn't fire when it is wedged); (2) no owner_pid (legacy) - falls back to this process's tracking. A pidless dir is only stale after the - grace period — deleting it immediately races the creator's first stdout open. - The daemon PID is verified as ours before a tree-kill (world-writable dir, - recycled PIDs), and refused without a start-time fingerprint. + immortal); (2) no owner_pid (legacy) falls back to this process's tracking. + A pidless dir is only stale after the grace period — deleting it immediately + races the creator's first stdout open. The daemon PID is identity-verified + before a tree-kill and refused without a start-time fingerprint. """ - _bt = _origin() owner_pid, owner_alive = _bt._owner_pid_alive(socket_dir, session_name) if owner_alive is True: if session_name in tracked_names: @@ -362,32 +344,23 @@ def _reap_socket_dir(socket_dir: str, session_name: str, tracked_names: set) -> shutil.rmtree(socket_dir, ignore_errors=True) return False - try: - daemon_pid = int(Path(pid_file).read_text(encoding="utf-8").strip()) - except (ValueError, OSError): - shutil.rmtree(socket_dir, ignore_errors=True) - return False - + daemon_pid = _read_pid_file(pid_file) from gateway.status import _pid_exists - if not _pid_exists(daemon_pid): + if daemon_pid is None or not _pid_exists(daemon_pid): shutil.rmtree(socket_dir, ignore_errors=True) return False if not _bt._verify_reapable_browser_daemon(daemon_pid, socket_dir, session_name): return False # leave process and dir for a later sweep once the imposter PID is gone - # Tree-kill so Chromium children (renderer, GPU, ...) go too, not just the daemon. + # Tree-kill so Chromium children (renderer, GPU, ...) go too. reaped = False try: - from gateway.status import get_process_start_time - from tools.process_registry import ProcessRegistry - daemon_start = get_process_start_time(daemon_pid) - if daemon_start is None: + if not _terminate_verified_daemon(daemon_pid): _bt.logger.warning( "Refusing to reap browser daemon PID %d (session %s): " "no start-time fingerprint available", daemon_pid, session_name) return False - ProcessRegistry._terminate_host_pid(daemon_pid, daemon_start) _bt.logger.info("Reaped orphaned browser daemon PID %d (session %s)", daemon_pid, session_name) reaped = True @@ -400,20 +373,15 @@ def _reap_socket_dir(socket_dir: str, session_name: str, tracked_names: set) -> def _reap_orphaned_browser_sessions(): """Kill agent-browser daemons whose owning hermes process is gone. - When the process that created a session exits uncleanly (SIGKILL, crash, - gateway restart) the in-memory ``_active_sessions`` tracking is lost but the - node + Chromium processes keep running. Scans the tmp dir for - ``agent-browser-*`` socket dirs and applies ``_reap_socket_dir``'s ownership - rules (owner_pid file first — cross-process safe, two hermes instances never - reap each other — then in-process tracking for legacy daemons). - Safe to call from any context — atexit, cleanup thread, or on demand. + An unclean exit (SIGKILL, crash, gateway restart) loses ``_active_sessions`` + but node + Chromium keep running. Scans the tmp dir for ``agent-browser-*`` + socket dirs and applies ``_reap_socket_dir``'s ownership rules. Safe from any + context — atexit, cleanup thread, or on demand. """ - _bt = _origin() import glob - # Lightpanda servers (Browser Use mode) keep their own records (no - # agent-browser socket dir); sweep them with the same owner-liveness rule - # BEFORE the daemon scan, which may return early. + # Lightpanda servers keep their own records (no socket dir); sweep them with the + # same owner-liveness rule BEFORE the daemon scan, which may return early. try: from tools.browser_lightpanda import reap_orphaned_lightpanda @@ -446,19 +414,13 @@ def _reap_orphaned_browser_sessions(): def _browser_cleanup_thread_worker(): - """Every 30s: close sessions idle past BROWSER_SESSION_INACTIVITY_TIMEOUT. - - Also reaps orphaned daemons on startup AND every BROWSER_ORPHAN_REAP_INTERVAL - seconds — a daemon can fall out of in-memory tracking at any point in a - long-lived process, and a startup-only reap could never recover from that. - """ - _bt = _origin() + """Every 30s: close sessions idle past BROWSER_SESSION_INACTIVITY_TIMEOUT; reap + orphans on startup AND every BROWSER_ORPHAN_REAP_INTERVAL seconds.""" reap_every_cycles = max(1, round(_bt.BROWSER_ORPHAN_REAP_INTERVAL / 30)) cycle = 0 while _bt._cleanup_running: - # cycle 0 is the startup reap; then every reap_every_cycles. - if cycle % reap_every_cycles == 0: + if cycle % reap_every_cycles == 0: # cycle 0 is the startup reap try: _bt._reap_orphaned_browser_sessions() except Exception as e: @@ -470,8 +432,7 @@ def _browser_cleanup_thread_worker(): except Exception as e: _bt.logger.warning("Cleanup thread error: %s", e) - # Sleep in 1-second intervals so we can stop quickly if needed - for _ in range(30): + for _ in range(30): # 1s granularity so stop is quick if not _bt._cleanup_running: break time.sleep(1) @@ -479,8 +440,6 @@ def _browser_cleanup_thread_worker(): def _start_browser_cleanup_thread(): """Start the background cleanup thread if not already running.""" - _bt = _origin() - with _bt._cleanup_lock: if _bt._cleanup_thread is None or not _bt._cleanup_thread.is_alive(): _bt._cleanup_running = True @@ -493,39 +452,30 @@ def _start_browser_cleanup_thread(): def _stop_browser_cleanup_thread(): """Stop the background cleanup thread.""" - _bt = _origin() _bt._cleanup_running = False if _bt._cleanup_thread is not None: _bt._cleanup_thread.join(timeout=5) def _update_session_activity(task_id: str): - """Update the last activity timestamp for a session. - - Also records the owning Hermes home on first sight so the process-global - janitor can tear the session down under its owner's scope. An - activity touch deliberately does NOT reset ``_cleanup_failures`` — only a - successful cleanup does. - """ - _bt = _origin() + """Touch the activity timestamp; records the owning Hermes home on first sight so + the process-global janitor tears down under the owner's scope. Deliberately does + NOT reset ``_cleanup_failures`` — only a successful cleanup does.""" with _bt._cleanup_lock: _bt._session_last_activity[task_id] = time.time() _bt._session_owner_homes.setdefault(task_id, str(get_hermes_home())) def _kill_process_tree(proc: "subprocess.Popen") -> None: - """Best-effort kill of *proc* and every descendant it spawned; never raises. + """Best-effort kill of *proc* and every descendant; never raises. - ``Popen.kill()`` only signals the direct child. npm/npx fork helpers and - agent-browser's detached daemon grandchild, which survive a plain kill and - keep a capture pipe open so ``communicate()`` never sees EOF — on Windows - there is no non-blocking read to poll around that, so the whole tree must - go. No grace period: the caller already burned its full timeout waiting. - Delegates to :func:`agent.deadline.kill_process_tree` (taskkill /T /F, - killpg, plus a psutil sweep that reaches ``setsid``'d descendants) and - falls back to :func:`_legacy_kill_process_tree` on any failure. + ``Popen.kill()`` only signals the direct child; npm/npx helpers and + agent-browser's detached daemon grandchild keep a capture pipe open so + ``communicate()`` never sees EOF (no non-blocking read on Windows), so the whole + tree must go. No grace period: the caller already burned its timeout. + Delegates to :func:`agent.deadline.kill_process_tree`, falling back to + :func:`_legacy_kill_process_tree` on any failure. """ - _bt = _origin() try: from agent.deadline import kill_process_tree as _deadline_kill_tree @@ -547,12 +497,8 @@ def _legacy_kill_process_tree(proc: "subprocess.Popen") -> None: except Exception: pass return - # os.killpg/signal.SIGKILL don't exist on Windows; this branch is - # POSIX-only (the `os.name == "nt"` check above already returns first - # on Windows), but resolve them defensively via getattr anyway so an - # accidental future refactor that drops that guard degrades to a plain - # kill() instead of AttributeError — same discipline as - # tools/mcp_stdio_watchdog.py's _terminate_process_group. + # POSIX-only below (the nt guard returned), but resolve killpg/SIGKILL via + # getattr so a future refactor dropping that guard degrades to plain kill(). killpg = getattr(os, "killpg", None) if killpg is None: # windows-footgun: ok - non-POSIX fallback try: @@ -594,59 +540,48 @@ def _pid_exists(pid: int) -> bool: return True -def _cleanup_old_screenshots(screenshots_dir, max_age_hours=24): - """Remove browser screenshots older than max_age_hours to prevent disk bloat. +def _unlink_older_than(directory: Path, pattern: str, max_age_hours: float, label: str) -> None: + """Delete ``directory/pattern`` files older than ``max_age_hours``; never raises.""" + try: + cutoff = time.time() - (max_age_hours * 3600) + for f in directory.glob(pattern): + try: + if f.stat().st_mtime < cutoff: + f.unlink() + except Exception as e: + _bt.logger.debug("Failed to clean old %s %s: %s", label, f, e) + except Exception as e: + _bt.logger.debug("%s cleanup error (non-critical): %s", label.capitalize(), e) - Throttled to run at most once per hour per directory to avoid repeated - scans on screenshot-heavy workflows. - """ - _bt = _origin() + +def _cleanup_old_screenshots(screenshots_dir, max_age_hours=24): + """Prune old browser screenshots; throttled to once per hour per directory.""" key = str(screenshots_dir) now = time.time() if now - _bt._last_screenshot_cleanup_by_dir.get(key, 0.0) < 3600: return _bt._last_screenshot_cleanup_by_dir[key] = now - - try: - cutoff = time.time() - (max_age_hours * 3600) - for f in screenshots_dir.glob("browser_screenshot_*.png"): - try: - if f.stat().st_mtime < cutoff: - f.unlink() - except Exception as e: - _bt.logger.debug("Failed to clean old screenshot %s: %s", f, e) - except Exception as e: - _bt.logger.debug("Screenshot cleanup error (non-critical): %s", e) + _unlink_older_than(screenshots_dir, "browser_screenshot_*.png", max_age_hours, "screenshot") def _cleanup_old_recordings(max_age_hours=72): - """Remove browser recordings older than max_age_hours to prevent disk bloat.""" - _bt = _origin() + """Prune old browser recordings.""" try: - hermes_home = get_hermes_home() - recordings_dir = hermes_home / "browser_recordings" - if not recordings_dir.exists(): - return - cutoff = time.time() - (max_age_hours * 3600) - for f in recordings_dir.glob("session_*.webm"): - try: - if f.stat().st_mtime < cutoff: - f.unlink() - except Exception as e: - _bt.logger.debug("Failed to clean old recording %s: %s", f, e) + recordings_dir = get_hermes_home() / "browser_recordings" except Exception as e: _bt.logger.debug("Recording cleanup error (non-critical): %s", e) + return + if recordings_dir.exists(): + _unlink_older_than(recordings_dir, "session_*.webm", max_age_hours, "recording") def _drop_last_active_binding(task_id: str) -> None: """Drop stale last-active ownership after cleaning ``task_id``. - Cleaning a bare task drops its binding; cleaning a sidecar drops the binding - only if that sidecar was still the recorded owner — so a later - click/snapshot can't resurrect a cleaned sidecar on about:blank while a - primary-session binding is preserved. + A bare task drops its binding; a sidecar drops it only if that sidecar was + still the recorded owner — a later click can't resurrect a cleaned sidecar on + about:blank while a primary-session binding is preserved. """ - _bt = _origin() if _bt._is_local_sidecar_key(task_id): bare_task_id = _bt._bare_task_id_for_session_key(task_id) if _bt._last_active_session_key.get(bare_task_id) == task_id: @@ -656,13 +591,8 @@ def _drop_last_active_binding(task_id: str) -> None: def cleanup_browser(task_id: Optional[str] = None) -> None: - """Clean up browser session(s) for a task (task completion / inactivity timeout). - - A bare task id reaps BOTH the primary session and any hybrid local sidecar - spawned for it; a key already carrying ``::local`` (inactivity loop) reaps - only that one. - """ - _bt = _origin() + """Clean up browser session(s) for a task: a bare task id reaps BOTH the primary + session and any hybrid local sidecar; a ``::local`` key reaps only that one.""" if task_id is None: task_id = "default" @@ -679,33 +609,23 @@ def cleanup_browser(task_id: Optional[str] = None) -> None: def _kill_verified_daemon(socket_dir: str, session_name: str) -> bool: - """Tree-kill the daemon recorded in ``/.pid`` if it is verifiably ours. - - The .pid file lives in a world-writable temp dir and PIDs recycle: the - process must pass ``_verify_reapable_browser_daemon`` and have a start-time - fingerprint (so the kill refuses if the PID is swapped between check and - kill). Returns True when a kill was issued. Never raises. - """ - _bt = _origin() + """Tree-kill the daemon in ``/.pid`` if verifiably ours + (identity check + start-time fingerprint). True when a kill was issued. Never raises.""" pid_file = os.path.join(socket_dir, f"{session_name}.pid") if not os.path.isfile(pid_file): return False try: - from tools.process_registry import ProcessRegistry daemon_pid = int(Path(pid_file).read_text(encoding="utf-8").strip()) if not _bt._verify_reapable_browser_daemon(daemon_pid, socket_dir, session_name): _bt.logger.debug( "Skipped daemon kill for %s: pid %s failed identity " "verification", session_name, daemon_pid) return False - from gateway.status import get_process_start_time - daemon_start = get_process_start_time(daemon_pid) - if daemon_start is None: + if not _terminate_verified_daemon(daemon_pid): _bt.logger.debug( "Skipped daemon kill for %s: no start-time " "fingerprint for pid %s", session_name, daemon_pid) return False - ProcessRegistry._terminate_host_pid(daemon_pid, daemon_start) _bt.logger.debug("Killed daemon pid %s for %s", daemon_pid, session_name) return True except (ProcessLookupError, ValueError, PermissionError, OSError): @@ -716,21 +636,13 @@ def _kill_verified_daemon(socket_dir: str, session_name: str) -> bool: def _release_session_resources(task_id: str, session_info: Dict[str, Any]) -> None: """Untrack ``task_id``, close its cloud provider session, kill its daemon. - The unconditional tail of ``_cleanup_single_browser_session``; also the - whole of the janitor's force-reap path, which skips the polite - agent-browser/Camofox ``close`` that kept failing but must still release - the cloud session and the local Chromium. + Unconditional tail of ``_cleanup_single_browser_session``; also the whole of the + janitor's force-reap path, which skips the polite ``close`` that kept failing. """ - _bt = _origin() bb_session_id = session_info.get("bb_session_id", "unknown") - with _bt._cleanup_lock: - _bt._active_sessions.pop(task_id, None) - _bt._session_last_activity.pop(task_id, None) - _bt._session_owner_homes.pop(task_id, None) - _bt._cleanup_failures.pop(task_id, None) + _forget_session_tracking(task_id, session=True) - # Cloud mode only — local sidecars have bb_session_id=None. - if bb_session_id: + if bb_session_id: # cloud only — local sidecars have bb_session_id=None provider = _bt._get_cloud_provider() if provider is not None: try: @@ -747,12 +659,8 @@ def _release_session_resources(task_id: str, session_info: Dict[str, Any]) -> No def _force_reap_browser_session(task_id: str) -> None: - """Janitor last resort after repeated cleanup failures. - - Skips the ``close`` round-trips that keep failing and goes straight to - ``_release_session_resources`` (cloud close + daemon kill + untrack). - """ - _bt = _origin() + """Janitor last resort after repeated failures: skip the failing ``close`` + round-trips, go straight to ``_release_session_resources``.""" _bt._stop_cdp_supervisor(task_id) with _bt._cleanup_lock: session_info = _bt._active_sessions.get(task_id) @@ -764,16 +672,13 @@ def _force_reap_browser_session(task_id: str) -> None: def _cleanup_single_browser_session(task_id: str) -> None: - """Internal: reap a single browser session by its exact session key.""" - # Stop the CDP supervisor for this task FIRST so we close our WebSocket - # before the backend tears down the underlying CDP endpoint. - _bt = _origin() + """Reap a single browser session by its exact session key.""" + # Stop the CDP supervisor FIRST so our WebSocket closes before the backend + # tears down the CDP endpoint. _bt._stop_cdp_supervisor(task_id) - # Also clean up Camofox session if running in Camofox mode. - # Skip full close when managed persistence is enabled — the browser - # profile (and its session cookies) must survive across agent tasks. - # The inactivity reaper still frees idle resources. + # Camofox: skip the full close when managed persistence is on — the profile + # (session cookies) must survive across tasks; the inactivity reaper still frees idle resources. if _bt._is_camofox_mode(): try: from tools.browser_camofox import camofox_close, camofox_soft_cleanup @@ -785,57 +690,47 @@ def _cleanup_single_browser_session(task_id: str) -> None: _bt.logger.debug("cleanup_browser called for task_id: %s", task_id) _bt.logger.debug("Active sessions: %s", list(_bt._active_sessions.keys())) - # Check if session exists (under lock), but don't remove yet - - # _run_browser_command needs it to build the close command. + # Look up (under lock) but don't remove yet — _run_browser_command needs the + # entry to build the close command. with _bt._cleanup_lock: session_info = _bt._active_sessions.get(task_id) - if session_info: - bb_session_id = session_info.get("bb_session_id", "unknown") - _bt.logger.debug("Found session for task %s: bb_session_id=%s", task_id, bb_session_id) - - # Stop auto-recording before closing (saves the file) - _bt._maybe_stop_recording(task_id) - - # A Lightpanda session is a process Hermes spawned itself (Browser - # Use mode); there is no agent-browser daemon to send ``close`` to. - # An expired cloud CDP URL cannot accept an agent-browser close command. - # Avoid feeding it back through _get_session_info(), which would try to - # renew the session recursively while cleanup is still in progress. - if (session_info.get("features") or {}).get("lightpanda"): - try: - from tools.browser_lightpanda import stop_lightpanda - - stop_lightpanda(session_info.get("session_name", "")) - except Exception as e: - _bt.logger.warning("lightpanda stop failed for task %s: %s", task_id, e) - elif _bt._session_has_expired(session_info): - _bt.logger.debug( - "Skipping agent-browser close for expired session %s", task_id - ) - else: - try: - _bt._run_browser_command(task_id, "close", [], timeout=10) - _bt.logger.debug( - "agent-browser close command completed for task %s", task_id - ) - except Exception as e: - _bt.logger.warning("agent-browser close failed for task %s: %s", task_id, e) - - _bt._release_session_resources(task_id, session_info) - - _bt.logger.debug("Removed task %s from active sessions", task_id) - else: + if not session_info: _bt.logger.debug("No active session found for task_id: %s", task_id) + return + + _bt.logger.debug("Found session for task %s: bb_session_id=%s", task_id, session_info.get("bb_session_id", "unknown")) + _bt._maybe_stop_recording(task_id) # saves the file before close + + # Lightpanda sessions are processes Hermes spawned (no daemon to ``close``). + # An expired cloud CDP URL cannot accept a close, and feeding it through + # _get_session_info() would try to renew the session recursively mid-cleanup. + if (session_info.get("features") or {}).get("lightpanda"): + try: + from tools.browser_lightpanda import stop_lightpanda + + stop_lightpanda(session_info.get("session_name", "")) + except Exception as e: + _bt.logger.warning("lightpanda stop failed for task %s: %s", task_id, e) + elif _bt._session_has_expired(session_info): + _bt.logger.debug( + "Skipping agent-browser close for expired session %s", task_id + ) + else: + try: + _bt._run_browser_command(task_id, "close", [], timeout=10) + _bt.logger.debug( + "agent-browser close command completed for task %s", task_id + ) + except Exception as e: + _bt.logger.warning("agent-browser close failed for task %s: %s", task_id, e) + + _bt._release_session_resources(task_id, session_info) + _bt.logger.debug("Removed task %s from active sessions", task_id) def cleanup_all_browsers() -> None: - """ - Clean up all active browser sessions. - - Useful for cleanup on shutdown. - """ - _bt = _origin() + """Clean up all active browser sessions (shutdown) and reset cached lookups.""" with _bt._cleanup_lock: task_ids = list(_bt._active_sessions.keys()) for task_id in task_ids: @@ -848,12 +743,11 @@ def cleanup_all_browsers() -> None: except Exception: pass - # Reset cached lookups so they are re-evaluated on next use. _bt._cached_agent_browser = None _bt._agent_browser_resolved = False _bt._discover_homebrew_node_dirs.cache_clear() - # Flip the resolved flag BEFORE nulling the cache so a concurrent - # reader never sees ``resolved=True`` with ``cache=None``. + # Flip the resolved flag BEFORE nulling the cache so a concurrent reader never + # sees ``resolved=True`` with ``cache=None``. _bt._command_timeout_resolved = False _bt._cached_command_timeout = None _bt._snapshot_threshold_resolved = False diff --git a/tools/browser_tool_origin.py b/tools/browser_tool_origin.py index b6cf05fec7..f12f3c1472 100644 --- a/tools/browser_tool_origin.py +++ b/tools/browser_tool_origin.py @@ -1,10 +1,9 @@ """Origin-module lookup shared by the ``tools.browser_tool_*`` extraction modules. -Code extracted from ``tools/browser_tool.py`` must keep reading its origin's -symbols (helpers, module state dicts, ``logger``) *through* ``tools.browser_tool`` -so ``patch("tools.browser_tool.X")`` in tests is honoured. The extraction modules -must not import ``tools.browser_tool`` at import time (cycle), so they call -:func:`origin_module` lazily per call. +Extracted code must keep reading its origin's symbols (helpers, module state, +``logger``) *through* ``tools.browser_tool`` so ``patch("tools.browser_tool.X")`` +in tests is honoured. The extraction modules must not import ``tools.browser_tool`` +at import time (cycle), so they call :func:`origin_module` lazily per call. """ import sys @@ -14,12 +13,9 @@ _ORIGIN_NAME = "tools.browser_tool" class _NamespaceView: - """Live attribute view over a module namespace whose module object is gone. - - Used when a test purged ``sys.modules`` after importing the origin: the old - module's code still runs with its own globals, so reads must see *those* - (patched) globals, not a fresh re-import. - """ + """Live attribute view over a module namespace whose module object is gone (a + test purged ``sys.modules`` after importing the origin): the old code still runs + with its own globals, so reads/writes must hit *those*, not a fresh re-import.""" __slots__ = ("_g",) @@ -38,30 +34,26 @@ class _NamespaceView: def _module_for_globals(g: dict): """The module object whose namespace is ``g`` (or a live view over it).""" - mod = sys.modules.get(_ORIGIN_NAME) - if mod is not None and mod.__dict__ is g: - return mod import tools - mod = getattr(tools, "browser_tool", None) - if mod is not None and mod.__dict__ is g: - return mod + for mod in (sys.modules.get(_ORIGIN_NAME), getattr(tools, "browser_tool", None)): + if mod is not None and mod.__dict__ is g: + return mod return _NamespaceView(g) -def origin_module(): +def origin_module(_depth: int = 2): """Return the ``tools.browser_tool`` instance the *calling* moved function belongs to. - Resolution order, mirroring what in-file code would see: - 1. the nearest enclosing frame executing ``tools.browser_tool`` code — a moved - helper called from origin code binds to that exact module copy, even after a - test purged/reloaded ``sys.modules`` (several copies may be alive); - 2. a ``tools.browser_tool`` module referenced from a calling frame's globals - (``from tools import browser_tool`` in a test that calls the helper directly); - 3. ``sys.modules`` / the ``tools`` package attribute / a fresh import. + Resolution order, mirroring what in-file code would see: (1) the nearest + enclosing frame executing ``tools.browser_tool`` code — binds to that exact + module copy even after a test purged/reloaded ``sys.modules``; (2) a + ``tools.browser_tool`` module referenced from a calling frame's globals; + (3) ``sys.modules`` / the ``tools`` package attribute / a fresh import. + ``_depth`` is the frame of the moved function's caller. """ try: - start = sys._getframe(2) + start = sys._getframe(_depth) except ValueError: # called directly by the interpreter (atexit callback) start = None frame = start @@ -83,3 +75,20 @@ def origin_module(): if mod is None: import tools.browser_tool as mod return mod + + +class _OriginProxy: + """Module-level ``_bt`` stand-in: every attribute read/write forwards to the + origin module resolved at that moment, so moved functions need no per-call + ``_bt = _origin()`` line.""" + + __slots__ = () + + def __getattr__(self, name): + return getattr(origin_module(3), name) + + def __setattr__(self, name, value): + setattr(origin_module(3), name, value) + + +origin = _OriginProxy() diff --git a/tools/browser_tool_session.py b/tools/browser_tool_session.py index bf6d3f56a3..3341a94720 100644 --- a/tools/browser_tool_session.py +++ b/tools/browser_tool_session.py @@ -1,9 +1,10 @@ -"""agent-browser session management: daemon spawn, per-backend session creation (local/lightpanda/cdp/cloud), cached session lookup, command execution with timeout handling and output interpretation. +"""agent-browser session management: daemon spawn, per-backend session creation +(local/lightpanda/cdp/cloud), cached session lookup, command execution with timeout +handling and output interpretation. -Split out of ``tools/browser_tool.py``; every name is re-imported there so -``tools.browser_tool.`` keeps resolving (and monkeypatching). Origin +Split out of ``tools/browser_tool.py``; every name is re-imported there. Origin symbols and module state are read/written through ``_bt`` (the origin module, -resolved per call by :func:`tools.browser_tool_origin.origin_module`) so +the :data:`tools.browser_tool_origin.origin` proxy) so ``patch("tools.browser_tool.X")`` is honoured and no import cycle exists. """ @@ -12,32 +13,39 @@ import logging import os import shutil import subprocess +import uuid from pathlib import Path from typing import Any, Dict, List, Optional -from tools.browser_tool_origin import origin_module as _origin +from tools.browser_tool_origin import origin as _bt + +_CHROMIUM_MISSING_DOCKER_HINT = ( + "Chromium browser is missing. You're running in Docker — pull " + "the latest image to get the bundled Chromium: " + "docker pull ghcr.io/nousresearch/hermes-agent:latest" +) +_CHROMIUM_MISSING_HINT = ( + "Chromium browser is missing. Install it with: " + "npx agent-browser install --with-deps " + "(or: npx playwright install --with-deps chromium)" +) def _needs_chromium_sandbox_bypass() -> bool: - """Return True when Chromium needs --no-sandbox to start reliably.""" - _bt = _origin() + """True when Chromium needs --no-sandbox to start reliably (root, Docker, AppArmor userns).""" if hasattr(os, "geteuid") and os.geteuid() == 0: return True if _bt._running_in_docker(): return True - userns_restrict = "/proc/sys/kernel/apparmor_restrict_unprivileged_userns" try: - with open(userns_restrict, encoding="utf-8") as f: - if f.read().strip() == "1": - return True + with open("/proc/sys/kernel/apparmor_restrict_unprivileged_userns", encoding="utf-8") as f: + return f.read().strip() == "1" except OSError: - pass - return False + return False def _apply_chromium_sandbox_args(browser_env: Dict[str, str]) -> None: """Add required Chromium sandbox flags without overriding user settings.""" - _bt = _origin() if ( "AGENT_BROWSER_ARGS" not in browser_env and "AGENT_BROWSER_CHROME_FLAGS" not in browser_env @@ -52,18 +60,14 @@ def _apply_chromium_sandbox_args(browser_env: Dict[str, str]) -> None: def _read_command_output_files(stdout_path: str, stderr_path: str) -> tuple[str, str]: """Best-effort read of agent-browser stdout/stderr temp files.""" - stdout = stderr = "" - for path, slot in ((stdout_path, "stdout"), (stderr_path, "stderr")): + out = [] + for path in (stdout_path, stderr_path): try: with open(path, "r", encoding="utf-8") as f: - text = f.read().strip() + out.append(f.read().strip()) except OSError: - continue - if slot == "stdout": - stdout = text - else: - stderr = text - return stdout, stderr + out.append("") + return out[0], out[1] def _unlink_command_output_files(*paths: str) -> None: @@ -77,53 +81,46 @@ def _unlink_command_output_files(*paths: str) -> None: def _format_browser_timeout_error( command: str, timeout: int, stdout: str, stderr: str ) -> str: - """Build an actionable timeout message from captured daemon output.""" - _bt = _origin() + """Actionable timeout message from captured daemon output.""" parts = [f"Command timed out after {timeout} seconds"] detail = (stderr or stdout or "").strip() if detail: parts.append(detail[:1500]) combined = f"{stderr}\n{stdout}".lower() - hints: list[str] = [] if "sandbox" in combined: - hints.append( + parts.append( "Chromium sandbox launch failed. Set AGENT_BROWSER_ARGS=" "'--no-sandbox,--disable-dev-shm-usage' in your environment, " "or run: npx agent-browser install --with-deps" ) elif command == "open" and _bt._is_local_mode(): if _bt._running_in_docker(): - hints.append( + parts.append( "The browser daemon may still be starting or Chromium may be " "missing. Pull the latest image: " "docker pull ghcr.io/nousresearch/hermes-agent:latest" ) else: - hints.append( + parts.append( "The browser daemon may still be starting, or Chromium may be " "missing system libraries. Install/repair with: " "npx agent-browser install --with-deps " "(or: npx playwright install --with-deps chromium)" ) - if hints: - parts.extend(hints) return "\n".join(parts) def _agent_browser_argv(browser_cmd: str) -> list: """Command prefix to invoke agent-browser (concrete binary or npx sentinel). - Concrete executable paths stay a single argv item (spaces intact); only the - synthetic npx sentinel expands. npx is resolved through the same - PATH + extended-PATH cascade ``_find_agent_browser`` uses — a bare - ``shutil.which("npx")`` would let a broken system npx shadow a healthy - Hermes-managed one. If npx isn't found at all (Termux, bare container) the - bare name is used so Popen raises a readable ``FileNotFoundError: 'npx'``. - ``--ignore-scripts``: AGENT_BROWSER_NPX_SPEC is a floating range, not an - exact pin — a compromised future patch must not run install-time scripts. + Concrete paths stay a single argv item; only the npx sentinel expands. npx is + resolved through the same PATH cascade as ``_find_agent_browser`` (a bare + ``shutil.which("npx")`` would let a broken system npx shadow a healthy managed + one); if absent the bare name gives a readable ``FileNotFoundError: 'npx'``. + ``--ignore-scripts``: AGENT_BROWSER_NPX_SPEC is a floating range — a compromised + future patch must not run install-time scripts. """ - _bt = _origin() if _bt._is_npx_agent_browser_sentinel(browser_cmd): _npx_bin = _bt._resolve_npx_bin() or "npx" return [_npx_bin, "--ignore-scripts", "--prefer-offline", "-y", _bt.AGENT_BROWSER_NPX_SPEC] @@ -131,15 +128,12 @@ def _agent_browser_argv(browser_cmd: str) -> list: def _prepare_session_socket_dir(session_name: str) -> str: - """Create the per-session agent-browser socket dir and claim it with our PID. + """Create the per-session socket dir and claim it with our PID. - Each session gets its own dir so parallel workers don't fight over the - default socket path ("Failed to create socket directory: Permission - denied"). The owner_pid file is written BEFORE first use: another hermes - process's orphan reaper rmtree's any agent-browser-* dir in the shared - tmpdir that carries no live owner, which would delete this one mid-command. + Per-session dirs keep parallel workers from fighting over the default socket + path. The owner_pid file is written BEFORE first use: another hermes process's + orphan reaper rmtree's any ownerless agent-browser-* dir in the shared tmpdir. """ - _bt = _origin() socket_dir = os.path.join(_bt._socket_safe_tmpdir(), f"agent-browser-{session_name}") os.makedirs(socket_dir, mode=0o700, exist_ok=True) _bt._write_owner_pid(socket_dir, session_name) @@ -147,14 +141,10 @@ def _prepare_session_socket_dir(session_name: str) -> str: def _agent_browser_command_env(socket_dir: str) -> Dict[str, str]: - """Credential-scrubbed env for one agent-browser command. - - Adds the discovery-time PATH fallbacks, the session socket dir, and the - daemon-side idle self-termination (``AGENT_BROWSER_IDLE_TIMEOUT_MS``, - agent-browser 0.24+) mirroring the Python-side inactivity janitor — - unless the user set the idle timeout explicitly. - """ - _bt = _origin() + """Credential-scrubbed env for one agent-browser command: discovery-time PATH + fallbacks, the session socket dir, and daemon-side idle self-termination + (``AGENT_BROWSER_IDLE_TIMEOUT_MS``, agent-browser 0.24+) mirroring the Python + janitor — unless the user set it explicitly.""" env = _bt._build_browser_env() env["PATH"] = _bt._merge_browser_path(env.get("PATH", "")) env["AGENT_BROWSER_SOCKET_DIR"] = socket_dir @@ -166,16 +156,13 @@ def _agent_browser_command_env(socket_dir: str) -> Dict[str, str]: def _popen_agent_browser(argv: List[str], env: Dict[str, str], socket_dir: str, tag: str) -> "subprocess.Popen": """Spawn agent-browser with stdout/stderr redirected to ``socket_dir/_stdout_``. - Temp files instead of pipes: the CLI forks a background daemon that inherits - its fds, so with pipes ``communicate()`` never sees EOF until the timeout. - Windows: CREATE_NO_WINDOW only (NOT CREATE_NEW_PROCESS_GROUP, which on - Python 3.11 cancels asyncio's running loop task and surfaces as - KeyboardInterrupt in the CLI), STARTF_USESTDHANDLES so CreateProcess hands - the child ONLY our three handles (leaked parent console handles make the - Rust binary's daemon grandchild die silently), close_fds=True for the rest. - Returns the Popen; the caller reads/unlinks the two files. + Temp files instead of pipes: the CLI forks a daemon that inherits its fds, so + with pipes ``communicate()`` never sees EOF until the timeout. Windows: + CREATE_NO_WINDOW only (NOT CREATE_NEW_PROCESS_GROUP — on 3.11 it cancels + asyncio's running task and surfaces as KeyboardInterrupt), STARTF_USESTDHANDLES + so the child gets ONLY our three handles (leaked console handles make the Rust + daemon grandchild die silently), close_fds=True for the rest. """ - _bt = _origin() stdout_path = os.path.join(socket_dir, f"_stdout_{tag}") stderr_path = os.path.join(socket_dir, f"_stderr_{tag}") stdout_fd = os.open(stdout_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) @@ -197,57 +184,49 @@ def _popen_agent_browser(argv: List[str], env: Dict[str, str], socket_dir: str, os.close(stderr_fd) -def _create_local_session(task_id: str, allow_real_profile: bool = True) -> Dict[str, str]: - _bt = _origin() - import uuid +def _session_record(prefix: str, cdp_url: Optional[str], features: Dict[str, Any]) -> Dict[str, Any]: + """Fresh session dict with a random ``_`` session name.""" + return { + "session_name": f"{prefix}_{uuid.uuid4().hex[:10]}", + "bb_session_id": None, + "cdp_url": cdp_url, + "features": features, + } - # Real-profile consent: attach this local session (via CDP) to the user's - # browser running on a hermes-owned SNAPSHOT of their real profile, logins - # included. Fail closed on resolver/launch errors — a consented user must - # never be silently downgraded to a throwaway. The hybrid private-URL - # sidecar passes allow_real_profile=False: handing the user's cookie jar to - # an arbitrary internal host the model chose is a larger, unconsented - # exposure than the routing rule protects against (and a real-profile - # failure must not break private-URL routing). + +def _create_local_session(task_id: str, allow_real_profile: bool = True) -> Dict[str, str]: + """Local Chromium session; consented real-profile CDP attach when allowed. + + Real-profile fails closed on resolver/launch errors — a consented user must + never be silently downgraded to a throwaway. The hybrid private-URL sidecar + passes ``allow_real_profile=False``: handing the user's cookie jar to an + arbitrary internal host the model chose is a larger, unconsented exposure. + """ if allow_real_profile: cdp_url, err = _bt._real_profile_cdp() if err: raise RuntimeError(err) if cdp_url: - session_name = f"rp_{uuid.uuid4().hex[:10]}" + info = _session_record("rp", _bt._resolve_cdp_override(cdp_url), {"local": True, "real_profile": True}) _bt.logger.info( - "Created real-profile local session %s for task %s", session_name, task_id + "Created real-profile local session %s for task %s", info["session_name"], task_id ) - return { - "session_name": session_name, - "bb_session_id": None, - "cdp_url": _bt._resolve_cdp_override(cdp_url), - "features": {"local": True, "real_profile": True}, - } + return info # Browser Use mode drives whatever CDP endpoint it is handed; with - # ``browser.engine: lightpanda`` that endpoint is a Hermes-spawned - # ``lightpanda serve``. The built-in tools never reach this branch — - # they are hidden in Browser Use mode — and keep driving Lightpanda via - # ``agent-browser --engine lightpanda`` on the plain local session below. + # ``browser.engine: lightpanda`` that is a Hermes-spawned ``lightpanda serve``. + # The built-in tools never reach this branch (hidden in Browser Use mode). if _bt._is_browser_use_cli_mode() and _bt._using_lightpanda_engine(): return _bt._create_lightpanda_session(task_id) - session_name = f"h_{uuid.uuid4().hex[:10]}" + info = _session_record("h", None, {"local": True}) _bt.logger.info("Created local browser session %s for task %s", - session_name, task_id) - return { - "session_name": session_name, - "bb_session_id": None, - "cdp_url": None, - "features": {"local": True}, - } + info["session_name"], task_id) + return info def _create_lightpanda_session(task_id: str) -> Dict[str, Any]: """Spawn ``lightpanda serve`` for this session key (Browser Use mode).""" - _bt = _origin() - import uuid from tools.browser_lightpanda import launch_lightpanda session_name = f"lp_{uuid.uuid4().hex[:10]}" @@ -278,27 +257,19 @@ def _local_backend_process_dead(session_info: Dict[str, Any]) -> bool: def _create_cdp_session(task_id: str, cdp_url: str) -> Dict[str, str]: - """Create a session that connects to a user-supplied CDP endpoint.""" - _bt = _origin() - import uuid - session_name = f"cdp_{uuid.uuid4().hex[:10]}" + """Session connecting to a user-supplied CDP endpoint.""" + info = _session_record("cdp", cdp_url, {"cdp_override": True}) _bt.logger.info("Created CDP browser session %s → %s for task %s", - session_name, _bt._sanitize_url_for_logs(cdp_url), task_id) - return { - "session_name": session_name, - "bb_session_id": None, - "cdp_url": cdp_url, - "features": {"cdp_override": True}, - } + info["session_name"], _bt._sanitize_url_for_logs(cdp_url), task_id) + return info def _create_cloud_session_or_fallback(task_id: str, provider) -> Dict[str, Any]: - """Create a cloud session; fall back to local Chromium (marked degraded) on failure. + """Cloud session; fall back to local Chromium (marked degraded) on failure. - Some cloud providers (Browser-Use v3) return an HTTP CDP discovery URL - instead of a raw websocket endpoint, so ``cdp_url`` is resolved here. + Some providers (Browser-Use v3) return an HTTP CDP discovery URL instead of a + raw websocket endpoint, so ``cdp_url`` is resolved here. """ - _bt = _origin() try: session_info = provider.create_session(task_id) if not session_info or not isinstance(session_info, dict): @@ -322,8 +293,7 @@ def _create_cloud_session_or_fallback(task_id: str, provider) -> Dict[str, Any]: f"Cloud provider {provider_name} failed ({e}) and local " f"fallback also failed ({local_error})" ) from e - # Mark session as degraded for observability - if isinstance(session_info, dict): + if isinstance(session_info, dict): # mark degraded for observability session_info = dict(session_info) session_info["fallback_from_cloud"] = True session_info["fallback_reason"] = str(e) @@ -332,14 +302,11 @@ def _create_cloud_session_or_fallback(task_id: str, provider) -> Dict[str, Any]: def _create_session_for_key(task_id: str, force_local: bool) -> Dict[str, Any]: - """Create a fresh session for ``task_id`` (runs OUTSIDE the lock: cloud mode makes a network call). + """Fresh session for ``task_id`` (runs OUTSIDE the lock: cloud mode makes a network call). - Precedence: CDP override > hybrid local sidecar > cloud provider > local. - The hybrid private-URL sidecar NEVER gets the real profile — presenting real - cookies to an arbitrary LAN host the model routed there is unconsented - exposure (see ``_create_local_session``). + Precedence: CDP override > hybrid local sidecar > cloud provider > local. The + hybrid sidecar NEVER gets the real profile (see ``_create_local_session``). """ - _bt = _origin() cdp_override = _bt._get_cdp_override() if cdp_override and not force_local: return _bt._create_cdp_session(task_id, cdp_override) @@ -354,94 +321,72 @@ def _create_session_for_key(task_id: str, force_local: bool) -> Dict[str, Any]: def _get_session_info(task_id: Optional[str] = None) -> Dict[str, Any]: """Get or create session info for a session key (thread-safe). - ``task_id`` may carry the ``::local`` suffix (hybrid local sidecar), which - forces a local Chromium even when a cloud provider is configured. Also - starts the inactivity cleanup thread and touches activity tracking. - Returns a dict with ``session_name`` (always) plus ``bb_session_id`` / + A ``::local``-suffixed key (hybrid sidecar) forces local Chromium even with a + cloud provider configured. Also starts the inactivity thread and touches + activity tracking. Returns ``session_name`` (always) plus ``bb_session_id`` / ``cdp_url`` for cloud sessions. """ - _bt = _origin() if task_id is None: task_id = "default" - # Start the cleanup thread if not running (handles inactivity timeouts) _bt._start_browser_cleanup_thread() - - # Update activity timestamp for this session _bt._update_session_activity(task_id) with _bt._cleanup_lock: - # Check if we already have a session for this task existing_session = _bt._active_sessions.get(task_id) - # Suspect-session recycle: a previous command - # timeout marked this cached session suspect via the SuspectableBackend - # adapter. ensure_healthy() tears it down here, at next use, and we fall - # through to create a fresh session — the expensive recycle lives on this - # path, not on the timeout path (mark must stay cheap). - if existing_session is not None and not _bt._browser_session_backend(task_id).ensure_healthy(): - # Teardown removes the activity entry; the replacement must be - # tracked by the inactivity reaper like an initial session. + def _replacement_after_teardown() -> Optional[Dict[str, Any]]: + # Teardown removes the activity entry; the replacement must be tracked by + # the reaper like an initial session. Another thread may already have + # recycled and re-created it — return that live one instead of a third. _bt._update_session_activity(task_id) with _bt._cleanup_lock: replacement = _bt._active_sessions.get(task_id) if replacement is not None and replacement is not existing_session: - # Another thread already recycled and re-created it. return replacement - existing_session = None + return None if existing_session is not None: - if ( + # Suspect-session recycle: a previous command timeout marked this cached + # session via the SuspectableBackend adapter; the expensive recycle lives + # here at next use, not on the timeout path (mark must stay cheap). + if not _bt._browser_session_backend(task_id).ensure_healthy(): + replacement = _replacement_after_teardown() + if replacement is not None: + return replacement + existing_session = None + elif ( not _bt._session_has_expired(existing_session) and not _bt._local_backend_process_dead(existing_session) ): return existing_session + else: + _bt.logger.info( + "Replacing expired or dead browser session for task %s", task_id + ) + _bt._cleanup_single_browser_session(task_id) + replacement = _replacement_after_teardown() + if replacement is not None: + return replacement - _bt.logger.info( - "Replacing expired or dead browser session for task %s", task_id - ) - _bt._cleanup_single_browser_session(task_id) - # Cleanup removes the activity entry. The replacement session must be - # tracked by the inactivity reaper just like an initial session. - _bt._update_session_activity(task_id) - - # Guard against a concurrent replacement: another thread may have - # already cleaned up the expired session and created a fresh one - # while we were waiting. If so, return the live replacement instead - # of falling through to create yet another session. - with _bt._cleanup_lock: - replacement = _bt._active_sessions.get(task_id) - if replacement is not None and replacement is not existing_session: - return replacement - - # Hybrid routing: session keys ending with ``::local`` force a local - # Chromium regardless of the globally-configured cloud provider. Public - # URLs in the same conversation continue to use the cloud session under - # the bare task_id key. force_local = _bt._is_local_sidecar_key(task_id) session_info = _bt._create_session_for_key(task_id, force_local) with _bt._cleanup_lock: - # Double-check: another thread may have created a session while we - # were doing the network call. Use the existing one to avoid leaking - # orphan cloud sessions. + # Another thread may have created a session during the network call; use + # it to avoid leaking orphan cloud sessions. if task_id in _bt._active_sessions: return _bt._active_sessions[task_id] session_info = dict(session_info) session_info.setdefault("session_key", task_id) session_info.setdefault("owner_task_id", _bt._bare_task_id_for_session_key(task_id)) _bt._active_sessions[task_id] = session_info - # A brand-new session is healthy by definition — drop any stale - # suspect flag left by a wedged-path eviction of its predecessor. + # A brand-new session is healthy by definition — drop any stale suspect flag. _bt._suspect_browser_sessions.pop(task_id, None) - # Lazy-start the CDP supervisor now that the session exists (if the - # backend surfaces a CDP URL via override or session_info["cdp_url"]). - # Idempotent; swallows errors. See _ensure_cdp_supervisor for details. - # Skip for local sidecars — they have no CDP URL — and for Lightpanda - # sessions: those only exist in Browser Use mode, where the browser_* - # tools that consume supervisor state are hidden, so the supervisor - # would just hold an idle second CDP connection to the process. + # Lazy-start the CDP supervisor (idempotent; swallows errors). Skip for local + # sidecars (no CDP URL) and Lightpanda sessions (Browser Use mode hides the + # browser_* tools that consume supervisor state; it would just idle a second CDP connection). if not force_local and not (session_info.get("features") or {}).get("lightpanda"): _bt._ensure_cdp_supervisor(task_id) @@ -452,13 +397,11 @@ def _discard_timed_out_browser_session( task_id: str, session_info: Dict[str, Any], task_socket_dir: str ) -> None: """Drop a stuck client generation without losing cloud cleanup state.""" - _bt = _origin() with _bt._cleanup_lock: if _bt._active_sessions.get(task_id) is not session_info: return _bt._stop_cdp_supervisor(task_id) if session_info.get("bb_session_id") or session_info.get("cdp_url"): - import uuid replacement = dict(session_info) replacement["session_name"] = f"h_{uuid.uuid4().hex[:10]}" replacement.pop("_first_nav", None) @@ -472,23 +415,21 @@ def _discard_timed_out_browser_session( _bt._last_active_session_key.pop(bare_task_id, None) session_name = str(session_info.get("session_name") or "") - if session_name: - pid_file = os.path.join(task_socket_dir, f"{session_name}.pid") - if os.path.isfile(pid_file): - try: - daemon_pid = int(Path(pid_file).read_text(encoding="utf-8").strip()) - if not _bt._verify_reapable_browser_daemon(daemon_pid, task_socket_dir, session_name): - return - # Tree-kill: the daemon spawns Chromium - # children; terminating only the daemon PID leaks the whole - # Chromium tree. agent.deadline.kill_process_tree escalates - # SIGTERM → SIGKILL across the tree. - from agent import deadline as _deadline + if session_name and os.path.isfile(os.path.join(task_socket_dir, f"{session_name}.pid")): + daemon_pid = _bt._read_browser_daemon_pid(task_socket_dir, session_name) + if daemon_pid is None: # corrupt pid file + _bt.logger.debug("Could not kill timed-out browser daemon for %s", session_name) + return + if not _bt._verify_reapable_browser_daemon(daemon_pid, task_socket_dir, session_name): + return + try: + # Tree-kill: terminating only the daemon PID leaks the Chromium tree. + from agent import deadline as _deadline - _deadline.kill_process_tree(daemon_pid) - except (ProcessLookupError, ValueError, PermissionError, OSError): - _bt.logger.debug("Could not kill timed-out browser daemon for %s", session_name) - return + _deadline.kill_process_tree(daemon_pid) + except (ProcessLookupError, PermissionError, OSError): + _bt.logger.debug("Could not kill timed-out browser daemon for %s", session_name) + return shutil.rmtree(task_socket_dir, ignore_errors=True) @@ -504,9 +445,9 @@ def _read_browser_daemon_pid(task_socket_dir: str, session_name: str) -> Optiona def _browser_daemon_responsive(task_socket_dir: str, probe_timeout_s: float = 1.0) -> bool: """Cheap liveness probe: connect to the daemon's unix control socket. - A successful connect proves the accept loop is alive (the command wedged on - the page/CDP side, not the daemon). Windows uses named pipes — no probe is - possible, so report unresponsive (tree-kill + respawn is the safe recovery). + A successful connect proves the accept loop is alive (the command wedged on the + page/CDP side). Windows uses named pipes — no probe possible, so report + unresponsive (tree-kill + respawn is the safe recovery). """ if os.name == "nt": return False @@ -518,14 +459,13 @@ def _browser_daemon_responsive(task_socket_dir: str, probe_timeout_s: float = 1. entries = os.listdir(task_socket_dir) except OSError: return False - sock_paths = [ - os.path.join(task_socket_dir, e) for e in entries if e.endswith(".sock") - ] - for sock_path in sock_paths: + for entry in entries: + if not entry.endswith(".sock"): + continue try: with socket_mod.socket(socket_mod.AF_UNIX, socket_mod.SOCK_STREAM) as s: s.settimeout(probe_timeout_s) - s.connect(sock_path) + s.connect(os.path.join(task_socket_dir, entry)) return True except OSError: continue @@ -537,20 +477,16 @@ def _handle_browser_command_timeout( ) -> None: """Recover session state after a browser command timeout. - * Cloud / CDP sessions: no local daemon to probe — replace the stuck client - generation now (fresh ``session_name``, same ``bb_session_id`` so cloud - cleanup still works). - * Local daemon alive (PID live, identity-verified, control socket accepts): - only the *command* wedged; mark the session suspect and let the next use - recycle it via ``ensure_healthy`` → clean ``close`` → fresh session. + * Cloud / CDP: no local daemon to probe — replace the stuck client generation + now (fresh ``session_name``, same ``bb_session_id`` so cloud cleanup works). + * Local daemon alive (PID live, identity-verified, control socket accepts): only + the *command* wedged; mark suspect and let next use recycle via ``ensure_healthy``. * Local daemon wedged/dead: it cannot service a clean close and its Chromium - children would leak — tree-kill and evict now; the next call respawns. + children would leak — tree-kill and evict now. Both local branches ``mark_suspect`` first (cheap, lock-free) so the - poisoned-cache invariant holds even if eviction races another thread's - replacement (the flag then costs one harmless no-op teardown). + poisoned-cache invariant holds even if eviction races another thread's replacement. """ - _bt = _origin() if session_info.get("bb_session_id") or session_info.get("cdp_url"): _bt._discard_timed_out_browser_session(task_id, session_info, task_socket_dir) return @@ -579,10 +515,8 @@ def _handle_browser_command_timeout( "tree-killing and evicting the session", task_id, ) _bt._discard_timed_out_browser_session(task_id, session_info, task_socket_dir) - # The poisoned entry is gone (evicted, or superseded by a concurrent - # replacement discard refused to touch) — either way the cache no longer - # holds the timed-out session, so drop the flag: it must not poison a - # session created later under the same key. + # The poisoned entry is gone either way; the flag must not poison a session + # created later under the same key. _bt._suspect_browser_sessions.pop(task_id, None) @@ -590,11 +524,10 @@ def _interpret_browser_command_output(command: str, stdout: str, stderr: str, re """Turn a finished agent-browser process's output into a result dict. Empty stdout with rc=0 is a broken state (stale daemon) and is reported as - failure rather than a silent ``{"success": True, "data": {}}`` — except for - commands in ``_EMPTY_OK_COMMANDS``. Non-JSON output is an error, except - for ``screenshot`` where the saved path is recovered from the prose. + failure rather than a silent success — except for ``_EMPTY_OK_COMMANDS``. + Non-JSON output is an error, except ``screenshot`` where the saved path is + recovered from the prose. """ - _bt = _origin() if stderr and stderr.strip(): level = logging.WARNING if returncode != 0 else logging.DEBUG _bt.logger.log(level, "browser '%s' stderr: %s", command, stderr.strip()[:500]) @@ -637,6 +570,93 @@ def _interpret_browser_command_output(command: str, stdout: str, stderr: str, re return parsed +def _browser_command_preflight() -> Dict[str, Any]: + """Fail fast before spawning: missing CLI, Termux install gap, interrupt, or no + Chromium in local mode (else every call hangs for command_timeout). Returns an + error result, or ``{"browser_cmd": path}`` on success.""" + try: + browser_cmd = _bt._find_agent_browser() + except FileNotFoundError as e: + _bt.logger.warning("agent-browser CLI not found: %s", e) + return {"success": False, "error": str(e)} + + if _bt._requires_real_termux_browser_install(browser_cmd): + error = _bt._termux_browser_install_error() + _bt.logger.warning("browser command blocked on Termux: %s", error) + return {"success": False, "error": error} + + # Skip when engine=lightpanda — LP doesn't need Chromium for navigation. + if ( + _bt._is_local_mode() + and not _bt._chromium_installed() + and _bt._get_browser_engine() != "lightpanda" + and not _bt._maybe_autoinstall_chromium() + ): + hint = _CHROMIUM_MISSING_DOCKER_HINT if _bt._running_in_docker() else _CHROMIUM_MISSING_HINT + _bt.logger.warning("browser command blocked: %s", hint) + return {"success": False, "error": hint} + + from tools.interrupt import is_interrupted + if is_interrupted(): + return {"success": False, "error": "Interrupted"} + return {"browser_cmd": browser_cmd} + + +def _spawn_and_collect( + task_id: str, session_info: Dict[str, Any], cmd_parts: List[str], + command: str, engine: str, timeout: int, +) -> Dict[str, Any]: + """Run the prepared agent-browser argv once and interpret its output (handles timeout).""" + task_socket_dir = _bt._prepare_session_socket_dir(session_info["session_name"]) + _bt.logger.debug("browser cmd=%s task=%s socket_dir=%s (%d chars)", + command, task_id, task_socket_dir, len(task_socket_dir)) + browser_env = _bt._agent_browser_command_env(task_socket_dir) + + # Lightpanda rejects Chromium-only launch flags: strip current and legacy vars; + # Chrome commands and fallback use the shared Chromium policy. + if engine == "lightpanda": + _stripped_args = browser_env.pop("AGENT_BROWSER_ARGS", None) + _stripped_flags = browser_env.pop("AGENT_BROWSER_CHROME_FLAGS", None) + if _stripped_args is not None or _stripped_flags is not None: + _bt.logger.debug( + "browser: stripped Chromium-only AGENT_BROWSER_ARGS/" + "AGENT_BROWSER_CHROME_FLAGS for Lightpanda command %s " + "(agent-browser rejects them with --engine lightpanda)", + command, + ) + else: + _bt._apply_chromium_sandbox_args(browser_env) + + stdout_path = os.path.join(task_socket_dir, f"_stdout_{command}") + stderr_path = os.path.join(task_socket_dir, f"_stderr_{command}") + proc = _bt._popen_agent_browser(cmd_parts, browser_env, task_socket_dir, command) + + try: + proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + stdout, stderr = _bt._read_command_output_files(stdout_path, stderr_path) + _bt._unlink_command_output_files(stdout_path, stderr_path) + _bt._handle_browser_command_timeout(task_id, session_info, task_socket_dir) + if stderr and stderr.strip(): + _bt.logger.warning( + "browser '%s' stderr after timeout: %s", command, stderr.strip()[:500] + ) + _bt.logger.warning("browser '%s' timed out after %ds (task=%s, socket_dir=%s)", + command, timeout, task_id, task_socket_dir) + return { + "success": False, + "error": _bt._format_browser_timeout_error(command, timeout, stdout, stderr), + } + with open(stdout_path, "r", encoding="utf-8") as f: + stdout = f.read() + with open(stderr_path, "r", encoding="utf-8") as f: + stderr = f.read() + _bt._unlink_command_output_files(stdout_path, stderr_path) + return _bt._interpret_browser_command_output(command, stdout, stderr, proc.returncode) + + def _run_browser_command( task_id: str, command: str, @@ -650,52 +670,15 @@ def _run_browser_command( ``_engine_override`` forces an engine for this call only (the Lightpanda fallback uses it to retry with Chrome without touching global state). """ - _bt = _origin() if timeout is None: timeout = _bt._safe_command_timeout() args = args or [] - # Build the command - try: - browser_cmd = _bt._find_agent_browser() - except FileNotFoundError as e: - _bt.logger.warning("agent-browser CLI not found: %s", e) - return {"success": False, "error": str(e)} + preflight = _browser_command_preflight() + if "browser_cmd" not in preflight: + return preflight + browser_cmd = preflight["browser_cmd"] - if _bt._requires_real_termux_browser_install(browser_cmd): - error = _bt._termux_browser_install_error() - _bt.logger.warning("browser command blocked on Termux: %s", error) - return {"success": False, "error": error} - - # Local mode with no Chromium on disk: fail fast with an actionable - # message instead of hanging for _command_timeout seconds per call. - # Skip when engine=lightpanda — LP doesn't need Chromium for navigation. - if ( - _bt._is_local_mode() - and not _bt._chromium_installed() - and _bt._get_browser_engine() != "lightpanda" - and not _bt._maybe_autoinstall_chromium() - ): - if _bt._running_in_docker(): - hint = ( - "Chromium browser is missing. You're running in Docker — pull " - "the latest image to get the bundled Chromium: " - "docker pull ghcr.io/nousresearch/hermes-agent:latest" - ) - else: - hint = ( - "Chromium browser is missing. Install it with: " - "npx agent-browser install --with-deps " - "(or: npx playwright install --with-deps chromium)" - ) - _bt.logger.warning("browser command blocked: %s", hint) - return {"success": False, "error": hint} - - from tools.interrupt import is_interrupted - if is_interrupted(): - return {"success": False, "error": "Interrupted"} - - # Get session info (creates Browserbase session with proxies if needed) try: session_info = _bt._get_session_info(task_id) except Exception as e: @@ -705,25 +688,17 @@ def _run_browser_command( if command != "close" and session_info.get("cdp_url"): _bt._ensure_cdp_supervisor(task_id) - # Build the command with the appropriate backend flag. - # Cloud mode: --cdp connects to Browserbase. - # Local mode: --session launches a local headless Chromium. - # The rest of the command (--json, command, args) is identical. + # Cloud/CDP: ``--cdp `` (NEVER with --session: agent-browser >=0.13 + # would create a local browser and silently ignore --cdp). Local: ``--session ``. if session_info.get("cdp_url"): - # Cloud mode — connect to remote Browserbase browser via CDP - # IMPORTANT: Do NOT use --session with --cdp. In agent-browser >=0.13, - # --session creates a local browser instance and silently ignores --cdp. backend_args = ["--cdp", session_info["cdp_url"]] else: - # Local mode — launch Chromium (headless by default, headed when configured) backend_args = ["--session", session_info["session_name"]] if _bt._is_headed_mode(): backend_args.append("--headed") - # Lightpanda engine injection (local mode only, agent-browser v0.25.3+). - # Use the resolved session backend rather than global cloud-provider state: - # hybrid private-URL routing can create a local sidecar while a cloud - # provider remains configured for public URLs. + # Engine injection keys off the resolved session backend, not global provider + # state: hybrid routing can create a local sidecar while a cloud provider stays configured. engine = _engine_override or _bt._get_browser_engine() if engine != "auto" and not _bt._is_camofox_mode() and not session_info.get("cdp_url"): backend_args += ["--engine", engine] @@ -731,65 +706,13 @@ def _run_browser_command( cmd_parts = _bt._agent_browser_argv(browser_cmd) + backend_args + ["--json", command] + args try: - task_socket_dir = _bt._prepare_session_socket_dir(session_info["session_name"]) - _bt.logger.debug("browser cmd=%s task=%s socket_dir=%s (%d chars)", - command, task_id, task_socket_dir, len(task_socket_dir)) - browser_env = _bt._agent_browser_command_env(task_socket_dir) - - # Chromium-only launch flags are rejected by Lightpanda. Strip both - # the current and legacy variables for Lightpanda commands; explicit - # Chrome commands and fallback use the shared Chromium policy. - if engine == "lightpanda": - _stripped_args = browser_env.pop("AGENT_BROWSER_ARGS", None) - _stripped_flags = browser_env.pop("AGENT_BROWSER_CHROME_FLAGS", None) - if _stripped_args is not None or _stripped_flags is not None: - _bt.logger.debug( - "browser: stripped Chromium-only AGENT_BROWSER_ARGS/" - "AGENT_BROWSER_CHROME_FLAGS for Lightpanda command %s " - "(agent-browser rejects them with --engine lightpanda)", - command, - ) - else: - _bt._apply_chromium_sandbox_args(browser_env) - - stdout_path = os.path.join(task_socket_dir, f"_stdout_{command}") - stderr_path = os.path.join(task_socket_dir, f"_stderr_{command}") - proc = _bt._popen_agent_browser(cmd_parts, browser_env, task_socket_dir, command) - - try: - proc.wait(timeout=timeout) - except subprocess.TimeoutExpired: - proc.kill() - proc.wait() - stdout, stderr = _bt._read_command_output_files(stdout_path, stderr_path) - _bt._unlink_command_output_files(stdout_path, stderr_path) - _bt._handle_browser_command_timeout(task_id, session_info, task_socket_dir) - if stderr and stderr.strip(): - _bt.logger.warning( - "browser '%s' stderr after timeout: %s", command, stderr.strip()[:500] - ) - _bt.logger.warning("browser '%s' timed out after %ds (task=%s, socket_dir=%s)", - command, timeout, task_id, task_socket_dir) - result = { - "success": False, - "error": _bt._format_browser_timeout_error(command, timeout, stdout, stderr), - } - # Fall through to fallback check below - else: - with open(stdout_path, "r", encoding="utf-8") as f: - stdout = f.read() - with open(stderr_path, "r", encoding="utf-8") as f: - stderr = f.read() - _bt._unlink_command_output_files(stdout_path, stderr_path) - result = _bt._interpret_browser_command_output(command, stdout, stderr, proc.returncode) - + result = _spawn_and_collect(task_id, session_info, cmd_parts, command, engine, timeout) except Exception as e: _bt.logger.warning("browser '%s' exception: %s", command, e, exc_info=True) result = {"success": False, "error": str(e)} - # --- Lightpanda automatic Chrome fallback --- - # If engine is lightpanda and the result looks broken, retry with Chrome. - # This runs for ALL exit paths (timeout, empty, non-JSON, nonzero rc, parsed). + # Lightpanda automatic Chrome fallback — runs for ALL exit paths (timeout, + # empty, non-JSON, nonzero rc, parsed). fallback_reason = _bt._lightpanda_fallback_reason(engine, command, result) if fallback_reason: _bt.logger.info( @@ -798,9 +721,7 @@ def _run_browser_command( task_id, fallback_reason, ) - # For screenshots, use the dedicated Chrome fallback helper - # (spins up a separate Chrome session to the same URL). - if command == "screenshot": + if command == "screenshot": # separate Chrome session to the same URL fallback_result = _bt._chrome_fallback_screenshot(task_id, args or [], timeout) else: fallback_result = _bt._run_chrome_fallback_command(task_id, command, args, timeout) diff --git a/tools/browser_tool_snapshot.py b/tools/browser_tool_snapshot.py index 0192e1e5f9..fcbe0c7c42 100644 --- a/tools/browser_tool_snapshot.py +++ b/tools/browser_tool_snapshot.py @@ -8,7 +8,14 @@ at import time (cycle). import re from typing import Any, Optional -from tools.browser_tool_origin import origin_module as _origin +from tools.browser_tool_origin import origin as _bt + + +_SCREENSHOT_PATH_PATTERNS = ( + r"Screenshot saved to ['\"](?P/[^'\"]+?\.png)['\"]", + r"Screenshot saved to (?P/\S+?\.png)(?:\s|$)", + r"(?P/\S+?\.png)(?:\s|$)", +) def _extract_screenshot_path_from_text(text: str) -> Optional[str]: @@ -16,19 +23,12 @@ def _extract_screenshot_path_from_text(text: str) -> Optional[str]: if not text: return None - patterns = [ - r"Screenshot saved to ['\"](?P/[^'\"]+?\.png)['\"]", - r"Screenshot saved to (?P/\S+?\.png)(?:\s|$)", - r"(?P/\S+?\.png)(?:\s|$)", - ] - - for pattern in patterns: + for pattern in _SCREENSHOT_PATH_PATTERNS: match = re.search(pattern, text) if match: path = match.group("path").strip().strip("'\"") if path: return path - return None @@ -40,7 +40,6 @@ def _store_full_snapshot(snapshot_text: str) -> Optional[str]: backend. The stored copy is force-redacted (page-rendered secrets must not hit disk unmasked) and named by content hash so identical snapshots dedupe. """ - _bt = _origin() try: import hashlib from hermes_constants import get_hermes_dir @@ -59,11 +58,9 @@ def _store_full_snapshot(snapshot_text: str) -> Optional[str]: ensure_spill_dir(cache_dir, private=False) digest = hashlib.sha256(content.encode("utf-8")).hexdigest()[:10] path = cache_dir / f"browser-snapshot-{digest}.txt" - # Deterministic filename in a well-known dir: refuse symlinks via - # lstat-unlink + exclusive create. Re-snapshotting the same page - # state legitimately overwrites (same content-hash name). Not - # private: cache/web is bind-mounted into remote backends whose - # container UID must be able to read it. + # Deterministic filename in a well-known dir: refuse symlinks (lstat-unlink + + # exclusive create); same-content re-snapshots legitimately overwrite. Not + # private: cache/web is bind-mounted into remote backends' container UID. write_text_exclusive(path, content, private=False, overwrite=True) return str(path) except Exception as exc: # noqa: BLE001 @@ -78,7 +75,6 @@ def _truncate_snapshot(snapshot_text: str, max_chars: Optional[int] = None) -> s cache/web and the appended note tells the agent how to page through it with read_file — element refs beyond the cut are in the file, not lost. """ - _bt = _origin() if max_chars is None: max_chars = _bt.get_browser_snapshot_threshold() if len(snapshot_text) <= max_chars: @@ -111,13 +107,9 @@ def _truncate_snapshot(snapshot_text: str, max_chars: Optional[int] = None) -> s def _redact_browser_output(value: Any) -> Any: - """Redact secrets from browser-originated data before returning to the model. - - Browser snapshots, console messages, JS exceptions, and eval results can - contain page-rendered API keys, cookies, bearer tokens, or pasted secrets. - Tool output is a model boundary, so force redaction here even if global log - redaction is disabled for debugging. - """ + """Force-redact secrets in browser-originated data (snapshots, console, eval + results can carry page-rendered keys/cookies/tokens). Tool output is a model + boundary, so this applies even when global log redaction is disabled.""" from agent.redact import redact_sensitive_text if isinstance(value, str): diff --git a/tools/browser_tool_vision.py b/tools/browser_tool_vision.py index 10430fb37d..f6d4732afd 100644 --- a/tools/browser_tool_vision.py +++ b/tools/browser_tool_vision.py @@ -1,10 +1,9 @@ """browser_vision helpers: Lightpanda pre-route, native provider vision, auxiliary-LLM screenshot analysis. -Split out of ``tools/browser_tool.py``; every name is re-imported there so -``tools.browser_tool.`` keeps resolving (and monkeypatching). Origin -symbols and module state are read/written through ``_bt`` (the origin module, -resolved per call by :func:`tools.browser_tool_origin.origin_module`) so -``patch("tools.browser_tool.X")`` is honoured and no import cycle exists. +Split out of ``tools/browser_tool.py``; every name is re-imported there. Origin +symbols are read through ``_bt`` (the +:data:`tools.browser_tool_origin.origin` proxy) so ``patch("tools.browser_tool.X")`` +is honoured and no import cycle exists. """ import os @@ -12,11 +11,10 @@ import shutil from pathlib import Path from typing import Any, Dict, Optional, Tuple -from tools.browser_tool_origin import origin_module as _origin +from tools.browser_tool_origin import origin as _bt def _vision_mode_label() -> str: - _bt = _origin() _cp = _bt._get_cloud_provider() return "local" if _cp is None else f"cloud ({_cp.provider_name()})" @@ -24,15 +22,10 @@ def _vision_mode_label() -> str: def _lightpanda_vision_preroute( effective_task_id: str, annotate: bool, screenshot_path: Path, ) -> Tuple[bool, Optional[str], Path]: - """Capture the vision screenshot through the Chrome fallback when Lightpanda is the engine. - - Lightpanda has no graphical renderer, so the normal path would fail with a - CDP error or return a placeholder PNG. Returns ``(prerouted, fallback_warning, - screenshot_path)``; on fallback failure ``prerouted`` is False and the caller - takes the normal screenshot path (forcing Chrome) so ``_run_browser_command`` - still produces the standard fallback metadata/error. - """ - _bt = _origin() + """Capture the vision screenshot via the Chrome fallback when Lightpanda is the engine + (it has no graphical renderer). Returns ``(prerouted, fallback_warning, path)``; + on fallback failure ``prerouted`` is False and the caller takes the normal + screenshot path (forcing Chrome) so the standard fallback metadata still applies.""" engine = _bt._get_browser_engine() if engine != "lightpanda" or not _bt._should_inject_engine(engine): return False, None, screenshot_path @@ -62,11 +55,9 @@ def _native_vision_result( ) -> Dict[str, Any]: """Multimodal tool-result envelope: the main model inspects the pixels itself. - History-reuse cap: this embed is baked into the tool result and re-sent on - every later turn, exactly like vision_analyze's native path — apply the same - proactive resize so full-res screenshots can't enter immutable history - uncapped. The helper's stat/dimension quick-estimate skips the resize when - already under both caps; without Pillow it fails open to the raw bytes. + The embed is baked into history and re-sent every later turn, so apply the same + proactive resize as vision_analyze's native path (skipped when already under + both caps; without Pillow it fails open to the raw bytes). """ from tools.vision_tools import ( _EMBED_MAX_DIMENSION, @@ -103,12 +94,10 @@ def _native_vision_result( def _analyze_screenshot_with_aux_llm(screenshot_path: Path, question: str) -> str: """One-shot aux vision-LLM analysis (not baked into history), secret-redacted. - Encodes at full resolution; on a size-related provider rejection the image - is downscaled once and retried. Timeout/temperature come from - ``auxiliary.vision.*`` — local vision models (llama.cpp, ollama) can take - well over 30s, so the default timeout is generous. + Full resolution first; on a size-related provider rejection the image is + downscaled once and retried. ``auxiliary.vision.timeout/temperature`` — local + vision models can take well over 30s, so the default timeout is generous. """ - _bt = _origin() import base64 vision_prompt = ( @@ -131,12 +120,10 @@ def _analyze_screenshot_with_aux_llm(screenshot_path: Path, question: str) -> st try: from hermes_cli.config import load_config _vision_cfg = _bt.cfg_get(load_config(), "auxiliary", "vision", default={}) - _vt = _vision_cfg.get("timeout") - if _vt is not None: - vision_timeout = float(_vt) - _vtemp = _vision_cfg.get("temperature") - if _vtemp is not None: - vision_temperature = float(_vtemp) + if _vision_cfg.get("timeout") is not None: + vision_timeout = float(_vision_cfg["timeout"]) + if _vision_cfg.get("temperature") is not None: + vision_temperature = float(_vision_cfg["temperature"]) except Exception: pass @@ -157,7 +144,7 @@ def _analyze_screenshot_with_aux_llm(screenshot_path: Path, question: str) -> st if vision_model: call_kwargs["model"] = vision_model try: - response = _bt._lazy_call_llm(**call_kwargs) + response = _bt.call_llm(**call_kwargs) except Exception as _api_err: from tools.vision_tools import ( _is_image_size_error, _resize_image_for_vision, _RESIZE_TARGET_BYTES, @@ -172,9 +159,7 @@ def _analyze_screenshot_with_aux_llm(screenshot_path: Path, question: str) -> st ) data_url = _resize_image_for_vision(screenshot_path, mime_type="image/png") call_kwargs["messages"][0]["content"][1]["image_url"]["url"] = data_url - response = _bt._lazy_call_llm(**call_kwargs) + response = _bt.call_llm(**call_kwargs) - analysis = (response.choices[0].message.content or "").strip() - # Redact secrets the vision LLM may have read from the screenshot. - from agent.redact import redact_sensitive_text - return redact_sensitive_text(analysis) + from agent.redact import redact_sensitive_text # the LLM may have read secrets off the screenshot + return redact_sensitive_text((response.choices[0].message.content or "").strip())