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

refactor(tools/browser): extract eval-policy/lightpanda-fallback/real-profile/snapshot modules from browser_tool; split supervisor dialogs/frames; dedupe camofox/cli

Browse Source
This commit is contained in:
Teknium
2026-09-02 13:55:32 -07:00
parent c1f8af1e86
commit 9f6335bc44
17 changed files with 3790 additions and 4832 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/browser_supervisor.py
+75 -705
View File
@@ -10,6 +10,10 @@ Not in the agent's tool schema. Output reaches the agent via
``browser_snapshot`` (merges supervisor state, see ``tools/browser_tool.py``)
and ``browser_dialog`` (calls ``respond_to_dialog()``).
Design spec: ``website/docs/developer-guide/browser-supervisor.md``.
Dialog capture lives in ``tools.browser_supervisor_dialogs``, frame tracking in
``tools.browser_supervisor_frames``; both are mixed into ``CDPSupervisor`` and
their public names are re-exported here.
"""
from __future__ import annotations
@@ -19,8 +23,32 @@ import json
import logging
import threading
import time
from dataclasses import dataclass, replace
from typing import Any, Callable, Coroutine, Dict, List, Optional, Tuple, TYPE_CHECKING
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional, Tuple, TYPE_CHECKING
from tools.browser_supervisor_dialogs import ( # noqa: F401 — re-exported
DEFAULT_DIALOG_POLICY,
DEFAULT_DIALOG_TIMEOUT_S,
DIALOG_BRIDGE_HOST,
DIALOG_BRIDGE_URL_PATTERN,
DIALOG_POLICY_AUTO_ACCEPT,
DIALOG_POLICY_AUTO_DISMISS,
DIALOG_POLICY_MUST_RESPOND,
RECENT_DIALOGS_MAX,
_DIALOG_BRIDGE_SCRIPT,
_VALID_POLICIES,
DialogRecord,
DialogSupervisionMixin,
PendingDialog,
_redact_supervisor_text,
_trim_ring,
)
from tools.browser_supervisor_frames import ( # noqa: F401 — re-exported
FRAME_TREE_MAX_ENTRIES,
FRAME_TREE_MAX_OOPIF_DEPTH,
FrameInfo,
FrameTrackingMixin,
)
# ``websockets`` costs ~22 ms at import and is only needed once a supervisor
# connects; with postponed annotations the type import stays under TYPE_CHECKING.
@@ -29,6 +57,9 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
# Ring buffer of recent console-level events.
CONSOLE_HISTORY_MAX = 50
def _redact_cdp_error_text(exc: object) -> str:
"""Redact CDP endpoint credentials from an exception's string form.
@@ -46,187 +77,23 @@ def _redact_cdp_error_text(exc: object) -> str:
return "<error redacted>"
def _redact_supervisor_text(value: str) -> str:
"""Redact page-originated text before exposing supervisor snapshots."""
from agent.redact import redact_sensitive_text
return redact_sensitive_text(value, force=True)
class _LoopUnavailable(RuntimeError):
"""The supervisor loop refused new work (closed / shutting down)."""
# ── Config defaults ───────────────────────────────────────────────────────────
def _schedule(coro, loop, *, timeout: float):
"""Run ``coro`` on the supervisor loop from a sync caller and wait for its result."""
from agent.async_utils import safe_schedule_threadsafe
DIALOG_POLICY_MUST_RESPOND = "must_respond"
DIALOG_POLICY_AUTO_DISMISS = "auto_dismiss"
DIALOG_POLICY_AUTO_ACCEPT = "auto_accept"
_VALID_POLICIES = frozenset(
{DIALOG_POLICY_MUST_RESPOND, DIALOG_POLICY_AUTO_DISMISS, DIALOG_POLICY_AUTO_ACCEPT}
)
DEFAULT_DIALOG_POLICY = DIALOG_POLICY_MUST_RESPOND
DEFAULT_DIALOG_TIMEOUT_S = 300.0
# Snapshot caps for frame_tree — keep payloads bounded on ad-heavy pages.
FRAME_TREE_MAX_ENTRIES = 30
FRAME_TREE_MAX_OOPIF_DEPTH = 2
# Ring buffer of recent console-level events.
CONSOLE_HISTORY_MAX = 50
# Last N closed dialogs kept in ``recent_dialogs`` so agents on backends that
# auto-dismiss server-side (Browserbase) can still observe that a dialog fired.
RECENT_DIALOGS_MAX = 20
# Magic host the injected dialog bridge XHRs to. Intercepted via the CDP Fetch
# domain before any network resolution, so it never has to exist. Keep ASCII +
# URL-safe; Fetch patterns are gated on it.
DIALOG_BRIDGE_HOST = "hermes-dialog-bridge.invalid"
DIALOG_BRIDGE_URL_PATTERN = f"http://{DIALOG_BRIDGE_HOST}/*"
# Injected into every frame via Page.addScriptToEvaluateOnNewDocument. Overrides
# alert/confirm/prompt to round-trip through a sync XHR we intercept via
# Fetch.requestPaused. Works on Browserbase (whose CDP proxy auto-dismisses REAL
# native dialogs) because the native dialogs never fire.
_DIALOG_BRIDGE_SCRIPT = r"""
(() => {
if (window.__hermesDialogBridgeInstalled) return;
window.__hermesDialogBridgeInstalled = true;
const ENDPOINT = "http://hermes-dialog-bridge.invalid/";
function ask(kind, message, defaultPrompt) {
try {
const xhr = new XMLHttpRequest();
// Use GET with query params so we don't need to worry about request
// body encoding in the Fetch interceptor.
const params = new URLSearchParams({
kind: String(kind || ""),
message: String(message == null ? "" : message),
default_prompt: String(defaultPrompt == null ? "" : defaultPrompt),
});
xhr.open("GET", ENDPOINT + "?" + params.toString(), false); // sync
xhr.send(null);
if (xhr.status !== 200) return null;
const body = xhr.responseText || "";
let parsed;
try { parsed = JSON.parse(body); } catch (e) { return null; }
if (kind === "alert") return undefined;
if (kind === "confirm") return Boolean(parsed && parsed.accept);
if (kind === "prompt") {
if (!parsed || !parsed.accept) return null;
return parsed.prompt_text == null ? "" : String(parsed.prompt_text);
}
return null;
} catch (e) {
// If the bridge is unreachable, fall back to the native call so the
// page still sees *some* behavior (the backend will auto-dismiss).
return null;
}
}
const realAlert = window.alert;
const realConfirm = window.confirm;
const realPrompt = window.prompt;
window.alert = function(message) { ask("alert", message, ""); };
window.confirm = function(message) {
const r = ask("confirm", message, "");
return r === null ? false : Boolean(r);
};
window.prompt = function(message, def) {
const r = ask("prompt", message, def == null ? "" : def);
return r === null ? null : String(r);
};
// onbeforeunload — we can't really synchronously prompt the user from this
// event without racing navigation. Leave native behavior for now; the
// supervisor's native-dialog fallback path still surfaces them in
// recent_dialogs.
})();
"""
fut = safe_schedule_threadsafe(coro, loop)
if fut is None:
raise _LoopUnavailable("Browser supervisor loop unavailable")
return fut.result(timeout=timeout)
# ── Data model ────────────────────────────────────────────────────────────────
@dataclass
class PendingDialog:
"""A JS dialog currently open on some frame's session."""
id: str
type: str # "alert" | "confirm" | "prompt" | "beforeunload"
message: str
default_prompt: str
opened_at: float
cdp_session_id: str # which attached CDP session the dialog fired in
frame_id: Optional[str] = None
# Set when captured via the bridge XHR path: respond via Fetch.fulfillRequest,
# NOT Page.handleJavaScriptDialog — the native dialog never fired.
bridge_request_id: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"type": self.type,
"message": _redact_supervisor_text(self.message),
"default_prompt": _redact_supervisor_text(self.default_prompt),
"opened_at": self.opened_at,
"frame_id": self.frame_id,
}
@dataclass
class DialogRecord:
"""A dialog that was opened and then handled (kept briefly in ``recent_dialogs``)."""
id: str
type: str
message: str
opened_at: float
closed_at: float
closed_by: str # "agent" | "auto_policy" | "remote" | "watchdog"
frame_id: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return {
"id": self.id,
"type": self.type,
"message": _redact_supervisor_text(self.message),
"opened_at": self.opened_at,
"closed_at": self.closed_at,
"closed_by": self.closed_by,
"frame_id": self.frame_id,
}
@dataclass
class FrameInfo:
"""One frame in the page's frame tree.
``is_oopif`` frames have their own CDP target (reachable via
``cdp_session_id``); same-origin / srcdoc iframes share the parent process
and have ``is_oopif=False`` + ``cdp_session_id=None``.
"""
frame_id: str
url: str
origin: str
parent_frame_id: Optional[str]
is_oopif: bool
cdp_session_id: Optional[str] = None
name: str = ""
def to_dict(self) -> Dict[str, Any]:
d = {
"frame_id": self.frame_id,
"url": self.url,
"origin": self.origin,
"is_oopif": self.is_oopif,
}
if self.cdp_session_id:
d["session_id"] = self.cdp_session_id
if self.parent_frame_id:
d["parent_frame_id"] = self.parent_frame_id
if self.name:
d["name"] = self.name
return d
@dataclass
class ConsoleEvent:
"""Ring buffer entry for console + exception traffic."""
@@ -263,7 +130,7 @@ class SupervisorSnapshot:
# ── Supervisor core ───────────────────────────────────────────────────────────
class CDPSupervisor:
class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin):
"""One supervisor per (task_id, cdp_url) pair.
``start()`` spawns a daemon thread running its own asyncio loop, connects,
@@ -311,7 +178,6 @@ class CDPSupervisor:
self._pending_calls: Dict[int, asyncio.Future] = {}
self._ws: Optional[ClientConnection] = None
self._page_session_id: Optional[str] = None
self._child_sessions: Dict[str, Dict[str, Any]] = {} # session_id -> info
# Dialog auto-dismiss watchdog handles (per dialog id) + id generator.
self._dialog_watchdogs: Dict[str, asyncio.TimerHandle] = {}
@@ -366,15 +232,9 @@ class CDPSupervisor:
# self._ws`` returns cleanly, ``_run`` hits its ``finally``, pending
# tasks cancel in order, THEN the thread exits.
try:
from agent.async_utils import safe_schedule_threadsafe
fut = safe_schedule_threadsafe(self._close_ws(), loop)
if fut is not None:
try:
fut.result(timeout=2.0)
except Exception:
pass
except RuntimeError:
pass # loop already shutting down
_schedule(self._close_ws(), loop, timeout=2.0)
except Exception:
pass # loop already shutting down / close timed out
if self._thread is not None:
self._thread.join(timeout=timeout)
with self._state_lock:
@@ -383,20 +243,15 @@ class CDPSupervisor:
def snapshot(self) -> SupervisorSnapshot:
"""Return an immutable snapshot of current state."""
with self._state_lock:
dialogs = tuple(self._pending_dialogs.values())
recent = tuple(self._recent_dialogs[-RECENT_DIALOGS_MAX:])
frames_tree = self._build_frame_tree_locked()
console = tuple(self._console_events[-CONSOLE_HISTORY_MAX:])
active = self._active
return SupervisorSnapshot(
pending_dialogs=dialogs,
recent_dialogs=recent,
frame_tree=frames_tree,
console_errors=console,
active=active,
cdp_url=self.cdp_url,
task_id=self.task_id,
)
return SupervisorSnapshot(
pending_dialogs=tuple(self._pending_dialogs.values()),
recent_dialogs=tuple(self._recent_dialogs[-RECENT_DIALOGS_MAX:]),
frame_tree=self._build_frame_tree_locked(),
console_errors=tuple(self._console_events[-CONSOLE_HISTORY_MAX:]),
active=self._active,
cdp_url=self.cdp_url,
task_id=self.task_id,
)
def respond_to_dialog(
self,
@@ -438,26 +293,24 @@ class CDPSupervisor:
}
else:
dialog = pending[0]
snapshot_copy = dialog
loop = self._loop
if loop is None:
return {"ok": False, "error": "supervisor loop is not running"}
async def _do_respond():
return await self._handle_dialog_cdp(
snapshot_copy, accept=(action == "accept"), prompt_text=prompt_text or ""
)
try:
from agent.async_utils import safe_schedule_threadsafe
fut = safe_schedule_threadsafe(_do_respond(), loop)
if fut is None:
return {"ok": False, "error": "Browser supervisor loop unavailable"}
fut.result(timeout=timeout)
_schedule(
self._handle_dialog_cdp(
dialog, accept=(action == "accept"), prompt_text=prompt_text or ""
),
loop,
timeout=timeout,
)
except _LoopUnavailable as e:
return {"ok": False, "error": str(e)}
except Exception as e:
return {"ok": False, "error": f"{type(e).__name__}: {e}"}
return {"ok": True, "dialog": snapshot_copy.to_dict()}
return {"ok": True, "dialog": dialog.to_dict()}
def evaluate_runtime(
self,
@@ -487,8 +340,8 @@ class CDPSupervisor:
if not session_id:
return {"ok": False, "error": "supervisor has no attached page session"}
async def _do_eval(by_value: bool) -> Dict[str, Any]:
return await self._cdp(
def _run_eval(by_value: bool) -> Dict[str, Any]:
coro = self._cdp(
"Runtime.evaluate",
{
"expression": expression,
@@ -500,14 +353,7 @@ class CDPSupervisor:
session_id=session_id,
timeout=timeout,
)
from agent.async_utils import safe_schedule_threadsafe
def _run_eval(by_value: bool) -> Dict[str, Any]:
fut = safe_schedule_threadsafe(_do_eval(by_value), loop)
if fut is None:
raise RuntimeError("Browser supervisor loop unavailable")
return fut.result(timeout=timeout + 1)
return _schedule(coro, loop, timeout=timeout + 1)
try:
response = _run_eval(return_by_value)
@@ -529,8 +375,7 @@ class CDPSupervisor:
exception_details = result_payload.get("exceptionDetails")
if exception_details:
exc_text = exception_details.get("text") or "JavaScript exception"
exc_obj = exception_details.get("exception") or {}
description = exc_obj.get("description")
description = (exception_details.get("exception") or {}).get("description")
if description:
exc_text = f"{exc_text}: {description}"
return {"ok": False, "error": exc_text}
@@ -626,12 +471,11 @@ class CDPSupervisor:
reader_task = asyncio.create_task(self._read_loop(), name="cdp-reader")
try:
# Reset per-connection session ids. ``_pending_dialogs`` and
# ``_frames`` are deliberately kept — they reconcile as fresh
# Reset the per-connection page session id. ``_pending_dialogs``
# and ``_frames`` are deliberately kept — they reconcile as fresh
# events arrive; worst case a stale dialog entry is rejected
# with "no dialog is showing" (logged, not surfaced).
self._page_session_id = None
self._child_sessions.clear()
await self._attach_initial_page()
with self._state_lock:
self._active = True
@@ -691,68 +535,9 @@ class CDPSupervisor:
{"targetId": target_id, "flatten": True},
)
self._page_session_id = attach["result"]["sessionId"]
await self._cdp("Page.enable", session_id=self._page_session_id)
await self._cdp("Runtime.enable", session_id=self._page_session_id)
await self._cdp(
"Target.setAutoAttach",
{"autoAttach": True, "waitForDebuggerOnStart": False, "flatten": True},
session_id=self._page_session_id,
)
await self._enable_page_domains(self._page_session_id, timeout=10.0)
await self._install_dialog_bridge(self._page_session_id)
async def _install_dialog_bridge(self, session_id: str) -> None:
"""Install the dialog-bridge init script + Fetch interceptor on a session.
The JS override runs in every frame before page scripts; Fetch.enable
scoped to the bridge URL catches the XHRs, which surface as pending
dialogs and are fulfilled when the agent responds. Idempotent at the CDP
level (Chromium de-dupes identical add-script calls; Fetch.enable
replaces prior patterns).
"""
try:
await self._cdp(
"Page.addScriptToEvaluateOnNewDocument",
{"source": _DIALOG_BRIDGE_SCRIPT, "runImmediately": True},
session_id=session_id,
timeout=5.0,
)
except Exception as e:
logger.debug(
"dialog bridge: addScriptToEvaluateOnNewDocument failed on sid=%s: %s",
(session_id or "")[:16], e,
)
try:
await self._cdp(
"Fetch.enable",
{
"patterns": [
{
"urlPattern": DIALOG_BRIDGE_URL_PATTERN,
"requestStage": "Request",
}
],
"handleAuthRequests": False,
},
session_id=session_id,
timeout=5.0,
)
except Exception as e:
logger.debug(
"dialog bridge: Fetch.enable failed on sid=%s: %s",
(session_id or "")[:16], e,
)
# Best-effort inject into the already-loaded document so existing pages
# pick up the override on reconnect.
try:
await self._cdp(
"Runtime.evaluate",
{"expression": _DIALOG_BRIDGE_SCRIPT, "returnByValue": True},
session_id=session_id,
timeout=3.0,
)
except Exception:
pass
async def _cdp(
self,
method: str,
@@ -817,369 +602,6 @@ class CDPSupervisor:
if result is not None:
await result
async def _on_dialog_opening(
self, params: Dict[str, Any], session_id: Optional[str]
) -> None:
dialog = self._new_dialog(
type=str(params.get("type") or ""),
message=str(params.get("message") or ""),
default_prompt=str(params.get("defaultPrompt") or ""),
session_id=session_id,
frame_id=params.get("frameId"),
)
self._admit_dialog(dialog, self._auto_handle_dialog)
def _new_dialog(
self,
*,
type: str,
message: str,
default_prompt: str,
session_id: Optional[str],
frame_id: Optional[str],
bridge_request_id: Optional[str] = None,
) -> PendingDialog:
self._dialog_seq += 1
return PendingDialog(
id=f"d-{self._dialog_seq}",
type=type,
message=message,
default_prompt=default_prompt,
opened_at=time.time(),
cdp_session_id=session_id or self._page_session_id or "",
frame_id=frame_id,
bridge_request_id=bridge_request_id,
)
def _admit_dialog(
self,
dialog: PendingDialog,
responder: Callable[..., Coroutine[Any, Any, None]],
) -> None:
"""Apply the dialog policy: auto-respond via ``responder`` or queue + arm watchdog.
Auto policies archive FIRST (tagged ``auto_policy``) so the ``closed``
event that follows our own response isn't re-archived as ``remote``.
"""
if self.dialog_policy == DIALOG_POLICY_AUTO_DISMISS:
with self._state_lock:
self._archive_dialog_locked(dialog, "auto_policy")
asyncio.create_task(responder(dialog, accept=False, prompt_text=""))
elif self.dialog_policy == DIALOG_POLICY_AUTO_ACCEPT:
with self._state_lock:
self._archive_dialog_locked(dialog, "auto_policy")
asyncio.create_task(
responder(dialog, accept=True, prompt_text=dialog.default_prompt)
)
else:
# must_respond → add to pending and arm watchdog.
with self._state_lock:
self._pending_dialogs[dialog.id] = dialog
loop = asyncio.get_running_loop()
handle = loop.call_later(
self.dialog_timeout_s,
lambda: asyncio.create_task(self._dialog_timeout_expired(dialog.id)),
)
self._dialog_watchdogs[dialog.id] = handle
async def _native_handle_dialog(
self, dialog: PendingDialog, *, accept: bool, prompt_text: Optional[str]
) -> None:
"""Page.handleJavaScriptDialog; ``promptText`` sent only for prompt dialogs
when ``prompt_text`` is given. Raises on CDP failure."""
params: Dict[str, Any] = {"accept": accept}
if prompt_text is not None and dialog.type == "prompt":
params["promptText"] = prompt_text
await self._cdp(
"Page.handleJavaScriptDialog",
params,
session_id=dialog.cdp_session_id or None,
timeout=5.0,
)
async def _auto_handle_dialog(
self, dialog: PendingDialog, *, accept: bool, prompt_text: str
) -> None:
"""Auto-policy response for a native dialog (already archived by the caller)."""
try:
await self._native_handle_dialog(dialog, accept=accept, prompt_text=prompt_text)
except Exception as e:
logger.debug("auto-handle CDP call failed for %s: %s", dialog.id, e)
def _retire_dialog(self, dialog_id: str, closed_by: str) -> None:
"""Remove a pending dialog (archiving it with ``closed_by``) and cancel its watchdog."""
with self._state_lock:
dialog = self._pending_dialogs.pop(dialog_id, None)
if dialog is not None:
self._archive_dialog_locked(dialog, closed_by)
handle = self._dialog_watchdogs.pop(dialog_id, None)
if handle is not None:
handle.cancel()
async def _dialog_timeout_expired(self, dialog_id: str) -> None:
with self._state_lock:
dialog = self._pending_dialogs.get(dialog_id)
if dialog is None:
return
logger.warning(
"CDP supervisor %s: dialog %s (%s) auto-dismissed after %ss timeout",
self.task_id,
dialog_id,
dialog.type,
self.dialog_timeout_s,
)
try:
# Archive with watchdog tag BEFORE unblocking the page.
with self._state_lock:
if self._pending_dialogs.pop(dialog_id, None) is not None:
self._archive_dialog_locked(dialog, "watchdog")
if dialog.bridge_request_id:
await self._fulfill_bridge_request(dialog, accept=False, prompt_text="")
else:
await self._native_handle_dialog(dialog, accept=False, prompt_text=None)
except Exception as e:
logger.debug("auto-dismiss failed for %s: %s", dialog_id, e)
def _archive_dialog_locked(self, dialog: PendingDialog, closed_by: str) -> None:
"""Move a pending dialog to the recent_dialogs ring buffer. Must hold state_lock."""
record = DialogRecord(
id=dialog.id,
type=dialog.type,
message=dialog.message,
opened_at=dialog.opened_at,
closed_at=time.time(),
closed_by=closed_by,
frame_id=dialog.frame_id,
)
self._recent_dialogs.append(record)
if len(self._recent_dialogs) > RECENT_DIALOGS_MAX * 2:
self._recent_dialogs = self._recent_dialogs[-RECENT_DIALOGS_MAX:]
async def _handle_dialog_cdp(
self, dialog: PendingDialog, *, accept: bool, prompt_text: str
) -> None:
"""Agent response path: bridge-fulfill for XHR-captured dialogs, else native CDP.
The dialog is retired regardless of outcome — a CDP error usually means
it already closed (browser auto-dismissed after navigation, etc.).
"""
try:
if dialog.bridge_request_id:
await self._fulfill_bridge_request(
dialog, accept=accept, prompt_text=prompt_text
)
else:
await self._native_handle_dialog(dialog, accept=accept, prompt_text=prompt_text)
finally:
self._retire_dialog(dialog.id, "agent")
async def _on_dialog_closed(
self, params: Dict[str, Any], session_id: Optional[str]
) -> None:
# ``Page.javascriptDialogClosed`` carries only ``result``/``userInput``, not
# the message. Match by session id and clear the oldest native dialog on
# it — the JS thread blocks while a dialog is up, so at most one is in
# flight per session. Bridge dialogs resolve via Fetch.fulfillRequest.
with self._state_lock:
candidate_ids = [
d.id
for d in self._pending_dialogs.values()
if d.cdp_session_id == session_id and d.bridge_request_id is None
]
if candidate_ids:
self._retire_dialog(candidate_ids[0], "remote")
async def _on_fetch_paused(
self, params: Dict[str, Any], session_id: Optional[str]
) -> None:
"""Bridge XHR captured mid-flight — materialize as a pending dialog.
The page's JS thread is blocked on the XHR until we Fetch.fulfillRequest
(from ``respond_to_dialog`` or the watchdog).
"""
url = str(params.get("request", {}).get("url") or "")
request_id = params.get("requestId")
if not request_id:
return
if DIALOG_BRIDGE_HOST not in url:
# Not ours — forward unchanged so the page sees its own request.
try:
await self._cdp(
"Fetch.continueRequest", {"requestId": request_id},
session_id=session_id, timeout=3.0,
)
except Exception:
pass
return
from urllib.parse import urlparse, parse_qs
q = parse_qs(urlparse(url).query)
def _q(name: str) -> str:
v = q.get(name, [""])
return v[0] if v else ""
dialog = self._new_dialog(
type=_q("kind") or "alert",
message=_q("message"),
default_prompt=_q("default_prompt"),
session_id=session_id,
frame_id=params.get("frameId"),
bridge_request_id=str(request_id),
)
self._admit_dialog(dialog, self._fulfill_bridge_request)
async def _fulfill_bridge_request(
self, dialog: PendingDialog, *, accept: bool, prompt_text: str
) -> None:
"""Resolve a bridge XHR via Fetch.fulfillRequest so the page unblocks."""
if not dialog.bridge_request_id:
return
payload = {
"accept": bool(accept),
"prompt_text": prompt_text if dialog.type == "prompt" else "",
"dialog_id": dialog.id,
}
body = json.dumps(payload).encode()
try:
import base64 as _b64
await self._cdp(
"Fetch.fulfillRequest",
{
"requestId": dialog.bridge_request_id,
"responseCode": 200,
"responseHeaders": [
{"name": "Content-Type", "value": "application/json"},
{"name": "Access-Control-Allow-Origin", "value": "*"},
],
"body": _b64.b64encode(body).decode(),
},
session_id=dialog.cdp_session_id or None,
timeout=5.0,
)
except Exception as e:
logger.debug("bridge fulfill failed for %s: %s", dialog.id, e)
# ── Frame / target tracking ─────────────────────────────────────────────
def _on_frame_attached(
self, params: Dict[str, Any], session_id: Optional[str]
) -> None:
frame_id = params.get("frameId")
if not frame_id:
return
with self._state_lock:
self._frames[frame_id] = FrameInfo(
frame_id=frame_id,
url="",
origin="",
parent_frame_id=params.get("parentFrameId"),
is_oopif=False,
cdp_session_id=session_id,
)
def _on_frame_navigated(
self, params: Dict[str, Any], session_id: Optional[str]
) -> None:
frame = params.get("frame") or {}
frame_id = frame.get("id")
if not frame_id:
return
with self._state_lock:
existing = self._frames.get(frame_id)
self._frames[frame_id] = FrameInfo(
frame_id=frame_id,
url=str(frame.get("url") or ""),
origin=str(frame.get("securityOrigin") or frame.get("origin") or ""),
parent_frame_id=frame.get("parentId") or (existing.parent_frame_id if existing else None),
is_oopif=bool(existing.is_oopif if existing else False),
cdp_session_id=existing.cdp_session_id if existing else session_id,
name=str(frame.get("name") or (existing.name if existing else "")),
)
def _on_frame_detached(
self, params: Dict[str, Any], session_id: Optional[str]
) -> None:
"""Drop a frame only when it's truly gone.
``reason="swap"`` means the frame is migrating processes (e.g. promoted
to an OOPIF) — dropping it would hide the iframe, so it's a no-op. Even
with ``reason="remove"`` the parent only knows the child left ITS
process; if we hold a live child session for that frame_id it is still
alive, so keep it until Target.detached + a later frameDetached clear it.
"""
frame_id = params.get("frameId")
if not frame_id:
return
reason = str(params.get("reason") or "remove").lower()
if reason == "swap":
return
with self._state_lock:
existing = self._frames.get(frame_id)
if existing and existing.is_oopif and existing.cdp_session_id:
return
self._frames.pop(frame_id, None)
async def _on_target_attached(self, params: Dict[str, Any], session_id: Optional[str] = None) -> None:
info = params.get("targetInfo") or {}
sid = params.get("sessionId")
target_type = info.get("type")
if not sid or target_type not in {"iframe", "worker"}:
return
self._child_sessions[sid] = {"info": info, "type": target_type}
# Record the frame with its OOPIF session id for interaction routing.
if target_type == "iframe":
target_id = info.get("targetId")
with self._state_lock:
existing = self._frames.get(target_id)
self._frames[target_id] = FrameInfo(
frame_id=target_id,
url=str(info.get("url") or ""),
origin="", # filled by frameNavigated on the child session
parent_frame_id=(existing.parent_frame_id if existing else None),
is_oopif=True,
cdp_session_id=sid,
name=str(info.get("title") or (existing.name if existing else "")),
)
# Enable child domains off-loop: awaiting the replies here would deadlock
# because only the reader can resolve those Futures.
asyncio.create_task(self._enable_child_domains(sid))
async def _enable_child_domains(self, sid: str) -> None:
"""Enable Page+Runtime (+nested setAutoAttach) and the dialog bridge on a child session."""
try:
await self._cdp("Page.enable", session_id=sid, timeout=3.0)
await self._cdp("Runtime.enable", session_id=sid, timeout=3.0)
await self._cdp(
"Target.setAutoAttach",
{"autoAttach": True, "waitForDebuggerOnStart": False, "flatten": True},
session_id=sid,
timeout=3.0,
)
except Exception as e:
logger.debug("child session %s setup failed: %s", sid[:16], e)
await self._install_dialog_bridge(sid)
def _on_target_detached(self, params: Dict[str, Any], session_id: Optional[str] = None) -> None:
"""Clear the session binding of frames on a detached child session.
Frames are deliberately NOT dropped: Browserbase fires transient detaches
during page transitions while the iframe is still visible, and dropping
would hide OOPIFs until the next ``Target.attachedToTarget``. Clearing
``cdp_session_id`` just stops stale routing; ``Page.frameDetached``
cleans up if the iframe truly goes away.
"""
sid = params.get("sessionId")
if not sid:
return
self._child_sessions.pop(sid, None)
with self._state_lock:
for fid, frame in list(self._frames.items()):
if frame.cdp_session_id == sid:
self._frames[fid] = replace(frame, cdp_session_id=None)
# ── Console / exception ring buffer ─────────────────────────────────────
def _on_console(self, params: Dict[str, Any], *, level_from: str) -> None:
@@ -1201,69 +623,17 @@ class CDPSupervisor:
event = ConsoleEvent(ts=time.time(), level=level, text=" ".join(parts))
with self._state_lock:
self._console_events.append(event)
if len(self._console_events) > CONSOLE_HISTORY_MAX * 2:
# Keep last CONSOLE_HISTORY_MAX; 2x slack reduces churn.
self._console_events = self._console_events[-CONSOLE_HISTORY_MAX:]
self._console_events = _trim_ring(self._console_events, CONSOLE_HISTORY_MAX)
# CDP event → handler(self, params, session_id). Async handlers return an
# awaitable that ``_on_event`` awaits; sync handlers return None.
_EVENT_HANDLERS: Dict[str, Callable[..., Any]] = {
"Page.javascriptDialogOpening": _on_dialog_opening,
"Page.javascriptDialogClosed": _on_dialog_closed,
"Fetch.requestPaused": _on_fetch_paused,
"Page.frameAttached": _on_frame_attached,
"Page.frameNavigated": _on_frame_navigated,
"Page.frameDetached": _on_frame_detached,
"Target.attachedToTarget": _on_target_attached,
"Target.detachedFromTarget": _on_target_detached,
**DialogSupervisionMixin.EVENT_HANDLERS,
**FrameTrackingMixin.EVENT_HANDLERS,
"Runtime.consoleAPICalled": lambda self, p, _sid: self._on_console(p, level_from="api"),
"Runtime.exceptionThrown": lambda self, p, _sid: self._on_console(p, level_from="exception"),
}
# ── Frame tree building (bounded) ───────────────────────────────────────
def _build_frame_tree_locked(self) -> Dict[str, Any]:
"""Build the capped frame_tree payload. Must be called under state lock."""
frames = self._frames
empty = {"top": None, "children": [], "truncated": False}
if not frames:
return empty
# Top frame: one with no parent, preferring oopif=False.
tops = [f for f in frames.values() if not f.parent_frame_id]
top = next((f for f in tops if not f.is_oopif), tops[0] if tops else None)
if top is None:
return empty
# BFS from top, capped by FRAME_TREE_MAX_ENTRIES and
# FRAME_TREE_MAX_OOPIF_DEPTH for OOPIF branches.
children: List[Dict[str, Any]] = []
truncated = False
queue: List[Tuple[FrameInfo, int]] = [
(f, 1) for f in frames.values() if f.parent_frame_id == top.frame_id
]
visited: set[str] = {top.frame_id}
while queue and len(children) < FRAME_TREE_MAX_ENTRIES:
frame, depth = queue.pop(0)
if frame.frame_id in visited:
continue
visited.add(frame.frame_id)
if frame.is_oopif and depth > FRAME_TREE_MAX_OOPIF_DEPTH:
truncated = True
continue
children.append(frame.to_dict())
for f in frames.values():
if f.parent_frame_id == frame.frame_id and f.frame_id not in visited:
queue.append((f, depth + 1))
if queue:
truncated = True
return {
"top": top.to_dict(),
"children": children,
"truncated": truncated,
}
# ── Registry ─────────────────────────────────────────────────────────────────