browser-use__browser-use/browser_use/skill_cli/actions.py

"""Direct action execution for CLI daemon — no event bus dispatch.

Wraps DefaultActionWatchdog methods and DomService for direct calling.
The watchdog instance is NOT registered on the event bus — it's just
used as a library of action implementations.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any

from bubus import EventBus

from browser_use.browser.events import (
	GoBackEvent,
	SelectDropdownOptionEvent,
	SendKeysEvent,
	TypeTextEvent,
	UploadFileEvent,
)
from browser_use.browser.watchdogs.default_action_watchdog import DefaultActionWatchdog
from browser_use.dom.service import DomService
from browser_use.dom.views import EnhancedDOMTreeNode, SerializedDOMState

if TYPE_CHECKING:
	from browser_use.browser.session import BrowserSession
	from browser_use.browser.views import BrowserStateSummary, PageInfo

logger = logging.getLogger('browser_use.skill_cli.actions')


class ActionHandler:
	"""Execute browser actions directly without the event bus.

	Uses DefaultActionWatchdog methods for complex actions (click, type, keys, etc.)
	and DomService for DOM snapshots. All other actions use direct CDP calls.
	"""

	def __init__(self, browser_session: BrowserSession) -> None:
		self.bs = browser_session
		# Create watchdog instance — NOT registered on event bus
		self._watchdog = DefaultActionWatchdog(
			event_bus=EventBus(),  # dummy, never dispatched to
			browser_session=browser_session,
		)
		self._dom_service: DomService | None = None

	async def navigate(self, url: str) -> None:
		"""Navigate the focused tab to a URL."""
		assert self.bs.agent_focus_target_id is not None, 'No focused tab'
		await self.bs._navigate_and_wait(url, self.bs.agent_focus_target_id)

	async def click_element(self, node: EnhancedDOMTreeNode) -> dict[str, Any] | None:
		"""Click an element using the watchdog's full implementation (with fallbacks)."""
		return await self._watchdog._click_element_node_impl(node)

	async def click_coordinate(self, x: int, y: int) -> dict[str, Any] | None:
		"""Click at coordinates."""
		from browser_use.browser.events import ClickCoordinateEvent

		event = ClickCoordinateEvent(coordinate_x=x, coordinate_y=y)
		return await self._watchdog.on_ClickCoordinateEvent(event)

	async def type_text(self, node: EnhancedDOMTreeNode, text: str) -> dict[str, Any] | None:
		"""Type text into an element."""
		event = TypeTextEvent(node=node, text=text)
		return await self._watchdog.on_TypeTextEvent(event)

	async def scroll(self, direction: str, amount: int) -> None:
		"""Scroll the page using JS (CDP gesture doesn't work in --connect mode)."""
		if direction in ('down', 'up'):
			x, y = 0, (amount if direction == 'down' else -amount)
		else:
			x, y = (amount if direction == 'right' else -amount), 0
		cdp_session = await self.bs.get_or_create_cdp_session()
		assert cdp_session is not None, 'No CDP session for scroll'
		await cdp_session.cdp_client.send.Runtime.evaluate(
			params={'expression': f'window.scrollBy({x}, {y})', 'awaitPromise': False},
			session_id=cdp_session.session_id,
		)

	async def go_back(self) -> None:
		"""Go back in history."""
		event = GoBackEvent()
		await self._watchdog.on_GoBackEvent(event)

	async def send_keys(self, keys: str) -> None:
		"""Send keyboard keys."""
		event = SendKeysEvent(keys=keys)
		await self._watchdog.on_SendKeysEvent(event)

	async def select_dropdown(self, node: EnhancedDOMTreeNode, text: str) -> dict[str, str]:
		"""Select a dropdown option."""
		event = SelectDropdownOptionEvent(node=node, text=text)
		return await self._watchdog.on_SelectDropdownOptionEvent(event)

	async def upload_file(self, node: EnhancedDOMTreeNode, file_path: str) -> None:
		"""Upload a file to a file input element."""
		event = UploadFileEvent(node=node, file_path=file_path)
		await self._watchdog.on_UploadFileEvent(event)

	async def get_state(self) -> BrowserStateSummary:
		"""Build DOM via DomService directly (no DOMWatchdog, no event bus)."""
		from browser_use.browser.views import BrowserStateSummary, PageInfo

		if self._dom_service is None:
			self._dom_service = DomService(browser_session=self.bs)

		page_url = await self.bs.get_current_page_url()

		# Fast path for non-http pages
		if page_url.lower().split(':', 1)[0] not in ('http', 'https'):
			return BrowserStateSummary(
				dom_state=SerializedDOMState(_root=None, selector_map={}),
				url=page_url,
				title='Empty Tab',
				tabs=await self.bs.get_tabs(),
				screenshot=None,
				page_info=None,
			)

		# Build DOM and take screenshot in parallel
		import asyncio

		dom_task = asyncio.create_task(self._dom_service.get_serialized_dom_tree())
		screenshot_task = asyncio.create_task(self.bs.take_screenshot())

		dom_state: SerializedDOMState | None = None
		screenshot_b64: str | None = None

		try:
			dom_state, _tree, _timing = await dom_task
		except Exception as e:
			logger.warning(f'DOM build failed: {e}')
			dom_state = SerializedDOMState(_root=None, selector_map={})

		try:
			screenshot_bytes = await screenshot_task
			import base64

			screenshot_b64 = base64.b64encode(screenshot_bytes).decode() if screenshot_bytes else None
		except Exception as e:
			logger.warning(f'Screenshot failed: {e}')

		# Update cached selector map for element lookups
		if dom_state and dom_state.selector_map:
			self.bs.update_cached_selector_map(dom_state.selector_map)

		# Get page info
		page_info: PageInfo | None = None
		try:
			cdp_session = await self.bs.get_or_create_cdp_session(target_id=None, focus=False)
			if cdp_session:
				metrics = await cdp_session.cdp_client.send.Page.getLayoutMetrics(session_id=cdp_session.session_id)
				css_metrics = metrics.get('cssLayoutViewport', {})
				content_size = metrics.get('cssContentSize', metrics.get('contentSize', {}))
				visual_viewport = metrics.get('cssVisualViewport', metrics.get('visualViewport', {}))
				page_info = PageInfo(
					viewport_width=int(css_metrics.get('clientWidth', 0)),
					viewport_height=int(css_metrics.get('clientHeight', 0)),
					page_width=int(content_size.get('width', 0)),
					page_height=int(content_size.get('height', 0)),
					scroll_x=int(visual_viewport.get('pageX', 0)),
					scroll_y=int(visual_viewport.get('pageY', 0)),
					pixels_above=int(visual_viewport.get('pageY', 0)),
					pixels_below=max(
						0,
						int(content_size.get('height', 0))
						- int(css_metrics.get('clientHeight', 0))
						- int(visual_viewport.get('pageY', 0)),
					),
					pixels_left=0,
					pixels_right=0,
				)
		except Exception as e:
			logger.debug(f'Failed to get page info: {e}')

		tabs = await self.bs.get_tabs()

		# Use focused tab's title, not tabs[0]
		title = ''
		focused_id = self.bs.agent_focus_target_id
		found_focused = False
		for tab in tabs:
			if tab.target_id == focused_id:
				title = tab.title
				found_focused = True
				break
		if not found_focused and tabs:
			title = tabs[0].title

		return BrowserStateSummary(
			dom_state=dom_state,
			url=page_url,
			title=title,
			tabs=tabs,
			screenshot=screenshot_b64,
			page_info=page_info,
			closed_popup_messages=self.bs._closed_popup_messages.copy(),
		)