hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 32 Packages Projects Releases Wiki Activity

refactor(tools): browser_tool — consolidate config caches, navigate/eval/console helpers, lifecycle best-effort wrapper, compact session/vision bodies

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:27:55 -07:00
parent 408cafe06f
commit ed9476cf40
4 changed files with 207 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files

306
tools/browser_tool.py
View File

@@ -1,15 +1,12 @@
#!/usr/bin/env python3
"""Browser automation tools driven by the agent-browser CLI.
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.
Backends — local headless Chromium, 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. 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
@@ -147,7 +144,6 @@ _SANE_PATH_DIRS = (
)
_SANE_PATH = os.pathsep.join(_SANE_PATH_DIRS)
from tools.browser_tool_install import ( # noqa: F401 (re-exported; tests patch tools.browser_tool.<name>)
_discover_homebrew_node_dirs, _browser_candidate_path_dirs, _merge_browser_path,
_browser_install_hint, _is_npx_agent_browser_sentinel, _requires_real_termux_browser_install,
@@ -157,13 +153,11 @@ from tools.browser_tool_install import ( # noqa: F401 (re-exported; tests patc
check_browser_requirements, check_browser_vision_requirements,
)
# Throttle screenshot cleanup to avoid repeated full directory scans.
_last_screenshot_cleanup_by_dir: dict[str, float] = {}
_last_screenshot_cleanup_by_dir: dict[str, float] = {} # throttles full directory scans
# ----------------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------------
DEFAULT_COMMAND_TIMEOUT = 30 # seconds
# Floors for ``open``: cold daemon + first Chromium launch can exceed the
@@ -177,18 +171,41 @@ MIN_FIRST_OPEN_TIMEOUT = 120
DEFAULT_SNAPSHOT_THRESHOLD = 15000
MIN_SNAPSHOT_THRESHOLD = 1000
SNAPSHOT_SUMMARIZE_THRESHOLD = DEFAULT_SNAPSHOT_THRESHOLD # legacy import surface
# 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
_EMPTY_OK_COMMANDS: frozenset = frozenset({"close", "record"}) # legitimately empty stdout
# Commands that legitimately return empty stdout.
_EMPTY_OK_COMMANDS: frozenset = frozenset({"close", "record"})
# Sentinel _find_agent_browser returns/caches to mean "resolve via npx" rather
# 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 / 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"
# Process caches (``_cached_X`` + ``_X_resolved`` pairs) for config-derived lookups;
# reset by ``cleanup_all_browsers``. Written/read by the sibling modules via the origin.
_cached_command_timeout: Optional[int] = None
_command_timeout_resolved = False
_cached_snapshot_threshold: Optional[int] = None
_snapshot_threshold_resolved = False
_cached_cloud_provider: Optional[CloudBrowserProvider] = None
_cloud_provider_resolved = False
_cached_cloud_provider_scope: Optional[str] = None
_cached_cloud_providers: Dict[tuple[str, tuple[int, int]], Optional[CloudBrowserProvider]] = {}
_cloud_provider_cache_lock = threading.RLock()
_allow_private_urls_resolved = False
_cached_allow_private_urls: Optional[bool] = None
_cached_agent_browser: Optional[str] = None
_agent_browser_resolved = False
_cached_browser_engine: Optional[str] = None # agent-browser v0.25.3+ ``--engine lightpanda``
_browser_engine_resolved = False
_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
_cached_chromium_installed: Optional[bool] = None
_chromium_autoinstall_attempted = False # one-shot: a failed 170MB download must not retry per call
# Mask secrets in logged CDP URLs; agent.redact.redact_cdp_url is the single policy.
_sanitize_url_for_logs = redact_cdp_url
@@ -289,20 +306,6 @@ _PROVIDER_REGISTRY: Dict[str, type] = {
# 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
_cloud_provider_resolved = False
_cached_cloud_provider_scope: Optional[str] = None
_cached_cloud_providers: Dict[tuple[str, tuple[int, int]], Optional[CloudBrowserProvider]] = {}
_cloud_provider_cache_lock = threading.RLock()
_allow_private_urls_resolved = False
_cached_allow_private_urls: Optional[bool] = None
_cached_agent_browser: Optional[str] = None
_agent_browser_resolved = False
# Lightpanda engine (agent-browser v0.25.3+ ``--engine lightpanda``), cached like the provider.
_cached_browser_engine: Optional[str] = None
_browser_engine_resolved = False
from tools.browser_tool_cloud import ( # noqa: F401 (re-exported; tests patch tools.browser_tool.<name>)
_is_legacy_provider_registry_overridden, _ensure_browser_plugins_loaded, _get_cloud_provider,
_instantiate_explicit_cloud_provider, _autodetect_cloud_provider,
@@ -312,23 +315,6 @@ from tools.browser_tool_cloud import ( # noqa: F401 (re-exported; tests patch
)
from hermes_constants import is_termux as _is_termux_environment # noqa: F401 (read via origin)
# Sentinel _find_agent_browser returns/caches to mean "resolve via npx" rather
# 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 / 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
from tools.browser_tool_lightpanda_fallback import ( # noqa: F401
_using_lightpanda_engine, lightpanda_engine_status, _lightpanda_fallback_reason,
_needs_lightpanda_fallback, _annotate_lightpanda_fallback, _copy_fallback_warning,
@@ -343,7 +329,6 @@ _real_profile_cdp_lock = threading.Lock()
_real_profile_cdp_cache: dict = {}
_real_profile_chrome_procs: list = [] # Popen handles of directly-launched real browsers
from tools.browser_tool_real_profile import ( # noqa: F401
_terminate_real_profile_chrome, _cdp_http_ready, _agent_browser_get_cdp, _cdp_on_data_dir,
_agent_browser_close_session, _REAL_PROFILE_CHROME_FLAGS, _real_profile_unsupported_reason,
@@ -367,10 +352,7 @@ def _url_is_private(url: str) -> bool:
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")
)
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:
hostname = (urlparse(url).hostname or "").strip().lower().rstrip(".")
@@ -419,12 +401,10 @@ def _navigation_session_key(task_id: str, url: str) -> str:
def _is_local_sidecar_key(session_key: str) -> bool:
"""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:
"""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
@@ -452,11 +432,8 @@ def _last_session_key(task_id: str) -> str:
if session_info and _session_info_owned_by_task(session_info, task_id, recorded_key):
return recorded_key
_last_active_session_key.pop(task_id, None)
logger.debug(
"browser session ownership: dropping stale/mismatched last-active binding %s -> %s",
task_id,
recorded_key,
)
logger.debug("browser session ownership: dropping stale/mismatched last-active binding %s -> %s",
task_id, recorded_key)
return task_id
@@ -471,19 +448,15 @@ def _socket_safe_tmpdir() -> str:
# 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, 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"
_cleanup_done = False
# 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)
)
DEFAULT_SESSION_INACTIVITY_TIMEOUT = int(DEFAULT_CONFIG.get("browser", {}).get("inactivity_timeout", 120))
def _get_session_inactivity_timeout() -> int:
@@ -496,11 +469,9 @@ def _get_session_inactivity_timeout() -> int:
BROWSER_SESSION_INACTIVITY_TIMEOUT = _get_session_inactivity_timeout()
# 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 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.
@@ -551,23 +522,16 @@ class _BrowserSessionBackend:
try:
_cleanup_single_browser_session(self._session_key)
except Exception:
logger.warning(
"Teardown of suspect browser session %s failed; a fresh "
"session will be created anyway", self._session_key,
exc_info=True,
)
logger.warning("Teardown of suspect browser session %s failed; a fresh "
"session will be created anyway", self._session_key, exc_info=True)
return False
def _browser_session_backend(session_key: str) -> _BrowserSessionBackend:
return _BrowserSessionBackend(session_key)
_browser_session_backend = _BrowserSessionBackend
_cleanup_thread = None
_cleanup_running = False
# Protects _session_last_activity AND _active_sessions (subagents run concurrently).
_cleanup_lock = threading.Lock()
_cleanup_lock = threading.Lock() # protects _session_last_activity AND _active_sessions
from tools.browser_tool_lifecycle import ( # noqa: F401 (re-exported; tests patch tools.browser_tool.<name>)
_session_expiry_timestamp, _session_has_expired, _emergency_cleanup_all_sessions,
@@ -586,11 +550,9 @@ from tools.browser_tool_lifecycle import ( # noqa: F401 (re-exported; tests pa
atexit.register(_emergency_cleanup_all_sessions)
atexit.register(_stop_browser_cleanup_thread)
# ----------------------------------------------------------------------------
# Tool Schemas
# ----------------------------------------------------------------------------
BROWSER_TOOL_SCHEMAS = [
{
"name": "browser_navigate",
@@ -733,13 +695,10 @@ BROWSER_TOOL_SCHEMAS = [
},
]
from tools.browser_tool_snapshot import ( # noqa: F401
_store_full_snapshot, _truncate_snapshot, _redact_browser_output,
_extract_screenshot_path_from_text,
_store_full_snapshot, _truncate_snapshot, _redact_browser_output, _extract_screenshot_path_from_text,
)
# ----------------------------------------------------------------------------
# Browser Tool Functions
# ----------------------------------------------------------------------------
@@ -893,17 +852,12 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
return json.dumps(safety_error)
if _is_camofox_mode():
from tools.browser_camofox import camofox_navigate
return camofox_navigate(url, task_id)
return _camofox("camofox_navigate", url, task_id)
if auto_local_this_nav:
logger.info(
"browser_navigate: auto-routing %s to local Chromium sidecar "
"(cloud provider %s stays on cloud for public URLs; "
"set browser.auto_local_for_private_urls: false to disable)",
url,
type(_get_cloud_provider()).__name__ if _get_cloud_provider() else "none",
)
logger.info("browser_navigate: auto-routing %s to local Chromium sidecar (cloud provider %s stays on "
"cloud for public URLs; set browser.auto_local_for_private_urls: false to disable)",
url, type(_get_cloud_provider()).__name__ if _get_cloud_provider() else "none")
session_info = _get_session_info(nav_session_key)
is_first_nav = session_info.get("_first_nav", True)
@@ -911,32 +865,33 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
session_info["_first_nav"] = False
_maybe_start_recording(nav_session_key)
result = _run_browser_command(
nav_session_key, "open", [url], timeout=_get_open_command_timeout(first_open=is_first_nav)
)
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 _dumps(_err(result.get("error", "Navigation failed")))
data = result.get("data", {})
title = data.get("title", "")
final_url = data.get("url", url)
blocked = _post_redirect_block(nav_session_key, url, final_url, auto_local_this_nav)
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.
try:
if (session_info.get("features") or {}).get("real_profile"):
response["used_real_profile"] = True
except Exception:
pass
features = session_info.get("features") or {}
if features.get("real_profile"): # auditability: this ran on the user's real-profile copy-browser
response["used_real_profile"] = True
# 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)
_add_navigate_warnings(response, title, session_info if is_first_nav else None)
_attach_auto_snapshot(response, nav_session_key)
return _dumps(response)
def _add_navigate_warnings(response: Dict[str, Any], title: str, first_nav_session: Optional[Dict[str, Any]]) -> None:
"""Bot-detection hint from the page title; on first navigation, the session's stealth features."""
title_lower = title.lower()
if any(pattern in title_lower for pattern in _BOT_DETECTION_TITLE_PATTERNS):
response["bot_detection_warning"] = (
@@ -945,9 +900,8 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
"3) Enable advanced stealth (BROWSERBASE_ADVANCED_STEALTH=true, requires Scale plan), "
"4) Some sites have very aggressive bot detection that may be unavoidable."
)
if is_first_nav and "features" in session_info:
features = session_info["features"]
if first_nav_session is not None and "features" in first_nav_session:
features = first_nav_session["features"]
if not features.get("proxies"):
response["stealth_warning"] = (
"Running WITHOUT residential proxies. Bot detection may be more aggressive. "
@@ -955,9 +909,6 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
)
response["stealth_features"] = [k for k, v in features.items() if v]
_attach_auto_snapshot(response, nav_session_key)
return _dumps(response)
def browser_snapshot(
full: bool = False, task_id: Optional[str] = None, user_task: Optional[str] = None
@@ -965,9 +916,7 @@ def browser_snapshot(
"""Text snapshot of the page's accessibility tree (compact unless ``full``).
``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)
return _camofox("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"):
@@ -1027,12 +976,15 @@ def _guarded_action(task_id: Optional[str], action: str, command: str, args: lis
return _tool_response(_run_browser_command(effective_task_id, command, args), ok, err)
def _at_ref(ref: str) -> str:
return ref if ref.startswith("@") else f"@{ref}"
def browser_click(ref: str, task_id: Optional[str] = None) -> str:
"""Click the element ``ref`` (e.g. "@e5")."""
if _is_camofox_mode():
return _camofox("camofox_click", ref, task_id)
if not ref.startswith("@"):
ref = f"@{ref}"
ref = _at_ref(ref)
return _guarded_action(task_id, "click", "click", [ref], {"clicked": ref}, f"Failed to click {ref}")
@@ -1044,10 +996,8 @@ def browser_type(ref: str, text: str, task_id: Optional[str] = None) -> str:
blocked = _blocked_private_page_action(effective_task_id, "type")
if blocked is not None:
return blocked
if not ref.startswith("@"):
ref = f"@{ref}"
ref = _at_ref(ref)
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 already went to the browser).
@@ -1056,8 +1006,7 @@ def browser_type(ref: str, text: str, task_id: Optional[str] = None) -> str:
response = {"success": True, "typed": display_text, "element": ref}
else:
response = _err(result.get("error", f"Failed to type into {ref}"))
response = _copy_fallback_warning(response, result)
return _dumps(redact_browser_typed_text_for_display(response, text))
return _dumps(redact_browser_typed_text_for_display(_copy_fallback_warning(response, result), text))
def browser_scroll(direction: str, task_id: Optional[str] = None) -> str:
@@ -1065,12 +1014,8 @@ def browser_scroll(direction: str, task_id: Optional[str] = None) -> str:
if direction not in {"up", "down"}:
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():
# Camofox REST API has no pixel argument; use repeated calls.
result = None
for _ in range(5):
result = _camofox("camofox_scroll", direction, task_id)
return result
if _is_camofox_mode(): # Camofox REST API has no pixel argument; use repeated calls
return [_camofox("camofox_scroll", direction, task_id) for _ in range(5)][-1]
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}")
@@ -1085,8 +1030,7 @@ def browser_back(task_id: Optional[str] = None) -> str:
if result.get("success"):
# 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.")
blocked = _blocked_private_page(effective_task_id, "Browser history navigation (back) landed on this address.")
if blocked is not None:
return blocked
return _tool_response(result, {"url": result.get("data", {}).get("url", "")}, "Failed to go back")
@@ -1115,8 +1059,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]:
"""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.")
return _blocked_private_page(effective_task_id, f"Refusing to {action} on this page in this browser mode.")
_EVAL_NAVIGATED_WHY = "This may have been caused by a JavaScript navigation via browser_console."
@@ -1149,25 +1092,17 @@ def browser_console(clear: bool = False, expression: Optional[str] = None, task_
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"):
for msg in console_result.get("data", {}).get("messages", []):
messages.append({
"type": msg.get("type", "log"),
"text": _redact_browser_output(msg.get("text", "")),
"source": "console",
})
errors = []
if errors_result.get("success"):
for err in errors_result.get("data", {}).get("errors", []):
errors.append({"message": _redact_browser_output(err.get("message", "")), "source": "exception"})
messages = [
{"type": msg.get("type", "log"), "text": _redact_browser_output(msg.get("text", "")), "source": "console"}
for msg in console_result.get("data", {}).get("messages", [])
] if console_result.get("success") else []
errors = [
{"message": _redact_browser_output(err.get("message", "")), "source": "exception"}
for err in errors_result.get("data", {}).get("errors", [])
] if errors_result.get("success") else []
response = {
"success": True,
"console_messages": messages,
"js_errors": errors,
"total_messages": len(messages),
"total_errors": len(errors),
"success": True, "console_messages": messages, "js_errors": errors,
"total_messages": len(messages), "total_errors": len(errors),
}
_copy_fallback_warning(response, console_result)
_merge_fallback_warning(response, errors_result)
@@ -1194,12 +1129,16 @@ def _parse_eval_value(raw_result: Any) -> Any:
def _eval_ok_response(parsed: Any, **extra) -> Dict[str, Any]:
return {
"success": True,
"result": _redact_browser_output(parsed),
"result_type": type(parsed).__name__,
**extra,
}
return {"success": True, "result": _redact_browser_output(parsed), "result_type": type(parsed).__name__, **extra}
def _eval_result_or_blocked(effective_task_id: str, parsed: Any, result: Dict[str, Any], **extra) -> str:
"""Eval tool JSON, unless the post-eval page-URL recheck finds an eval navigated the
page to a private address — then the result is withheld."""
blocked = _blocked_private_page_content(effective_task_id)
if blocked is not None:
return blocked
return _dumps(_copy_fallback_warning(_eval_ok_response(parsed, **extra), result), default=str)
def _eval_supervisor_fast_path(effective_task_id: str, expression: str) -> Optional[str]:
@@ -1217,12 +1156,8 @@ def _eval_supervisor_fast_path(effective_task_id: str, expression: str) -> Optio
return None
sup_result = supervisor.evaluate_runtime(expression)
if sup_result.get("ok"):
parsed = _parse_eval_value(sup_result.get("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
return _dumps(_eval_ok_response(parsed, method="cdp_supervisor"), default=str)
return _eval_result_or_blocked(
effective_task_id, _parse_eval_value(sup_result.get("result")), {}, method="cdp_supervisor")
err = sup_result.get("error") or "evaluate_runtime failed"
if "supervisor" not in err.lower():
return _dumps(_err(err))
@@ -1281,12 +1216,7 @@ def _browser_eval(expression: str, task_id: Optional[str] = None) -> str:
result = _run_browser_command(effective_task_id, "eval", [expression])
if not result.get("success"):
return _eval_failure_response(result)
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 _dumps(_copy_fallback_warning(response, result), default=str)
return _eval_result_or_blocked(effective_task_id, _parse_eval_value(result.get("data", {}).get("result")), result)
def _camofox_eval(expression: str, task_id: Optional[str] = None) -> str:
@@ -1394,9 +1324,7 @@ def browser_get_images(task_id: Optional[str] = None) -> str:
{"success": True, "images": [], "count": 0, "warning": "Could not parse image data"}, result)
_LP_VISION_FALLBACK_REASON = (
"Lightpanda has no graphical renderer for screenshots; used Chrome for vision capture."
)
_LP_VISION_FALLBACK_REASON = "Lightpanda has no graphical renderer for screenshots; used Chrome for vision capture."
from tools.browser_tool_vision import ( # noqa: F401 (re-exported; tests patch tools.browser_tool.<name>)
@@ -1452,9 +1380,9 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str]
)
if not result.get("success"):
error_detail = result.get("error", "Unknown error")
return _json_with_fallback(
_err(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): {result.get('error', 'Unknown error')}"
), result)
actual_screenshot_path = result.get("data", {}).get("path")
if actual_screenshot_path:
@@ -1476,11 +1404,8 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str]
return _native_vision_result(screenshot_path, question, annotate, result, _lp_fallback_warning)
analysis = _analyze_screenshot_with_aux_llm(screenshot_path, question)
response_data = {
"success": True,
"analysis": analysis or "Vision analysis returned no content.",
"screenshot_path": str(screenshot_path),
}
response_data = {"success": True, "analysis": analysis or "Vision analysis returned no content.",
"screenshot_path": str(screenshot_path)}
_copy_fallback_warning(response_data, result)
if annotate and result.get("data", {}).get("annotations"):
response_data["annotations"] = result["data"]["annotations"]
@@ -1498,12 +1423,6 @@ def browser_vision(question: str, annotate: bool = False, task_id: Optional[str]
return _dumps(error_info)
# 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
_chromium_autoinstall_attempted = False
# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------
@@ -1530,10 +1449,9 @@ def _fallback_call(fn_name: str, arg_defaults: Dict[str, Any], extra_kw: tuple =
return call
# (tool name, emoji, availability gate, schema-arg defaults[, extra kw names]) — the tool
# function is the module global of the same name. —
# routed-through-extension tools (gate None) use the per-action gate; get_images/console/vision
# keep the plain requirement checks.
# (tool name, emoji, availability gate, schema-arg defaults[, extra kw names]); the tool
# function is the module global of the same name. Routed-through-extension tools (gate None)
# use the per-action gate; get_images/console/vision keep the plain requirement checks.
_BROWSER_TOOL_TABLE = (
("browser_navigate", "🌐", None, {"url": ""}),
("browser_snapshot", "📸", None, {"full": False}, ("user_task",)),
@@ -1558,20 +1476,14 @@ 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),
task_id=kw.get("task_id"), session_id=kw.get("session_id"),
)
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); also looked up by the
# registration loop below via globals().
for _name, _emoji, _check_fn, _defaults, *_extra in _BROWSER_TOOL_TABLE:
if _check_fn is None:
if _check_fn is None: # also binds the legacy check_browser_<x>_requirements globals (tests + callers)
_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_call(_name, _defaults, *_extra)),
check_fn=_check_fn, emoji=_emoji,
)
registry.register(name=_name, toolset="browser", schema=_BROWSER_SCHEMA_MAP[_name],
handler=_routed_handler(_name, _fallback_call(_name, _defaults, *_extra)),
check_fn=_check_fn, emoji=_emoji)

