"""Browser control via CDP. Core helpers live here. Agent-editable helpers live in BH_AGENT_WORKSPACE/agent_helpers.py. [WHO]: Provides Browser Harness navigation, input, screenshot, tab, JavaScript, upload, and HTTP helper functions [FROM]: Depends on browser_harness._ipc, CDP daemon requests, urllib, base64, importlib, and optional Pillow [TO]: Consumed by browser_harness.run snippets and Catui browser tool calls [HERE]: extensions/builtin/browser/src/browser_harness/helpers.py within vendored Browser Harness package """ import base64, importlib.util, json, math, os, time, urllib.request from pathlib import Path from urllib.parse import urlparse from . import _ipc as ipc CORE_DIR = Path(__file__).resolve().parent REPO_ROOT = CORE_DIR.parent.parent AGENT_WORKSPACE = Path(os.environ.get("BH_AGENT_WORKSPACE", REPO_ROOT / "agent-workspace")).expanduser() def _load_env(): paths = [REPO_ROOT / ".env", AGENT_WORKSPACE / ".env"] for p in paths: if not p.exists(): continue _load_env_file(p) def _load_env_file(p): for line in p.read_text().splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) _load_env() NAME = os.environ.get("BU_NAME", "default") SOCK = ipc.sock_addr(NAME) INTERNAL = ("chrome://", "chrome-untrusted://", "devtools://", "chrome-extension://", "about:") def _send(req): c = ipc.connect(NAME, timeout=5.0) c.sendall((json.dumps(req) + "\n").encode()) data = b"" while not data.endswith(b"\n"): chunk = c.recv(1 << 20) if not chunk: break data += chunk c.close() r = json.loads(data) if "error" in r: raise RuntimeError(r["error"]) return r def cdp(method, session_id=None, **params): """Raw CDP. cdp('Page.navigate', url='...'), cdp('DOM.getDocument', depth=-1).""" return _send({"method": method, "params": params, "session_id": session_id}).get("result", {}) def drain_events(): return _send({"meta": "drain_events"})["events"] def _js_snippet(expression, limit=160): snippet = expression.strip().replace("\n", "\\n") return snippet[:limit - 3] + "..." if len(snippet) > limit else snippet def _js_exception_description(result, details): desc = result.get("description") exc = details.get("exception") if details else None if not desc and isinstance(exc, dict): desc = exc.get("description") if desc is None and "value" in exc: desc = str(exc["value"]) if desc is None: desc = exc.get("className") if not desc and details: desc = details.get("text") return desc or "JavaScript evaluation failed" def _decode_unserializable_js_value(value): if value == "NaN": return math.nan if value == "Infinity": return math.inf if value == "-Infinity": return -math.inf if value == "-0": return -0.0 if value.endswith("n"): return int(value[:-1]) return value def _runtime_value(response, expression): result = response.get("result", {}) details = response.get("exceptionDetails") if details or result.get("subtype") == "error": desc = _js_exception_description(result, details) if details: line = details.get("lineNumber") col = details.get("columnNumber") loc = f" at line {line}, column {col}" if line is not None and col is not None else "" else: loc = "" raise RuntimeError(f"JavaScript evaluation failed{loc}: {desc}; expression: {_js_snippet(expression)}") if "value" in result: return result["value"] if "unserializableValue" in result: return _decode_unserializable_js_value(result["unserializableValue"]) return None def _runtime_evaluate(expression, session_id=None, await_promise=False): try: r = cdp("Runtime.evaluate", session_id=session_id, expression=expression, returnByValue=True, awaitPromise=await_promise) except TimeoutError as e: raise RuntimeError(f"Runtime.evaluate timed out; expression: {_js_snippet(expression)}") from e return _runtime_value(r, expression) def _has_return_statement(expression): i = 0 n = len(expression) state = "code" quote = "" while i < n: ch = expression[i] nxt = expression[i + 1] if i + 1 < n else "" if state == "code": if ch in ("'", '"', "`"): state = "string"; quote = ch; i += 1; continue if ch == "/" and nxt == "/": state = "line_comment"; i += 2; continue if ch == "/" and nxt == "*": state = "block_comment"; i += 2; continue if expression.startswith("return", i): before = expression[i - 1] if i > 0 else "" after = expression[i + 6] if i + 6 < n else "" if not (before == "_" or before.isalnum()) and not (after == "_" or after.isalnum()): return True i += 1; continue if state == "line_comment": if ch == "\n": state = "code" i += 1; continue if state == "block_comment": if ch == "*" and nxt == "/": state = "code"; i += 2; continue i += 1; continue if state == "string": if ch == "\\": i += 2; continue if ch == quote: state = "code"; quote = "" i += 1; continue return False # --- navigation / page --- def goto_url(url): r = cdp("Page.navigate", 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 page_info(): """{url, title, w, h, sx, sy, pw, ph} — viewport + scroll + page size. If a native dialog (alert/confirm/prompt/beforeunload) is open, returns {dialog: {type, message, ...}} instead — the page's JS thread is frozen until the dialog is handled (see interaction-skills/dialogs.md).""" dialog = _send({"meta": "pending_dialog"}).get("dialog") if dialog: return {"dialog": dialog} expression = "JSON.stringify({url:location.href,title:document.title,w:innerWidth,h:innerHeight,sx:scrollX,sy:scrollY,pw:document.documentElement.scrollWidth,ph:document.documentElement.scrollHeight})" return json.loads(_runtime_evaluate(expression)) # --- input --- _debug_click_counter = 0 def click_at_xy(x, y, button="left", clicks=1): if os.environ.get("BH_DEBUG_CLICKS"): global _debug_click_counter try: from PIL import Image, ImageDraw dpr = js("window.devicePixelRatio") or 1 path = capture_screenshot(str(ipc._TMP / f"debug_click_{_debug_click_counter}.png")) img = Image.open(path) draw = ImageDraw.Draw(img) px, py = int(x * dpr), int(y * dpr) r = int(15 * dpr) draw.ellipse([px - r, py - r, px + r, py + r], outline="red", width=int(3 * dpr)) draw.line([px - r - int(5 * dpr), py, px + r + int(5 * dpr), py], fill="red", width=int(2 * dpr)) draw.line([px, py - r - int(5 * dpr), px, py + r + int(5 * dpr)], fill="red", width=int(2 * dpr)) img.save(path) print(f"[debug_click] saved {path} (x={x}, y={y}, dpr={dpr})") except Exception as e: print(f"[debug_click] overlay failed: {e}") _debug_click_counter += 1 cdp("Input.dispatchMouseEvent", type="mousePressed", x=x, y=y, button=button, clickCount=clicks) cdp("Input.dispatchMouseEvent", type="mouseReleased", x=x, y=y, button=button, clickCount=clicks) def type_text(text): cdp("Input.insertText", text=text) _KEYS = { # key → (windowsVirtualKeyCode, code, text) "Enter": (13, "Enter", "\r"), "Tab": (9, "Tab", "\t"), "Backspace": (8, "Backspace", ""), "Escape": (27, "Escape", ""), "Delete": (46, "Delete", ""), " ": (32, "Space", " "), "ArrowLeft": (37, "ArrowLeft", ""), "ArrowUp": (38, "ArrowUp", ""), "ArrowRight": (39, "ArrowRight", ""), "ArrowDown": (40, "ArrowDown", ""), "Home": (36, "Home", ""), "End": (35, "End", ""), "PageUp": (33, "PageUp", ""), "PageDown": (34, "PageDown", ""), } def press_key(key, modifiers=0): """Modifiers bitfield: 1=Alt, 2=Ctrl, 4=Meta(Cmd), 8=Shift. Special keys (Enter, Tab, Arrow*, Backspace, etc.) carry their virtual key codes 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 {})) 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) def scroll(x, y, dy=-300, dx=0): cdp("Input.dispatchMouseEvent", type="mouseWheel", x=x, y=y, deltaX=dx, deltaY=dy) # --- visual --- def capture_screenshot(path=None, full=False, max_dim=None): """Save a PNG of the current viewport. Set max_dim=1800 on a 2× display to keep the file under the 2000px-per-side limit some image-aware LLMs enforce.""" path = path or str(ipc._TMP / "shot.png") r = cdp("Page.captureScreenshot", format="png", captureBeyondViewport=full) open(path, "wb").write(base64.b64decode(r["data"])) if max_dim: from PIL import Image img = Image.open(path) if max(img.size) > max_dim: img.thumbnail((max_dim, max_dim)) img.save(path) return path # --- tabs --- def list_tabs(include_chrome=True): out = [] for t in cdp("Target.getTargets")["targetInfos"]: if t["type"] != "page": continue url = t.get("url", "") if not include_chrome and url.startswith(INTERNAL): continue out.append({"targetId": t["targetId"], "title": t.get("title", ""), "url": url}) return out def current_tab(): t = cdp("Target.getTargetInfo").get("targetInfo", {}) return {"targetId": t.get("targetId"), "url": t.get("url", ""), "title": t.get("title", "")} def _mark_tab(): """Prepend 🟢 to tab title so the user can see which tab the agent controls.""" try: cdp("Runtime.evaluate", expression="if(!document.title.startsWith('\U0001F7E2'))document.title='\U0001F7E2 '+document.title") except Exception: pass def switch_tab(target): # Accept either a raw targetId string or the dict returned by current_tab() / list_tabs(), # so `switch_tab(current_tab())` works without a manual ["targetId"] dance. target_id = target.get("targetId") if isinstance(target, dict) else target # Unmark old tab try: cdp("Runtime.evaluate", expression="if(document.title.startsWith('\U0001F7E2 '))document.title=document.title.slice(2)") except Exception: pass cdp("Target.activateTarget", targetId=target_id) sid = cdp("Target.attachToTarget", targetId=target_id, flatten=True)["sessionId"] _send({"meta": "set_session", "session_id": sid, "target_id": target_id}) _mark_tab() return sid def new_tab(url="about:blank"): # Always create blank, then goto: passing url to createTarget races with # attach, so the brief about:blank is "complete" by the time the caller # polls and wait_for_load() returns before navigation actually starts. tid = cdp("Target.createTarget", url="about:blank")["targetId"] switch_tab(tid) if url != "about:blank": goto_url(url) return tid def ensure_real_tab(): """Switch to a real user tab if current is chrome:// / internal / stale.""" tabs = list_tabs(include_chrome=False) if not tabs: return None try: cur = current_tab() if cur["url"] and not cur["url"].startswith(INTERNAL): return cur except Exception: pass switch_tab(tabs[0]["targetId"]) return tabs[0] def iframe_target(url_substr): """First iframe target whose URL contains `url_substr`. Use with js(..., target_id=...).""" for t in cdp("Target.getTargets")["targetInfos"]: if t["type"] == "iframe" and url_substr in t.get("url", ""): return t["targetId"] return None # --- utility --- def wait(seconds=1.0): time.sleep(seconds) def wait_for_load(timeout=15.0): """Poll document.readyState == 'complete' or timeout.""" deadline = time.time() + timeout while time.time() < deadline: if js("document.readyState") == "complete": return True time.sleep(0.3) return False def js(expression, target_id=None): """Run JS in the attached tab (default) or inside an iframe target (via iframe_target()). Expressions with top-level `return` are automatically wrapped in an IIFE, so both `document.title` and `const x = 1; return x` are valid inputs. """ sid = cdp("Target.attachToTarget", targetId=target_id, flatten=True)["sessionId"] if target_id else None if _has_return_statement(expression) and not expression.strip().startswith("("): expression = f"(function(){{{expression}}})()" return _runtime_evaluate(expression, session_id=sid, await_promise=True) _KC = {"Enter": 13, "Tab": 9, "Escape": 27, "Backspace": 8, " ": 32, "ArrowLeft": 37, "ArrowUp": 38, "ArrowRight": 39, "ArrowDown": 40} def dispatch_key(selector, key="Enter", event="keypress"): """Dispatch a DOM KeyboardEvent on the matched element. Use this when a site reacts to synthetic DOM key events on an element more reliably than to raw CDP input events. """ kc = _KC.get(key, ord(key) if len(key) == 1 else 0) js( f"(()=>{{const e=document.querySelector({json.dumps(selector)});if(e){{e.focus();e.dispatchEvent(new KeyboardEvent({json.dumps(event)},{{key:{json.dumps(key)},code:{json.dumps(key)},keyCode:{kc},which:{kc},bubbles:true}}));}}}})()" ) 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).""" 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) def http_get(url, headers=None, timeout=20.0): """Pure HTTP — no browser. Use for static pages / APIs. Wrap in ThreadPoolExecutor for bulk. When BROWSER_USE_API_KEY is set, routes through the fetch-use proxy (handles bot detection, residential proxies, retries). Falls back to local urllib otherwise.""" if os.environ.get("BROWSER_USE_API_KEY"): try: from fetch_use import fetch_sync return fetch_sync(url, headers=headers, timeout_ms=int(timeout * 1000)).text except ImportError: pass import gzip h = {"User-Agent": "Mozilla/5.0", "Accept-Encoding": "gzip"} if headers: h.update(headers) with urllib.request.urlopen(urllib.request.Request(url, headers=h), timeout=timeout) as r: data = r.read() if r.headers.get("Content-Encoding") == "gzip": data = gzip.decompress(data) return data.decode() def _load_agent_helpers(): p = AGENT_WORKSPACE / "agent_helpers.py" if not p.exists(): return spec = importlib.util.spec_from_file_location("browser_harness_agent_helpers", p) if not spec or not spec.loader: return module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) for name, value in vars(module).items(): if name.startswith("_"): continue globals()[name] = value _load_agent_helpers()