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

refactor(browser): table-driven registry.register loop for the 10 browser tools; bracket-hug compaction across browser_* modules (AST-identical)

Browse Source
This commit is contained in:
Teknium
2026-09-02 16:26:33 -07:00
parent fdaaa87ea4
commit 6a9387f6f4
10 changed files with 110 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/browser_lightpanda.py
+3 -6
View File
@@ -25,8 +25,7 @@ logger = logging.getLogger(__name__)
LIGHTPANDA_INSTALL_URL = "https://lightpanda.io/docs/run-locally/installation/one-liner"
LIGHTPANDA_INSTALL_HINT = (
f"Install Lightpanda from {LIGHTPANDA_INSTALL_URL} and make sure "
"`lightpanda` is on PATH"
f"Install Lightpanda from {LIGHTPANDA_INSTALL_URL} and make sure " "`lightpanda` is on PATH"
)
_READY_TIMEOUT_S = 10.0
@@ -57,8 +56,7 @@ class LightpandaServer:
def _home_candidates() -> list:
home = Path.home()
candidates = [
home / ".lightpanda" / "lightpanda",
home / ".local" / "bin" / "lightpanda",
home / ".lightpanda" / "lightpanda", home / ".local" / "bin" / "lightpanda"
]
try:
from hermes_constants import get_hermes_home
@@ -285,8 +283,7 @@ def launch_lightpanda(
with _servers_lock:
_servers[session_name] = server
logger.info(
"Started lightpanda serve (pid %s, port %s) for session %s",
proc.pid, port, session_name,
"Started lightpanda serve (pid %s, port %s) for session %s", proc.pid, port, session_name
)
return server, None
tools/browser_supervisor.py
+1 -2
View File
@@ -555,8 +555,7 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin):
# CDP event → handler(self, params, session_id). Async handlers return an
# awaitable that ``_read_loop`` awaits; sync handlers return None.
_EVENT_HANDLERS: Dict[str, Callable[..., Any]] = {
**DialogSupervisionMixin.EVENT_HANDLERS,
**FrameTrackingMixin.EVENT_HANDLERS,
**DialogSupervisionMixin.EVENT_HANDLERS, **FrameTrackingMixin.EVENT_HANDLERS
}
tools/browser_tool.py
+81 -210
View File
@@ -722,8 +722,7 @@ class _BrowserSessionBackend:
if reason is None:
return True
logger.info(
"Recycling suspect browser session %s before reuse (%s)",
self._session_key, reason,
"Recycling suspect browser session %s before reuse (%s)", self._session_key, reason
)
try:
_cleanup_single_browser_session(self._session_key)
@@ -852,8 +851,7 @@ BROWSER_TOOL_SCHEMAS = [
"description": "The element reference from the snapshot (e.g., '@e3')"
},
"text": {
"type": "string",
"description": "The text to type into the field"
"type": "string", "description": "The text to type into the field"
}
},
"required": ["ref", "text"]
@@ -866,9 +864,7 @@ BROWSER_TOOL_SCHEMAS = [
"type": "object",
"properties": {
"direction": {
"type": "string",
"enum": ["up", "down"],
"description": "Direction to scroll"
"type": "string", "enum": ["up", "down"], "description": "Direction to scroll"
}
},
"required": ["direction"]
@@ -878,9 +874,7 @@ BROWSER_TOOL_SCHEMAS = [
"name": "browser_back",
"description": "Navigate back to the previous page in browser history. Requires browser_navigate to be called first.",
"parameters": {
"type": "object",
"properties": {},
"required": []
"type": "object", "properties": {}, "required": []
}
},
{
@@ -901,9 +895,7 @@ BROWSER_TOOL_SCHEMAS = [
"name": "browser_get_images",
"description": "Get a list of all images on the current page with their URLs and alt text. Useful for finding images to analyze with the vision tool. Requires browser_navigate to be called first.",
"parameters": {
"type": "object",
"properties": {},
"required": []
"type": "object", "properties": {}, "required": []
}
},
{
@@ -1043,8 +1035,7 @@ def _post_redirect_block(nav_session_key: str, url: str, final_url: str, auto_lo
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",
"success": False, "error": "Blocked: redirect landed on a cloud metadata endpoint"
})
if (
not _is_local_backend()
@@ -1054,8 +1045,7 @@ def _post_redirect_block(nav_session_key: str, url: str, final_url: str, auto_lo
):
_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",
"success": False, "error": "Blocked: redirect landed on a private/internal address"
})
return None
@@ -1125,16 +1115,12 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
_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),
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")
"success": False, "error": result.get("error", "Navigation failed")
}, ensure_ascii=False)
data = result.get("data", {})
@@ -1146,9 +1132,7 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
return blocked
response = {
"success": True,
"url": final_url,
"title": title
"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.
@@ -1188,9 +1172,7 @@ def browser_navigate(url: str, task_id: Optional[str] = None) -> str:
def browser_snapshot(
full: bool = False,
task_id: Optional[str] = None,
user_task: Optional[str] = None
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``).
@@ -1252,8 +1234,7 @@ def browser_snapshot(
return json.dumps(response, ensure_ascii=False)
else:
response = {
"success": False,
"error": result.get("error", "Failed to get snapshot")
"success": False, "error": result.get("error", "Failed to get snapshot")
}
return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False)
@@ -1306,8 +1287,7 @@ def browser_type(ref: str, text: str, task_id: Optional[str] = None) -> str:
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,
redact_browser_typed_text_for_display, redact_tool_args_for_display
)
# Typed text goes through the secret-pattern redactor so API keys / tokens
@@ -1327,8 +1307,7 @@ 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'."
"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.
@@ -1472,8 +1451,7 @@ def browser_console(clear: bool = False, expression: Optional[str] = None, task_
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",
"message": _redact_browser_output(err.get("message", "")), "source": "exception"
})
response = {
@@ -1552,8 +1530,7 @@ def _eval_supervisor_fast_path(effective_task_id: str, expression: str) -> Optio
if "supervisor" not in err.lower():
return json.dumps({"success": False, "error": err}, ensure_ascii=False)
logger.debug(
"browser_eval: supervisor path unavailable (%s), falling back to subprocess",
err,
"browser_eval: supervisor path unavailable (%s), falling back to subprocess", err
)
except ImportError:
pass
@@ -1758,23 +1735,17 @@ def browser_get_images(task_id: Optional[str] = None) -> str:
images = raw_result
response = {
"success": True,
"images": _redact_browser_output(images),
"count": len(images)
"success": True, "images": _redact_browser_output(images), "count": len(images)
}
return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False)
except json.JSONDecodeError:
response = {
"success": True,
"images": [],
"count": 0,
"warning": "Could not parse image data"
"success": True, "images": [], "count": 0, "warning": "Could not parse image data"
}
return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False)
else:
response = {
"success": False,
"error": result.get("error", "Failed to get images")
"success": False, "error": result.get("error", "Failed to get images")
}
return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False)
@@ -1976,8 +1947,7 @@ if __name__ == "__main__":
# ---------------------------------------------------------------------------
from tools.registry import registry, tool_error
from tools.browser_extension_router import (
extension_controller_available,
routed_browser_handler,
extension_controller_available, routed_browser_handler
)
_BROWSER_SCHEMA_MAP = {s["name"]: s for s in BROWSER_TOOL_SCHEMAS}
@@ -1986,8 +1956,7 @@ _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"),
"task_id": kw.get("task_id"), "session_id": kw.get("session_id")
}
@@ -1996,163 +1965,65 @@ def check_browser_routed_requirements(action: str = "browser_snapshot") -> bool:
return check_browser_requirements() or extension_controller_available(action)
def check_browser_navigate_requirements() -> bool:
return check_browser_routed_requirements("browser_navigate")
def check_browser_snapshot_requirements() -> bool:
return check_browser_routed_requirements("browser_snapshot")
def check_browser_click_requirements() -> bool:
return check_browser_routed_requirements("browser_click")
def check_browser_type_requirements() -> bool:
return check_browser_routed_requirements("browser_type")
def check_browser_scroll_requirements() -> bool:
return check_browser_routed_requirements("browser_scroll")
def check_browser_back_requirements() -> bool:
return check_browser_routed_requirements("browser_back")
def check_browser_press_requirements() -> bool:
return check_browser_routed_requirements("browser_press")
registry.register(
name="browser_navigate",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_navigate"],
handler=lambda args, **kw: routed_browser_handler(
"browser_navigate",
args,
fallback=lambda: browser_navigate(url=args.get("url", ""), task_id=kw.get("task_id")),
**_browser_router_kw(kw),
),
check_fn=check_browser_navigate_requirements,
emoji="🌐",
)
registry.register(
name="browser_snapshot",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_snapshot"],
handler=lambda args, **kw: routed_browser_handler(
"browser_snapshot",
args,
fallback=lambda: browser_snapshot(
full=args.get("full", False), task_id=kw.get("task_id"), user_task=kw.get("user_task")),
**_browser_router_kw(kw),
),
check_fn=check_browser_snapshot_requirements,
emoji="📸",
)
registry.register(
name="browser_click",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_click"],
handler=lambda args, **kw: routed_browser_handler(
"browser_click",
args,
fallback=lambda: browser_click(ref=args.get("ref", ""), task_id=kw.get("task_id")),
**_browser_router_kw(kw),
),
check_fn=check_browser_click_requirements,
emoji="👆",
)
registry.register(
name="browser_type",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_type"],
handler=lambda args, **kw: routed_browser_handler(
"browser_type",
args,
fallback=lambda: browser_type(ref=args.get("ref", ""), text=args.get("text", ""), task_id=kw.get("task_id")),
**_browser_router_kw(kw),
),
check_fn=check_browser_type_requirements,
emoji="⌨️",
)
registry.register(
name="browser_scroll",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_scroll"],
handler=lambda args, **kw: routed_browser_handler(
"browser_scroll",
args,
fallback=lambda: browser_scroll(direction=args.get("direction", "down"), task_id=kw.get("task_id")),
**_browser_router_kw(kw),
),
check_fn=check_browser_scroll_requirements,
emoji="📜",
)
registry.register(
name="browser_back",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_back"],
handler=lambda args, **kw: routed_browser_handler(
"browser_back",
args,
fallback=lambda: browser_back(task_id=kw.get("task_id")),
**_browser_router_kw(kw),
),
check_fn=check_browser_back_requirements,
emoji="◀️",
)
registry.register(
name="browser_press",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_press"],
handler=lambda args, **kw: routed_browser_handler(
"browser_press",
args,
fallback=lambda: browser_press(key=args.get("key", ""), task_id=kw.get("task_id")),
**_browser_router_kw(kw),
),
check_fn=check_browser_press_requirements,
emoji="⌨️",
# (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).
_BROWSER_TOOL_TABLE = (
("browser_navigate", "🌐", None,
lambda args, kw: browser_navigate(url=args.get("url", ""), task_id=kw.get("task_id"))),
("browser_snapshot", "📸", None,
lambda args, kw: browser_snapshot(
full=args.get("full", False), task_id=kw.get("task_id"), user_task=kw.get("user_task"))),
("browser_click", "👆", None,
lambda args, kw: browser_click(ref=args.get("ref", ""), task_id=kw.get("task_id"))),
("browser_type", "⌨️", None,
lambda args, kw: browser_type(ref=args.get("ref", ""), text=args.get("text", ""), task_id=kw.get("task_id"))),
("browser_scroll", "📜", None,
lambda args, kw: browser_scroll(direction=args.get("direction", "down"), task_id=kw.get("task_id"))),
("browser_back", "◀️", None,
lambda args, kw: browser_back(task_id=kw.get("task_id"))),
("browser_press", "⌨️", None,
lambda args, kw: browser_press(key=args.get("key", ""), task_id=kw.get("task_id"))),
("browser_get_images", "🖼️", check_browser_requirements,
lambda args, kw: browser_get_images(task_id=kw.get("task_id"))),
("browser_vision", "👁️", check_browser_vision_requirements,
lambda args, kw: browser_vision(
question=args.get("question", ""), annotate=args.get("annotate", False), task_id=kw.get("task_id"))),
("browser_console", "🖥️", check_browser_requirements,
lambda args, kw: browser_console(
clear=args.get("clear", False), expression=args.get("expression"), task_id=kw.get("task_id"))),
)
registry.register(
name="browser_get_images",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_get_images"],
handler=lambda args, **kw: routed_browser_handler(
"browser_get_images",
args,
fallback=lambda: browser_get_images(task_id=kw.get("task_id")),
**_browser_router_kw(kw),
),
check_fn=check_browser_requirements,
emoji="🖼️",
)
registry.register(
name="browser_vision",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_vision"],
handler=lambda args, **kw: routed_browser_handler(
"browser_vision",
args,
fallback=lambda: browser_vision(question=args.get("question", ""), annotate=args.get("annotate", False), task_id=kw.get("task_id")),
**_browser_router_kw(kw),
),
check_fn=check_browser_vision_requirements,
emoji="👁️",
)
registry.register(
name="browser_console",
toolset="browser",
schema=_BROWSER_SCHEMA_MAP["browser_console"],
handler=lambda args, **kw: routed_browser_handler(
"browser_console",
args,
fallback=lambda: browser_console(clear=args.get("clear", False), expression=args.get("expression"), task_id=kw.get("task_id")),
**_browser_router_kw(kw),
),
check_fn=check_browser_requirements,
emoji="🖥️",
)
def _routed_check_fn(name: str):
"""Per-action availability gate (a named function, as the registry expects)."""
def check() -> bool:
return check_browser_routed_requirements(name)
check.__name__ = check.__qualname__ = f"check_{name}_requirements"
return check
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 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")
for _name, _emoji, _check_fn, _fallback in _BROWSER_TOOL_TABLE:
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"],
emoji=_emoji,
)
tools/browser_tool_cdp.py
+3 -10
View File
@@ -116,9 +116,7 @@ def _get_dialog_policy_config() -> Tuple[str, float]:
# Defer imports so browser_tool can be imported in minimal environments.
_bt = _origin()
from tools.browser_supervisor import (
DEFAULT_DIALOG_POLICY,
DEFAULT_DIALOG_TIMEOUT_S,
_VALID_POLICIES,
DEFAULT_DIALOG_POLICY, DEFAULT_DIALOG_TIMEOUT_S, _VALID_POLICIES
)
try:
@@ -171,16 +169,11 @@ def _ensure_cdp_supervisor(task_id: str) -> None:
policy, timeout_s = _bt._get_dialog_policy_config()
SUPERVISOR_REGISTRY.get_or_start(
task_id=task_id,
cdp_url=cdp_url,
dialog_policy=policy,
dialog_timeout_s=timeout_s,
task_id=task_id, cdp_url=cdp_url, dialog_policy=policy, dialog_timeout_s=timeout_s
)
except Exception as exc:
_bt.logger.debug(
"CDP supervisor attach for task=%s failed (non-fatal): %s",
task_id,
exc,
"CDP supervisor attach for task=%s failed (non-fatal): %s", task_id, exc
)
tools/browser_tool_eval_policy.py
+3 -6
View File
@@ -65,13 +65,11 @@ def _current_page_private_url(effective_task_id: str) -> Optional[str]:
_bt = _origin()
try:
url_result = _bt._run_browser_command(
effective_task_id, "eval", ["window.location.href"],
timeout=5, _engine_override="auto",
effective_task_id, "eval", ["window.location.href"], timeout=5, _engine_override="auto"
)
if url_result.get("success"):
current_url = (
url_result.get("data", {}).get("result", "")
.strip().strip('"').strip("'")
url_result.get("data", {}).get("result", "") .strip().strip('"').strip("'")
)
if current_url and (
_bt._is_always_blocked_url(current_url) or not _bt._is_safe_url(current_url)
@@ -96,8 +94,7 @@ _RISKY_BROWSER_EVAL_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
_JS_STRING_LITERAL_RE = re.compile(
r"""'(?:\\.|[^'\\])*'|\"(?:\\.|[^\"\\])*\"|`(?:\\.|[^`\\])*`""",
re.S,
r"""'(?:\\.|[^'\\])*'|\"(?:\\.|[^\"\\])*\"|`(?:\\.|[^`\\])*`""", re.S
)
tools/browser_tool_lifecycle.py
+4 -10
View File
@@ -131,9 +131,7 @@ def _session_owner_scope(task_id: str):
return
from agent.secret_scope import (
build_profile_secret_scope,
reset_secret_scope,
set_secret_scope,
build_profile_secret_scope, reset_secret_scope, set_secret_scope
)
from hermes_cli.env_loader import hydrate_profile_secret_sources
@@ -487,9 +485,7 @@ def _start_browser_cleanup_thread():
if _bt._cleanup_thread is None or not _bt._cleanup_thread.is_alive():
_bt._cleanup_running = True
_bt._cleanup_thread = threading.Thread(
target=_bt._browser_cleanup_thread_worker,
daemon=True,
name="browser-cleanup"
target=_bt._browser_cleanup_thread_worker, daemon=True, name="browser-cleanup"
)
_bt._cleanup_thread.start()
_bt.logger.info("Started inactivity cleanup thread (timeout: %ss)", _bt.BROWSER_SESSION_INACTIVITY_TIMEOUT)
@@ -815,15 +811,13 @@ def _cleanup_single_browser_session(task_id: str) -> None:
_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,
"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,
"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)
tools/browser_tool_lightpanda_fallback.py
+6 -16
View File
@@ -58,8 +58,7 @@ def lightpanda_engine_status() -> Tuple[bool, str]:
if bu_mode:
try:
from tools.browser_use_cli import (
_read_browser_cfg,
is_legacy_browser_use_cloud_config,
_read_browser_cfg, is_legacy_browser_use_cloud_config
)
if is_legacy_browser_use_cloud_config(_read_browser_cfg()):
@@ -131,16 +130,13 @@ def _needs_lightpanda_fallback(engine: str, command: str, result: Dict[str, Any]
def _annotate_lightpanda_fallback(result: Dict[str, Any], reason: str) -> Dict[str, Any]:
"""Add a user-visible Chrome fallback warning to a browser command result."""
warning = (
"⚠ Lightpanda fallback: Chrome was used for this browser action. "
f"{reason}"
"⚠ Lightpanda fallback: Chrome was used for this browser action. " f"{reason}"
)
annotated = dict(result)
annotated["fallback_warning"] = warning
annotated["browser_engine"] = "chrome"
annotated["browser_engine_fallback"] = {
"from": "lightpanda",
"to": "chrome",
"reason": reason,
"from": "lightpanda", "to": "chrome", "reason": reason
}
data = annotated.get("data")
if isinstance(data, dict):
@@ -148,8 +144,7 @@ def _annotate_lightpanda_fallback(result: Dict[str, Any], reason: str) -> Dict[s
data.setdefault("fallback_warning", warning)
data.setdefault("browser_engine", "chrome")
data.setdefault(
"browser_engine_fallback",
{"from": "lightpanda", "to": "chrome", "reason": reason},
"browser_engine_fallback", {"from": "lightpanda", "to": "chrome", "reason": reason}
)
annotated["data"] = data
return annotated
@@ -165,10 +160,7 @@ def _copy_fallback_warning(target: Dict[str, Any], result: Dict[str, Any]) -> Di
def _run_chrome_fallback_command(
task_id: str,
command: str,
args: List[str],
timeout: int,
task_id: str, command: str, args: List[str], timeout: int
) -> Dict[str, Any]:
"""Run a browser command in a temporary Chrome session at the current URL.
@@ -259,9 +251,7 @@ def _run_chrome_fallback_command(
def _chrome_fallback_screenshot(
task_id: str,
args: List[str],
timeout: int,
task_id: str, args: List[str], timeout: int
) -> Dict[str, Any]:
"""Take a screenshot using a temporary Chrome session."""
_bt = _origin()
tools/browser_tool_real_profile.py
+1 -4
View File
@@ -332,10 +332,7 @@ def _real_profile_cdp() -> tuple:
)
from hermes_cli.browser_connect import (
chromium_executable,
detect_default_chromium,
real_profile_copy_dir,
snapshot_real_profile,
chromium_executable, detect_default_chromium, real_profile_copy_dir, snapshot_real_profile
)
with _bt._real_profile_cdp_lock:
tools/browser_tool_session.py
+7 -19
View File
@@ -75,10 +75,7 @@ def _unlink_command_output_files(*paths: str) -> None:
def _format_browser_timeout_error(
command: str,
timeout: int,
stdout: str,
stderr: str,
command: str, timeout: int, stdout: str, stderr: str
) -> str:
"""Build an actionable timeout message from captured daemon output."""
_bt = _origin()
@@ -260,8 +257,7 @@ def _create_lightpanda_session(task_id: str) -> Dict[str, Any]:
if err:
raise RuntimeError(err)
_bt.logger.info(
"Created Lightpanda session %s (port %s) for task %s",
session_name, server.port, task_id,
"Created Lightpanda session %s (port %s) for task %s", session_name, server.port, task_id
)
return {
"session_name": session_name,
@@ -402,8 +398,7 @@ def _get_session_info(task_id: Optional[str] = None) -> Dict[str, Any]:
return existing_session
_bt.logger.info(
"Replacing expired or dead browser session for task %s",
task_id,
"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
@@ -454,9 +449,7 @@ def _get_session_info(task_id: Optional[str] = None) -> Dict[str, Any]:
def _discard_timed_out_browser_session(
task_id: str,
session_info: Dict[str, Any],
task_socket_dir: str,
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()
@@ -540,9 +533,7 @@ def _browser_daemon_responsive(task_socket_dir: str, probe_timeout_s: float = 1.
def _handle_browser_command_timeout(
task_id: str,
session_info: Dict[str, Any],
task_socket_dir: str,
task_id: str, session_info: Dict[str, Any], task_socket_dir: str
) -> None:
"""Recover session state after a browser command timeout.
@@ -631,8 +622,7 @@ def _interpret_browser_command_output(command: str, stdout: str, stderr: str, re
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,
"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}"}
@@ -776,9 +766,7 @@ def _run_browser_command(
_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],
"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)
tools/browser_tool_vision.py
+1 -2
View File
@@ -95,8 +95,7 @@ def _native_vision_result(
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}"
f"{native_result.get('text_summary', '')} " f"Screenshot path: {screenshot_path}"
).strip()
return native_result