diff --git a/pyproject.toml b/pyproject.toml index f812a6ab..2540434d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "browser-harness" -version = "0.1.0" +version = "0.2.0" description = "The simplest, thinnest, and most powerful harness to control your real browser with your agent." requires-python = ">=3.11" dependencies = [ diff --git a/src/browser_harness/daemon.py b/src/browser_harness/daemon.py index 01edad1c..481b6b16 100644 --- a/src/browser_harness/daemon.py +++ b/src/browser_harness/daemon.py @@ -1,10 +1,10 @@ -"""CDP WS holder + IPC relay (Unix socket on POSIX, TCP loopback on Windows). One daemon per BU_NAME.""" -import asyncio, json, os, socket, sys, time, urllib.error, urllib.request +"""Browser protocol holder + IPC relay. One daemon per BU_NAME.""" +import asyncio, copy, json, os, socket, sys, time, urllib.error, urllib.request, uuid from collections import deque from pathlib import Path from . import _ipc as ipc -from cdp_use.client import CDPClient +from .transport import transport_from_environment def _load_env(): @@ -32,6 +32,51 @@ def _load_env_file(p): LOG = str(ipc.log_path(NAME)) PID = str(ipc.pid_path(NAME)) BUF = 500 +HEALTH_CAPABILITY = "health_events_v2" +HEALTH_SCHEMA_VERSION = 2 +HEALTH_OBSERVATION_CONTROL_METHODS = { + "Page.enable", + "Page.disable", + "Runtime.enable", + "Runtime.disable", + "Log.enable", + "Log.disable", + "Network.enable", + "Network.disable", + "Console.enable", + "Console.disable", + "Target.setAutoAttach", + "Target.autoAttachRelated", +} +HEALTH_EVENT_SCHEMA_VERSION = 1 +HEALTH_SESSION_DOMAINS = ("Page", "Runtime", "Log", "Network") +HEALTH_EVENT_MAX_COUNT = int(os.environ.get("BH_HEALTH_EVENT_MAX_COUNT", BUF)) +HEALTH_EVENT_MAX_BYTES = int(os.environ.get("BH_HEALTH_EVENT_MAX_BYTES", 8 * 1024 * 1024)) +HEALTH_EVENT_MAX_ITEM_BYTES = int(os.environ.get("BH_HEALTH_EVENT_MAX_ITEM_BYTES", 256 * 1024)) +_BODY_KEYS = frozenset(("body", "postData", "postDataEntries", "payloadData")) +_NETWORK_DATA_PAYLOAD_METHODS = frozenset(( + "Network.dataReceived", + "Network.eventSourceMessageReceived", + "Network.directTCPSocketChunkReceived", + "Network.directTCPSocketChunkSent", + "Network.directUDPSocketChunkReceived", + "Network.directUDPSocketChunkSent", +)) +_HEALTH_CDP_METHODS = frozenset(( + "Runtime.exceptionThrown", + "Runtime.consoleAPICalled", + "Console.messageAdded", + "Log.entryAdded", + "Inspector.targetCrashed", + "Target.targetCrashed", + "Security.certificateError", + "Network.requestWillBeSent", + "Network.responseReceived", + "Network.loadingFailed", + "Target.attachedToTarget", + "Target.detachedFromTarget", + "Target.targetDestroyed", +)) PROFILES = [ Path.home() / "Library/Application Support/Google/Chrome", Path.home() / "Library/Application Support/Comet", @@ -148,16 +193,740 @@ def is_real_page(t): return t["type"] == "page" and not t.get("url", "").startswith(INTERNAL) +def _without_bodies(value, strip_root_data=False): + if isinstance(value, dict): + return { + key: _without_bodies(child) + for key, child in value.items() + if key not in _BODY_KEYS and not (strip_root_data and key == "data") + } + if isinstance(value, list): + return [_without_bodies(child) for child in value] + return value + + +def _bounded_text(value, limit=4096): + return str(value)[:limit] + + +def _project_call_frames(stack_trace): + if not isinstance(stack_trace, dict): + return None + frames = [] + for frame in stack_trace.get("callFrames", [])[:16]: + if not isinstance(frame, dict): + continue + frames.append({ + "functionName": _bounded_text(frame.get("functionName", ""), 256), + "url": _bounded_text(frame.get("url", ""), 2048), + "lineNumber": frame.get("lineNumber"), + "columnNumber": frame.get("columnNumber"), + }) + return {"callFrames": frames} + + +def _project_remote_object(value): + if not isinstance(value, dict): + return {} + projected = { + key: value[key] + for key in ("type", "subtype") + if key in value + } + primitive = value.get("value") + if isinstance(primitive, (str, int, float, bool)) or primitive is None: + if "value" in value: + projected["value"] = ( + _bounded_text(primitive) + if isinstance(primitive, str) + else primitive + ) + for key in ("description", "unserializableValue"): + if key in value: + projected[key] = _bounded_text(value[key]) + return projected + + +def _project_health_params(method, params): + if method == "Runtime.exceptionThrown": + details = params.get("exceptionDetails") or {} + exception = details.get("exception") or {} + projected_details = { + key: details[key] + for key in ("lineNumber", "columnNumber") + if key in details + } + for key in ("text", "url"): + if key in details: + projected_details[key] = _bounded_text(details[key]) + if exception.get("description") is not None: + projected_details["exception"] = { + "description": _bounded_text(exception["description"]), + } + stack = _project_call_frames(details.get("stackTrace")) + if stack is not None: + projected_details["stackTrace"] = stack + return {"exceptionDetails": projected_details} + if method == "Runtime.consoleAPICalled": + projected = { + "type": params.get("type"), + "args": [ + _project_remote_object(argument) + for argument in params.get("args", [])[:32] + ], + } + stack = _project_call_frames(params.get("stackTrace")) + if stack is not None: + projected["stackTrace"] = stack + return projected + if method == "Console.messageAdded": + message = params.get("message") or {} + projected_message = { + key: message[key] + for key in ("level", "line", "column") + if key in message + } + for key in ("source", "text", "url"): + if key in message: + projected_message[key] = _bounded_text(message[key]) + stack = _project_call_frames(message.get("stack") or message.get("stackTrace")) + if stack is not None: + projected_message["stack"] = stack + return {"message": projected_message} + if method == "Log.entryAdded": + entry = params.get("entry") or {} + projected_entry = { + key: entry[key] + for key in ("level", "lineNumber") + if key in entry + } + for key in ("source", "text", "url"): + if key in entry: + projected_entry[key] = _bounded_text(entry[key]) + stack = _project_call_frames(entry.get("stackTrace")) + if stack is not None: + projected_entry["stackTrace"] = stack + return {"entry": projected_entry} + if method in ("Inspector.targetCrashed",): + return {} + if method == "Target.targetCrashed": + return { + key: params[key] + for key in ("targetId", "status", "errorCode") + if key in params + } + if method == "Security.certificateError": + return { + key: _bounded_text(params[key]) + for key in ("eventId", "errorType", "requestURL") + if key in params + } + if method == "Network.requestWillBeSent": + request = params.get("request") or {} + return { + **{ + key: params[key] + for key in ("requestId", "type") + if key in params + }, + "request": { + key: _bounded_text(request[key]) + for key in ("method", "url") + if key in request + }, + } + if method == "Network.responseReceived": + response = params.get("response") or {} + projected_response = { + key: response[key] + for key in ("status",) + if key in response + } + for key in ("url", "statusText"): + if key in response: + projected_response[key] = _bounded_text(response[key]) + return { + **{ + key: params[key] + for key in ("requestId", "type") + if key in params + }, + "response": projected_response, + } + if method == "Network.loadingFailed": + projected = { + key: params[key] + for key in ("requestId", "type", "canceled", "blockedReason") + if key in params + } + if "errorText" in params: + projected["errorText"] = _bounded_text(params["errorText"]) + cors = params.get("corsErrorStatus") + if isinstance(cors, dict): + projected["corsErrorStatus"] = { + key: _bounded_text(cors[key]) + for key in ("corsError", "failedParameter") + if key in cors + } + return projected + if method == "Target.attachedToTarget": + target = params.get("targetInfo") or {} + return { + **{ + key: params[key] + for key in ("sessionId", "waitingForDebugger") + if key in params + }, + "targetInfo": { + key: _bounded_text(target[key]) + for key in ("targetId", "type", "url", "parentId") + if key in target + }, + } + if method == "Target.detachedFromTarget": + return { + key: _bounded_text(params[key]) + for key in ("sessionId", "targetId") + if key in params + } + if method == "Target.targetDestroyed": + return ( + {"targetId": _bounded_text(params["targetId"])} + if "targetId" in params + else {} + ) + return {} + + +class _HealthEventRing: + def __init__(self, max_count, max_bytes, max_item_bytes): + if min(max_count, max_bytes, max_item_bytes) <= 0: + raise ValueError("health event retention limits must be positive") + self.max_count = max_count + self.max_bytes = max_bytes + self.max_item_bytes = max_item_bytes + self.sequence = 0 + self.total_bytes = 0 + self.events = deque() + self.lost_ranges = deque() + + def _record_loss(self, sequence): + ordered = sorted((*self.lost_ranges, (sequence, sequence))) + merged = [] + for start, end in ordered: + if merged and start <= merged[-1][1] + 1: + previous_start, previous_end = merged[-1] + merged[-1] = (previous_start, max(previous_end, end)) + else: + merged.append((start, end)) + while len(merged) > self.max_count: + first, second = merged[:2] + merged[:2] = [(first[0], second[1])] + self.lost_ranges = deque(merged) + + def append(self, event): + self.sequence += 1 + retained = {"sequence": self.sequence, **event} + encoded_size = len( + json.dumps(retained, separators=(",", ":"), sort_keys=True).encode("utf-8") + ) + if encoded_size > self.max_item_bytes or encoded_size > self.max_bytes: + self._record_loss(self.sequence) + return self.sequence + while self.events and ( + len(self.events) >= self.max_count + or self.total_bytes + encoded_size > self.max_bytes + ): + evicted, size = self.events.popleft() + self.total_bytes -= size + self._record_loss(evicted["sequence"]) + self.events.append((retained, encoded_size)) + self.total_bytes += encoded_size + return self.sequence + + def read_after(self, after_sequence, through_sequence=None): + upper = self.sequence + if through_sequence is not None: + upper = min(upper, through_sequence) + events = [ + copy.deepcopy(event) + for event, _ in self.events + if after_sequence < event["sequence"] <= upper + ] + intersecting_losses = [ + (start, end) + for start, end in self.lost_ranges + if end > after_sequence and start <= upper + ] + available_from = self.events[0][0]["sequence"] if self.events else self.sequence + 1 + overflow = None + if intersecting_losses: + overflow = { + "kind": "retention_or_truncation", + "requested_after_sequence": after_sequence, + "lost_through_sequence": max(end for _, end in intersecting_losses), + "available_from_sequence": available_from, + } + returned_through = events[-1]["sequence"] if events else min(after_sequence, upper) + return { + "events": events, + "range": { + "requested_after_sequence": after_sequence, + "available_from_sequence": available_from, + "current_sequence": self.sequence, + "returned_through_sequence": returned_through, + "complete": overflow is None, + }, + "overflow": overflow, + } + + class Daemon: - def __init__(self): + def __init__( + self, + event_max_count=HEALTH_EVENT_MAX_COUNT, + event_max_bytes=HEALTH_EVENT_MAX_BYTES, + event_max_item_bytes=HEALTH_EVENT_MAX_ITEM_BYTES, + ): self.cdp = None self.session = None self.target_id = None - self.events = deque(maxlen=BUF) + self.daemon_fingerprint = uuid.uuid4().hex + self.target_epoch = 0 + self.session_epoch = 0 + self.attachment_generation = 0 + self.attachment_transition = False + self.pending_session_handoff = None + self.prepared_auto_session = None + self.superseded_health_sessions = set() + self.background_tasks = set() + self.subscriptions = {} + self.health_events = _HealthEventRing( + event_max_count, + event_max_bytes, + event_max_item_bytes, + ) + self.compatibility_events = _HealthEventRing( + event_max_count, + event_max_bytes, + event_max_item_bytes, + ) + self.compatibility_cursor = 0 + self.health_attempts = {} self.dialog = None self.stop = None # asyncio.Event, set inside start() - async def attach_first_page(self): + def _observation(self): + subscriptions = copy.deepcopy(self.subscriptions) + required = ("Target", *HEALTH_SESSION_DOMAINS) + ready = not self.attachment_transition and bool(self.target_id and self.session) and all( + subscriptions.get(domain, {}).get("enabled") for domain in required + ) + return { + "ready": ready, + "target_id": self.target_id, + "session_id": self.session, + "target_epoch": self.target_epoch, + "session_epoch": self.session_epoch, + "subscriptions": subscriptions, + } + + def _capabilities(self): + ring = self.health_events + capabilities = { + HEALTH_CAPABILITY: { + "schema_version": HEALTH_SCHEMA_VERSION, + "event_schema_version": 2, + "operations": ["begin", "events_since", "seal"], + "sequence_origin": 1, + "continuity_proofs": ["same_target_paused_session_handoff_v1"], + "retention": { + "max_events": ring.max_count, + "max_total_bytes": ring.max_bytes, + "max_event_bytes": ring.max_item_bytes, + }, + "identity": ["engine", "protocol", "browser_session"], + } + } + if self.cdp is not None and callable(getattr(self.cdp, "capabilities", None)): + capabilities["browser_transport_v1"] = { + "schema_version": 1, + **self.cdp.capabilities(), + } + return { + "daemon_fingerprint": self.daemon_fingerprint, + "capabilities": capabilities, + "observation": self._observation(), + } + + def _event_record(self, method, params): + kind = { + "Runtime.exceptionThrown": "script-exception", + "Runtime.consoleAPICalled": "console", + "Console.messageAdded": "console", + "Log.entryAdded": "console", + "Inspector.targetCrashed": "context-crash", + "Target.targetCrashed": "context-crash", + "Security.certificateError": "certificate-error", + "Network.requestWillBeSent": "network-request", + "Network.responseReceived": "network-response", + "Network.loadingFailed": "network-failure", + "Target.targetDestroyed": "context-destroyed", + }.get(method, "lifecycle" if method.startswith("BrowserHarness.") else "protocol-event") + return { + "method": method, + "params": params, + "kind": kind, + "engine": getattr(self.cdp, "engine", None), + "source_protocol": getattr(self.cdp, "protocol", None), + "browser_session_id": getattr(self.cdp, "session_id", None) or self.session, + "session_id": self.session, + "daemon_fingerprint": self.daemon_fingerprint, + "target_id": self.target_id, + "target_epoch": self.target_epoch, + "session_epoch": self.session_epoch, + } + + def _record_health_event(self, method, params, _transport_session_id): + if not method.startswith("BrowserHarness.") and method not in _HEALTH_CDP_METHODS: + return None + if ( + not method.startswith("BrowserHarness.") + and _transport_session_id in self.superseded_health_sessions + ): + return None + projected = ( + params + if method.startswith("BrowserHarness.") + else _project_health_params(method, params) + ) + return self.health_events.append(self._event_record(method, projected)) + + async def _enable_target_observation(self, target_id): + enabled = True + operations = ( + ("Target.setDiscoverTargets", {"discover": True}), + ( + "Target.setAutoAttach", + { + "autoAttach": True, + "waitForDebuggerOnStart": True, + "flatten": True, + "filter": [ + {"type": "page", "exclude": False}, + {"exclude": True}, + ], + }, + ), + ) + for method, params in operations: + try: + await asyncio.wait_for( + self.cdp.send_raw(method, params, session_id=None), + timeout=5, + ) + except Exception as exc: + enabled = False + log(f"enable Target observation ({method}): {exc}") + return { + "enabled": enabled, + "scope": "browser", + "session_id": None, + "auto_attach": "browser_pages", + "wait_for_debugger_on_start": True, + } + + async def _enable_session_observation(self, session_id): + proof = {} + for domain in HEALTH_SESSION_DOMAINS: + try: + await asyncio.wait_for( + self.cdp.send_raw( + f"{domain}.enable", + session_id=session_id, + ), + timeout=5, + ) + enabled = True + except Exception as exc: + enabled = False + log(f"enable {domain}: {exc}") + proof[domain] = { + "enabled": enabled, + "scope": "session", + "session_id": session_id, + } + try: + await asyncio.wait_for( + self.cdp.send_raw("DOM.enable", session_id=session_id), + timeout=5, + ) + except Exception as exc: + log(f"enable DOM: {exc}") + return proof + + async def _enable_observation(self): + return { + "Target": await self._enable_target_observation(self.target_id), + **await self._enable_session_observation(self.session), + } + + async def _set_attachment(self, target_id, session_id, reason): + self.attachment_generation += 1 + generation = self.attachment_generation + previous = { + "target_id": self.target_id, + "session_id": self.session, + "target_epoch": self.target_epoch, + "session_epoch": self.session_epoch, + } + if target_id != self.target_id: + self.pending_session_handoff = None + self.prepared_auto_session = None + self.superseded_health_sessions.clear() + if target_id != self.target_id: + self.target_epoch += 1 + if session_id != self.session: + self.session_epoch += 1 + self.target_id = target_id + self.session = session_id + self.attachment_transition = True + self.subscriptions = {} + proof = await self._enable_observation() + if ( + generation != self.attachment_generation + or target_id != self.target_id + or session_id != self.session + ): + return False + self.subscriptions = proof + self.attachment_transition = False + prepared = self.prepared_auto_session + if ( + prepared + and prepared["target_id"] == target_id + and previous["target_id"] == target_id + and previous["session_id"] + ): + return self._adopt_prepared_overlap(prepared, previous) + previous_session_id = previous["session_id"] + if self.pending_session_handoff: + previous_session_id = self.pending_session_handoff[ + "previous_session_id" + ] + if previous_session_id and previous_session_id != session_id: + self.superseded_health_sessions.add(previous_session_id) + self.pending_session_handoff = None + self._record_health_event( + "BrowserHarness.attachmentChanged", + { + "reason": reason, + "previous": previous, + "current": self._observation(), + }, + session_id, + ) + return True + + def _invalidate_attachment(self, reason, clear_target): + self.attachment_generation += 1 + self.pending_session_handoff = None + self.prepared_auto_session = None + if clear_target: + self.superseded_health_sessions.clear() + previous = self._observation() + if self.session is not None: + self.session_epoch += 1 + self.session = None + if clear_target and self.target_id is not None: + self.target_epoch += 1 + self.target_id = None + self.attachment_transition = False + self.subscriptions = {} + self._record_health_event( + "BrowserHarness.attachmentInvalidated", + { + "reason": reason, + "previous": previous, + "current": self._observation(), + }, + None, + ) + + def _begin_session_handoff(self): + self.attachment_generation += 1 + self.pending_session_handoff = { + "target_id": self.target_id, + "target_epoch": self.target_epoch, + "previous_session_id": self.session, + "previous_session_epoch": self.session_epoch, + } + self.session = None + self.attachment_transition = True + self.subscriptions = {} + prepared = self.prepared_auto_session + if prepared and prepared["target_id"] == self.target_id: + self._adopt_prepared_auto_session(prepared) + + def _adopt_prepared_auto_session(self, prepared): + pending = self.pending_session_handoff + if not pending or prepared["target_id"] != pending["target_id"]: + return False + previous_session_id = pending["previous_session_id"] + if previous_session_id and previous_session_id != prepared["session_id"]: + self.superseded_health_sessions.add(previous_session_id) + self.attachment_generation += 1 + self.target_id = pending["target_id"] + self.target_epoch = pending["target_epoch"] + self.session_epoch = pending["previous_session_epoch"] + 1 + self.session = prepared["session_id"] + self.subscriptions = copy.deepcopy(prepared["subscriptions"]) + self.attachment_transition = False + self.pending_session_handoff = None + self.prepared_auto_session = None + self._record_health_event( + "BrowserHarness.sameTargetSessionHandoff", + { + "proof": "same_target_paused_session_handoff_v1", + "target_id": self.target_id, + "previous_session_id": pending["previous_session_id"], + "previous_session_epoch": pending["previous_session_epoch"], + "session_id": self.session, + "session_epoch": self.session_epoch, + "waiting_for_debugger_on_start": True, + "required_domains": ["Target", *HEALTH_SESSION_DOMAINS], + "subscriptions_before_resume": True, + "resume_acknowledged": True, + }, + self.session, + ) + return True + + def _adopt_prepared_overlap(self, prepared, previous): + if ( + prepared["target_id"] != previous["target_id"] + or not previous["session_id"] + ): + return False + self.pending_session_handoff = { + "target_id": previous["target_id"], + "target_epoch": previous["target_epoch"], + "previous_session_id": previous["session_id"], + "previous_session_epoch": previous["session_epoch"], + } + return self._adopt_prepared_auto_session(prepared) + + async def _resume_auto_attached_session(self, session_id): + try: + await asyncio.wait_for( + self.cdp.send_raw( + "Runtime.runIfWaitingForDebugger", + session_id=session_id, + ), + timeout=5, + ) + return True + except Exception as exc: + log(f"resume auto-attached session {session_id}: {exc}") + return False + + async def _prepare_auto_attached_session(self, params): + session_id = params.get("sessionId") + target_info = params.get("targetInfo") or {} + target_id = target_info.get("targetId") + waiting = params.get("waitingForDebugger") is True + if not session_id or not waiting: + return False + if target_id != self.target_id or target_info.get("type") != "page": + await self._resume_auto_attached_session(session_id) + return False + session_proof = await self._enable_session_observation(session_id) + subscriptions_ready = all( + session_proof.get(domain, {}).get("enabled") + for domain in HEALTH_SESSION_DOMAINS + ) + resumed = await self._resume_auto_attached_session(session_id) + if ( + not subscriptions_ready + or not resumed + or target_id != self.target_id + ): + return False + prepared = { + "target_id": target_id, + "session_id": session_id, + "subscriptions": { + "Target": { + "enabled": True, + "scope": "browser", + "session_id": None, + "auto_attach": "browser_pages", + "wait_for_debugger_on_start": True, + }, + **session_proof, + }, + } + self.prepared_auto_session = prepared + if self.pending_session_handoff: + return self._adopt_prepared_auto_session(prepared) + observation = self._observation() + if observation["ready"]: + return self._adopt_prepared_overlap( + prepared, + { + "target_id": observation["target_id"], + "session_id": observation["session_id"], + "target_epoch": observation["target_epoch"], + "session_epoch": observation["session_epoch"], + }, + ) + return True + + def _record_cdp_event(self, method, params, session_id): + compatibility_params = _without_bodies( + params, + strip_root_data=method in _NETWORK_DATA_PAYLOAD_METHODS, + ) + self.compatibility_events.append( + self._event_record(method, compatibility_params) + ) + sequence = self._record_health_event(method, params, session_id) + if method == "Target.detachedFromTarget": + detached_session = params.get("sessionId") + detached_target = params.get("targetId") + prepared = self.prepared_auto_session + if prepared and detached_session == prepared["session_id"]: + self.prepared_auto_session = None + elif ( + detached_session == self.session + if detached_session is not None + else detached_target == self.target_id + ): + self._begin_session_handoff() + if detached_session: + self.superseded_health_sessions.discard(detached_session) + elif method in ("Target.targetDestroyed", "Target.targetCrashed"): + if params.get("targetId") == self.target_id: + self._invalidate_attachment( + "target_destroyed" if method == "Target.targetDestroyed" else "target_crashed", + clear_target=True, + ) + return sequence + + def _spawn_background(self, coroutine): + task = asyncio.create_task(_silent(coroutine)) + self.background_tasks.add(task) + task.add_done_callback(self.background_tasks.discard) + return task + + async def _settle_background_tasks(self): + while self.background_tasks: + pending = tuple(self.background_tasks) + await asyncio.gather(*pending, return_exceptions=True) + for task in pending: + self.background_tasks.discard(task) + + async def attach_first_page(self, reason="attach_first_page"): """Attach to a real page (or any page). Sets self.session. Returns attached target or None.""" targets = (await self.cdp.send_raw("Target.getTargets"))["targetInfos"] pages = [t for t in targets if is_real_page(t)] @@ -166,29 +935,28 @@ async def attach_first_page(self): tid = (await self.cdp.send_raw("Target.createTarget", {"url": "about:blank"}))["targetId"] log(f"no real pages found, created about:blank ({tid})") pages = [{"targetId": tid, "url": "about:blank", "type": "page"}] - self.session = (await self.cdp.send_raw( + session_id = (await self.cdp.send_raw( "Target.attachToTarget", {"targetId": pages[0]["targetId"], "flatten": True} ))["sessionId"] - self.target_id = pages[0]["targetId"] + established = await self._set_attachment(pages[0]["targetId"], session_id, reason) + if not established: + return None log(f"attached {pages[0]['targetId']} ({pages[0].get('url','')[:80]}) session={self.session}") - for d in ("Page", "DOM", "Runtime", "Network"): - try: - await asyncio.wait_for( - self.cdp.send_raw(f"{d}.enable", session_id=self.session), - timeout=5 - ) - except Exception as e: - log(f"enable {d}: {e}") return pages[0] async def start(self): self.stop = asyncio.Event() - url = get_ws_url() - log(f"connecting to {url}") - self.cdp = CDPClient(url) + protocol = os.environ.get("BU_BROWSER_PROTOCOL", "cdp").strip().lower() + endpoint = None if protocol == "bidi" else get_ws_url() + self.cdp = transport_from_environment(cdp_endpoint=endpoint) + log(f"connecting to {self.cdp.endpoint} ({self.cdp.engine}/{self.cdp.protocol})") try: await self.cdp.start() except Exception as e: + if protocol == "bidi": + raise RuntimeError( + f"WebDriver BiDi connection failed: {e} -- verify the dedicated Firefox lane and endpoint" + ) if os.environ.get("BU_CDP_WS"): raise RuntimeError( f"CDP WS handshake failed: {e} -- remote browser WebSocket connection failed. " @@ -196,11 +964,12 @@ async def start(self): "If you use Browser Use cloud, verify BROWSER_USE_API_KEY and get a fresh URL via start_remote_daemon()." ) raise RuntimeError(f"CDP WS handshake failed: {e} -- click Allow in Chrome if prompted, then retry") - await self.attach_first_page() orig = self.cdp._event_registry.handle_event mark_js = "if(!document.title.startsWith('\U0001F7E2'))document.title='\U0001F7E2 '+document.title" async def tap(method, params, session_id=None): - self.events.append({"method": method, "params": params, "session_id": session_id}) + self._record_cdp_event(method, params, session_id) + if method == "Target.attachedToTarget": + self._spawn_background(self._prepare_auto_attached_session(params)) if method == "Page.javascriptDialogOpening": self.dialog = params elif method == "Page.javascriptDialogClosed": @@ -209,6 +978,14 @@ async def tap(method, params, session_id=None): asyncio.create_task(_silent(asyncio.wait_for(self.cdp.send_raw("Runtime.evaluate", {"expression": mark_js}, session_id=self.session), timeout=2))) return await orig(method, params, session_id) self.cdp._event_registry.handle_event = tap + await self.attach_first_page(reason="initial_attach") + await self._settle_background_tasks() + + async def close(self): + await self._settle_background_tasks() + if self.cdp is not None: + await self.cdp.close() + self.cdp = None async def handle(self, req): # Token guard for Windows TCP loopback: any local process can otherwise @@ -222,14 +999,101 @@ async def handle(self, req): # daemon and not an unrelated process that reused our port post-crash. if meta == "ping": return {"pong": True} if meta == "drain_events": - out = list(self.events); self.events.clear() - return {"events": out} + result = self.compatibility_events.read_after(self.compatibility_cursor) + self.compatibility_cursor = self.compatibility_events.sequence + return {"events": result["events"], "overflow": result["overflow"]} + if meta == "health_capabilities": + return self._capabilities() + if meta == "health_begin": + requested_capability = req.get("capability") or HEALTH_CAPABILITY + if requested_capability != HEALTH_CAPABILITY: + return {"error": "unsupported_health_capability"} + schema_version = HEALTH_SCHEMA_VERSION + attempt_id = req.get("attempt_id") + if not isinstance(attempt_id, str) or not attempt_id or len(attempt_id) > 256: + return {"error": "invalid_health_attempt_id"} + existing = self.health_attempts.get(attempt_id) + if existing: + return copy.deepcopy(existing["begin"]) + await self._settle_background_tasks() + begin = { + "capability": requested_capability, + "schema_version": schema_version, + "attempt_id": attempt_id, + "daemon_fingerprint": self.daemon_fingerprint, + "start_sequence": self.health_events.sequence, + "sealed_through_sequence": None, + "observation": self._observation(), + } + self.health_attempts[attempt_id] = {"begin": begin, "seal": None} + return copy.deepcopy(begin) + if meta == "health_events_since": + attempt_id = req.get("attempt_id") + attempt = self.health_attempts.get(attempt_id) + if not attempt: + return {"error": "unknown_health_attempt"} + after_sequence = req.get("after_sequence") + if ( + not isinstance(after_sequence, int) + or isinstance(after_sequence, bool) + or after_sequence < attempt["begin"]["start_sequence"] + ): + return {"error": "invalid_health_sequence"} + seal = attempt["seal"] + through_sequence = ( + seal["sealed_through_sequence"] + if seal + else req.get("through_sequence") + ) + if ( + through_sequence is not None + and ( + not isinstance(through_sequence, int) + or isinstance(through_sequence, bool) + or through_sequence < after_sequence + ) + ): + return {"error": "invalid_health_sequence"} + result = self.health_events.read_after(after_sequence, through_sequence) + result["range"]["sealed_through_sequence"] = ( + seal["sealed_through_sequence"] if seal else None + ) + return { + "capability": attempt["begin"]["capability"], + "schema_version": attempt["begin"]["schema_version"], + "attempt_id": attempt_id, + "daemon_fingerprint": self.daemon_fingerprint, + **result, + } + if meta == "health_seal": + attempt_id = req.get("attempt_id") + attempt = self.health_attempts.get(attempt_id) + if not attempt: + return {"error": "unknown_health_attempt"} + if attempt["seal"]: + return copy.deepcopy(attempt["seal"]) + seal = { + "capability": attempt["begin"]["capability"], + "schema_version": attempt["begin"]["schema_version"], + "attempt_id": attempt_id, + "daemon_fingerprint": self.daemon_fingerprint, + "start_sequence": attempt["begin"]["start_sequence"], + "sealed_through_sequence": self.health_events.sequence, + "observation": self._observation(), + } + attempt["seal"] = seal + return copy.deepcopy(seal) if meta == "session": return {"session_id": self.session} if meta == "connection_status": if not self.target_id: return {"error": "not_attached"} try: - info = (await self.cdp.send_raw("Target.getTargetInfo", {"targetId": self.target_id}))["targetInfo"] + targets = (await self.cdp.send_raw("Target.getTargets"))["targetInfos"] + info = next( + target + for target in targets + if target.get("targetId") == self.target_id + ) except Exception: return {"error": "cdp_disconnected"} page = None @@ -241,10 +1105,10 @@ async def handle(self, req): } return {"target_id": self.target_id, "session_id": self.session, "page": page} if meta == "set_session": - self.session = req.get("session_id") - self.target_id = req.get("target_id") or self.target_id + session_id = req.get("session_id") + target_id = req.get("target_id") or self.target_id try: - await asyncio.wait_for(self.cdp.send_raw("Page.enable", session_id=self.session), timeout=3) + await self._set_attachment(target_id, session_id, "set_session") await asyncio.wait_for(self.cdp.send_raw("Runtime.evaluate", {"expression": "if(!document.title.startsWith('\U0001F7E2'))document.title='\U0001F7E2 '+document.title"}, session_id=self.session), timeout=2) except Exception: pass return {"session_id": self.session} @@ -253,6 +1117,11 @@ async def handle(self, req): method = req["method"] params = req.get("params") or {} + if ( + method in HEALTH_OBSERVATION_CONTROL_METHODS + and any(attempt["seal"] is None for attempt in self.health_attempts.values()) + ): + return {"error": "health_observation_control_is_guarded"} # Browser-level Target.* calls must not use a session (stale or otherwise). # For everything else, explicit session in req wins; else default. sid = None if method.startswith("Target.") else (req.get("session_id") or self.session) @@ -260,10 +1129,57 @@ async def handle(self, req): return {"result": await self.cdp.send_raw(method, params, session_id=sid)} except Exception as e: msg = str(e) - if "Session with given id not found" in msg and sid == self.session and sid: - log(f"stale session {sid}, re-attaching") - if await self.attach_first_page(): - return {"result": await self.cdp.send_raw(method, params, session_id=self.session)} + if "Session with given id not found" in msg and sid: + if self.session != sid and self._observation()["ready"]: + log(f"session handoff replaced stale session {sid} with {self.session}") + return { + "result": await self.cdp.send_raw( + method, + params, + session_id=self.session, + ) + } + if sid == self.session: + await self._settle_background_tasks() + if self.session != sid and self._observation()["ready"]: + log( + f"paused auto-attach replaced stale session {sid} " + f"with {self.session}" + ) + return { + "result": await self.cdp.send_raw( + method, + params, + session_id=self.session, + ) + } + prepared = self.prepared_auto_session + observation = self._observation() + if ( + prepared + and prepared["target_id"] == observation["target_id"] + and observation["ready"] + and self._adopt_prepared_overlap( + prepared, + { + "target_id": observation["target_id"], + "session_id": observation["session_id"], + "target_epoch": observation["target_epoch"], + "session_epoch": observation["session_epoch"], + }, + ) + ): + log(f"prepared overlap replaced stale session {sid} with {self.session}") + return { + "result": await self.cdp.send_raw( + method, + params, + session_id=self.session, + ) + } + log(f"stale session {sid}, re-attaching") + if await self.attach_first_page(): + return {"result": await self.cdp.send_raw(method, params, session_id=self.session)} return {"error": msg} @@ -297,6 +1213,7 @@ async def handler(reader, writer): t.cancel() try: await t except (asyncio.CancelledError, Exception): pass + await d.close() ipc.cleanup_endpoint(NAME) diff --git a/src/browser_harness/helpers.py b/src/browser_harness/helpers.py index aa897da9..aa46c009 100644 --- a/src/browser_harness/helpers.py +++ b/src/browser_harness/helpers.py @@ -1,4 +1,4 @@ -"""Browser control via CDP. +"""Browser control through the persistent protocol-neutral session. Core helpers live here. Agent-editable helpers live in BH_AGENT_WORKSPACE/agent_helpers.py. @@ -57,6 +57,55 @@ def cdp(method, session_id=None, **params): def drain_events(): return _send({"meta": "drain_events"})["events"] +def health_capabilities(): + return _send({"meta": "health_capabilities"}) + + +def browser_capabilities(): + """Return the live engine/protocol action and event capability manifest.""" + capabilities = health_capabilities().get("capabilities", {}) + manifest = capabilities.get("browser_transport_v1") + if not isinstance(manifest, dict): + raise RuntimeError("Browser-Harness daemon does not expose browser_transport_v1") + return {key: value for key, value in manifest.items() if key != "schema_version"} + + +def health_begin(attempt_id, capability=None): + request = {"meta": "health_begin", "attempt_id": attempt_id} + selected_capability = capability or os.environ.get("BH_HEALTH_CAPABILITY") or "health_events_v2" + if selected_capability: + request["capability"] = selected_capability + return _send(request) + + +def health_events_since(attempt_id, after_sequence, through_sequence=None): + request = { + "meta": "health_events_since", + "attempt_id": attempt_id, + "after_sequence": after_sequence, + } + if through_sequence is not None: + request["through_sequence"] = through_sequence + return _send(request) + + +def health_seal(attempt_id): + return _send({"meta": "health_seal", "attempt_id": attempt_id}) + +def health_v2_begin(attempt_id): + return _send({ + "meta": "health_begin", + "attempt_id": attempt_id, + "capability": "health_events_v2", + }) + +def health_v2_events_since(attempt_id, after_sequence, through_sequence=None): + return health_events_since(attempt_id, after_sequence, through_sequence) + +def health_v2_seal(attempt_id): + return health_seal(attempt_id) + + def _js_snippet(expression, limit=160): snippet = expression.strip().replace("\n", "\\n") return snippet[:limit - 3] + "..." if len(snippet) > limit else snippet @@ -163,6 +212,10 @@ def goto_url(url): d = (AGENT_WORKSPACE / "domain-skills" / (urlparse(url).hostname or "").removeprefix("www.").split(".")[0]) return {**r, "domain_skills": sorted(p.name for p in d.rglob("*.md"))[:10]} if d.is_dir() else r +def stop_loading(): + """Stop the current document load through the active browser adapter.""" + return cdp("Page.stopLoading") + def page_info(): """{url, title, w, h, sx, sy, pw, ph} — viewport + scroll + page size. @@ -256,7 +309,11 @@ def press_key(key, modifiers=0): so listeners checking e.keyCode / e.key all fire.""" vk, code, text = _KEYS.get(key, (ord(key[0]) if len(key) == 1 else 0, key, key if len(key) == 1 else "")) base = {"key": key, "code": code, "modifiers": modifiers, "windowsVirtualKeyCode": vk, "nativeVirtualKeyCode": vk} - cdp("Input.dispatchKeyEvent", type="keyDown", **base, **({"text": text} if text else {})) + # CDP inserts printable text on the dedicated `char` event. Including the + # same text on keyDown inserts every character twice in Chromium. BiDi's + # adapter uses the keyDown value and deliberately ignores the compatibility + # `char`, so this sequence has one insertion in either protocol. + cdp("Input.dispatchKeyEvent", type="keyDown", **base) if text and len(text) == 1: cdp("Input.dispatchKeyEvent", type="char", text=text, **{k: v for k, v in base.items() if k != "text"}) cdp("Input.dispatchKeyEvent", type="keyUp", **base) @@ -292,8 +349,13 @@ def list_tabs(include_chrome=True): return out def current_tab(): - t = cdp("Target.getTargetInfo").get("targetInfo", {}) - return {"targetId": t.get("targetId"), "url": t.get("url", ""), "title": t.get("title", "")} + status = _send({"meta": "connection_status"}) + page = status.get("page") or {} + return { + "targetId": page.get("targetId") or status.get("target_id"), + "url": page.get("url", ""), + "title": page.get("title", ""), + } def _mark_tab(): """Prepend 🟢 to tab title so the user can see which tab the agent controls.""" @@ -441,11 +503,15 @@ def dispatch_key(selector, key="Enter", event="keypress"): ) def upload_file(selector, path): - """Set files on a file input via CDP DOM.setFileInputFiles. `path` is an absolute filepath (use tempfile.mkstemp if needed).""" + """Set absolute file paths on a file input in the attached browser context.""" + files = [path] if isinstance(path, str) else list(path) + if browser_capabilities().get("protocol") == "bidi": + cdp("BrowserHarness.setFiles", selector=selector, files=files) + return doc = cdp("DOM.getDocument", depth=-1) nid = cdp("DOM.querySelector", nodeId=doc["root"]["nodeId"], selector=selector)["nodeId"] if not nid: raise RuntimeError(f"no element for {selector}") - cdp("DOM.setFileInputFiles", files=[path] if isinstance(path, str) else list(path), nodeId=nid) + cdp("DOM.setFileInputFiles", files=files, nodeId=nid) def http_get(url, headers=None, timeout=20.0): """Pure HTTP — no browser. Use for static pages / APIs. Wrap in ThreadPoolExecutor for bulk. diff --git a/src/browser_harness/run.py b/src/browser_harness/run.py index 2c6ac957..6802957b 100644 --- a/src/browser_harness/run.py +++ b/src/browser_harness/run.py @@ -1,4 +1,4 @@ -import os, sys, urllib.request +import json, os, sys, urllib.request # Windows default stdout encoding is cp1252, which can't encode the 🟢 marker # helpers prepend to tab titles (or anything else outside Latin-1). Force UTF-8 @@ -38,6 +38,7 @@ Commands: browser-harness --version print the installed version + browser-harness --capabilities print live daemon capabilities as JSON browser-harness --doctor diagnose install, daemon, and browser state browser-harness --update [-y] pull the latest version (agents: pass -y) browser-harness --reload stop the daemon so next call picks up code changes @@ -64,6 +65,10 @@ def main(): if args and args[0] == "--version": print(_version() or "unknown") return + if args and args[0] == "--capabilities": + ensure_daemon() + print(json.dumps(health_capabilities(), separators=(",", ":"), sort_keys=True)) + return if args and args[0] == "--doctor": sys.exit(run_doctor()) if args and args[0] == "--update": diff --git a/src/browser_harness/transport.py b/src/browser_harness/transport.py new file mode 100644 index 00000000..fa3d99e3 --- /dev/null +++ b/src/browser_harness/transport.py @@ -0,0 +1,670 @@ +"""Browser protocol adapters used by the persistent Browser-Harness daemon. + +The daemon owns attachment, IPC, and health sequencing. Transports own browser +protocol syntax and expose the small CDP-shaped compatibility surface needed +during the protocol-neutral helper migration. +""" + +from __future__ import annotations + +import asyncio +import json +import os +from urllib.parse import urlsplit, urlunsplit + +from cdp_use.client import CDPClient + + +BIDI_EVENTS = ( + "browsingContext.contextCreated", + "browsingContext.contextDestroyed", + "browsingContext.navigationStarted", + "browsingContext.domContentLoaded", + "browsingContext.load", + "browsingContext.userPromptOpened", + "browsingContext.userPromptClosed", + "log.entryAdded", + "network.beforeRequestSent", + "network.responseStarted", + "network.fetchError", +) + +BIDI_ACTIONS = ( + "contexts", + "navigate", + "evaluate", + "screenshot", + "pointer", + "keyboard", + "set_files", + "preload_script", + "cache", +) + +_NOOP_ENABLE_METHODS = frozenset( + { + "Console.enable", + "Console.disable", + "DOM.enable", + "DOM.disable", + "Log.enable", + "Log.disable", + "Network.enable", + "Network.disable", + "Page.enable", + "Page.disable", + "Runtime.enable", + "Runtime.disable", + "Target.setDiscoverTargets", + "Target.setAutoAttach", + "Target.autoAttachRelated", + } +) + +_BIDI_KEY_VALUES = { + "Backspace": "\ue003", + "Tab": "\ue004", + "Enter": "\ue007", + "Escape": "\ue00c", + " ": "\ue00d", + "PageUp": "\ue00e", + "PageDown": "\ue00f", + "End": "\ue010", + "Home": "\ue011", + "ArrowLeft": "\ue012", + "ArrowUp": "\ue013", + "ArrowRight": "\ue014", + "ArrowDown": "\ue015", + "Delete": "\ue017", +} + +_BIDI_MODIFIERS = ( + (1, "\ue00a"), # Alt + (2, "\ue009"), # Control + (4, "\ue03d"), # Meta + (8, "\ue008"), # Shift +) + + +class _EventRegistry: + async def handle_event(self, _method, _params, _session_id=None): + return None + + +class CDPTransport: + protocol = "cdp" + engine = "edge" + + def __init__(self, endpoint): + self.endpoint = endpoint + self._client = None + self._event_registry = _EventRegistry() + + async def start(self): + self._client = CDPClient(self.endpoint) + await self._client.start() + self._event_registry = self._client._event_registry + + async def send_raw(self, method, params=None, session_id=None): + return await self._client.send_raw(method, params, session_id=session_id) + + async def close(self): + if self._client is not None: + await self._client.stop() + self._client = None + + def capabilities(self): + return { + "engine": self.engine, + "protocol": self.protocol, + "browser_version": None, + "actions": list(BIDI_ACTIONS), + "events": {"supported": ["cdp_health_events_v1"], "unsupported": {}}, + "raw_protocol": True, + } + + +def _remote_value(value): + if not isinstance(value, dict): + return value + kind = value.get("type") + payload = value.get("value") + if kind == "array": + return [_remote_value(item) for item in payload or []] + if kind in {"object", "map"}: + return { + str(_remote_value(key)): _remote_value(child) + for key, child in payload or [] + } + if kind == "set": + return [_remote_value(item) for item in payload or []] + if kind == "null": + return None + if kind == "undefined": + return None + if kind == "bigint" and payload is not None: + try: + return int(payload) + except (TypeError, ValueError): + return str(payload) + return payload + + +def _cdp_remote_object(value): + kind = value.get("type") if isinstance(value, dict) else None + decoded = _remote_value(value) + if kind == "null": + return {"type": "object", "subtype": "null", "value": None} + if kind == "array": + return {"type": "object", "subtype": "array", "value": decoded} + if kind in {"object", "map", "set", "node", "window"}: + return {"type": "object", "value": decoded} + if kind == "undefined": + return {"type": "undefined"} + return {"type": kind or type(decoded).__name__, "value": decoded} + + +def normalize_bidi_event(method, params): + """Translate one BiDi event to the compatibility event Guardian v1 reads.""" + context = params.get("context") or (params.get("source") or {}).get("context") + if method == "log.entryAdded": + entry_type = params.get("type") + if entry_type == "javascript": + return ( + "Runtime.exceptionThrown", + { + "exceptionDetails": { + "text": params.get("text") or "JavaScript exception", + "url": (params.get("source") or {}).get("realm", ""), + } + }, + context, + ) + level = params.get("method") or params.get("level") or "log" + return ( + "Runtime.consoleAPICalled", + { + "type": level, + "args": [ + _cdp_remote_object(argument) + for argument in params.get("args", []) + ] or [{"type": "string", "value": params.get("text", "")}], + }, + context, + ) + if method == "network.beforeRequestSent": + request = params.get("request") or {} + return ( + "Network.requestWillBeSent", + { + "requestId": request.get("request"), + "type": "Fetch", + "request": { + "method": request.get("method"), + "url": request.get("url"), + }, + }, + context, + ) + if method in {"network.responseStarted", "network.responseCompleted"}: + request = params.get("request") or {} + response = params.get("response") or {} + return ( + "Network.responseReceived", + { + "requestId": request.get("request"), + "type": "Fetch", + "response": { + "url": response.get("url") or request.get("url"), + "status": response.get("status"), + "statusText": response.get("statusText"), + }, + }, + context, + ) + if method == "network.fetchError": + request = params.get("request") or {} + error_text = params.get("errorText") or "WebDriver BiDi fetch error" + canceled = any( + marker in error_text.lower() + for marker in ("abort", "cancel", "interrupted") + ) + return ( + "Network.loadingFailed", + { + "requestId": request.get("request"), + "type": "Fetch", + "errorText": error_text, + "canceled": canceled, + }, + context, + ) + if method == "browsingContext.contextDestroyed": + return "Target.targetDestroyed", {"targetId": context}, context + if method == "browsingContext.contextCreated": + return ( + "BrowserHarness.contextCreated", + {"target_id": context, "url": params.get("url", "")}, + context, + ) + if method == "browsingContext.navigationStarted": + return ( + "BrowserHarness.navigationStarted", + {"target_id": context, "url": params.get("url", "")}, + context, + ) + if method == "browsingContext.domContentLoaded": + return "Page.domContentEventFired", {}, context + if method == "browsingContext.load": + return "Page.loadEventFired", {}, context + if method == "browsingContext.userPromptOpened": + return ( + "Page.javascriptDialogOpening", + { + "type": params.get("type"), + "message": params.get("message", ""), + "defaultPrompt": params.get("defaultValue", ""), + }, + context, + ) + if method == "browsingContext.userPromptClosed": + return "Page.javascriptDialogClosed", {}, context + return f"BrowserHarness.bidi.{method}", params, context + + +class WebDriverBiDiTransport: + protocol = "bidi" + engine = "firefox" + + def __init__(self, endpoint, *, connect_host=None, websocket_connect=None): + self.endpoint = endpoint.rstrip("/") + self.connect_host = connect_host + self._websocket_connect = websocket_connect + self._event_registry = _EventRegistry() + self._socket = None + self._reader_task = None + self._next_id = 0 + self._pending = {} + self.session_id = None + self.current_context = None + self.browser_capabilities = {} + self.supported_events = set() + self.unsupported_events = {} + self._pressed_modifiers = [] + + @property + def websocket_url(self): + parsed = urlsplit(self.endpoint) + if parsed.scheme in {"ws", "wss"}: + path = parsed.path.rstrip("/") + if not path.endswith("/session"): + path = f"{path}/session" if path else "/session" + return urlunsplit((parsed.scheme, parsed.netloc, path, parsed.query, parsed.fragment)) + scheme = "wss" if parsed.scheme == "https" else "ws" + path = parsed.path.rstrip("/") + if not path.endswith("/session"): + path = f"{path}/session" if path else "/session" + return urlunsplit((scheme, parsed.netloc, path, parsed.query, parsed.fragment)) + + def capabilities(self): + return { + "engine": self.engine, + "protocol": self.protocol, + "browser_version": self.browser_capabilities.get("browserVersion"), + "actions": list(BIDI_ACTIONS), + "events": { + "supported": sorted(self.supported_events), + "unsupported": dict(sorted(self.unsupported_events.items())), + }, + "raw_protocol": False, + } + + async def start(self): + if self._websocket_connect is None: + from websockets.asyncio.client import connect + + self._websocket_connect = connect + connection_options = {} + if self.connect_host: + connection_options = { + "host": self.connect_host, + "port": urlsplit(self.websocket_url).port, + } + self._socket = await self._websocket_connect( + self.websocket_url, **connection_options + ) + self._reader_task = asyncio.create_task(self._read_messages()) + created = await self._command( + "session.new", {"capabilities": {"alwaysMatch": {}}} + ) + self.session_id = created.get("sessionId") + self.browser_capabilities = created.get("capabilities") or {} + if not self.session_id: + raise RuntimeError("WebDriver BiDi session.new omitted sessionId") + for event_name in BIDI_EVENTS: + try: + await self._command("session.subscribe", {"events": [event_name]}) + self.supported_events.add(event_name) + except Exception as exc: + self.unsupported_events[event_name] = str(exc)[:512] + + async def close(self): + socket = self._socket + if socket is None: + return + try: + if self.session_id: + await self._command("session.end", {}) + except Exception: + # Firefox may close the WebSocket before acknowledging session.end. + pass + finally: + if self._reader_task is not None: + self._reader_task.cancel() + try: + await self._reader_task + except asyncio.CancelledError: + pass + self._reader_task = None + await socket.close() + self._socket = None + self.session_id = None + + async def _read_messages(self): + try: + async for raw in self._socket: + message = json.loads(raw) + request_id = message.get("id") + if request_id is not None: + future = self._pending.pop(request_id, None) + if future and not future.done(): + future.set_result(message) + continue + method = message.get("method") + if method: + translated = normalize_bidi_event(method, message.get("params") or {}) + await self._event_registry.handle_event(*translated) + except Exception as exc: + for future in self._pending.values(): + if not future.done(): + future.set_exception(RuntimeError(f"WebDriver BiDi disconnected: {exc}")) + self._pending.clear() + + async def _command(self, method, params=None): + if self._socket is None: + raise RuntimeError("WebDriver BiDi transport is not connected") + self._next_id += 1 + request_id = self._next_id + future = asyncio.get_running_loop().create_future() + self._pending[request_id] = future + await self._socket.send( + json.dumps({"id": request_id, "method": method, "params": params or {}}) + ) + try: + response = await asyncio.wait_for(future, timeout=30) + finally: + self._pending.pop(request_id, None) + if response.get("type") == "error" or response.get("error"): + error = response.get("error") or "unknown error" + message = response.get("message") or response.get("value", {}).get("message") or "" + raise RuntimeError(f"{error}: {message}".rstrip()) + return response.get("result") or {} + + def _context(self, session_id=None): + context = session_id or self.current_context + if not context: + raise RuntimeError("WebDriver BiDi has no attached browsing context") + return context + + async def _targets(self): + result = await self._command("browsingContext.getTree", {}) + targets = [] + + def visit(node, kind="page"): + context = node.get("context") + targets.append( + { + "targetId": context, + "type": kind, + "url": node.get("url", ""), + "title": node.get("originalOpener") or "", + } + ) + for child in node.get("children") or []: + visit(child, "iframe") + + for context in result.get("contexts") or []: + visit(context) + return targets + + async def _perform(self, actions, context=None): + return await self._command( + "input.performActions", + {"context": context or self._context(), "actions": actions}, + ) + + async def _dispatch_key(self, params): + event_type = params.get("type") + if event_type == "char": + return {} + key = params.get("key") or params.get("text") or "" + value = _BIDI_KEY_VALUES.get(key, key) + modifiers = params.get("modifiers", 0) + actions = [] + if event_type in {"keyDown", "rawKeyDown"}: + self._pressed_modifiers = [ + value for bit, value in _BIDI_MODIFIERS if modifiers & bit + ] + actions.extend({"type": "keyDown", "value": item} for item in self._pressed_modifiers) + actions.append({"type": "keyDown", "value": value}) + elif event_type == "keyUp": + actions.append({"type": "keyUp", "value": value}) + actions.extend( + {"type": "keyUp", "value": item} + for item in reversed(self._pressed_modifiers) + ) + self._pressed_modifiers = [] + if not actions: + return {} + return await self._perform([{"type": "key", "id": "keyboard", "actions": actions}]) + + async def send_raw(self, method, params=None, session_id=None): + params = params or {} + if method in _NOOP_ENABLE_METHODS: + return {} + if method == "Browser.getVersion": + return { + "product": f"Firefox/{self.browser_capabilities.get('browserVersion', 'unknown')}", + "protocolVersion": "WebDriver BiDi", + } + if method == "Target.getTargets": + return {"targetInfos": await self._targets()} + if method == "Target.createTarget": + result = await self._command( + "browsingContext.create", + {"type": "tab", "background": False}, + ) + self.current_context = result.get("context") + return {"targetId": self.current_context} + if method == "Target.attachToTarget": + self.current_context = params.get("targetId") or self._context(session_id) + return {"sessionId": self.current_context} + if method == "Target.activateTarget": + context = params.get("targetId") or self._context(session_id) + await self._command("browsingContext.activate", {"context": context}) + self.current_context = context + return {} + if method == "Runtime.evaluate": + context = self._context(session_id) + result = await self._command( + "script.evaluate", + { + "expression": params.get("expression", ""), + "target": {"context": context}, + "awaitPromise": bool(params.get("awaitPromise")), + "resultOwnership": "none", + "serializationOptions": {"maxObjectDepth": 10}, + "userActivation": bool(params.get("userGesture", False)), + }, + ) + if result.get("type") == "exception": + details = result.get("exceptionDetails") or {} + text = details.get("text") or "JavaScript evaluation failed" + return { + "result": {"type": "object", "subtype": "error", "description": text}, + "exceptionDetails": {"text": text}, + } + return {"result": _cdp_remote_object(result.get("result") or {})} + if method == "Page.navigate": + context = self._context(session_id) + result = await self._command( + "browsingContext.navigate", + {"context": context, "url": params["url"], "wait": "none"}, + ) + return {"frameId": context, "loaderId": result.get("navigation")} + if method == "Page.stopLoading": + await self._command( + "script.evaluate", + { + "expression": "window.stop();", + "target": {"context": self._context(session_id)}, + "awaitPromise": False, + "resultOwnership": "none", + }, + ) + return {} + if method == "Page.captureScreenshot": + result = await self._command( + "browsingContext.captureScreenshot", + { + "context": self._context(session_id), + "origin": "document" if params.get("captureBeyondViewport") else "viewport", + "format": {"type": params.get("format", "png")}, + }, + ) + return {"data": result.get("data")} + if method == "Page.addScriptToEvaluateOnNewDocument": + result = await self._command( + "script.addPreloadScript", {"functionDeclaration": f"() => {{{params.get('source', '')}}}"} + ) + return {"identifier": result.get("script")} + if method == "Page.removeScriptToEvaluateOnNewDocument": + await self._command("script.removePreloadScript", {"script": params.get("identifier")}) + return {} + if method == "Input.dispatchMouseEvent": + event_type = params.get("type") + if event_type == "mouseWheel": + return await self._perform( + [ + { + "type": "wheel", + "id": "wheel", + "actions": [ + { + "type": "scroll", + "x": round(params.get("x", 0)), + "y": round(params.get("y", 0)), + "deltaX": round(params.get("deltaX", 0)), + "deltaY": round(params.get("deltaY", 0)), + "duration": 0, + "origin": "viewport", + } + ], + } + ] + ) + button = {"left": 0, "middle": 1, "right": 2}.get(params.get("button"), 0) + pointer_action = { + "mouseMoved": None, + "mousePressed": {"type": "pointerDown", "button": button}, + "mouseReleased": {"type": "pointerUp", "button": button}, + }.get(event_type) + if event_type not in {"mouseMoved", "mousePressed", "mouseReleased"}: + raise RuntimeError(f"unsupported_browser_capability:{method}:{event_type}") + pointer_actions = [ + { + "type": "pointerMove", + "x": round(params.get("x", 0)), + "y": round(params.get("y", 0)), + "duration": 0, + "origin": "viewport", + } + ] + if pointer_action is not None: + pointer_actions.append(pointer_action) + return await self._perform( + [ + { + "type": "pointer", + "id": "mouse", + "parameters": {"pointerType": "mouse"}, + "actions": pointer_actions, + } + ] + ) + if method == "Input.dispatchKeyEvent": + return await self._dispatch_key(params) + if method == "Input.setIgnoreInputEvents" and not params.get("ignore", False): + return {} + if method == "Input.insertText": + actions = [] + for value in params.get("text", ""): + actions.extend( + ({"type": "keyDown", "value": value}, {"type": "keyUp", "value": value}) + ) + return await self._perform([{"type": "key", "id": "keyboard", "actions": actions}]) + if method == "BrowserHarness.setFiles": + expression = ( + "document.querySelector(" + json.dumps(params.get("selector")) + ")" + ) + located = await self._command( + "script.evaluate", + { + "expression": expression, + "target": {"context": self._context(session_id)}, + "awaitPromise": False, + "resultOwnership": "root", + }, + ) + element = located.get("result") or {} + shared_id = element.get("sharedId") + if not shared_id: + raise RuntimeError(f"no element for {params.get('selector')}") + return await self._command( + "input.setFiles", + { + "context": self._context(session_id), + "element": {"sharedId": shared_id}, + "files": params.get("files") or [], + }, + ) + if method == "Network.setCacheDisabled": + return await self._command( + "network.setCacheBehavior", + {"cacheBehavior": "bypass" if params.get("cacheDisabled") else "default"}, + ) + if method == "Network.clearBrowserCache": + # BiDi has no destructive cache-clear command. Functional tests use + # this immediately after setCacheDisabled, whose standard `bypass` + # behavior gives the same fresh-resource contract for the run. + return {} + raise RuntimeError(f"unsupported_browser_capability:{method}") + + +def transport_from_environment(env=None, *, cdp_endpoint=None): + env = os.environ if env is None else env + protocol = str(env.get("BU_BROWSER_PROTOCOL", "cdp")).strip().lower() + if protocol == "bidi": + endpoint = env.get("BU_BIDI_URL") or env.get("BU_BROWSER_ENDPOINT") + if not endpoint: + raise RuntimeError("BU_BIDI_URL is required for WebDriver BiDi") + return WebDriverBiDiTransport( + endpoint, + connect_host=env.get("BU_BIDI_CONNECT_HOST"), + ) + if protocol != "cdp": + raise RuntimeError(f"Unsupported browser protocol: {protocol}") + endpoint = env.get("BU_CDP_WS") or cdp_endpoint or env.get("BU_CDP_URL") + if not endpoint: + raise RuntimeError("A CDP endpoint is required") + return CDPTransport(endpoint) diff --git a/tests/unit/test_daemon.py b/tests/unit/test_daemon.py new file mode 100644 index 00000000..f0111488 --- /dev/null +++ b/tests/unit/test_daemon.py @@ -0,0 +1,1268 @@ +import asyncio + +from browser_harness.daemon import Daemon + + +def run(coro): + return asyncio.run(coro) + + +def begin(daemon, attempt_id="attempt-1"): + return run(daemon.handle({"meta": "health_begin", "attempt_id": attempt_id})) + + +def events_since(daemon, after_sequence, attempt_id="attempt-1"): + return run( + daemon.handle( + { + "meta": "health_events_since", + "attempt_id": attempt_id, + "after_sequence": after_sequence, + } + ) + ) + + +def test_health_capabilities_advertise_exact_schema_and_daemon_identity(): + daemon = Daemon(event_max_count=7, event_max_bytes=4096, event_max_item_bytes=512) + + result = run(daemon.handle({"meta": "health_capabilities"})) + + assert result["capabilities"] == { + "health_events_v2": { + "schema_version": 2, + "event_schema_version": 2, + "operations": ["begin", "events_since", "seal"], + "sequence_origin": 1, + "continuity_proofs": ["same_target_paused_session_handoff_v1"], + "retention": { + "max_events": 7, + "max_total_bytes": 4096, + "max_event_bytes": 512, + }, + "identity": ["engine", "protocol", "browser_session"], + }, + } + assert result["daemon_fingerprint"] + assert result["observation"] == { + "ready": False, + "target_id": None, + "session_id": None, + "target_epoch": 0, + "session_epoch": 0, + "subscriptions": {}, + } + + +def test_health_capabilities_include_the_live_browser_transport_manifest(): + class FirefoxTransport: + protocol = "bidi" + engine = "firefox" + + def capabilities(self): + return { + "engine": "firefox", + "protocol": "bidi", + "browser_version": "153.0", + "actions": ["evaluate", "navigate"], + "events": { + "supported": ["log.entryAdded"], + "unsupported": {}, + }, + "raw_protocol": False, + } + + daemon = Daemon() + daemon.cdp = FirefoxTransport() + + result = run(daemon.handle({"meta": "health_capabilities"})) + + assert result["capabilities"]["browser_transport_v1"] == { + "schema_version": 1, + "engine": "firefox", + "protocol": "bidi", + "browser_version": "153.0", + "actions": ["evaluate", "navigate"], + "events": { + "supported": ["log.entryAdded"], + "unsupported": {}, + }, + "raw_protocol": False, + } + + +def test_health_v2_binds_normalized_events_to_transport_identity(): + class FirefoxTransport: + protocol = "bidi" + engine = "firefox" + session_id = "browser-session-1" + + daemon = Daemon() + daemon.cdp = FirefoxTransport() + begun = run(daemon.handle({ + "meta": "health_begin", + "attempt_id": "attempt-v2", + "capability": "health_events_v2", + })) + daemon._record_health_event( + "Network.loadingFailed", + {"requestId": "request-1", "errorText": "offline"}, + "context-1", + ) + observed = run(daemon.handle({ + "meta": "health_events_since", + "attempt_id": "attempt-v2", + "after_sequence": begun["start_sequence"], + })) + + assert begun["schema_version"] == 2 + assert observed["schema_version"] == 2 + assert observed["events"][0]["kind"] == "network-failure" + assert observed["events"][0]["engine"] == "firefox" + assert observed["events"][0]["source_protocol"] == "bidi" + assert observed["events"][0]["browser_session_id"] == "browser-session-1" + + +def test_connection_status_resolves_the_exact_attached_target_from_inventory(): + class TargetInventoryCDP: + async def send_raw(self, method, params=None, session_id=None): + assert method == "Target.getTargets" + assert params is None + assert session_id is None + return { + "targetInfos": [ + { + "targetId": "target-other", + "type": "page", + "url": "https://example.test/other", + "title": "Other", + }, + { + "targetId": "target-current", + "type": "page", + "url": "https://example.test/current", + "title": "Current", + }, + ] + } + + daemon = ready_daemon() + daemon.cdp = TargetInventoryCDP() + + result = run(daemon.handle({"meta": "connection_status"})) + + assert result == { + "target_id": "target-current", + "session_id": "session-current", + "page": { + "targetId": "target-current", + "title": "Current", + "url": "https://example.test/current", + }, + } + + +def test_daemon_fingerprint_changes_when_daemon_is_replaced(): + first = Daemon() + replacement = Daemon() + + assert first.daemon_fingerprint != replacement.daemon_fingerprint + + +def test_begin_and_seal_are_idempotent_and_fix_the_terminal_sequence(): + daemon = Daemon() + + first_begin = begin(daemon) + daemon._record_health_event("Runtime.consoleAPICalled", {"type": "error"}, "session-1") + repeated_begin = begin(daemon) + first_seal = run( + daemon.handle({"meta": "health_seal", "attempt_id": "attempt-1"}) + ) + daemon._record_health_event("Page.loadEventFired", {}, "session-1") + repeated_seal = run( + daemon.handle({"meta": "health_seal", "attempt_id": "attempt-1"}) + ) + + assert first_begin == repeated_begin + assert first_begin["start_sequence"] == 0 + assert first_seal == repeated_seal + assert first_seal["sealed_through_sequence"] == 1 + result = events_since(daemon, first_begin["start_sequence"]) + assert [event["sequence"] for event in result["events"]] == [1] + assert result["range"]["sealed_through_sequence"] == 1 + + +def test_begin_waits_for_inflight_attachment_work_before_fencing_identity(): + async def scenario(): + daemon = ready_daemon() + + async def finish_handoff(): + await asyncio.sleep(0) + daemon.session = "session-next" + daemon.session_epoch = 2 + daemon._record_health_event( + "BrowserHarness.sameTargetSessionHandoff", + { + "proof": "same_target_paused_session_handoff_v1", + "target_id": "target-current", + "previous_session_id": "session-current", + "previous_session_epoch": 1, + "session_id": "session-next", + "session_epoch": 2, + }, + "session-next", + ) + + task = asyncio.create_task(finish_handoff()) + daemon.background_tasks.add(task) + task.add_done_callback(daemon.background_tasks.discard) + begun = await daemon.handle( + {"meta": "health_begin", "attempt_id": "attempt-stable"} + ) + return daemon, begun + + daemon, begun = run(scenario()) + + assert begun["start_sequence"] == 1 + assert begun["observation"]["session_id"] == "session-next" + assert begun["observation"]["session_epoch"] == 2 + assert daemon.background_tasks == set() + + +def test_active_health_attempt_guards_observation_control_at_daemon_boundary(): + daemon = ready_daemon() + daemon.cdp = FakeCDP() + begin(daemon) + + for method in ( + "Runtime.enable", + "Runtime.disable", + "Network.disable", + "Target.setAutoAttach", + ): + result = run(daemon.handle({"method": method, "params": {}})) + assert result == {"error": "health_observation_control_is_guarded"} + + run(daemon.handle({"meta": "health_seal", "attempt_id": "attempt-1"})) + assert run(daemon.handle({ + "method": "Runtime.enable", + "params": {}, + })) == {"result": {}} + + +def test_events_since_is_non_destructive_and_returns_sequenced_identity(): + daemon = Daemon() + daemon.target_id = "target-1" + daemon.session = "session-1" + daemon.target_epoch = 2 + daemon.session_epoch = 3 + started = begin(daemon) + daemon._record_health_event( + "Runtime.exceptionThrown", + {"exceptionDetails": {"text": "boom"}}, + "session-1", + ) + + first = events_since(daemon, started["start_sequence"]) + repeated = events_since(daemon, started["start_sequence"]) + + assert first == repeated + assert first["overflow"] is None + assert first["range"] == { + "requested_after_sequence": 0, + "available_from_sequence": 1, + "current_sequence": 1, + "returned_through_sequence": 1, + "sealed_through_sequence": None, + "complete": True, + } + assert first["events"] == [ + { + "sequence": 1, + "method": "Runtime.exceptionThrown", + "params": {"exceptionDetails": {"text": "boom"}}, + "kind": "script-exception", + "engine": None, + "source_protocol": None, + "browser_session_id": "session-1", + "session_id": "session-1", + "daemon_fingerprint": daemon.daemon_fingerprint, + "target_id": "target-1", + "target_epoch": 2, + "session_epoch": 3, + } + ] + + +def test_count_eviction_reports_an_explicit_gap(): + daemon = Daemon(event_max_count=2) + started = begin(daemon) + for index in range(3): + daemon._record_health_event("Log.entryAdded", {"index": index}, None) + + result = events_since(daemon, started["start_sequence"]) + + assert [event["sequence"] for event in result["events"]] == [2, 3] + assert result["range"]["complete"] is False + assert result["overflow"] == { + "kind": "retention_or_truncation", + "requested_after_sequence": 0, + "lost_through_sequence": 1, + "available_from_sequence": 2, + } + + +def test_total_byte_eviction_is_independent_of_count_limit(): + daemon = Daemon( + event_max_count=20, + event_max_bytes=900, + event_max_item_bytes=800, + ) + started = begin(daemon) + for index in range(4): + daemon._record_health_event( + "Runtime.consoleAPICalled", + { + "type": "error", + "args": [{"type": "string", "value": f"{index}-" + "x" * 260}], + }, + None, + ) + + result = events_since(daemon, started["start_sequence"]) + + assert result["overflow"] is not None + assert result["overflow"]["lost_through_sequence"] >= 1 + assert result["events"][-1]["params"]["args"][0]["value"].startswith("3-") + + +def test_irrelevant_cdp_event_flood_does_not_advance_or_overflow_health_ring(): + daemon = Daemon(event_max_count=3, event_max_bytes=2048) + started = begin(daemon) + irrelevant = ( + "Network.dataReceived", + "Network.webSocketFrameReceived", + "Page.lifecycleEvent", + "Runtime.executionContextCreated", + ) + + for index in range(100): + daemon._record_cdp_event( + irrelevant[index % len(irrelevant)], + {"requestId": str(index), "data": "x" * 500}, + "transport-session", + ) + + result = events_since(daemon, started["start_sequence"]) + assert result["events"] == [] + assert result["range"]["current_sequence"] == 0 + assert result["range"]["complete"] is True + assert result["overflow"] is None + + +def test_relevant_request_response_and_error_remain_complete_across_noise(): + daemon = Daemon(event_max_count=4, event_max_bytes=4096) + started = begin(daemon) + daemon._record_cdp_event( + "Network.requestWillBeSent", + { + "requestId": "request-1", + "type": "Fetch", + "request": { + "method": "POST", + "url": "https://example.test/api/items?token=secret", + "headers": { + "Authorization": "Bearer secret", + "Cookie": "session=secret", + }, + "postData": "password=secret", + "postDataEntries": [{"bytes": "secret"}], + "hasPostData": True, + }, + "associatedCookies": [{"cookie": {"name": "session", "value": "secret"}}], + }, + "transport-session", + ) + for index in range(100): + daemon._record_cdp_event( + "Network.dataReceived", + {"requestId": "request-1", "dataLength": index, "data": "x" * 500}, + "transport-session", + ) + daemon._record_cdp_event( + "Network.responseReceived", + { + "requestId": "request-1", + "type": "Fetch", + "response": { + "url": "https://example.test/api/items?token=secret", + "status": 500, + "statusText": "Internal Server Error", + "headers": { + "Set-Cookie": "session=secret", + "X-Secret": "secret", + }, + "requestHeaders": {"Authorization": "Bearer secret"}, + "securityDetails": {"issuer": "private"}, + }, + }, + "transport-session", + ) + daemon._record_cdp_event( + "Runtime.consoleAPICalled", + { + "type": "error", + "args": [{"type": "string", "value": "request failed"}], + "executionContextId": 99, + }, + "transport-session", + ) + + result = events_since(daemon, started["start_sequence"]) + assert [event["sequence"] for event in result["events"]] == [1, 2, 3] + assert [event["method"] for event in result["events"]] == [ + "Network.requestWillBeSent", + "Network.responseReceived", + "Runtime.consoleAPICalled", + ] + assert result["range"]["complete"] is True + assert result["overflow"] is None + assert result["events"][0]["params"] == { + "requestId": "request-1", + "type": "Fetch", + "request": { + "method": "POST", + "url": "https://example.test/api/items?token=secret", + }, + } + assert result["events"][1]["params"] == { + "requestId": "request-1", + "type": "Fetch", + "response": { + "url": "https://example.test/api/items?token=secret", + "status": 500, + "statusText": "Internal Server Error", + }, + } + assert result["events"][2]["params"] == { + "type": "error", + "args": [{"type": "string", "value": "request failed"}], + } + + +def test_out_of_order_losses_keep_a_monotonic_union_when_gap_storage_is_one(): + daemon = Daemon(event_max_count=1, event_max_item_bytes=300) + started = begin(daemon) + daemon._record_health_event("Log.entryAdded", {"text": "retained-1"}, None) + daemon._record_health_event( + "Runtime.consoleAPICalled", + {"args": [{"value": "x" * 600}]}, + None, + ) + daemon._record_health_event("Log.entryAdded", {"text": "retained-3"}, None) + + result = events_since(daemon, 1) + + assert [event["sequence"] for event in result["events"]] == [3] + assert result["range"]["complete"] is False + assert result["overflow"]["lost_through_sequence"] == 2 + + +def test_oversized_event_is_not_retained_and_makes_the_range_incomplete(): + daemon = Daemon(event_max_item_bytes=300) + started = begin(daemon) + daemon._record_health_event( + "Runtime.consoleAPICalled", + {"args": [{"type": "string", "value": "x" * 600}]}, + None, + ) + + result = events_since(daemon, started["start_sequence"]) + + assert result["events"] == [] + assert result["range"]["current_sequence"] == 1 + assert result["range"]["complete"] is False + assert result["overflow"]["lost_through_sequence"] == 1 + + +def test_response_and_request_bodies_are_removed_before_retention(): + daemon = Daemon() + begin(daemon) + daemon._record_health_event( + "Network.requestWillBeSent", + { + "request": { + "url": "https://example.test/", + "postData": "password=not-retained", + }, + "response": { + "status": 500, + "body": "not-retained", + "payloadData": "not-retained", + }, + }, + None, + ) + + event = events_since(daemon, 0)["events"][0] + + assert event["params"] == {"request": {"url": "https://example.test/"}} + + +def test_non_health_network_payload_variants_are_not_retained(): + daemon = Daemon() + begin(daemon) + daemon._record_health_event( + "Network.requestWillBeSent", + { + "request": { + "url": "https://example.test/", + "postDataEntries": [{"bytes": "encoded-request-body"}], + "metadata": {"data": "keep-metadata"}, + } + }, + None, + ) + daemon._record_health_event( + "Network.eventSourceMessageReceived", + {"eventName": "message", "eventId": "1", "data": "event-stream-body"}, + None, + ) + daemon._record_health_event( + "Network.dataReceived", + {"requestId": "1", "dataLength": 12, "data": "response-body"}, + None, + ) + daemon._record_health_event( + "Network.directTCPSocketChunkReceived", + {"identifier": "socket-1", "data": "socket-body"}, + None, + ) + daemon._record_health_event( + "Runtime.consoleAPICalled", + { + "type": "log", + "args": [{"type": "string", "value": "ordinary runtime metadata"}], + }, + None, + ) + + result = events_since(daemon, 0) + + assert [event["method"] for event in result["events"]] == [ + "Network.requestWillBeSent", + "Runtime.consoleAPICalled", + ] + assert [event["params"] for event in result["events"]] == [ + { + "request": { + "url": "https://example.test/", + } + }, + { + "type": "log", + "args": [{"type": "string", "value": "ordinary runtime metadata"}], + }, + ] + + +def test_compatibility_drain_advances_its_cursor_without_deleting_evidence(): + daemon = Daemon() + begin(daemon) + daemon._record_cdp_event("Network.requestWillBeSent", {"requestId": "1"}, None) + + first_drain = run(daemon.handle({"meta": "drain_events"})) + second_drain = run(daemon.handle({"meta": "drain_events"})) + guardian_read = events_since(daemon, 0) + + assert [event["sequence"] for event in first_drain["events"]] == [1] + assert second_drain["events"] == [] + assert [event["sequence"] for event in guardian_read["events"]] == [1] + + +class FakeCDP: + def __init__(self): + self.attachments = [ + ("target-1", "session-1"), + ("target-2", "session-2"), + ] + self.calls = [] + + async def send_raw(self, method, params=None, session_id=None): + self.calls.append((method, params, session_id)) + if method == "Target.getTargets": + target_id, _ = self.attachments[0] + return { + "targetInfos": [ + { + "targetId": target_id, + "type": "page", + "url": "https://example.test/", + } + ] + } + if method == "Target.attachToTarget": + _, session_id = self.attachments.pop(0) + return {"sessionId": session_id} + return {} + + +def test_reattachment_restores_all_health_subscriptions_and_emits_continuity(): + daemon = Daemon() + daemon.cdp = FakeCDP() + begin_result = begin(daemon) + + run(daemon.attach_first_page(reason="initial_attach")) + run(daemon.attach_first_page(reason="reattach")) + + enabled = [ + (method, session_id) + for method, _, session_id in daemon.cdp.calls + if method.endswith(".enable") or method == "Target.setDiscoverTargets" + ] + assert ("Target.setDiscoverTargets", None) in enabled + auto_attach_calls = [ + (params, session_id) + for method, params, session_id in daemon.cdp.calls + if method == "Target.setAutoAttach" + ] + assert auto_attach_calls == [ + ( + { + "autoAttach": True, + "waitForDebuggerOnStart": True, + "flatten": True, + "filter": [ + {"type": "page", "exclude": False}, + {"exclude": True}, + ], + }, + None, + ), + ( + { + "autoAttach": True, + "waitForDebuggerOnStart": True, + "flatten": True, + "filter": [ + {"type": "page", "exclude": False}, + {"exclude": True}, + ], + }, + None, + ), + ] + for session_id in ("session-1", "session-2"): + for domain in ("Page", "Runtime", "Log", "Network"): + assert (f"{domain}.enable", session_id) in enabled + + capability = run(daemon.handle({"meta": "health_capabilities"})) + assert capability["observation"]["ready"] is True + assert capability["observation"]["target_id"] == "target-2" + assert capability["observation"]["session_id"] == "session-2" + assert capability["observation"]["target_epoch"] == 2 + assert capability["observation"]["session_epoch"] == 2 + assert all( + proof["enabled"] + for proof in capability["observation"]["subscriptions"].values() + ) + + continuity = events_since(daemon, begin_result["start_sequence"])["events"] + assert [event["method"] for event in continuity] == [ + "BrowserHarness.attachmentChanged", + "BrowserHarness.attachmentChanged", + ] + assert continuity[-1]["params"]["reason"] == "reattach" + + +class BlockingSubscriptionCDP: + def __init__(self): + self.subscription_started = asyncio.Event() + self.allow_subscription = asyncio.Event() + + async def send_raw(self, method, params=None, session_id=None): + if method == "Target.setDiscoverTargets": + self.subscription_started.set() + await self.allow_subscription.wait() + return {} + + +def test_attachment_transition_never_attests_new_identity_with_old_ready_proof(): + async def scenario(): + daemon = Daemon() + daemon.cdp = BlockingSubscriptionCDP() + daemon.target_id = "target-old" + daemon.session = "session-old" + daemon.target_epoch = 1 + daemon.session_epoch = 1 + daemon.subscriptions = { + "Target": {"enabled": True, "scope": "browser", "session_id": None}, + **{ + domain: { + "enabled": True, + "scope": "session", + "session_id": "session-old", + } + for domain in ("Page", "Runtime", "Log", "Network") + }, + } + + transition = asyncio.create_task( + daemon._set_attachment("target-new", "session-new", "switch") + ) + await daemon.cdp.subscription_started.wait() + + during = await daemon.handle( + {"meta": "health_begin", "attempt_id": "during-transition"} + ) + assert during["observation"]["target_id"] == "target-new" + assert during["observation"]["session_id"] == "session-new" + assert during["observation"]["ready"] is False + assert during["observation"]["subscriptions"] == {} + assert daemon.health_events.sequence == 0 + + daemon.cdp.allow_subscription.set() + await transition + after = await daemon.handle({"meta": "health_capabilities"}) + assert after["observation"]["ready"] is True + assert all( + proof["session_id"] == "session-new" + for domain, proof in after["observation"]["subscriptions"].items() + if domain != "Target" + ) + assert daemon.health_events.sequence == 1 + + run(scenario()) + + +def test_invalidation_during_subscription_discards_stale_attachment_proof(): + async def scenario(): + daemon = Daemon() + daemon.cdp = BlockingSubscriptionCDP() + + transition = asyncio.create_task( + daemon._set_attachment("target-new", "session-new", "switch") + ) + await daemon.cdp.subscription_started.wait() + daemon._record_cdp_event( + "Target.targetDestroyed", + {"targetId": "target-new"}, + None, + ) + daemon.cdp.allow_subscription.set() + established = await transition + + assert established is False + assert daemon.target_id is None + assert daemon.session is None + assert daemon.subscriptions == {} + assert [ + event["method"] + for event in daemon.health_events.read_after(0)["events"] + ] == [ + "Target.targetDestroyed", + "BrowserHarness.attachmentInvalidated", + ] + + run(scenario()) + + +def ready_daemon(): + daemon = Daemon() + daemon.target_id = "target-current" + daemon.session = "session-current" + daemon.target_epoch = 1 + daemon.session_epoch = 1 + daemon.subscriptions = { + "Target": {"enabled": True, "scope": "browser", "session_id": None}, + **{ + domain: { + "enabled": True, + "scope": "session", + "session_id": "session-current", + } + for domain in ("Page", "Runtime", "Log", "Network") + }, + } + return daemon + + +def test_browser_level_target_attachment_uses_logical_observed_page_session_identity(): + daemon = ready_daemon() + begin(daemon) + + daemon._record_cdp_event( + "Target.attachedToTarget", + { + "sessionId": "transient-session", + "waitingForDebugger": False, + "targetInfo": { + "targetId": "target-current", + "type": "page", + "url": "https://example.test/next", + } + }, + None, + ) + + event = events_since(daemon, 0)["events"][0] + assert event["session_id"] == "session-current" + assert event["session_epoch"] == 1 + + +def test_transient_child_transport_event_uses_logical_observed_page_session_identity(): + daemon = ready_daemon() + begin(daemon) + + daemon._record_cdp_event( + "Runtime.consoleAPICalled", + {"type": "error"}, + "transient-child-session", + ) + + event = events_since(daemon, 0)["events"][0] + assert event["session_id"] == "session-current" + assert event["session_epoch"] == 1 + + +def test_current_session_detach_begins_unproven_handoff_but_unrelated_detach_does_not(): + daemon = ready_daemon() + started = begin(daemon) + + daemon._record_cdp_event( + "Target.detachedFromTarget", + {"sessionId": "session-other", "targetId": "target-other"}, + None, + ) + assert daemon._observation()["ready"] is True + + daemon._record_cdp_event( + "Target.detachedFromTarget", + {"sessionId": "session-current", "targetId": "target-current"}, + None, + ) + sealed = run(daemon.handle({"meta": "health_seal", "attempt_id": "attempt-1"})) + + assert daemon.target_id == "target-current" + assert daemon.session is None + assert daemon.session_epoch == 1 + assert sealed["observation"]["ready"] is False + assert sealed["observation"]["subscriptions"] == {} + methods = [ + event["method"] + for event in events_since(daemon, started["start_sequence"])["events"] + ] + assert methods == [ + "Target.detachedFromTarget", + "Target.detachedFromTarget", + ] + + +def test_current_target_destroy_invalidates_target_session_and_proof(): + daemon = ready_daemon() + begin(daemon) + + daemon._record_cdp_event( + "Target.targetDestroyed", + {"targetId": "target-current"}, + None, + ) + + assert daemon.target_id is None + assert daemon.session is None + assert daemon.target_epoch == 2 + assert daemon.session_epoch == 2 + assert daemon._observation()["ready"] is False + assert daemon._observation()["subscriptions"] == {} + + +class HandoffCDP: + def __init__(self): + self.calls = [] + + async def send_raw(self, method, params=None, session_id=None): + self.calls.append((method, params, session_id)) + return {} + + +def auto_attached(session_id="session-new", target_id="target-current"): + return { + "sessionId": session_id, + "targetInfo": { + "targetId": target_id, + "type": "page", + "url": "https://example.test/next", + }, + "waitingForDebugger": True, + } + + +def test_same_target_paused_auto_attach_proves_gap_free_session_handoff(): + daemon = ready_daemon() + daemon.cdp = HandoffCDP() + begin(daemon) + daemon._record_cdp_event( + "Target.detachedFromTarget", + {"sessionId": "session-current", "targetId": "target-current"}, + None, + ) + assert daemon._observation()["ready"] is False + assert daemon.session is None + + attached = auto_attached() + daemon._record_cdp_event("Target.attachedToTarget", attached, None) + run(daemon._prepare_auto_attached_session(attached)) + + assert daemon.target_id == "target-current" + assert daemon.session == "session-new" + assert daemon.session_epoch == 2 + assert daemon._observation()["ready"] is True + assert daemon.cdp.calls[-1] == ( + "Runtime.runIfWaitingForDebugger", + None, + "session-new", + ) + for domain in ("Page", "Runtime", "Log", "Network"): + assert (f"{domain}.enable", None, "session-new") in daemon.cdp.calls[:-1] + + result = events_since(daemon, 0) + assert [event["method"] for event in result["events"]] == [ + "Target.detachedFromTarget", + "Target.attachedToTarget", + "BrowserHarness.sameTargetSessionHandoff", + ] + proof = result["events"][-1] + assert proof["session_id"] == "session-new" + assert proof["session_epoch"] == 2 + assert proof["params"] == { + "proof": "same_target_paused_session_handoff_v1", + "target_id": "target-current", + "previous_session_id": "session-current", + "previous_session_epoch": 1, + "session_id": "session-new", + "session_epoch": 2, + "waiting_for_debugger_on_start": True, + "required_domains": ["Target", "Page", "Runtime", "Log", "Network"], + "subscriptions_before_resume": True, + "resume_acknowledged": True, + } + + +def test_prepared_auto_attached_session_is_adopted_immediately_with_overlap_proof(): + daemon = ready_daemon() + daemon.cdp = HandoffCDP() + begin(daemon) + attached = auto_attached() + daemon._record_cdp_event("Target.attachedToTarget", attached, None) + run(daemon._prepare_auto_attached_session(attached)) + + assert daemon.session == "session-new" + assert daemon.session_epoch == 2 + assert daemon._observation()["ready"] is True + assert events_since(daemon, 0)["events"][-1]["method"] == ( + "BrowserHarness.sameTargetSessionHandoff" + ) + + daemon._record_cdp_event( + "Target.detachedFromTarget", + {"sessionId": "session-current", "targetId": "target-current"}, + None, + ) + + assert daemon.session == "session-new" + assert daemon._observation()["ready"] is True + assert "BrowserHarness.attachmentChanged" not in [ + event["method"] for event in events_since(daemon, 0)["events"] + ] + + +def test_manual_reattach_uses_prepared_overlap_proof_instead_of_attachment_changed(): + daemon = ready_daemon() + daemon.cdp = HandoffCDP() + begin(daemon) + attached = auto_attached() + daemon._record_cdp_event("Target.attachedToTarget", attached, None) + daemon.attachment_transition = True + run(daemon._prepare_auto_attached_session(attached)) + assert daemon.prepared_auto_session["session_id"] == "session-new" + + established = run( + daemon._set_attachment( + "target-current", + "unproven-manual-session", + "stale_session_reattach", + ) + ) + + assert established is True + assert daemon.session == "session-new" + assert daemon.session_epoch == 2 + assert daemon._observation()["ready"] is True + methods = [event["method"] for event in events_since(daemon, 0)["events"]] + assert methods == [ + "Target.attachedToTarget", + "BrowserHarness.sameTargetSessionHandoff", + ] + + +def test_detach_without_paused_auto_attach_remains_not_ready_and_unproven(): + daemon = ready_daemon() + begin(daemon) + + daemon._record_cdp_event( + "Target.detachedFromTarget", + {"sessionId": "session-current", "targetId": "target-current"}, + None, + ) + sealed = run(daemon.handle({"meta": "health_seal", "attempt_id": "attempt-1"})) + + assert sealed["observation"]["ready"] is False + assert daemon.session is None + assert "BrowserHarness.sameTargetSessionHandoff" not in [ + event["method"] for event in events_since(daemon, 0)["events"] + ] + + +def test_auto_attach_for_replacement_target_cannot_prove_session_continuity(): + daemon = ready_daemon() + daemon.cdp = HandoffCDP() + begin(daemon) + daemon._record_cdp_event( + "Target.detachedFromTarget", + {"sessionId": "session-current", "targetId": "target-current"}, + None, + ) + replacement = auto_attached( + session_id="session-replacement", + target_id="target-replacement", + ) + daemon._record_cdp_event("Target.attachedToTarget", replacement, None) + run(daemon._prepare_auto_attached_session(replacement)) + + assert daemon.target_id == "target-current" + assert daemon.session is None + assert daemon._observation()["ready"] is False + assert daemon.cdp.calls == [ + ( + "Runtime.runIfWaitingForDebugger", + None, + "session-replacement", + ) + ] + assert "BrowserHarness.sameTargetSessionHandoff" not in [ + event["method"] for event in events_since(daemon, 0)["events"] + ] + + +class SessionChangedDuringCommandCDP: + def __init__(self, daemon): + self.daemon = daemon + self.calls = [] + + async def send_raw(self, method, params=None, session_id=None): + self.calls.append((method, session_id)) + if session_id == "session-current": + self.daemon.session = "session-new" + self.daemon.session_epoch = 2 + self.daemon.subscriptions = { + "Target": { + "enabled": True, + "scope": "browser", + "session_id": None, + "auto_attach": "browser_pages", + "wait_for_debugger_on_start": True, + }, + **{ + domain: { + "enabled": True, + "scope": "session", + "session_id": "session-new", + } + for domain in ("Page", "Runtime", "Log", "Network") + }, + } + raise RuntimeError("Session with given id not found") + return {"value": "retried-on-proven-session"} + + +def test_command_retries_on_new_ready_session_when_handoff_wins_the_race(): + daemon = ready_daemon() + daemon.cdp = SessionChangedDuringCommandCDP(daemon) + + result = run( + daemon.handle( + { + "method": "Runtime.evaluate", + "params": {"expression": "1"}, + } + ) + ) + + assert result == {"result": {"value": "retried-on-proven-session"}} + assert daemon.cdp.calls == [ + ("Runtime.evaluate", "session-current"), + ("Runtime.evaluate", "session-new"), + ] + + +class StaleCommandWithPreparedSessionCDP: + def __init__(self): + self.calls = [] + + async def send_raw(self, method, params=None, session_id=None): + self.calls.append((method, session_id)) + if session_id == "session-current": + raise RuntimeError("Session with given id not found") + return {"value": "retried-without-manual-reattach"} + + +def test_stale_command_adopts_prepared_overlap_without_manual_reattach(): + daemon = ready_daemon() + daemon.cdp = StaleCommandWithPreparedSessionCDP() + daemon.prepared_auto_session = { + "target_id": "target-current", + "session_id": "session-new", + "subscriptions": { + "Target": { + "enabled": True, + "scope": "browser", + "session_id": None, + "auto_attach": "browser_pages", + "wait_for_debugger_on_start": True, + }, + **{ + domain: { + "enabled": True, + "scope": "session", + "session_id": "session-new", + } + for domain in ("Page", "Runtime", "Log", "Network") + }, + }, + } + begin(daemon) + + result = run( + daemon.handle( + { + "method": "Runtime.evaluate", + "params": {"expression": "1"}, + } + ) + ) + + assert result == {"result": {"value": "retried-without-manual-reattach"}} + assert daemon.session == "session-new" + assert daemon.cdp.calls == [ + ("Runtime.evaluate", "session-current"), + ("Runtime.evaluate", "session-new"), + ] + assert events_since(daemon, 0)["events"][-1]["method"] == ( + "BrowserHarness.sameTargetSessionHandoff" + ) + + +class StaleCommandWhileAutoAttachPreparesCDP: + def __init__(self): + self.calls = [] + + async def send_raw(self, method, params=None, session_id=None): + self.calls.append((method, session_id)) + if method == "Runtime.evaluate" and session_id == "session-current": + raise RuntimeError("Session with given id not found") + if method == "Runtime.evaluate": + return {"value": "retried-after-paused-handoff"} + return {} + + +def test_stale_command_awaits_inflight_auto_attach_before_manual_fallback(): + async def scenario(): + daemon = ready_daemon() + daemon.cdp = StaleCommandWhileAutoAttachPreparesCDP() + begin_result = await daemon.handle( + {"meta": "health_begin", "attempt_id": "attempt-1"} + ) + attached = auto_attached() + daemon._record_cdp_event("Target.attachedToTarget", attached, None) + + async def prepare_after_command_failure(): + await asyncio.sleep(0) + return await daemon._prepare_auto_attached_session(attached) + + task = asyncio.create_task(prepare_after_command_failure()) + daemon.background_tasks.add(task) + task.add_done_callback(daemon.background_tasks.discard) + + async def unexpected_manual_attach(*_args, **_kwargs): + raise AssertionError("manual reattach raced the paused auto-attach proof") + + daemon.attach_first_page = unexpected_manual_attach + result = await daemon.handle( + { + "method": "Runtime.evaluate", + "params": {"expression": "1"}, + } + ) + events = await daemon.handle( + { + "meta": "health_events_since", + "attempt_id": "attempt-1", + "after_sequence": begin_result["start_sequence"], + } + ) + return daemon, result, events + + daemon, result, events = run(scenario()) + + assert result == {"result": {"value": "retried-after-paused-handoff"}} + assert daemon.session == "session-new" + assert daemon.cdp.calls[-1] == ("Runtime.evaluate", "session-new") + assert [event["method"] for event in events["events"]][-1] == ( + "BrowserHarness.sameTargetSessionHandoff" + ) + + +def test_superseded_page_session_cannot_duplicate_active_session_health_event(): + daemon = ready_daemon() + daemon.cdp = HandoffCDP() + begin(daemon) + attached = auto_attached() + daemon._record_cdp_event("Target.attachedToTarget", attached, None) + run(daemon._prepare_auto_attached_session(attached)) + cursor = daemon.health_events.sequence + duplicate = { + "type": "error", + "args": [{"type": "string", "value": "one browser event"}], + } + + daemon._record_cdp_event( + "Runtime.consoleAPICalled", + duplicate, + "session-current", + ) + daemon._record_cdp_event( + "Runtime.consoleAPICalled", + duplicate, + "session-new", + ) + + events = events_since(daemon, cursor)["events"] + assert len(events) == 1 + assert events[0]["method"] == "Runtime.consoleAPICalled" + assert events[0]["session_id"] == "session-new" + + +def test_identical_events_from_active_session_remain_distinct(): + daemon = ready_daemon() + begin(daemon) + event = { + "type": "error", + "args": [{"type": "string", "value": "legitimate repeated error"}], + } + + daemon._record_cdp_event("Runtime.consoleAPICalled", event, "session-current") + daemon._record_cdp_event("Runtime.consoleAPICalled", event, "session-current") + + events = events_since(daemon, 0)["events"] + assert [entry["sequence"] for entry in events] == [1, 2] + assert events[0]["params"] == events[1]["params"] diff --git a/tests/unit/test_helpers.py b/tests/unit/test_helpers.py index e90602b9..88436c34 100644 --- a/tests/unit/test_helpers.py +++ b/tests/unit/test_helpers.py @@ -77,6 +77,119 @@ def fake_cdp(method, **kwargs): helpers.page_info() +def test_health_helpers_send_the_versioned_protocol_operations(): + responses = [ + {"capabilities": {"health_events_v2": {"schema_version": 2}}}, + {"attempt_id": "attempt-1", "start_sequence": 4}, + {"attempt_id": "attempt-1", "events": [], "range": {"complete": True}}, + {"attempt_id": "attempt-1", "sealed_through_sequence": 9}, + ] + requests = [] + + def fake_send(request): + requests.append(request) + return responses.pop(0) + + with patch("browser_harness.helpers._send", side_effect=fake_send): + assert helpers.health_capabilities()["capabilities"]["health_events_v2"][ + "schema_version" + ] == 2 + assert helpers.health_begin("attempt-1")["start_sequence"] == 4 + assert helpers.health_events_since("attempt-1", 4)["events"] == [] + assert helpers.health_seal("attempt-1")["sealed_through_sequence"] == 9 + + assert requests == [ + {"meta": "health_capabilities"}, + {"meta": "health_begin", "attempt_id": "attempt-1", "capability": "health_events_v2"}, + { + "meta": "health_events_since", + "attempt_id": "attempt-1", + "after_sequence": 4, + }, + {"meta": "health_seal", "attempt_id": "attempt-1"}, + ] + + +def test_health_v2_helpers_select_the_normalized_capability(): + requests = [] + + with patch("browser_harness.helpers._send", side_effect=lambda request: requests.append(request) or {}): + helpers.health_v2_begin("attempt-2") + + assert requests == [{ + "meta": "health_begin", + "attempt_id": "attempt-2", + "capability": "health_events_v2", + }] + + +def test_browser_capabilities_returns_the_protocol_neutral_manifest(): + with patch( + "browser_harness.helpers._send", + return_value={ + "capabilities": { + "browser_transport_v1": { + "engine": "firefox", + "protocol": "bidi", + } + } + }, + ): + assert helpers.browser_capabilities() == { + "engine": "firefox", + "protocol": "bidi", + } + + +def test_upload_file_uses_protocol_neutral_set_files_for_bidi(): + calls = [] + + def fake_cdp(method, **params): + calls.append((method, params)) + return {} + + with patch( + "browser_harness.helpers.browser_capabilities", + return_value={"protocol": "bidi"}, + ), patch("browser_harness.helpers.cdp", side_effect=fake_cdp): + helpers.upload_file("#upload", ["/tmp/one.glb", "/tmp/two.glb"]) + + assert calls == [ + ( + "BrowserHarness.setFiles", + { + "selector": "#upload", + "files": ["/tmp/one.glb", "/tmp/two.glb"], + }, + ) + ] + + +def test_current_tab_uses_daemon_attachment_without_creating_a_manual_session(): + with patch( + "browser_harness.helpers._send", + return_value={ + "target_id": "target-current", + "session_id": "session-current", + "page": { + "targetId": "target-current", + "url": "https://example.test/current", + "title": "Current", + }, + }, + ) as send, patch( + "browser_harness.helpers.cdp", + side_effect=AssertionError("current_tab must not manually query or attach"), + ): + assert helpers.current_tab() == { + "targetId": "target-current", + "url": "https://example.test/current", + "title": "Current", + } + + send.assert_called_once_with({"meta": "connection_status"}) + + # --- fill_input --- def test_fill_input_focuses_types_and_fires_events(): @@ -165,6 +278,23 @@ def fake_js(expr, **kwargs): assert "Backspace" not in keys_seen +def test_printable_key_inserts_text_once_per_protocol_sequence(): + key_events = [] + + def fake_cdp(method, **kwargs): + if method == "Input.dispatchKeyEvent": + key_events.append(kwargs) + return {} + + with patch("browser_harness.helpers.cdp", side_effect=fake_cdp): + helpers.press_key("x") + + key_down = next(event for event in key_events if event["type"] == "keyDown") + char = next(event for event in key_events if event["type"] == "char") + assert "text" not in key_down + assert char["text"] == "x" + + # --- wait_for_element --- def test_wait_for_element_returns_true_when_found_immediately(): @@ -302,4 +432,3 @@ def fake_send(req): result = helpers.wait_for_network_idle(timeout=10.0, idle_ms=500) assert result is False - diff --git a/tests/unit/test_run.py b/tests/unit/test_run.py index 31cb4e1d..58601ff8 100644 --- a/tests/unit/test_run.py +++ b/tests/unit/test_run.py @@ -1,4 +1,5 @@ import sys +import json from io import StringIO from unittest.mock import patch @@ -15,6 +16,22 @@ def test_c_flag_executes_code(): assert stdout.getvalue().strip() == "hello from -c" +def test_capabilities_prints_live_daemon_attestation_as_json(): + stdout = StringIO() + capability = { + "daemon_fingerprint": "daemon-1", + "capabilities": {"health_events_v1": {"schema_version": 1}}, + } + + with patch.object(sys, "argv", ["browser-harness", "--capabilities"]), \ + patch("browser_harness.run.ensure_daemon"), \ + patch("browser_harness.run.health_capabilities", return_value=capability), \ + patch("sys.stdout", stdout): + run.main() + + assert json.loads(stdout.getvalue()) == capability + + def test_cloud_bootstrap_on_headless_server(monkeypatch): """No daemon, no local Chrome, API key + BU_AUTOSPAWN set -> auto-provision cloud daemon.""" monkeypatch.setenv("BROWSER_USE_API_KEY", "test-key") diff --git a/tests/unit/test_transport.py b/tests/unit/test_transport.py new file mode 100644 index 00000000..90688fb5 --- /dev/null +++ b/tests/unit/test_transport.py @@ -0,0 +1,250 @@ +import asyncio + +import pytest + +from browser_harness.transport import ( + WebDriverBiDiTransport, + normalize_bidi_event, + transport_from_environment, +) + + +def run(coro): + return asyncio.run(coro) + + +def test_transport_factory_preserves_cdp_and_selects_bidi_explicitly(): + cdp = transport_from_environment({"BU_CDP_WS": "ws://edge.test/devtools/browser/1"}) + bidi = transport_from_environment( + { + "BU_BROWSER_PROTOCOL": "bidi", + "BU_BIDI_URL": "http://localhost:9222", + "BU_BIDI_CONNECT_HOST": "firefox.test", + } + ) + + assert cdp.protocol == "cdp" + assert cdp.engine == "edge" + assert bidi.protocol == "bidi" + assert bidi.engine == "firefox" + assert bidi.endpoint == "http://localhost:9222" + assert bidi.connect_host == "firefox.test" + assert bidi.websocket_url == "ws://localhost:9222/session" + + +def test_bidi_transport_translates_navigation_and_script_evaluation(): + transport = WebDriverBiDiTransport("http://firefox.test:9222") + transport.current_context = "context-1" + calls = [] + + async def command(method, params=None): + calls.append((method, params)) + if method == "browsingContext.navigate": + return {"navigation": "navigation-1", "url": params["url"]} + if method == "script.evaluate": + return { + "type": "success", + "result": {"type": "string", "value": "Angel3"}, + } + raise AssertionError(method) + + transport._command = command + + navigation = run( + transport.send_raw("Page.navigate", {"url": "https://angel3.test/tactical"}) + ) + evaluation = run( + transport.send_raw( + "Runtime.evaluate", + {"expression": "document.title", "awaitPromise": True}, + ) + ) + + assert navigation == { + "frameId": "context-1", + "loaderId": "navigation-1", + } + assert evaluation == { + "result": {"type": "string", "value": "Angel3"}, + } + assert calls == [ + ( + "browsingContext.navigate", + { + "context": "context-1", + "url": "https://angel3.test/tactical", + "wait": "none", + }, + ), + ( + "script.evaluate", + { + "expression": "document.title", + "target": {"context": "context-1"}, + "awaitPromise": True, + "resultOwnership": "none", + "serializationOptions": {"maxObjectDepth": 10}, + "userActivation": False, + }, + ), + ] + + +def test_bidi_transport_reports_unsupported_diagnostics_before_dispatch(): + transport = WebDriverBiDiTransport("http://firefox.test:9222") + transport.current_context = "context-1" + + with pytest.raises(RuntimeError, match="unsupported_browser_capability:HeapProfiler.collectGarbage"): + run(transport.send_raw("HeapProfiler.collectGarbage")) + + +def test_bidi_events_normalize_to_guardian_health_categories(): + assert normalize_bidi_event( + "log.entryAdded", + { + "type": "console", + "method": "error", + "text": "renderer failed", + "args": [{"type": "string", "value": "renderer failed"}], + "source": {"context": "context-1"}, + }, + ) == ( + "Runtime.consoleAPICalled", + { + "type": "error", + "args": [{"type": "string", "value": "renderer failed"}], + }, + "context-1", + ) + + assert normalize_bidi_event( + "network.fetchError", + { + "request": {"request": "request-2", "url": "https://angel3.test/old"}, + "errorText": "NS_BINDING_ABORTED", + "context": "context-1", + }, + )[1]["canceled"] is True + + assert normalize_bidi_event( + "network.responseStarted", + { + "request": {"request": "request-1", "url": "https://angel3.test/api"}, + "response": {"url": "https://angel3.test/api", "status": 503, "statusText": "Unavailable"}, + "context": "context-1", + }, + ) == ( + "Network.responseReceived", + { + "requestId": "request-1", + "type": "Fetch", + "response": { + "url": "https://angel3.test/api", + "status": 503, + "statusText": "Unavailable", + }, + }, + "context-1", + ) + + +def test_bidi_capability_manifest_uses_observed_subscriptions(): + transport = WebDriverBiDiTransport("http://firefox.test:9222") + transport.browser_capabilities = {"browserName": "firefox", "browserVersion": "153.0"} + transport.supported_events = {"log.entryAdded", "network.fetchError"} + transport.unsupported_events = {"browsingContext.navigationFailed": "invalid argument"} + + manifest = transport.capabilities() + + assert manifest["engine"] == "firefox" + assert manifest["protocol"] == "bidi" + assert manifest["browser_version"] == "153.0" + assert manifest["events"]["supported"] == ["log.entryAdded", "network.fetchError"] + assert manifest["events"]["unsupported"] == { + "browsingContext.navigationFailed": "invalid argument" + } + assert "evaluate" in manifest["actions"] + + +def test_bidi_transport_ends_session_before_closing_socket(): + calls = [] + + class Socket: + async def close(self): + calls.append(("socket.close", None)) + + transport = WebDriverBiDiTransport("http://firefox.test:9222") + transport._socket = Socket() + transport.session_id = "session-1" + + async def command(method, params=None): + calls.append((method, params)) + return {} + + transport._command = command + + run(transport.close()) + + assert calls == [("session.end", {}), ("socket.close", None)] + assert transport._socket is None + assert transport.session_id is None + + +def test_bidi_transport_supports_pointer_move_without_a_button_action(): + transport = WebDriverBiDiTransport("http://firefox.test:9222") + transport.current_context = "context-1" + performed = [] + + async def perform(actions, context=None): + performed.append((actions, context)) + return {} + + transport._perform = perform + + run(transport.send_raw("Input.dispatchMouseEvent", {"type": "mouseMoved", "x": 12.4, "y": 18.6})) + + assert performed[0][0][0]["actions"] == [{ + "type": "pointerMove", + "x": 12, + "y": 19, + "duration": 0, + "origin": "viewport", + }] + + +def test_bidi_transport_maps_cache_bypass_and_safe_input_restore(): + transport = WebDriverBiDiTransport("http://firefox.test:9222") + transport.current_context = "context-1" + calls = [] + + async def command(method, params=None): + calls.append((method, params)) + return {} + + transport._command = command + + run(transport.send_raw("Network.setCacheDisabled", {"cacheDisabled": True})) + assert run(transport.send_raw("Network.clearBrowserCache")) == {} + assert run(transport.send_raw("Input.setIgnoreInputEvents", {"ignore": False})) == {} + + assert calls == [("network.setCacheBehavior", {"cacheBehavior": "bypass"})] + + +def test_bidi_transport_stops_loading_without_a_nonstandard_context_command(): + transport = WebDriverBiDiTransport("http://firefox.test:9222") + transport.current_context = "context-1" + calls = [] + + async def command(method, params=None): + calls.append((method, params)) + return {} + + transport._command = command + + assert run(transport.send_raw("Page.stopLoading")) == {} + assert calls == [("script.evaluate", { + "expression": "window.stop();", + "target": {"context": "context-1"}, + "awaitPromise": False, + "resultOwnership": "none", + })]