126
tools/browser_tool_lifecycle.py
View File

@@ -53,6 +53,19 @@ def _session_has_expired(
return (time.time() if now is None else now) >= expires_at
def _best_effort(label: str, fn) -> None:
"""Run ``fn()``; log (debug) and swallow any exception — teardown must never abort."""
try:
fn()
except Exception as e:
_bt.logger.debug("%s failed: %s", label, e)
def _stop_all_lightpanda() -> None:
from tools.browser_lightpanda import stop_all_lightpanda
stop_all_lightpanda()
def _emergency_cleanup_all_sessions():
"""atexit: close this process's sessions, then sweep orphans left by crashed
hermes processes — every clean exit reaps accumulated orphans, not only
@@ -64,10 +77,7 @@ def _emergency_cleanup_all_sessions():
# 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:
_bt.logger.debug("Real-profile chrome cleanup on exit failed: %s", e)
_best_effort("Real-profile chrome cleanup on exit", _bt._terminate_real_profile_chrome)
if _bt._active_sessions:
_bt.logger.info("Emergency cleanup: closing %s active session(s)...", len(_bt._active_sessions))
try:
@@ -81,21 +91,11 @@ def _emergency_cleanup_all_sessions():
_bt._session_owner_homes.clear()
_bt._cleanup_failures.clear()
_bt._recording_sessions.clear()
# Lightpanda servers we spawned that fell out of ``_active_sessions``.
try:
from tools.browser_lightpanda import stop_all_lightpanda
stop_all_lightpanda()
except Exception as e:
_bt.logger.debug("Lightpanda cleanup on exit failed: %s", e)
# 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()
except Exception as e:
_bt.logger.debug("Orphan reap on exit failed: %s", e)
_best_effort("Lightpanda cleanup on exit", _stop_all_lightpanda)
# Safe even if we never used the browser — owner_pid liveness protects daemons
# owned by other live hermes processes.
_best_effort("Orphan reap on exit", _bt._reap_orphaned_browser_sessions)
@contextlib.contextmanager
@@ -147,10 +147,8 @@ def _cleanup_inactive_browser_sessions():
current_time = time.time()
with _bt._cleanup_lock:
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
]
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))
@@ -200,14 +198,15 @@ def _verify_reapable_browser_daemon(daemon_pid: int, socket_dir: str,
(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.
"""
def refuse(reason: str, *args) -> bool:
_bt.logger.warning("Refusing to reap browser daemon PID %d (session %s): " + reason,
daemon_pid, session_name, *args)
return False
try:
import psutil
except ImportError: # psutil is a hard dep; defensive only
_bt.logger.warning(
"Refusing to reap browser daemon PID %d (session %s): "
"psutil unavailable for identity verification",
daemon_pid, session_name)
return False
return refuse("psutil unavailable for identity verification")
try:
proc = psutil.Process(daemon_pid)
@@ -216,17 +215,10 @@ def _verify_reapable_browser_daemon(daemon_pid: int, socket_dir: str,
except psutil.NoSuchProcess:
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): "
"could not read process identity (%s)",
daemon_pid, session_name, exc)
return False
return refuse("could not read process identity (%s)", exc)
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
return refuse("not an agent-browser process (name=%r)", name)
socket_dir_l = socket_dir.lower()
socket_base_l = os.path.basename(socket_dir).lower()
@@ -237,14 +229,8 @@ 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):
bound = False # environ() can be denied even same-user; cmdline already failed — fail closed
if not bound:
_bt.logger.warning(
"Refusing to reap agent-browser PID %d: not bound to session "
"socket dir %s (possible recycled PID or planted pid file)",
daemon_pid, socket_dir)
return False
return refuse("not bound to session socket dir %s (possible recycled PID or planted pid file)", socket_dir)
return True
@@ -374,12 +360,10 @@ def _reap_orphaned_browser_sessions():
# 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:
def _reap_lp():
from tools.browser_lightpanda import reap_orphaned_lightpanda
reap_orphaned_lightpanda()
except Exception as e:
_bt.logger.debug("Lightpanda orphan reap failed: %s", e)
_best_effort("Lightpanda orphan reap", _reap_lp)
tmpdir = _bt._socket_safe_tmpdir()
socket_dirs = []
@@ -389,11 +373,7 @@ def _reap_orphaned_browser_sessions():
return
with _bt._cleanup_lock:
tracked_names = {
info.get("session_name")
for info in _bt._active_sessions.values()
if info.get("session_name")
}
tracked_names = {info.get("session_name") for info in _bt._active_sessions.values() if info.get("session_name")}
reaped = 0
for socket_dir in socket_dirs:
@@ -477,15 +457,13 @@ def _kill_process_tree(proc: "subprocess.Popen") -> None:
def _legacy_kill_process_tree(proc: "subprocess.Popen") -> None:
"""Local tree-kill — fallback when agent.deadline is unavailable."""
"""Local tree-kill (SIGTERM then SIGKILL to the process group) — fallback when
agent.deadline is unavailable. Differs from hermes_cli._subprocess_compat's
group-leader-only variant, and tests pin this sequence."""
if os.name == "nt":
try:
subprocess.run(
["taskkill", "/PID", str(proc.pid), "/T", "/F"],
check=False,
capture_output=True,
stdin=subprocess.DEVNULL,
)
subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"],
check=False, capture_output=True, stdin=subprocess.DEVNULL)
except Exception:
pass
return
@@ -502,8 +480,7 @@ def _legacy_kill_process_tree(proc: "subprocess.Popen") -> None:
pgid = os.getpgid(proc.pid)
except (ProcessLookupError, OSError):
return
sigkill = getattr(signal, "SIGKILL", signal.SIGTERM)
for sig in (signal.SIGTERM, sigkill):
for sig in (signal.SIGTERM, getattr(signal, "SIGKILL", signal.SIGTERM)):
try:
killpg(pgid, sig)
except (ProcessLookupError, PermissionError, OSError):
@@ -658,12 +635,11 @@ def _cleanup_single_browser_session(task_id: str) -> None:
# 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:
def _camofox_cleanup():
from tools.browser_camofox import camofox_close, camofox_soft_cleanup
if not camofox_soft_cleanup(task_id):
camofox_close(task_id)
except Exception as e:
_bt.logger.debug("Camofox cleanup for task %s: %s", task_id, e)
_best_effort(f"Camofox cleanup for task {task_id}", _camofox_cleanup)
_bt.logger.debug("cleanup_browser called for task_id: %s", task_id)
_bt.logger.debug("Active sessions: %s", list(_bt._active_sessions.keys()))
@@ -686,7 +662,6 @@ def _cleanup_single_browser_session(task_id: str) -> None:
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)
@@ -717,16 +692,15 @@ def cleanup_all_browsers() -> None:
except Exception:
pass
_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
# Each resolved flag flips BEFORE its cache is nulled 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
_bt._cached_snapshot_threshold = None
_bt._cached_chromium_installed = None
_bt._chromium_autoinstall_attempted = False
_bt._cached_browser_engine = None
_bt._browser_engine_resolved = False
for flag, cache in (
("_agent_browser_resolved", "_cached_agent_browser"),
("_command_timeout_resolved", "_cached_command_timeout"),
("_snapshot_threshold_resolved", "_cached_snapshot_threshold"),
("_chromium_autoinstall_attempted", "_cached_chromium_installed"),
("_browser_engine_resolved", "_cached_browser_engine"),
):
setattr(_bt, flag, False)
setattr(_bt, cache, None)

