diff --git a/tests/tools/test_browser_coordinate_click.py b/tests/tools/test_browser_coordinate_click.py new file mode 100644 index 000000000000..2f583693bd50 --- /dev/null +++ b/tests/tools/test_browser_coordinate_click.py @@ -0,0 +1,474 @@ +"""Tests for native (compositor-level) ref-based click in browser_click. + +Covers: +- browser_click requires a ref +- Private-page action guard blocks the click (regression test for #62991 review) +- CDP native click path (resolve ref box -> Input.dispatchMouseEvent at center) +- agent-browser mouse fallback path (no CDP endpoint) +- box-resolution failure degrades gracefully to plain ref click +- Camofox passthrough still works with ref +- Schema reflects ref-only (no x/y) +- Session caching + stale-session reattach +""" +from __future__ import annotations + +import asyncio +import json +import threading +from typing import Any, Dict, List +import pytest + +import websockets +from websockets.asyncio.server import serve + + +class _CDPServer: + """Tiny CDP mock - replies to registered method handlers.""" + + def __init__(self) -> None: + self._handlers: Dict[str, Any] = {} + self._responses: List[Dict[str, Any]] = [] + self._loop: asyncio.AbstractEventLoop | None = None + self._server: Any = None + self._thread: threading.Thread | None = None + self._host = "127.0.0.1" + self._port = 0 + self._url: str = "" + + def on(self, method: str, handler): + self._handlers[method] = handler + + def start(self) -> str: + ready = threading.Event() + + def _run() -> None: + self._loop = asyncio.new_event_loop() + asyncio.set_event_loop(self._loop) + + async def _handler(ws): + try: + async for raw in ws: + msg = json.loads(raw) + call_id = msg.get("id") + method = msg.get("method", "") + params = msg.get("params", {}) or {} + session_id = msg.get("sessionId") + self._responses.append(msg) + + fn = self._handlers.get(method) + if fn is None: + reply = { + "id": call_id, + "error": {"code": -32601, "message": f"No handler for {method}"}, + } + else: + try: + result = fn(params, session_id) + reply = {"id": call_id, "result": result} + except Exception as exc: + reply = {"id": call_id, "error": {"code": -1, "message": str(exc)}} + if session_id: + reply["sessionId"] = session_id + await ws.send(json.dumps(reply)) + except websockets.exceptions.ConnectionClosed: + pass + + async def _serve() -> None: + self._server = await serve(_handler, self._host, 0) + sock = next(iter(self._server.sockets)) + self._port = sock.getsockname()[1] + ready.set() + await self._server.wait_closed() + + try: + self._loop.run_until_complete(_serve()) + finally: + self._loop.close() + + self._thread = threading.Thread(target=_run, daemon=True) + self._thread.start() + if not ready.wait(timeout=5.0): + raise RuntimeError("CDP mock server failed to start") + self._url = f"ws://{self._host}:{self._port}/devtools/browser/mock" + return self._url + + def stop(self) -> None: + if self._loop and self._server: + self._loop.call_soon_threadsafe(self._server.close) + if self._thread: + self._thread.join(timeout=3.0) + + def received(self) -> List[Dict[str, Any]]: + return list(self._responses) + + +@pytest.fixture +def cdp_server(monkeypatch): + server = _CDPServer() + ws_url = server.start() + + import tools.browser_cdp_tool as cdp_mod + monkeypatch.setattr(cdp_mod, "_resolve_cdp_endpoint", lambda: ws_url) + + from tools import browser_tool as _bt + _bt._CDP_SESSION_CACHE.clear() + + try: + yield server + finally: + _bt._CDP_SESSION_CACHE.clear() + server.stop() + + +def _wire_cdp_click_handlers(server: _CDPServer) -> None: + server.on( + "Target.getTargets", + lambda p, s: { + "targetInfos": [ + {"targetId": "page-1", "type": "page", "attached": True, "url": "https://example.com"}, + ] + }, + ) + server.on("Target.attachToTarget", lambda p, s: {"sessionId": f"sess-{p['targetId']}"}) + server.on("Input.dispatchMouseEvent", lambda p, s: {}) + + +def _mock_ref_box(monkeypatch, x: float, y: float, w: float, h: float) -> None: + from tools import browser_tool + + def mock_run_cmd(task_id, command, args=None, timeout=None): + if command == "get" and args and args[0] == "box": + return {"success": True, "data": {"x": x, "y": y, "width": w, "height": h}} + return {"success": True} + + monkeypatch.setattr(browser_tool, "_run_browser_command", mock_run_cmd) + monkeypatch.setattr(browser_tool, "_last_session_key", lambda tid: tid) + monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False) + + +def _box_data(x: float, y: float, w: float, h: float) -> dict: + return {"success": True, "data": {"x": x, "y": y, "width": w, "height": h}} + + +class TestClickInputValidation: + def test_missing_ref(self): + from tools.browser_tool import browser_click + + result = json.loads(browser_click()) + assert result["success"] is False + assert "ref" in result["error"].lower() + + def test_empty_ref_treated_as_missing(self): + from tools.browser_tool import browser_click + + result = json.loads(browser_click(ref="")) + assert result["success"] is False + assert "ref" in result["error"].lower() + + +class TestPrivatePageGuard: + def test_guard_blocks_native_click(self, monkeypatch): + from tools import browser_tool + + monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False) + monkeypatch.setattr(browser_tool, "_last_session_key", lambda tid: tid) + monkeypatch.setattr( + browser_tool, "_blocked_private_page_action", + lambda tid, action: json.dumps({"success": False, "error": "Blocked: private page"}), + ) + commands = [] + + def mock_run_cmd(task_id, command, args=None, timeout=None): + commands.append((command, args)) + return {"success": True, "data": {"x": 0, "y": 0, "width": 10, "height": 10}} + + monkeypatch.setattr(browser_tool, "_run_browser_command", mock_run_cmd) + + result = json.loads(browser_tool.browser_click(ref="@e5")) + assert result["success"] is False + assert "Blocked" in result["error"] + assert commands == [] + + +class TestCDPNativeClick: + def test_cdp_click_dispatches_press_and_release_at_center(self, cdp_server, monkeypatch): + from tools.browser_tool import browser_click + + _wire_cdp_click_handlers(cdp_server) + _mock_ref_box(monkeypatch, 100.0, 200.0, 40.0, 20.0) + + result = json.loads(browser_click(ref="@e1")) + assert result["success"] is True + assert result["clicked"] == "@e1" + assert result["clicked_at"] == {"x": 120, "y": 210} + assert result["method"] == "cdp_native" + + calls = cdp_server.received() + methods = [c["method"] for c in calls] + assert "Target.getTargets" in methods + assert "Input.dispatchMouseEvent" in methods + + mouse_events = [c for c in calls if c["method"] == "Input.dispatchMouseEvent"] + assert len(mouse_events) == 2 + assert mouse_events[0]["params"]["type"] == "mousePressed" + assert mouse_events[0]["params"]["x"] == 120 + assert mouse_events[0]["params"]["y"] == 210 + assert mouse_events[0]["params"]["button"] == "left" + assert mouse_events[1]["params"]["type"] == "mouseReleased" + + def test_cdp_click_rounds_center(self, cdp_server, monkeypatch): + from tools.browser_tool import browser_click + + _wire_cdp_click_handlers(cdp_server) + _mock_ref_box(monkeypatch, 10.2, 10.7, 3.0, 3.0) + + result = json.loads(browser_click(ref="@e1")) + assert result["success"] is True + assert result["clicked_at"] == {"x": 12, "y": 12} + + def test_cdp_dispatch_failure_returns_error(self, cdp_server, monkeypatch): + from tools.browser_tool import browser_click + + _wire_cdp_click_handlers(cdp_server) + cdp_server._handlers.pop("Input.dispatchMouseEvent", None) + _mock_ref_box(monkeypatch, 0.0, 0.0, 10.0, 10.0) + + result = json.loads(browser_click(ref="@e1")) + assert result["success"] is False + assert "CDP native click failed" in result["error"] + + +class TestAgentBrowserMouseFallback: + def test_falls_back_to_agent_browser_mouse(self, monkeypatch): + from tools import browser_tool, browser_cdp_tool + + monkeypatch.setattr(browser_cdp_tool, "_resolve_cdp_endpoint", lambda: "") + _mock_ref_box(monkeypatch, 100.0, 200.0, 40.0, 20.0) + + commands_sent = [] + + def mock_run_cmd(task_id, command, args=None, timeout=None): + if command == "get" and args and args[0] == "box": + return _box_data(100.0, 200.0, 40.0, 20.0) + commands_sent.append((command, args)) + return {"success": True} + + monkeypatch.setattr(browser_tool, "_run_browser_command", mock_run_cmd) + + result = json.loads(browser_tool.browser_click(ref="@e1")) + assert result["success"] is True + assert result["clicked_at"] == {"x": 120, "y": 210} + assert result["method"] == "agent_browser_mouse" + + assert commands_sent[0] == ("mouse", ["move", "120", "210"]) + assert commands_sent[1] == ("mouse", ["down"]) + assert commands_sent[2] == ("mouse", ["up"]) + + def test_mouse_down_failure_returns_error(self, monkeypatch): + from tools import browser_tool, browser_cdp_tool + + monkeypatch.setattr(browser_cdp_tool, "_resolve_cdp_endpoint", lambda: "") + _mock_ref_box(monkeypatch, 100.0, 200.0, 40.0, 20.0) + + def mock_run_cmd(task_id, command, args=None, timeout=None): + if command == "get" and args and args[0] == "box": + return _box_data(100.0, 200.0, 40.0, 20.0) + if command == "mouse" and args and args[0] == "down": + return {"success": False, "error": "mouse down failed"} + return {"success": True} + + monkeypatch.setattr(browser_tool, "_run_browser_command", mock_run_cmd) + + result = json.loads(browser_tool.browser_click(ref="@e1")) + assert result["success"] is False + assert "mouse down" in result["error"] + + +class TestBoxResolutionFailure: + def test_missing_size_falls_back(self, monkeypatch): + """If width/height is missing or zero, fall back to plain ref click.""" + from tools import browser_tool, browser_cdp_tool + + monkeypatch.setattr(browser_cdp_tool, "_resolve_cdp_endpoint", lambda: "") + monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False) + monkeypatch.setattr(browser_tool, "_last_session_key", lambda tid: tid) + + commands = [] + + def mock_run_cmd(task_id, command, args=None, timeout=None): + commands.append((command, args)) + if command == "get" and args and args[0] == "box": + return {"success": True, "data": {"x": 100.0, "y": 200.0, "width": 0, "height": 0}} + return {"success": True} + + monkeypatch.setattr(browser_tool, "_run_browser_command", mock_run_cmd) + + result = json.loads(browser_tool.browser_click(ref="@e5")) + assert result["success"] is True + assert result["method"] == "agent_browser_ref" + assert ("click", ["@e5"]) in commands + + def test_falls_back_to_plain_ref_click(self, monkeypatch): + from tools import browser_tool, browser_cdp_tool + + monkeypatch.setattr(browser_cdp_tool, "_resolve_cdp_endpoint", lambda: "") + monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False) + monkeypatch.setattr(browser_tool, "_last_session_key", lambda tid: tid) + + commands = [] + + def mock_run_cmd(task_id, command, args=None, timeout=None): + commands.append((command, args)) + if command == "get" and args and args[0] == "box": + return {"success": False, "error": "element not found"} + return {"success": True} + + monkeypatch.setattr(browser_tool, "_run_browser_command", mock_run_cmd) + + result = json.loads(browser_tool.browser_click(ref="@e9")) + assert result["success"] is True + assert result["clicked"] == "@e9" + assert result["method"] == "agent_browser_ref" + assert ("click", ["@e9"]) in commands + + +class TestRefClickPlumbing: + def test_ref_without_at_prefix_auto_added(self, monkeypatch): + from tools import browser_tool, browser_cdp_tool + + monkeypatch.setattr(browser_cdp_tool, "_resolve_cdp_endpoint", lambda: "") + monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False) + monkeypatch.setattr(browser_tool, "_last_session_key", lambda tid: tid) + + mouse_calls = [] + + def mock_run_cmd(task_id, command, args=None, timeout=None): + if command == "get" and args and args[0] == "box": + return _box_data(0.0, 0.0, 1.0, 1.0) + if command == "mouse": + mouse_calls.append(args) + return {"success": True} + + monkeypatch.setattr(browser_tool, "_run_browser_command", mock_run_cmd) + + browser_tool.browser_click(ref="e12") + # Native path: ref normalized to @e12; mouse click dispatched at the + # resolved center (box 0,0,1,1 -> center 0,0). + assert mouse_calls[0] == ["move", "0", "0"] + assert mouse_calls[1] == ["down"] + assert mouse_calls[2] == ["up"] + + def test_camofox_passthrough(self, monkeypatch): + from tools import browser_tool + import tools.browser_camofox as camofox_mod + + monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: True) + + captured = {} + + def mock_camofox_click(ref, task_id): + captured["ref"] = ref + return json.dumps({"success": True, "clicked": ref}) + + # browser_click does `from tools.browser_camofox import camofox_click` + monkeypatch.setattr(camofox_mod, "camofox_click", mock_camofox_click) + + result = json.loads(browser_tool.browser_click(ref="@e3")) + assert result["success"] is True + assert captured["ref"] == "@e3" + + +class TestSchemaUpdated: + def test_schema_has_only_ref_property(self): + from tools.browser_tool import _BROWSER_SCHEMA_MAP + + schema = _BROWSER_SCHEMA_MAP["browser_click"] + props = schema["parameters"]["properties"] + assert "ref" in props + assert "x" not in props + assert "y" not in props + + def test_ref_is_required(self): + from tools.browser_tool import _BROWSER_SCHEMA_MAP + + schema = _BROWSER_SCHEMA_MAP["browser_click"] + assert schema["parameters"]["required"] == ["ref"] + + +class TestRegistryIntegration: + def test_dispatch_with_ref(self, monkeypatch, cdp_server): + from tools.registry import registry + + _wire_cdp_click_handlers(cdp_server) + _mock_ref_box(monkeypatch, 50.0, 60.0, 20.0, 20.0) + + raw = registry.dispatch("browser_click", {"ref": "@e3"}, task_id="t1") + result = json.loads(raw) + assert result["success"] is True + assert result["clicked_at"] == {"x": 60, "y": 70} + + +class TestSessionCaching: + def test_second_click_skips_session_resolution(self, cdp_server, monkeypatch): + from tools import browser_tool + import tools.browser_cdp_tool as cdp_mod + + browser_tool._CDP_SESSION_CACHE.clear() + monkeypatch.setattr(cdp_mod, "_resolve_cdp_endpoint", lambda: cdp_server._url) + _mock_ref_box(monkeypatch, 0.0, 0.0, 10.0, 10.0) + + resolve_count = {"n": 0} + + def _getTargets(p, s): + resolve_count["n"] += 1 + return {"targetInfos": [{"targetId": "p1", "type": "page", "attached": True, "url": "..."}]} + + cdp_server.on("Target.getTargets", _getTargets) + cdp_server.on("Target.attachToTarget", lambda p, s: {"sessionId": "sess-cached"}) + cdp_server.on("Input.dispatchMouseEvent", lambda p, s: {}) + + r1 = json.loads(browser_tool.browser_click(ref="@e1")) + assert r1["success"] is True + assert resolve_count["n"] == 1 + + r2 = json.loads(browser_tool.browser_click(ref="@e2")) + assert r2["success"] is True + assert resolve_count["n"] == 1, "session resolution was repeated despite warm cache" + + def test_stale_session_triggers_reattach(self, cdp_server, monkeypatch): + from tools import browser_tool + import tools.browser_cdp_tool as cdp_mod + + browser_tool._CDP_SESSION_CACHE.clear() + monkeypatch.setattr(cdp_mod, "_resolve_cdp_endpoint", lambda: cdp_server._url) + _mock_ref_box(monkeypatch, 0.0, 0.0, 10.0, 10.0) + + call_count = {"mouse": 0, "resolve": 0} + + def _getTargets(p, s): + call_count["resolve"] += 1 + return {"targetInfos": [{"targetId": "px", "type": "page", "attached": True, "url": "..."}]} + + def _dispatch(p, s): + call_count["mouse"] += 1 + if call_count["mouse"] <= 2: + raise RuntimeError("Session with given id not found: stale-session-id") + return {} + + cdp_server.on("Target.getTargets", _getTargets) + cdp_server.on("Target.attachToTarget", lambda p, s: {"sessionId": f"sess-{call_count['resolve']}"}) + cdp_server.on("Input.dispatchMouseEvent", _dispatch) + + browser_tool._CDP_SESSION_CACHE[(cdp_server._url, "default")] = "stale-session-id" + + r = json.loads(browser_tool.browser_click(ref="@e1")) + assert r["success"] is True + assert call_count["resolve"] >= 1 + + def test_cache_cleared_on_endpoint_change(self, monkeypatch): + from tools import browser_tool + + browser_tool._CDP_SESSION_CACHE.clear() + browser_tool._CDP_SESSION_CACHE[("ws://endpoint-a/", "task-a")] = "sess-a" + + assert browser_tool._CDP_SESSION_CACHE.get(("ws://endpoint-a/", "task-b")) is None diff --git a/tests/tools/test_browser_private_page_action_guard.py b/tests/tools/test_browser_private_page_action_guard.py index ff01d26b036a..0281c838b441 100644 --- a/tests/tools/test_browser_private_page_action_guard.py +++ b/tests/tools/test_browser_private_page_action_guard.py @@ -42,21 +42,35 @@ def fail_run(*_args, **_kwargs): def test_click_still_runs_when_current_page_is_public(monkeypatch): + """Guard allows the native click when the current page is public. + + A1 dispatch: resolve the ref's box (get box) then a native mouse click at + the center. The guard must NOT block a public page. + """ calls = [] monkeypatch.setattr(browser_tool, "_eval_ssrf_guard_active", lambda task_id: True) monkeypatch.setattr(browser_tool, "_current_page_private_url", lambda task_id: None) + import tools.browser_cdp_tool as cdp_mod + monkeypatch.setattr(cdp_mod, "_resolve_cdp_endpoint", lambda: "") + monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False) def fake_run(task_id, command, args): calls.append((task_id, command, args)) + if command == "get" and args and args[0] == "box": + return {"success": True, "data": {"x": 0, "y": 0, "width": 10, "height": 10}} return {"success": True} monkeypatch.setattr(browser_tool, "_run_browser_command", fake_run) out = json.loads(browser_tool.browser_click("e1", task_id="task-1")) - assert out == {"success": True, "clicked": "@e1"} - assert calls == [("task-1", "click", ["@e1"])] + # Guard passed -> native click dispatched; ref normalized to @e1. + assert out["success"] is True + assert out["clicked"] == "@e1" + # box resolution + mouse move/down/up, with the normalized ref. + assert calls[0] == ("task-1", "get", ["box", "@e1"]) + assert calls[-1] == ("task-1", "mouse", ["up"]) def test_guard_inactive_does_not_block_or_probe(monkeypatch): @@ -67,6 +81,9 @@ def test_guard_inactive_does_not_block_or_probe(monkeypatch): calls = [] monkeypatch.setattr(browser_tool, "_eval_ssrf_guard_active", lambda task_id: False) + import tools.browser_cdp_tool as cdp_mod + monkeypatch.setattr(cdp_mod, "_resolve_cdp_endpoint", lambda: "") + monkeypatch.setattr(browser_tool, "_is_camofox_mode", lambda: False) def fail_probe(task_id): raise AssertionError("_current_page_private_url must not be probed when guard inactive") @@ -75,14 +92,18 @@ def fail_probe(task_id): def fake_run(task_id, command, args): calls.append((task_id, command, args)) + if command == "get" and args and args[0] == "box": + return {"success": True, "data": {"x": 0, "y": 0, "width": 10, "height": 10}} return {"success": True} monkeypatch.setattr(browser_tool, "_run_browser_command", fake_run) out = json.loads(browser_tool.browser_click("@e1", task_id="task-1")) - assert out == {"success": True, "clicked": "@e1"} - assert calls == [("task-1", "click", ["@e1"])] + # Guard inactive -> native click proceeds, no URL probe attempted. + assert out["success"] is True + assert out["clicked"] == "@e1" + assert calls[0] == ("task-1", "get", ["box", "@e1"]) def test_camofox_short_circuits_before_guard(monkeypatch): diff --git a/tools/browser_tool.py b/tools/browser_tool.py index 82c248a3f715..fa536291a3a0 100644 --- a/tools/browser_tool.py +++ b/tools/browser_tool.py @@ -1848,7 +1848,7 @@ def _update_session_activity(task_id: str): }, { "name": "browser_click", - "description": "Click on an element identified by its ref ID from the snapshot (e.g., '@e5'). The ref IDs are shown in square brackets in the snapshot output. Requires browser_navigate and browser_snapshot to be called first.", + "description": "Click on an element identified by its ref ID from the snapshot (e.g., '@e5'). The ref IDs are shown in square brackets in the snapshot output. Requires browser_navigate and browser_snapshot to be called first.\n\nThe click is dispatched as a real compositor-level (native) mouse event at the element's location so React onChange/onClick handlers fire the same way a real user click does.", "parameters": { "type": "object", "properties": { @@ -3017,22 +3017,337 @@ def browser_snapshot( return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False) -def browser_click(ref: str, task_id: Optional[str] = None) -> str: +# --------------------------------------------------------------------------- +# Session cache for CDP coordinate clicks +# +# Target.getTargets + Target.attachToTarget cost one round-trip each and +# their results (page targetId + session_id) are stable across clicks on +# the same page. We cache them keyed by CDP endpoint URL and invalidate +# automatically when the browser reports a stale session error. +# +# Pattern: browser-harness daemon keeps session_id on the daemon object and +# retries once on "Session with given id not found" to self-heal after +# navigation or crash. We replicate that here without a persistent daemon +# process by storing it in a module-level dict. +# --------------------------------------------------------------------------- + +_CDP_SESSION_CACHE: dict[tuple[str, str], str] = {} # (ws_url, task_id) → cached session_id + + +async def _cdp_resolve_session( + ws: Any, + ws_url: str, + task_id: str, + deadline: float, + msg_id_ref: list, +) -> Optional[str]: + """Resolve (and cache) the page-scoped CDP session ID. + + Sends Target.getTargets + Target.attachToTarget on *ws* and caches the + resulting session_id for future clicks. Returns None if no page target + is found (Input.dispatchMouseEvent will be sent at browser level, which + works for simple cases). The cache is keyed by (ws_url, task_id) so + concurrent tasks sharing one browser endpoint don't collide. + """ + import asyncio as _asyncio + + async def _send(method: str, params: dict, sid: Optional[str] = None) -> int: + msg_id_ref[0] += 1 + call_id = msg_id_ref[0] + req: dict = {"id": call_id, "method": method, "params": params} + if sid: + req["sessionId"] = sid + await ws.send(json.dumps(req)) + return call_id + + async def _recv_until(call_id: int) -> dict: + while True: + remaining = deadline - _asyncio.get_running_loop().time() + if remaining <= 0: + raise TimeoutError(f"CDP timed out waiting for id={call_id}") + raw = await _asyncio.wait_for(ws.recv(), timeout=remaining) + msg = json.loads(raw) + if msg.get("id") == call_id: + if "error" in msg: + raise RuntimeError(f"CDP error: {msg['error']}") + return msg.get("result", {}) + + gt_id = await _send("Target.getTargets", {}) + gt_result = await _recv_until(gt_id) + page_target_id: Optional[str] = None + for t in gt_result.get("targetInfos", []): + if t.get("type") == "page" and t.get("attached", True): + page_target_id = t["targetId"] + break + + if not page_target_id: + return None + + at_id = await _send("Target.attachToTarget", + {"targetId": page_target_id, "flatten": True}) + at_result = await _recv_until(at_id) + session_id = at_result.get("sessionId") or None + if session_id: + _CDP_SESSION_CACHE[(ws_url, task_id)] = session_id + return session_id + + +async def _cdp_native_click_async( + ws_url: str, + x: int, + y: int, + task_id: str, + button: str, + timeout: float, +) -> None: + """Dispatch a compositor-level (native, trusted) mouse click on a single + persistent CDP WebSocket connection. + + Trusted CDP ``Input.dispatchMouseEvent`` events are the same low-level + input Chrome generates for a real user click, so React ``onChange`` / + ``onClick`` handlers fire — unlike synthetic ``.click()`` calls. + """ + import asyncio as _asyncio + from tools.browser_cdp_tool import websockets as _ws + + async with _ws.connect( + ws_url, + max_size=None, + open_timeout=timeout, + close_timeout=5, + ping_interval=None, + compression=None, + ) as ws: + deadline = _asyncio.get_running_loop().time() + timeout + msg_id_ref = [0] + + def _next_id() -> int: + msg_id_ref[0] += 1 + return msg_id_ref[0] + + async def _send_mouse(event_type: str, sid: Optional[str]) -> int: + call_id = _next_id() + req: dict = { + "id": call_id, + "method": "Input.dispatchMouseEvent", + "params": {"type": event_type, "x": x, "y": y, + "button": button, "clickCount": 1}, + } + if sid: + req["sessionId"] = sid + await ws.send(json.dumps(req)) + return call_id + + async def _recv_until(call_id: int) -> dict: + while True: + remaining = deadline - _asyncio.get_running_loop().time() + if remaining <= 0: + raise TimeoutError(f"CDP timed out waiting for id={call_id}") + raw = await _asyncio.wait_for(ws.recv(), timeout=remaining) + msg = json.loads(raw) + if msg.get("id") == call_id: + if "error" in msg: + raise RuntimeError(f"CDP error: {msg['error']}") + return msg.get("result", {}) + + session_id: Optional[str] = _CDP_SESSION_CACHE.get((ws_url, task_id)) + if not session_id: + session_id = await _cdp_resolve_session(ws, ws_url, task_id, deadline, msg_id_ref) + + _press_id = await _send_mouse("mousePressed", session_id) + release_id = await _send_mouse("mouseReleased", session_id) + try: + await _recv_until(_press_id) + await _recv_until(release_id) + except RuntimeError as exc: + if "Session with given id not found" in str(exc) and session_id: + _CDP_SESSION_CACHE.pop((ws_url, task_id), None) + session_id = await _cdp_resolve_session(ws, ws_url, task_id, deadline, msg_id_ref) + _press_id = await _send_mouse("mousePressed", session_id) + release_id = await _send_mouse("mouseReleased", session_id) + await _recv_until(_press_id) + await _recv_until(release_id) + else: + raise + + +def _native_click_via_agent_browser( + x: float, + y: float, + task_id: str, + button: str = "left", + ref_for_response: str = "", +) -> str: + """Native click fallback via agent-browser low-level mouse subcommands. + + ``agent-browser mouse move/down/up`` drives real input at viewport + coordinates, which also produces trusted events (unlike ``click ``). + Used when no CDP endpoint is configured. """ - Click on an element. + effective_task_id = _last_session_key(task_id) + ix, iy = int(round(x)), int(round(y)) + + move_result = _run_browser_command(effective_task_id, "mouse", ["move", str(ix), str(iy)]) + if not move_result.get("success"): + return json.dumps({ + "success": False, + "error": f"mouse move failed: {move_result.get('error', 'unknown')}", + }, ensure_ascii=False) + + btn_arg = [] if button == "left" else [button] + down_result = _run_browser_command(effective_task_id, "mouse", ["down"] + btn_arg) + if not down_result.get("success"): + return json.dumps({ + "success": False, + "error": f"mouse down failed: {down_result.get('error', 'unknown')}", + }, ensure_ascii=False) + + up_result = _run_browser_command(effective_task_id, "mouse", ["up"] + btn_arg) + if not up_result.get("success"): + return json.dumps({ + "success": False, + "error": f"mouse up failed: {up_result.get('error', 'unknown')}", + }, ensure_ascii=False) + + return json.dumps({ + "success": True, + "clicked": ref_for_response, + "clicked_at": {"x": ix, "y": iy}, + "method": "agent_browser_mouse", + }, ensure_ascii=False) + + +def _resolve_ref_box(ref: str, task_id: str) -> Optional[dict]: + """Resolve a snapshot ref to its viewport bounding box via agent-browser. + + Returns ``{"x": cx, "y": cy}`` (element center) or ``None`` if the box + can't be resolved (element gone, off-screen, etc.). + """ + effective_task_id = _last_session_key(task_id) + box_result = _run_browser_command(effective_task_id, "get", ["box", ref]) + if not box_result.get("success"): + return None + data = box_result.get("data", {}) + # agent-browser returns x/y/width/height (numeric) or a string blob. + try: + if isinstance(data, dict): + # Require at least x and y; reject missing/zero size so we don't + # click at the top-left corner from a malformed box result. + if not any(k in data for k in ("x", "left")): + return None + if not any(k in data for k in ("y", "top")): + return None + x = float(data.get("x", data.get("left", 0))) + y = float(data.get("y", data.get("top", 0))) + w = float(data.get("width", 0)) + h = float(data.get("height", 0)) + elif isinstance(data, str): + # Parse "x=.. y=.. width=.. height=.." style output defensively. + parts = dict( + kv.split("=") for kv in data.replace(",", " ").split() + if "=" in kv + ) + if "x" not in parts or "y" not in parts: + return None + x = float(parts.get("x", 0)) + y = float(parts.get("y", 0)) + w = float(parts.get("width", 0)) + h = float(parts.get("height", 0)) + else: + return None + except (TypeError, ValueError): + return None + if w <= 0 or h <= 0: + return None + return {"x": x + w / 2.0, "y": y + h / 2.0} + + +def _native_click(ref: str, task_id: str, button: str = "left") -> str: + """Click ``ref`` with a real compositor-level (native, trusted) mouse + event so React handlers fire. + + Resolution order: + 1. Resolve the element's viewport center via ``agent-browser get box``. + 2. If a CDP endpoint is available, dispatch ``Input.dispatchMouseEvent`` + at that center (highest fidelity, trusted events). + 3. Else fall back to ``agent-browser mouse move/down/up`` at the center. + 4. If the box can't be resolved, fall back to the plain ``click `` + command so behavior degrades gracefully (no regression). + """ + center = _resolve_ref_box(ref, task_id) + if center is None: + # Can't get coordinates — use the plain ref click as a safe fallback. + result = _run_browser_command(_last_session_key(task_id), "click", [ref]) + if result.get("success"): + response = {"success": True, "clicked": ref, "method": "agent_browser_ref"} + return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False) + response = {"success": False, "error": result.get("error", f"Failed to click {ref}")} + return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False) + + cx, cy = center["x"], center["y"] + ix, iy = int(round(cx)), int(round(cy)) + + try: + from tools.browser_cdp_tool import _run_async, _resolve_cdp_endpoint, _WS_AVAILABLE + except ImportError: + return _native_click_via_agent_browser(cx, cy, task_id, button, ref) + + endpoint = _resolve_cdp_endpoint() + if not endpoint or not endpoint.startswith(("ws://", "wss://")): + return _native_click_via_agent_browser(cx, cy, task_id, button, ref) + + if not _WS_AVAILABLE: + return _native_click_via_agent_browser(cx, cy, task_id, button, ref) + + try: + _run_async(_cdp_native_click_async(endpoint, ix, iy, task_id, button, 10.0)) + except Exception as exc: + return json.dumps({ + "success": False, + "error": f"CDP native click failed: {type(exc).__name__}: {exc}", + }, ensure_ascii=False) + + return json.dumps({ + "success": True, + "clicked": ref, + "clicked_at": {"x": ix, "y": iy}, + "method": "cdp_native", + }, ensure_ascii=False) + + +def browser_click( + ref: Optional[str] = None, + task_id: Optional[str] = None, +) -> str: + """Click on an element identified by its ref ID from the snapshot. + + The click is dispatched as a real compositor-level (native, trusted) mouse + event at the element's location, so React ``onChange``/``onClick`` handlers + fire the same way a real user click does (unlike synthetic ``.click()`` + calls, which React ignores). Args: - ref: Element reference (e.g., "@e5") + ref: Element reference from the snapshot (e.g., "@e5") task_id: Task identifier for session isolation Returns: JSON string with click result """ + has_ref = ref is not None and str(ref).strip() != "" + + if not has_ref: + return json.dumps({ + "success": False, + "error": "Provide a 'ref' (element reference from the snapshot, e.g. '@e5').", + }, ensure_ascii=False) + if _is_camofox_mode(): from tools.browser_camofox import camofox_click return camofox_click(ref, task_id) effective_task_id = _last_session_key(task_id or "default") + + # --- Private-page action guard (must wrap ALL click dispatch) ----------- blocked = _blocked_private_page_action(effective_task_id, "click") if blocked is not None: return blocked @@ -3041,20 +3356,7 @@ def browser_click(ref: str, task_id: Optional[str] = None) -> str: if not ref.startswith("@"): ref = f"@{ref}" - result = _run_browser_command(effective_task_id, "click", [ref]) - - if result.get("success"): - response = { - "success": True, - "clicked": ref - } - return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False) - else: - response = { - "success": False, - "error": result.get("error", f"Failed to click {ref}") - } - return json.dumps(_copy_fallback_warning(response, result), ensure_ascii=False) + return _native_click(ref, task_id or "default") def browser_type(ref: str, text: str, task_id: Optional[str] = None) -> str: @@ -4740,7 +5042,7 @@ def check_browser_vision_requirements() -> bool: name="browser_click", toolset="browser", schema=_BROWSER_SCHEMA_MAP["browser_click"], - handler=lambda args, **kw: browser_click(ref=args.get("ref", ""), task_id=kw.get("task_id")), + handler=lambda args, **kw: browser_click(ref=args.get("ref"), task_id=kw.get("task_id")), check_fn=check_browser_requirements, emoji="👆", )