118
tools/browser_tool_session.py
View File

@@ -19,16 +19,12 @@ from typing import Any, Dict, List, Optional
from tools.browser_tool_origin import origin as _bt
_DOCKER_PULL = "docker pull ghcr.io/nousresearch/hermes-agent:latest"
_CHROMIUM_INSTALL = "npx agent-browser install --with-deps (or: npx playwright install --with-deps chromium)"
_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)"
f"Chromium browser is missing. You're running in Docker — pull the latest image to get the bundled Chromium: {_DOCKER_PULL}"
)
_CHROMIUM_MISSING_HINT = f"Chromium browser is missing. Install it with: {_CHROMIUM_INSTALL}"
def _needs_chromium_sandbox_bypass() -> bool:
@@ -51,10 +47,7 @@ def _apply_chromium_sandbox_args(browser_env: Dict[str, str]) -> None:
and "AGENT_BROWSER_CHROME_FLAGS" not in browser_env
and _bt._needs_chromium_sandbox_bypass()
):
_bt.logger.debug(
"browser: sandbox bypass needed (root/docker/AppArmor userns) — "
"injecting --no-sandbox"
)
_bt.logger.debug("browser: sandbox bypass needed (root/docker/AppArmor userns) — injecting --no-sandbox")
browser_env["AGENT_BROWSER_ARGS"] = "--no-sandbox,--disable-dev-shm-usage"
@@ -87,27 +80,17 @@ def _format_browser_timeout_error(
if detail:
parts.append(detail[:1500])
combined = f"{stderr}\n{stdout}".lower()
if "sandbox" in combined:
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"
)
if "sandbox" in f"{stderr}\n{stdout}".lower():
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():
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"
)
parts.append("The browser daemon may still be starting or Chromium may be "
f"missing. Pull the latest image: {_DOCKER_PULL}")
else:
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)"
)
parts.append("The browser daemon may still be starting, or Chromium may be "
f"missing system libraries. Install/repair with: {_CHROMIUM_INSTALL}")
return "\n".join(parts)
@@ -163,25 +146,18 @@ def _popen_agent_browser(argv: List[str], env: Dict[str, str], socket_dir: str,
so the child gets ONLY our three handles (leaked console handles make the Rust
daemon grandchild die silently), close_fds=True for the rest.
"""
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)
stderr_fd = os.open(stderr_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
fds = [os.open(os.path.join(socket_dir, f"_{slot}_{tag}"), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
for slot in ("stdout", "stderr")]
try:
_popen_extra: dict = {}
if os.name == "nt":
_popen_extra["creationflags"] = _bt.windows_hide_flags()
_popen_extra["close_fds"] = True
_si = subprocess.STARTUPINFO()
_si.dwFlags |= subprocess.STARTF_USESTDHANDLES
_popen_extra["startupinfo"] = _si
return subprocess.Popen(
argv, stdout=stdout_fd, stderr=stderr_fd,
stdin=subprocess.DEVNULL, env=env, **_popen_extra,
)
_popen_extra = {"creationflags": _bt.windows_hide_flags(), "close_fds": True, "startupinfo": _si}
return subprocess.Popen(argv, stdout=fds[0], stderr=fds[1], stdin=subprocess.DEVNULL, env=env, **_popen_extra)
finally:
os.close(stdout_fd)
os.close(stderr_fd)
for fd in fds:
os.close(fd)
def _session_record(prefix: str, cdp_url: Optional[str], features: Dict[str, Any]) -> Dict[str, Any]:
@@ -208,9 +184,7 @@ def _create_local_session(task_id: str, allow_real_profile: bool = True) -> Dict
raise RuntimeError(err)
if cdp_url:
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", info["session_name"], task_id
)
_bt.logger.info("Created real-profile local session %s for task %s", info["session_name"], task_id)
return info
# Browser Use mode drives whatever CDP endpoint it is handed; with
@@ -228,17 +202,13 @@ def _create_lightpanda_session(task_id: str) -> Dict[str, Any]:
"""Spawn ``lightpanda serve`` for this session key (Browser Use mode)."""
from tools.browser_lightpanda import launch_lightpanda
session_name = f"lp_{uuid.uuid4().hex[:10]}"
server, err = launch_lightpanda(session_name, block_private_networks=not _bt._is_local_backend())
info = _session_record("lp", None, {"local": True, "lightpanda": True})
server, err = launch_lightpanda(info["session_name"], block_private_networks=not _bt._is_local_backend())
if err:
raise RuntimeError(err)
_bt.logger.info("Created Lightpanda session %s (port %s) for task %s", session_name, server.port, task_id)
return {
"session_name": session_name,
"bb_session_id": None,
"cdp_url": server.cdp_url,
"features": {"local": True, "lightpanda": True},
}
info["cdp_url"] = server.cdp_url
_bt.logger.info("Created Lightpanda session %s (port %s) for task %s", info["session_name"], server.port, task_id)
return info
def _local_backend_process_dead(session_info: Dict[str, Any]) -> bool:
@@ -350,10 +320,7 @@ def _get_session_info(task_id: Optional[str] = None) -> Dict[str, Any]:
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)
):
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)
@@ -538,16 +505,13 @@ def _interpret_browser_command_output(command: str, stdout: str, stderr: str, re
parsed = json.loads(stdout_text)
except json.JSONDecodeError:
raw = stdout_text[:2000]
_bt.logger.warning("browser '%s' returned non-JSON output (rc=%s): %s",
command, returncode, raw[:500])
_bt.logger.warning("browser '%s' returned non-JSON output (rc=%s): %s", command, returncode, raw[:500])
if command == "screenshot":
stderr_text = (stderr or "").strip()
combined_text = "\n".join(part for part in [stdout_text, stderr_text] if part)
recovered_path = _bt._extract_screenshot_path_from_text(combined_text)
if recovered_path and Path(recovered_path).exists():
_bt.logger.info(
"browser 'screenshot' recovered file from non-JSON output: %s", recovered_path
)
_bt.logger.info("browser 'screenshot' recovered file from non-JSON output: %s", recovered_path)
return {"success": True, "data": {"path": recovered_path, "raw": raw}}
return {"success": False, "error": f"Non-JSON output from agent-browser for '{command}': {raw}"}
@@ -606,15 +570,11 @@ def _spawn_and_collect(
# 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,
)
stripped = [browser_env.pop(k, None) for k in ("AGENT_BROWSER_ARGS", "AGENT_BROWSER_CHROME_FLAGS")]
if any(v is not None for v in stripped):
_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)
@@ -634,10 +594,7 @@ def _spawn_and_collect(
_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),
}
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:
@@ -704,12 +661,7 @@ def _run_browser_command(
# empty, non-JSON, nonzero rc, parsed).
fallback_reason = _bt._lightpanda_fallback_reason(engine, command, result)
if fallback_reason:
_bt.logger.info(
"Lightpanda fallback: retrying '%s' with Chrome (task=%s): %s",
command,
task_id,
fallback_reason,
)
_bt.logger.info("Lightpanda fallback: retrying '%s' with Chrome (task=%s): %s", command, task_id, fallback_reason)
if command == "screenshot": # separate Chrome session to the same URL
fallback_result = _bt._chrome_fallback_screenshot(task_id, args or [], timeout)
else:

47
tools/browser_tool_vision.py
View File

@@ -66,28 +66,18 @@ def _native_vision_result(
_resize_image_for_vision,
)
data_url = _resize_image_for_vision(
screenshot_path,
mime_type="image/png",
max_base64_bytes=_EMBED_TARGET_BYTES,
max_dimension=_EMBED_MAX_DIMENSION,
force_jpeg=True,
)
native_result = _build_native_vision_tool_result(
image_url=str(screenshot_path),
question=question,
image_data_url=data_url,
image_size_bytes=screenshot_path.stat().st_size,
)
data_url = _resize_image_for_vision(screenshot_path, mime_type="image/png", max_base64_bytes=_EMBED_TARGET_BYTES,
max_dimension=_EMBED_MAX_DIMENSION, force_jpeg=True)
native_result = _build_native_vision_tool_result(image_url=str(screenshot_path), question=question,
image_data_url=data_url,
image_size_bytes=screenshot_path.stat().st_size)
meta = native_result.setdefault("meta", {})
meta["screenshot_path"] = str(screenshot_path)
if lp_fallback_warning:
meta["fallback_warning"] = lp_fallback_warning
if annotate and result.get("data", {}).get("annotations"):
meta["annotations"] = result["data"]["annotations"]
native_result["text_summary"] = (
f"{native_result.get('text_summary', '')} " f"Screenshot path: {screenshot_path}"
).strip()
native_result["text_summary"] = f"{native_result.get('text_summary', '')} Screenshot path: {screenshot_path}".strip()
return native_result
@@ -127,18 +117,11 @@ def _analyze_screenshot_with_aux_llm(screenshot_path: Path, question: str) -> st
pass
call_kwargs = {
"task": "vision",
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": vision_prompt},
{"type": "image_url", "image_url": {"url": data_url}},
],
}
],
"temperature": vision_temperature,
"timeout": vision_timeout,
"task": "vision", "temperature": vision_temperature, "timeout": vision_timeout,
"messages": [{"role": "user", "content": [
{"type": "text", "text": vision_prompt},
{"type": "image_url", "image_url": {"url": data_url}},
]}],
}
if vision_model:
call_kwargs["model"] = vision_model
@@ -148,12 +131,8 @@ def _analyze_screenshot_with_aux_llm(screenshot_path: Path, question: str) -> st
from tools.vision_tools import _is_image_size_error, _resize_image_for_vision, _RESIZE_TARGET_BYTES
if not (_is_image_size_error(_api_err) and len(data_url) > _RESIZE_TARGET_BYTES):
raise
_bt.logger.info(
"Vision API rejected screenshot (%.1f MB); "
"auto-resizing to ~%.0f MB and retrying...",
len(data_url) / (1024 * 1024),
_RESIZE_TARGET_BYTES / (1024 * 1024),
)
_bt.logger.info("Vision API rejected screenshot (%.1f MB); auto-resizing to ~%.0f MB and retrying...",
len(data_url) / (1024 * 1024), _RESIZE_TARGET_BYTES / (1024 * 1024))
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.call_llm(**call_kwargs)