diff --git a/src/leapflow/config.py b/src/leapflow/config.py index 6839f22..fdacb5f 100644 --- a/src/leapflow/config.py +++ b/src/leapflow/config.py @@ -487,7 +487,6 @@ class Settings: # ── Cua Driver ── use_cua_driver: bool = True cua_driver_cmd: str = "cua-driver" - desktop_tools_enabled: bool = True # ── Workflow Copilot ── copilot_enabled: bool = True @@ -996,7 +995,6 @@ def _tuple_env(key: str, default: tuple) -> tuple: # Cua Driver use_cua_driver = _bool("LEAPFLOW_USE_CUA_DRIVER", "true") cua_driver_cmd = os.getenv("LEAPFLOW_CUA_DRIVER_CMD", "cua-driver").strip() - desktop_tools_enabled = _bool("LEAPFLOW_DESKTOP_TOOLS_ENABLED", "true") # Workflow Copilot copilot_enabled = _bool("LEAPFLOW_COPILOT_ENABLED", "true") @@ -1322,7 +1320,6 @@ def _tuple_env(key: str, default: tuple) -> tuple: # Cua Driver use_cua_driver=use_cua_driver, cua_driver_cmd=cua_driver_cmd, - desktop_tools_enabled=desktop_tools_enabled, # Workflow Copilot copilot_enabled=copilot_enabled, copilot_min_idle_ms=copilot_min_idle_ms, diff --git a/src/leapflow/domain/__init__.py b/src/leapflow/domain/__init__.py index 5e655ec..c3a57f3 100644 --- a/src/leapflow/domain/__init__.py +++ b/src/leapflow/domain/__init__.py @@ -8,7 +8,7 @@ UIActionSubType, UNDO_SHORTCUTS, ) -from leapflow.domain.events import SystemEvent, UINode +from leapflow.domain.events import SystemEvent, UIElement, UISnapshot from leapflow.domain.platform import ( Capability, DEFAULT_DARWIN_CAPABILITIES, @@ -56,7 +56,8 @@ "SystemEvent", "Trajectory", "TrajectoryStep", - "UINode", + "UIElement", + "UISnapshot", "action_type_from_event", "capability_from_str", ] diff --git a/src/leapflow/domain/events.py b/src/leapflow/domain/events.py index 7feea49..629710e 100644 --- a/src/leapflow/domain/events.py +++ b/src/leapflow/domain/events.py @@ -3,7 +3,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Any, AsyncIterator, Dict, List, Optional, Protocol, runtime_checkable +from typing import Any, AsyncIterator, Dict, List, Optional, Protocol, Tuple, runtime_checkable # ── Event priority levels ── # Higher value = higher urgency. Used by downstream consumers (queues, @@ -37,18 +37,58 @@ class SystemEvent: priority: int = PRIORITY_NORMAL -@dataclass -class UINode: - """Normalized UI element tree node.""" +@dataclass(frozen=True) +class UIElement: + """One actionable UI element row from a window snapshot. + + Mirrors the driver's get_window_state element record. ``element_index`` + and ``element_token`` are the action addressing handles; the token is + preferred because the driver validates its staleness on every action. + """ - node_id: str + element_index: int role: str - label: str + label: str = "" value: str = "" - children: List["UINode"] = field(default_factory=list) - actions: List[str] = field(default_factory=list) + element_token: str = "" + enabled: bool = True + selected: Optional[bool] = None + depth: int = 0 + parent_index: Optional[int] = None frame: Optional[Dict[str, float]] = None - ax_props: Dict[str, Any] = field(default_factory=dict) + + @property + def target(self) -> str: + """Preferred action target: element_token, else the index.""" + return self.element_token or str(self.element_index) + + +@dataclass(frozen=True) +class UISnapshot: + """Immutable snapshot of one window's actionable elements. + + A snapshot is scoped to (pid, window_id) and superseded by the next + read of the same window; ``elements_complete`` and ``coverage`` carry + the driver's own statements about what this snapshot cannot see + (e.g. browser page content in window scope). + """ + + pid: int + window_id: int + snapshot_id: str = "" + elements: Tuple[UIElement, ...] = () + elements_complete: bool = True + total_element_count: int = 0 + degraded: bool = False + degraded_reason: str = "" + coverage: Dict[str, Any] = field(default_factory=dict) + + def find(self, element_index: int) -> Optional[UIElement]: + """Look up an element by its index.""" + for element in self.elements: + if element.element_index == element_index: + return element + return None @runtime_checkable @@ -57,7 +97,11 @@ class PerceptionPort(Protocol): async def subscribe_fs(self, paths: List[str]) -> str: ... - async def read_ui_tree(self, app_id: Optional[str] = None) -> UINode: ... + async def read_window_state( + self, pid: int, window_id: int, query: str = "" + ) -> UISnapshot: ... + + async def list_windows(self) -> Dict[str, Any]: ... async def get_clipboard(self) -> Dict[str, Any]: ... @@ -76,9 +120,13 @@ async def perform_ui_action( self, node_id: str, action: str, params: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: ... - async def launch_app(self, app_id: str) -> Dict[str, Any]: ... + async def launch_app( + self, app_id: str, urls: Optional[List[str]] = None + ) -> Dict[str, Any]: ... - async def activate_app(self, app_id: str) -> Dict[str, Any]: ... + async def activate_app( + self, pid: int, window_id: Optional[int] = None + ) -> Dict[str, Any]: ... async def run_intent( self, intent_name: str, params: Dict[str, Any] @@ -88,7 +136,7 @@ async def exec_shell(self, command: str) -> Dict[str, Any]: ... async def set_clipboard(self, text: str) -> Dict[str, Any]: ... - async def type_text(self, text: str, method: str = "paste") -> Dict[str, Any]: ... + async def type_text(self, text: str) -> Dict[str, Any]: ... async def send_shortcut(self, keys: str) -> Dict[str, Any]: ... diff --git a/src/leapflow/domain/ui_vocabulary.py b/src/leapflow/domain/ui_vocabulary.py index 0884fe6..db55d8e 100644 --- a/src/leapflow/domain/ui_vocabulary.py +++ b/src/leapflow/domain/ui_vocabulary.py @@ -1,43 +1,17 @@ -"""Shared UI vocabulary — role classifications and action type mappings. +"""Shared UI vocabulary — ActionType ↔ tool name mappings. -This module is the single source of truth for UI element semantics used by -both the Recording pipeline (EventNormalizer → ActionAbstractor) and the -Execution pipeline (SemanticAdapter → UITreeSummarizer). Keeping these -constants in one place ensures learn→run semantic coherence. +This module connects the Recording vocabulary (ActionType enum values) +with the Execution vocabulary (tool names registered in bridge_factory), +keeping learn→run semantic coherence. -Architecture: - Recording (forward): raw AX role → classify → filter/weight in analysis - Execution (reverse): AX role → classify → filter/prioritize in summarizer - Both share the same classification, preventing vocabulary drift. +Role classification tables were retired with the tree summarizer: the +driver's get_window_state already returns the filtered, actionable-only +element list, so no execution-side role filtering remains. """ from __future__ import annotations -from typing import Dict, FrozenSet - - -# ═══════════════════════════════════════════════════════════════════════════ -# Role classifications — what kind of UI element is this? -# ═══════════════════════════════════════════════════════════════════════════ - -INTERACTIVE_ROLES: FrozenSet[str] = frozenset({ - "AXButton", "AXTextField", "AXTextArea", "AXLink", - "AXMenuItem", "AXCheckBox", "AXRadioButton", "AXPopUpButton", - "AXTab", "AXSlider", "AXComboBox", "AXDisclosureTriangle", - "AXIncrementor", "AXColorWell", "AXMenuButton", -}) - -LAYOUT_ROLES: FrozenSet[str] = frozenset({ - "AXGroup", "AXScrollArea", "AXSplitGroup", "AXLayoutArea", - "AXList", "AXOutline", "AXTable", "AXRow", "AXColumn", - "AXBrowser", "AXScrollBar", "AXRuler", "AXGrowArea", - "AXMatte", "AXSplitter", -}) - -STRUCTURAL_ROLES: FrozenSet[str] = frozenset({ - "AXWindow", "AXSheet", "AXDialog", "AXToolbar", - "AXMenuBar", "AXMenu", -}) +from typing import Dict # ═══════════════════════════════════════════════════════════════════════════ @@ -56,7 +30,7 @@ "ui.shortcut": "shortcut", "clipboard.copy": "get_clipboard", "app.switch": "switch_app", - "ui.scroll": "shortcut", + "ui.scroll": "scroll", } TOOL_TO_ACTION: Dict[str, str] = { @@ -68,6 +42,7 @@ "switch_app": "app.switch", "open_url": "app.switch", "observe_ui": "ui.click", + "scroll": "ui.scroll", } diff --git a/src/leapflow/engine/engine.py b/src/leapflow/engine/engine.py index 6239450..0803905 100644 --- a/src/leapflow/engine/engine.py +++ b/src/leapflow/engine/engine.py @@ -4320,11 +4320,8 @@ def _semantic_tool_schemas(self) -> List[Dict[str, Any]]: Empty when the bridge is absent or carries no semantic tools (perception offline) — the bridge itself is the dynamic on/off switch. Cached by bridge object identity so a hot-swapped bridge rebuilds on - first access. ``desktop_tools_enabled`` is the process-level master - switch (off in the journey harness to keep replayed prompts stable). + first access. """ - if not getattr(self._settings, "desktop_tools_enabled", True): - return [] if self._tool_bridge is None: return [] if self._semantic_schema_bridge is not self._tool_bridge: @@ -4359,8 +4356,7 @@ def _unified_tool_handlers(self) -> Dict[str, Any]: from leapflow.skills.semantic_schema import build_semantic_handlers handlers: Dict[str, Any] = dict(TOOL_HANDLERS) - if getattr(self._settings, "desktop_tools_enabled", True): - handlers.update(build_semantic_handlers(self._tool_bridge)) + handlers.update(build_semantic_handlers(self._tool_bridge)) return handlers async def _approve_desktop_action(self, name: str, args: Any) -> tuple[bool, str]: diff --git a/src/leapflow/learning/codegen.py b/src/leapflow/learning/codegen.py index 496663f..04cf310 100644 --- a/src/leapflow/learning/codegen.py +++ b/src/leapflow/learning/codegen.py @@ -120,8 +120,10 @@ async def generate(self, candidate: Any, context: CodeGenContext) -> Optional[Ge ### PerceptionPort (read-only system observation) - `await perception.subscribe_fs(paths: List[str]) -> str` Subscribe to filesystem changes at given paths. Returns subscription ID. - - `await perception.read_ui_tree(app_id: Optional[str] = None) -> UINode` - Read the accessibility tree of the focused app (or specified app). + - `await perception.read_window_state(pid: int, window_id: int, query: str = "") -> UISnapshot` + Snapshot one window's actionable elements (flat, element_index-addressed). + - `await perception.list_windows() -> Dict[str, Any]` + List top-level windows with pid/window_id/title records. - `await perception.get_clipboard() -> Dict[str, Any]` Read current clipboard content. Returns {"text": ..., "type": ...}. - `async for event in perception.stream_events() -> AsyncIterator[SystemEvent]` @@ -132,9 +134,10 @@ async def generate(self, candidate: Any, context: CodeGenContext) -> Optional[Ge File operations. op: "copy"|"move"|"create"|"delete"|"rename" params: {"source": path, "destination": path} or {"path": path, "content": str} - `await execution.perform_ui_action(node_id: str, action: str, params: Optional[Dict] = None) -> Dict` - Interact with UI elements. action: "click"|"type"|"select"|"scroll" - - `await execution.launch_app(app_id: str) -> Dict[str, Any]` - Launch an application by bundle ID. + Interact with UI elements. action: "press"|"type_text"|"set_value"|"show_menu"|"double_click" + - `await execution.launch_app(app_id: str, urls: Optional[List[str]] = None) -> Dict[str, Any]` + Launch an application (backgrounded). Optional urls are file paths/URLs + the app opens on launch. Returns the launched pid and windows records. - `await execution.run_intent(intent_name: str, params: Dict[str, Any]) -> Dict[str, Any]` Execute a system intent (share, open-with, etc.). - `await execution.exec_shell(command: str) -> Dict[str, Any]` @@ -752,12 +755,13 @@ def build_default_context( return CodeGenContext( available_ports=[ "PerceptionPort.subscribe_fs(paths: List[str]) -> str", - "PerceptionPort.read_ui_tree(app_id: Optional[str]) -> UINode", + "PerceptionPort.read_window_state(pid: int, window_id: int, query: str = '') -> UISnapshot", + "PerceptionPort.list_windows() -> Dict[str, Any]", "PerceptionPort.get_clipboard() -> Dict[str, Any]", "PerceptionPort.stream_events() -> AsyncIterator[SystemEvent]", "ExecutionPort.perform_file_op(op: str, params: Dict) -> Dict", "ExecutionPort.perform_ui_action(node_id: str, action: str, params: Optional[Dict]) -> Dict", - "ExecutionPort.launch_app(app_id: str) -> Dict", + "ExecutionPort.launch_app(app_id: str, urls: Optional[List[str]] = None) -> Dict", "ExecutionPort.run_intent(intent_name: str, params: Dict) -> Dict", "ExecutionPort.exec_shell(command: str) -> Dict", ], diff --git a/src/leapflow/perception/state_snapshot.py b/src/leapflow/perception/state_snapshot.py index 8d39ea1..803c0d0 100644 --- a/src/leapflow/perception/state_snapshot.py +++ b/src/leapflow/perception/state_snapshot.py @@ -11,9 +11,8 @@ import time from dataclasses import dataclass from enum import IntEnum -from typing import Any, Dict, Optional, Tuple +from typing import Dict, Optional, Tuple -from leapflow.domain.events import UINode from leapflow.memory.providers.episodic import EpisodicMemoryProvider from leapflow.platform.protocol import HostRpc @@ -166,11 +165,13 @@ async def _get_clipboard(self) -> str: async def _get_ax_info(self, app_id: str) -> tuple[str, str]: try: + # NOTE: ax.tree now requires pid/window_id targets; without them + # the call returns a structured error and this snapshot facet + # degrades to empty digests. App→pid resolution is a follow-up. tree = await self._rpc.call("ax.tree", {"app_id": app_id} if app_id else None) if isinstance(tree, dict): - node = _dict_to_ui_node(tree) - digest = _compute_ax_digest(node) - summary = _compute_ax_summary(node) + digest = _compute_ax_digest(tree) + summary = _compute_ax_summary(tree) return digest, summary except Exception: logger.debug("ax.tree failed for snapshot", exc_info=True) @@ -186,50 +187,39 @@ async def _get_screenshot_phash(self) -> str: return "" -def _compute_ax_digest(node: UINode, max_depth: int = 3, max_width: int = 5) -> str: - """Compress AX tree into a structural fingerprint.""" +def _compute_ax_digest(payload: dict, max_items: int = 40) -> str: + """Compress a window-state payload into a structural fingerprint.""" + elements = payload.get("elements") + if not isinstance(elements, list): + return "" parts: list[str] = [] - - def _walk(n: UINode, depth: int = 0) -> None: - if depth > max_depth: - return - label_part = n.label[:20] if n.label else "" - parts.append(f"{n.role}:{label_part}") - for child in (n.children or [])[:max_width]: - _walk(child, depth + 1) - - _walk(node) + for record in elements[:max_items]: + if not isinstance(record, dict): + continue + label = str(record.get("label", "") or "")[:20] + parts.append(f"{record.get('role', '')}:{label}") + if not parts: + return "" return hashlib.md5("|".join(parts).encode()).hexdigest()[:16] -def _compute_ax_summary(node: UINode, max_items: int = 8) -> str: - """Generate a brief natural-language summary of the AX tree.""" +def _compute_ax_summary(payload: dict, max_items: int = 8) -> str: + """Generate a brief natural-language summary of a window-state payload.""" + elements = payload.get("elements") + if not isinstance(elements, list): + return "" items: list[str] = [] - - def _walk(n: UINode, depth: int = 0) -> None: - if len(items) >= max_items or depth > 2: - return - if n.label: - items.append(f"{n.role}({n.label})") - for child in (n.children or [])[:4]: - _walk(child, depth + 1) - - _walk(node) + for record in elements: + if len(items) >= max_items: + break + if not isinstance(record, dict): + continue + label = str(record.get("label", "") or "") + if label: + items.append(f"{record.get('role', '')}({label})") return ", ".join(items) -def _dict_to_ui_node(d: Any) -> UINode: - """Recursively convert a dict to UINode.""" - children = [_dict_to_ui_node(c) for c in (d.get("children") or [])] - return UINode( - node_id=d.get("node_id", ""), - role=d.get("role", ""), - label=d.get("label", ""), - value=d.get("value", ""), - children=children, - ) - - def _hamming_distance(a: str, b: str) -> int: """Hamming distance between two hex-encoded hashes.""" if len(a) != len(b): diff --git a/src/leapflow/platform/adapters/darwin.py b/src/leapflow/platform/adapters/darwin.py index 175c9f3..0667d94 100644 --- a/src/leapflow/platform/adapters/darwin.py +++ b/src/leapflow/platform/adapters/darwin.py @@ -4,12 +4,15 @@ import asyncio import logging +import os +import subprocess +import tempfile import uuid from collections import deque from pathlib import Path from typing import Any, AsyncIterator, Dict, List, Optional -from leapflow.domain.events import SystemEvent, UINode +from leapflow.domain.events import SystemEvent, UIElement, UISnapshot from leapflow.domain.platform import Capability, PlatformManifest from leapflow.platform.protocol import HostRpc, Methods @@ -28,27 +31,47 @@ async def subscribe_fs(self, paths: List[str]) -> str: result = await self._rpc.call(Methods.FS_SUBSCRIBE, {"path": paths[0] if paths else "~"}) return str(result.get("subscription_id", "")) - async def read_ui_tree(self, app_id: Optional[str] = None) -> UINode: - params: Dict[str, Any] = {} - if app_id: - params["bundle_id"] = app_id - - if self._manifest.supports(Capability.APP_INTENTS_DISCOVER): - params["prefer_intents"] = True - + async def read_window_state( + self, pid: int, window_id: int, query: str = "" + ) -> UISnapshot: + params: Dict[str, Any] = { + "pid": pid, + "window_id": window_id, + "include_screenshot": False, + } + if query: + params["query"] = query result = await self._rpc.call(Methods.AX_TREE, params) - return _parse_ui_tree(result.get("root", {})) + payload = result if isinstance(result, dict) else {} + return _snapshot_from_payload(payload, pid=pid, window_id=window_id) + + async def list_windows(self) -> Dict[str, Any]: + """List top-level windows; the source of pid/window_id targets.""" + result = await self._rpc.call(Methods.AX_LIST, {}) + return result if isinstance(result, dict) else {"windows": result} async def get_clipboard(self) -> Dict[str, Any]: - return await self._rpc.call(Methods.CLIPBOARD_GET, {}) + result = await self._rpc.call(Methods.CLIPBOARD_GET, {}) + if isinstance(result, dict): + return result + return {"text": str(result or ""), "change_count": 0, "change_ts": None} - async def capture_screenshot(self, region: str = "", app_id: str = "") -> Dict[str, Any]: - params: Dict[str, Any] = {} - if app_id: - params["bundle_id"] = app_id - elif region: - params["region"] = region - return await self._rpc.call(Methods.SCREEN_CAPTURE_FRAME, params) + async def capture_screenshot( + self, pid: Optional[int] = None, window_id: Optional[int] = None + ) -> Dict[str, Any]: + # Route the image to disk: base64 payloads must never enter context. + out_file = str( + Path(tempfile.gettempdir()) / f"leapflow_screenshot_{uuid.uuid4().hex[:8]}.png" + ) + params: Dict[str, Any] = {"screenshot_out_file": out_file} + if pid is not None and window_id is not None: + params["pid"] = pid + params["window_id"] = window_id + result = await self._rpc.call(Methods.SCREEN_CAPTURE_FRAME, params) + if not isinstance(result, dict): + return {"ok": True, "path": out_file} + result.setdefault("path", result.get("screenshot_file_path") or out_file) + return result async def stream_events(self) -> AsyncIterator[SystemEvent]: while True: @@ -128,8 +151,14 @@ async def perform_ui_action( {"node_id": node_id, "action": action, **(params or {})}, ) - async def launch_app(self, app_id: str) -> Dict[str, Any]: - return await self._rpc.call(Methods.APP_LAUNCH, {"bundle_id": app_id}) + async def launch_app( + self, app_id: str, urls: Optional[List[str]] = None + ) -> Dict[str, Any]: + params: Dict[str, Any] = {"bundle_id": app_id} + if urls: + params["urls"] = list(urls) + result = await self._rpc.call(Methods.APP_LAUNCH, params) + return result if isinstance(result, dict) else {"ok": True, "result": result} async def run_intent(self, intent_name: str, params: Dict[str, Any]) -> Dict[str, Any]: if not self._manifest.supports(Capability.APP_INTENTS_PERFORM): @@ -138,48 +167,89 @@ async def run_intent(self, intent_name: str, params: Dict[str, Any]) -> Dict[str "intent.perform", {"intent": intent_name, "params": params} ) - async def activate_app(self, app_id: str) -> Dict[str, Any]: - return await self._rpc.call(Methods.APP_ACTIVATE, {"bundle_id": app_id}) + async def activate_app( + self, pid: int, window_id: Optional[int] = None + ) -> Dict[str, Any]: + params: Dict[str, Any] = {"pid": pid} + if window_id is not None: + params["window_id"] = window_id + return await self._rpc.call(Methods.APP_ACTIVATE, params) async def open_url(self, url: str) -> Dict[str, Any]: """Open a URL in the default browser (local OS dispatch).""" result = await self._rpc.call(Methods.OPEN_URL, {"url": url}) return result if isinstance(result, dict) else {"ok": True, "result": result} - async def list_apps(self, filter: str = "", running_only: bool = False) -> Dict[str, Any]: + async def list_apps(self) -> Dict[str, Any]: """List available applications on the system.""" - return await self._rpc.call( - Methods.APP_LIST, {"filter": filter, "running_only": running_only} - ) + return await self._rpc.call(Methods.APP_LIST, {}) async def exec_shell(self, command: str) -> Dict[str, Any]: - return await self._rpc.call( - Methods.AX_PERFORM, - {"commands": [{"type": "shell", "cmd": command}]}, - ) + """Run a shell command locally — cua-driver exposes no shell tool.""" + timeout = float(os.environ.get("LEAPFLOW_SHELL_TIMEOUT", "60.0")) + try: + proc = await asyncio.create_subprocess_shell( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for( + proc.communicate(), timeout=timeout + ) + except asyncio.TimeoutError: + proc.kill() + return { + "ok": False, + "error": f"timeout after {timeout}s", + "stdout": "", + "stderr": "", + } + return { + "ok": proc.returncode == 0, + "stdout": stdout.decode(errors="replace"), + "stderr": stderr.decode(errors="replace"), + "exit_code": proc.returncode, + } + except Exception as exc: # noqa: BLE001 - boundary: report, never crash the turn + return {"ok": False, "error": str(exc), "stdout": "", "stderr": ""} async def set_clipboard(self, text: str) -> Dict[str, Any]: return await self._rpc.call(Methods.CLIPBOARD_SET, {"text": text}) - async def type_text(self, text: str, method: str = "paste") -> Dict[str, Any]: - return await self._rpc.call( - Methods.INPUT_TYPE_TEXT, {"text": text, "method": method} - ) + async def type_text(self, text: str) -> Dict[str, Any]: + return await self._rpc.call(Methods.INPUT_TYPE_TEXT, {"text": text}) async def send_shortcut(self, keys: str) -> Dict[str, Any]: return await self._rpc.call(Methods.INPUT_SHORTCUT, {"keys": keys}) - async def scroll(self, node_id: str, delta_x: int, delta_y: int) -> Dict[str, Any]: - return await self._rpc.call(Methods.AX_SCROLL, { - "node_id": node_id, "delta_x": delta_x, "delta_y": delta_y, - }) - - async def capture_screenshot(self, region: str = "", app_id: str = "") -> Dict[str, Any]: + async def scroll( + self, + node_id: str, + direction: str, + amount: int = 3, + pid: Optional[int] = None, + window_id: Optional[int] = None, + ) -> Dict[str, Any]: + # The driver's keystroke scroll path requires pid even without an + # element target (verified on 0.19.3 despite the doc marking it + # optional) — it must know which process receives the keystrokes. + params: Dict[str, Any] = {"direction": direction, "amount": amount} + if node_id: + params["node_id"] = node_id + if pid is not None: + params["pid"] = pid + if window_id is not None: + params["window_id"] = window_id + return await self._rpc.call(Methods.AX_SCROLL, params) + + async def capture_screenshot( + self, pid: Optional[int] = None, window_id: Optional[int] = None + ) -> Dict[str, Any]: params: Dict[str, Any] = {} - if app_id: - params["bundle_id"] = app_id - elif region: - params["region"] = region + if pid is not None and window_id is not None: + params["pid"] = pid + params["window_id"] = window_id return await self._rpc.call(Methods.SCREEN_CAPTURE_FRAME, params) async def undo(self, steps: int = 1) -> List[Dict[str, Any]]: @@ -198,7 +268,9 @@ async def undo_last(self) -> Dict[str, Any]: async def _backup_for_undo(self, path: str) -> str: """Copy file to temp before deletion for potential recovery.""" filename = Path(path).name if path else "file" - backup = f"/tmp/leap_undo_{uuid.uuid4().hex[:8]}_{filename}" + backup = str( + Path(tempfile.gettempdir()) / f"leap_undo_{uuid.uuid4().hex[:8]}_{filename}" + ) try: await self._rpc.call(Methods.FILE_COPY, {"source": path, "destination": backup}) except Exception as exc: @@ -237,28 +309,60 @@ async def _reverse_op(self, record: Dict[str, Any]) -> Dict[str, Any]: return {"ok": False, "error": f"not_reversible:{op_type}"} -def _parse_ui_tree(raw: Dict[str, Any]) -> UINode: - """Recursively parse a raw AX tree dict into UINode.""" - children_raw = raw.get("children", []) - children = [_parse_ui_tree(c) for c in children_raw] if children_raw else [] +def _snapshot_from_payload( + payload: Dict[str, Any], *, pid: int, window_id: int +) -> UISnapshot: + """Parse a get_window_state payload into a flat UISnapshot. - frame = raw.get("frame") - frame_dict = ( - {"x": frame["x"], "y": frame["y"], "w": frame["w"], "h": frame["h"]} - if isinstance(frame, dict) - else None - ) - - ax_props_raw = raw.get("ax_props") or {} - ax_props = ax_props_raw if isinstance(ax_props_raw, dict) else {} - - return UINode( - node_id=str(raw.get("id", raw.get("role", ""))), - role=str(raw.get("role", "")), - label=str(raw.get("title", "")), - value=str(raw.get("value", "")), - children=children, - actions=raw.get("actions", []), - frame=frame_dict, - ax_props=ax_props, + The driver already returns the filtered, actionable-only element list; + records are taken verbatim. ``elements_complete`` and ``capture_coverage`` + are the driver's own statements about blind spots (e.g. browser page + content is not observable in window scope) and must reach the caller. + """ + raw_elements = payload.get("elements") + records = [r for r in raw_elements if isinstance(r, dict)] if isinstance( + raw_elements, list + ) else [] + + elements: List[UIElement] = [] + for record in records: + index = record.get("element_index") + if not isinstance(index, int): + continue + frame_raw = record.get("frame") + frame = ( + {"x": float(frame_raw["x"]), "y": float(frame_raw["y"]), + "w": float(frame_raw["w"]), "h": float(frame_raw["h"])} + if isinstance(frame_raw, dict) + and all(k in frame_raw for k in ("x", "y", "w", "h")) + else None + ) + value = record.get("value") + parent = record.get("parent_index") + selected = record.get("selected") + elements.append(UIElement( + element_index=index, + role=str(record.get("role", "") or ""), + label=str(record.get("label", "") or ""), + value="" if value is None else str(value), + element_token=str(record.get("element_token", "") or ""), + enabled=bool(record.get("enabled", True)), + selected=bool(selected) if isinstance(selected, bool) else None, + depth=record.get("depth", 0) if isinstance(record.get("depth"), int) else 0, + parent_index=parent if isinstance(parent, int) else None, + frame=frame, + )) + + coverage = payload.get("capture_coverage") + total = payload.get("total_element_count", payload.get("element_count", len(elements))) + return UISnapshot( + pid=pid, + window_id=window_id, + snapshot_id=str(payload.get("snapshot_id", "") or ""), + elements=tuple(elements), + elements_complete=bool(payload.get("elements_complete", True)), + total_element_count=total if isinstance(total, int) else len(elements), + degraded=bool(payload.get("degraded", False)), + degraded_reason=str(payload.get("degraded_reason", "") or ""), + coverage=coverage if isinstance(coverage, dict) else {}, ) diff --git a/src/leapflow/platform/adapters/mock.py b/src/leapflow/platform/adapters/mock.py index 8a35840..434bb3c 100644 --- a/src/leapflow/platform/adapters/mock.py +++ b/src/leapflow/platform/adapters/mock.py @@ -8,7 +8,7 @@ import time from typing import Any, AsyncIterator, Dict, List, Optional -from leapflow.domain.events import SystemEvent, UINode +from leapflow.domain.events import SystemEvent, UIElement, UISnapshot _URL_OPEN_RE = re.compile( r"^\s*(?:open|xdg-open|start)\s+['\"]?https?://", re.IGNORECASE @@ -29,31 +29,72 @@ def __init__(self) -> None: async def subscribe_fs(self, paths: List[str]) -> str: return "mock-sub-001" - async def read_ui_tree(self, app_id: Optional[str] = None) -> UINode: - return UINode( - node_id="mock-root", - role="AXWindow", - label="Mock Window", - children=[ - UINode(node_id="mock-btn-save", role="AXButton", label="Save", actions=["AXPress"]), - UINode(node_id="mock-btn-cancel", role="AXButton", label="Cancel", actions=["AXPress"]), - UINode( - node_id="mock-scroll-area", - role="AXScrollArea", - label="", - children=[ - UINode(node_id="mock-text-field", role="AXTextField", label="Input", value="hello"), - ], - ), - ], - actions=["AXPress", "AXRaise"], + async def read_window_state( + self, pid: int, window_id: int, query: str = "" + ) -> UISnapshot: + elements = ( + UIElement( + element_index=0, role="Button", label="Save", + element_token="s00000001:0", depth=1, + frame={"x": 10.0, "y": 10.0, "w": 80.0, "h": 24.0}, + ), + UIElement( + element_index=1, role="Button", label="Cancel", + element_token="s00000001:1", depth=1, + frame={"x": 100.0, "y": 10.0, "w": 80.0, "h": 24.0}, + ), + UIElement( + element_index=2, role="Edit", label="Input", value="hello", + element_token="s00000001:2", depth=2, + ), + UIElement( + element_index=3, role="TabItem", label="Home", + element_token="s00000001:3", depth=2, selected=True, + ), + ) + if query: + needle = query.lower() + elements = tuple( + el for el in elements + if needle in el.label.lower() or needle in el.role.lower() + ) + return UISnapshot( + pid=pid, + window_id=window_id, + snapshot_id="s00000001", + elements=elements, + elements_complete=True, + total_element_count=4, ) + async def list_windows(self) -> Dict[str, Any]: + return { + "windows": [ + { + "window_id": 1, + "pid": 100, + "app_name": "Mock App", + "title": "Mock Window", + "bounds": {"x": 0, "y": 0, "width": 800, "height": 600}, + "z_index": 0, + "is_on_screen": True, + "minimized": False, + } + ] + } + async def get_clipboard(self) -> Dict[str, Any]: return {"text": "", "change_count": 0, "change_ts": time.time()} - async def capture_screenshot(self, region: str = "") -> Dict[str, Any]: - return {"ok": True, "path": "/tmp/mock_screenshot.png", "region": region} + async def capture_screenshot( + self, pid: Optional[int] = None, window_id: Optional[int] = None + ) -> Dict[str, Any]: + return { + "ok": True, + "path": "/tmp/mock_screenshot.png", + "pid": pid, + "window_id": window_id, + } async def stream_events(self) -> AsyncIterator[SystemEvent]: while True: @@ -115,26 +156,59 @@ async def perform_ui_action( self.history.append(record) return {"ok": True, "mock": True} - async def launch_app(self, app_id: str) -> Dict[str, Any]: - record = {"type": "launch_app", "app_id": app_id} + async def launch_app( + self, app_id: str, urls: Optional[List[str]] = None + ) -> Dict[str, Any]: + record = {"type": "launch_app", "app_id": app_id, "urls": list(urls or [])} self.history.append(record) - return {"ok": True, "mock": True} + return { + "ok": True, + "mock": True, + "pid": 100, + "bundle_id": app_id, + "launch_state": "window_ready", + "windows": [ + { + "window_id": 1, + "pid": 100, + "app_name": "Mock App", + "title": "Mock Window", + "z_index": 0, + "is_on_screen": True, + "minimized": False, + } + ], + } async def run_intent(self, intent_name: str, params: Dict[str, Any]) -> Dict[str, Any]: record = {"type": "intent", "intent": intent_name, "params": params} self.history.append(record) return {"ok": True, "mock": True, "stub": True} - async def activate_app(self, app_id: str) -> Dict[str, Any]: - record = {"type": "activate_app", "app_id": app_id} + async def activate_app( + self, pid: int, window_id: Optional[int] = None + ) -> Dict[str, Any]: + record = {"type": "activate_app", "pid": pid, "window_id": window_id} self.history.append(record) return {"ok": True, "mock": True} - async def list_apps(self, filter: str = "", running_only: bool = False) -> Dict[str, Any]: - """Return empty app list in mock mode.""" - record = {"type": "list_apps", "filter": filter, "running_only": running_only} + async def list_apps(self) -> Dict[str, Any]: + """Return a deterministic app list mirroring the driver's record shape.""" + record = {"type": "list_apps"} self.history.append(record) - return {"ok": True, "apps": []} + return { + "ok": True, + "apps": [ + { + "name": "Mock App", + "bundle_id": "com.mock.app", + "pid": 100, + "running": True, + "active": True, + "kind": "desktop", + } + ], + } async def exec_shell(self, command: str) -> Dict[str, Any]: record = {"type": "shell", "command": command} @@ -167,8 +241,8 @@ async def set_clipboard(self, text: str) -> Dict[str, Any]: self.history.append(record) return {"ok": True, "mock": True} - async def type_text(self, text: str, method: str = "paste") -> Dict[str, Any]: - record = {"type": "type_text", "text": text, "method": method} + async def type_text(self, text: str) -> Dict[str, Any]: + record = {"type": "type_text", "text": text} self.history.append(record) return {"ok": True, "mock": True} @@ -177,8 +251,18 @@ async def send_shortcut(self, keys: str) -> Dict[str, Any]: self.history.append(record) return {"ok": True, "mock": True} - async def scroll(self, node_id: str, delta_x: int, delta_y: int) -> Dict[str, Any]: - record = {"type": "scroll", "node_id": node_id, "delta_x": delta_x, "delta_y": delta_y} + async def scroll( + self, + node_id: str, + direction: str, + amount: int = 3, + pid: Optional[int] = None, + window_id: Optional[int] = None, + ) -> Dict[str, Any]: + record = { + "type": "scroll", "node_id": node_id, "direction": direction, + "amount": amount, "pid": pid, "window_id": window_id, + } self.history.append(record) return {"ok": True, "mock": True} diff --git a/src/leapflow/platform/cua_client.py b/src/leapflow/platform/cua_client.py index ac9d58d..e148fff 100644 --- a/src/leapflow/platform/cua_client.py +++ b/src/leapflow/platform/cua_client.py @@ -16,11 +16,13 @@ from __future__ import annotations import asyncio +import base64 import concurrent.futures import json import logging import os import platform +import re import shutil import subprocess import sys @@ -467,12 +469,28 @@ def _is_closed_session_error(exc: Exception) -> bool: def _clipboard_get() -> str: """Read clipboard via platform-native command.""" - if sys.platform == "darwin": - cmd = ["pbpaste"] - elif sys.platform == "win32": - cmd = ["powershell", "-command", "Get-Clipboard"] - else: - cmd = ["xclip", "-selection", "clipboard", "-o"] + if sys.platform == "win32": + # -Raw preserves the clipboard text verbatim (the default mode + # returns a line array and reflows newlines). Force UTF-8 output — + # the default console codepage mangles CJK. + cmd = [ + "powershell", "-NoProfile", "-Command", + "[Console]::OutputEncoding=[Text.Encoding]::UTF8; Get-Clipboard -Raw", + ] + try: + result = subprocess.run(cmd, capture_output=True, timeout=5.0) + text = result.stdout.decode("utf-8", errors="replace") + # PowerShell appends exactly one trailing newline to stdout; + # strip only that one so genuine trailing newlines survive. + if text.endswith("\r\n"): + return text[:-2] + if text.endswith("\n"): + return text[:-1] + return text + except Exception as e: + raise RpcError("clipboard_error", f"Failed to read clipboard: {e}", {}) + + cmd = ["pbpaste"] if sys.platform == "darwin" else ["xclip", "-selection", "clipboard", "-o"] try: result = subprocess.run( cmd, capture_output=True, text=True, @@ -485,20 +503,29 @@ def _clipboard_get() -> str: def _clipboard_set(text: str) -> None: """Write clipboard via platform-native command.""" - if sys.platform == "darwin": - cmd = ["pbcopy"] - elif sys.platform == "win32": - cmd = ["powershell", "-command", "Set-Clipboard", "-Value", text] - else: - cmd = ["xclip", "-selection", "clipboard"] + if sys.platform == "win32": + # PowerShell -Command re-parses the joined argv, stripping the + # quoting subprocess added — any text with spaces breaks parameter + # binding. Base64 transport is immune to spaces/quotes/CJK/$. + encoded = base64.b64encode(text.encode("utf-16-le")).decode("ascii") + script = ( + "$t=[Text.Encoding]::Unicode.GetString(" + f"[Convert]::FromBase64String('{encoded}'));" + "if ($t.Length -gt 0) { Set-Clipboard -Value $t }" + " else { Set-Clipboard -Value $null }" + ) + cmd = ["powershell", "-NoProfile", "-Command", script] + try: + subprocess.run(cmd, capture_output=True, timeout=5.0, check=True) + except Exception as e: + raise RpcError("clipboard_error", f"Failed to set clipboard: {e}", {}) + return + cmd = ["pbcopy"] if sys.platform == "darwin" else ["xclip", "-selection", "clipboard"] try: - if sys.platform == "win32": - subprocess.run(cmd, capture_output=True, timeout=5.0, check=True) - else: - subprocess.run( - cmd, input=text, capture_output=True, text=True, timeout=5.0, check=True - ) + subprocess.run( + cmd, input=text, capture_output=True, text=True, timeout=5.0, check=True + ) except Exception as e: raise RpcError("clipboard_error", f"Failed to set clipboard: {e}", {}) @@ -519,14 +546,14 @@ def _file_list(params: Dict[str, Any]) -> List[Dict[str, Any]]: return entries -def _file_move(params: Dict[str, Any]) -> Dict[str, str]: +def _file_move(params: Dict[str, Any]) -> Dict[str, Any]: src = Path(params["source"]) dst = Path(params["destination"]) src.rename(dst) - return {"moved": str(dst)} + return {"ok": True, "moved": str(dst)} -def _file_copy(params: Dict[str, Any]) -> Dict[str, str]: +def _file_copy(params: Dict[str, Any]) -> Dict[str, Any]: import shutil as _shutil src = Path(params["source"]) dst = Path(params["destination"]) @@ -534,17 +561,17 @@ def _file_copy(params: Dict[str, Any]) -> Dict[str, str]: _shutil.copytree(str(src), str(dst)) else: _shutil.copy2(str(src), str(dst)) - return {"copied": str(dst)} + return {"ok": True, "copied": str(dst)} -def _file_delete(params: Dict[str, Any]) -> Dict[str, str]: +def _file_delete(params: Dict[str, Any]) -> Dict[str, Any]: import shutil as _shutil target = Path(params["path"]) if target.is_dir(): _shutil.rmtree(str(target)) else: target.unlink() - return {"deleted": str(target)} + return {"ok": True, "deleted": str(target)} # ── Dispatch helpers ───────────────────────────────────────────────────────── @@ -552,65 +579,115 @@ def _file_delete(params: Dict[str, Any]) -> Dict[str, str]: def _launch_app_key(app: str) -> str: """Pick the launch_app schema field for an app identifier. - cua-driver 0.17 distinguishes AUMIDs (``bundle_id``), executable paths - (``path``), and plain aliases (``name``); sending the wrong one makes - resolution fail. + cua-driver 0.19.3 launch_app accepts ``bundle_id`` (preferred) or + ``name`` only. AUMIDs (``!``) and reverse-DNS identifiers (at least + two dots, no path separators or spaces) are bundle ids; everything + else — display names and executable paths — goes through ``name``. """ if "!" in app: return "bundle_id" - if "/" in app or "\\" in app: - return "path" + if ( + app.count(".") >= 2 + and "/" not in app + and "\\" not in app + and " " not in app + ): + return "bundle_id" return "name" -def _resolve_ax_perform_tool(params: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: - """Map ax.perform params to the appropriate cua-driver tool + args. +# ax.perform action → (cua tool, forced args). Covers the click tool's action +# vocabulary (press/show_menu/pick/confirm/cancel/open), the discrete +# double_click/right_click/type_text/set_value tools, and the legacy AX action +# names emitted by SemanticAdapter. Unknown actions fall back to a plain click. +_AX_ACTION_TABLE: Dict[str, Tuple[str, Dict[str, Any]]] = { + "click": ("click", {}), + "press": ("click", {}), + "AXPress": ("click", {}), + "AXShowDefaultUI": ("click", {}), + "open": ("click", {"action": "open"}), + "pick": ("click", {"action": "pick"}), + "select": ("click", {"action": "pick"}), + "confirm": ("click", {"action": "confirm"}), + "cancel": ("click", {"action": "cancel"}), + "double_click": ("double_click", {}), + "AXOpen": ("double_click", {}), + "right_click": ("right_click", {}), + "show_menu": ("right_click", {}), + "AXShowMenu": ("right_click", {}), + "type": ("type_text", {}), + "type_text": ("type_text", {}), + "set_value": ("set_value", {}), +} + + +def _element_target_args(params: Dict[str, Any]) -> Dict[str, Any]: + """Extract cua-driver element/pixel target args from neutral params. - cua-driver splits AX actions into discrete tools: click, type_text, - set_value. We infer the target from the `action` param. + ``element_token`` is preferred (it carries pid/window/snapshot); an + int-like ``node_id`` is treated as an ``element_index``, anything else + as a token. Pixel coordinates land as separate ``x``/``y`` fields — + 0.19.3 has no ``coordinates`` parameter. """ - action = params.get("action", "") - element_index = params.get("element_index") - element_token = params.get("element_token") - - # Common args shared across tools - base_args: Dict[str, Any] = {} - if element_index is not None: - base_args["element_index"] = element_index - if element_token is not None: - base_args["element_token"] = element_token - - # Delivery mode for Verify-Then-Escalate + args: Dict[str, Any] = {} + + if params.get("element_token"): + args["element_token"] = params["element_token"] + elif params.get("element_index") is not None: + args["element_index"] = params["element_index"] + else: + node_id = str(params.get("node_id", "") or "") + if node_id.isdigit(): + args["element_index"] = int(node_id) + elif node_id: + args["element_token"] = node_id + + for key in ("snapshot_id", "pid", "window_id"): + if key in params: + args[key] = params[key] + + coords = params.get("coordinates") + if isinstance(coords, dict) and "x" in coords and "y" in coords: + args["x"], args["y"] = coords["x"], coords["y"] + elif isinstance(coords, (list, tuple)) and len(coords) == 2: + args["x"], args["y"] = coords[0], coords[1] + for key in ("x", "y"): + if key in params: + args[key] = params[key] + return args + + +def _resolve_ax_perform_tool(params: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: + """Map ax.perform params to the appropriate cua-driver tool + args.""" + action = str(params.get("action", "") or "click") + tool, forced = _AX_ACTION_TABLE.get(action, ("click", {})) + + args = _element_target_args(params) + delivery_mode = params.get("delivery_mode", "background") if delivery_mode != "background": - base_args["delivery_mode"] = delivery_mode - - if action in ("click", "double_click", "right_click"): - args = {**base_args, "action": action} - if "coordinates" in params: - args["coordinates"] = params["coordinates"] - return "click", args - - elif action in ("type", "type_text"): - args = {**base_args} - if "text" in params: - args["text"] = params["text"] - return "type_text", args - - elif action == "set_value": - args = {**base_args} - if "value" in params: - args["value"] = params["value"] - return "set_value", args - - elif action == "select": - args = {**base_args} - return "click", args + args["delivery_mode"] = delivery_mode - else: - # Fallback: pass action directly as a click variant - args = {**base_args, "action": action} - return "click", args + if tool == "type_text" and "text" in params: + args["text"] = params["text"] + if tool == "set_value" and "value" in params: + args["value"] = params["value"] + + args.update(forced) + return tool, args + + +_SHORTCUT_SPLIT_RE = re.compile(r"[+\s]+") + + +def _normalize_shortcut_keys(keys: Any) -> List[str]: + """Normalize a shortcut spec ('cmd+c', 'cmd c', or a list) to a key list.""" + if isinstance(keys, (list, tuple)): + return [str(k).strip() for k in keys if str(k).strip()] + text = str(keys or "").strip() + if not text: + return [] + return [part for part in _SHORTCUT_SPLIT_RE.split(text) if part] # ── CuaDriverClient ────────────────────────────────────────────────────────── @@ -824,22 +901,39 @@ async def _call_cua_tool( def _map_to_cua_tool(self, method: str, params: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]: """Translate a LeapFlow Methods constant to (cua_tool_name, args).""" if method == Methods.AX_TREE: - args: Dict[str, Any] = {} - app = params.get("app") or params.get("bundle_id") - if app: - args["app"] = app - if "window_id" in params: - args["window_id"] = params["window_id"] + if "pid" not in params or "window_id" not in params: + raise RpcError( + "invalid_params", + "ax.tree requires pid and window_id (discover them via ax.list)", + {"provided": sorted(params.keys())}, + ) + args: Dict[str, Any] = { + "pid": params["pid"], + "window_id": params["window_id"], + } + for key in ("query", "include_screenshot", "max_elements", "max_depth"): + if key in params: + args[key] = params[key] return "get_window_state", args + elif method == Methods.AX_LIST: + return "list_windows", {} + elif method == Methods.AX_PERFORM: return _resolve_ax_perform_tool(params) elif method == Methods.AX_SCROLL: - args = {} - for key in ("x", "y", "direction", "amount", "coordinates", "element_index"): - if key in params: - args[key] = params[key] + direction = str(params.get("direction", "") or "") + if direction not in ("up", "down", "left", "right"): + raise RpcError( + "invalid_params", + f"scroll requires direction up/down/left/right, got '{direction}'", + {}, + ) + args = _element_target_args(params) + args["direction"] = direction + if "amount" in params: + args["amount"] = int(params["amount"]) return "scroll", args elif method == Methods.APP_LAUNCH: @@ -847,40 +941,68 @@ def _map_to_cua_tool(self, method: str, params: Dict[str, Any]) -> Tuple[str, Di args: Dict[str, Any] = {} if app: args[_launch_app_key(app)] = app + urls = params.get("urls") + if isinstance(urls, list) and urls: + # Driver-native: file paths/URLs handed to the app as open targets. + args["urls"] = [str(u) for u in urls] return "launch_app", args elif method == Methods.APP_ACTIVATE: - app = params.get("app_name") or params.get("name") or params.get("bundle_id", "") - args = {} - if app: - args[_launch_app_key(app)] = app - return "launch_app", args + # launch_app is explicitly backgrounded; foreground activation + # is bring_to_front, addressed by pid. + if "pid" not in params: + raise RpcError( + "invalid_params", + "app.activate requires pid (from ax.list or launch_app's response)", + {"provided": sorted(params.keys())}, + ) + args = {"pid": params["pid"]} + if "window_id" in params: + args["window_id"] = params["window_id"] + return "bring_to_front", args elif method == Methods.APP_LIST: return "list_apps", {} elif method == Methods.INPUT_TYPE_TEXT: args = {"text": params.get("text", "")} + args.update(_element_target_args(params)) + if "pid" not in args: + # Without a pid/window target, desktop scope is the documented + # way to type into the frontmost application. + args["scope"] = "desktop" return "type_text", args elif method == Methods.INPUT_SHORTCUT: - # Parse key combo into cua-driver hotkey format keys = params.get("keys", params.get("shortcut", "")) - args = {"keys": keys} if isinstance(keys, list) else {"key": keys} - return "hotkey", args + parts = _normalize_shortcut_keys(keys) + if not parts: + raise RpcError("invalid_params", "shortcut requires at least one key", {}) + if len(parts) == 1: + # hotkey requires modifiers + one key (>=2 items); a bare key + # (enter, escape, tab) is a press_key. + args = {"key": parts[0]} + tool = "press_key" + else: + args = {"keys": parts} + tool = "hotkey" + if "pid" in params: + args["pid"] = params["pid"] + else: + args["scope"] = "desktop" + return tool, args elif method == Methods.SCREEN_CAPTURE_FRAME: - app = params.get("app") or params.get("bundle_id") - if self._session.has_tool("screenshot"): - args = {} - if app: - args["app"] = app - return "screenshot", args - else: - args = {} - if app: - args["app"] = app + # 0.19.3 has no standalone screenshot tool: full-display capture + # is get_desktop_state; window capture rides on get_window_state. + args = {} + if "screenshot_out_file" in params: + args["screenshot_out_file"] = params["screenshot_out_file"] + if "pid" in params and "window_id" in params: + args["pid"] = params["pid"] + args["window_id"] = params["window_id"] return "get_window_state", args + return "get_desktop_state", args elif method == Methods.RECORDING_START: args: Dict[str, Any] = {} @@ -969,20 +1091,37 @@ def _apply_escalation( @staticmethod def _unwrap_result(result: Dict[str, Any]) -> Any: - """Unwrap the flattened tool result into caller-friendly form.""" + """Unwrap the flattened tool result into caller-friendly form. + + Dict payloads gain ``ok: True`` (errors already raised) so callers' + envelope checks hold, and MCP image blocks ride along as ``images`` + instead of being dropped when structured content is present. + """ if result.get("isError"): data = result.get("data", "unknown error") raise RpcError("cua_tool_error", str(data), result) - # Prefer structuredContent, then data, then images + + images = result.get("images") or [] + + def _finalize(payload: Dict[str, Any]) -> Dict[str, Any]: + out = dict(payload) + if images and "images" not in out: + out["images"] = images + out.setdefault("ok", True) + return out + structured = result.get("structuredContent") + if isinstance(structured, dict): + return _finalize(structured) if structured is not None: return structured data = result.get("data") + if isinstance(data, dict): + return _finalize(data) if data is not None: return data - images = result.get("images") if images: - return {"images": images} + return {"ok": True, "images": images} return None @@ -995,17 +1134,33 @@ def __init__(self, data: Any) -> None: self.data = data -def _local_clipboard_get(params: Dict[str, Any]) -> str: - return _clipboard_get() +def _local_clipboard_get(params: Dict[str, Any]) -> Dict[str, Any]: + """Return the PerceptionPort clipboard contract shape. + + The platform command cannot observe change counts, so change_count is 0 + and change_ts is the read time. + """ + return { + "ok": True, + "text": _clipboard_get(), + "change_count": 0, + "change_ts": time.time(), + } -def _local_clipboard_set(params: Dict[str, Any]) -> None: +def _local_clipboard_set(params: Dict[str, Any]) -> Dict[str, Any]: _clipboard_set(params.get("text", params.get("content", ""))) + return {"ok": True} def _local_clipboard_last_change(params: Dict[str, Any]) -> Dict[str, Any]: - # Best-effort: return current content with no timestamp - return {"content": _clipboard_get(), "timestamp": None} + # Best-effort: the platform command exposes no change counter. + return { + "ok": True, + "text": _clipboard_get(), + "change_count": 0, + "change_ts": time.time(), + } def _local_fs_subscribe(params: Dict[str, Any]) -> Dict[str, Any]: diff --git a/src/leapflow/platform/facade.py b/src/leapflow/platform/facade.py index 265a762..2894f71 100644 --- a/src/leapflow/platform/facade.py +++ b/src/leapflow/platform/facade.py @@ -123,13 +123,16 @@ def _parse_manifest(raw: dict) -> PlatformManifest: # matched nothing here and silently produced an empty capability set. _CUA_TOOL_TO_CAPABILITIES: dict[str, list[Capability]] = { "get_window_state": [Capability.AX_TREE_READ], + "list_windows": [Capability.AX_TREE_READ], "click": [Capability.AX_PERFORM_ACTION], "type_text": [Capability.AX_PERFORM_ACTION], "set_value": [Capability.AX_PERFORM_ACTION], "scroll": [Capability.AX_PERFORM_ACTION], "hotkey": [Capability.AX_PERFORM_ACTION], "screenshot": [Capability.SCREEN_CAPTURE], + "get_desktop_state": [Capability.SCREEN_CAPTURE], "launch_app": [Capability.APP_LAUNCH], + "bring_to_front": [Capability.APP_ACTIVATE], "list_apps": [Capability.APP_ACTIVATE], # No Capability member exists for recording yet; re-enable by adding e.g. # SCREEN_RECORD to the enum and uncommenting: diff --git a/src/leapflow/platform/mock.py b/src/leapflow/platform/mock.py index 9276b74..9605232 100644 --- a/src/leapflow/platform/mock.py +++ b/src/leapflow/platform/mock.py @@ -113,16 +113,128 @@ async def call(self, method: str, params: Optional[Dict[str, Any]] = None) -> An await self._emit_event(EventTypes.FS_CHANGE, {"path": path, "flags": 0x00000200, "ts": time.time()}) return {"ok": existed, "path": path} if method == Methods.APP_LAUNCH: - return {"ok": True, "bundle_id": p.get("bundle_id"), "mock": True} + return { + "ok": True, + "bundle_id": p.get("bundle_id") or p.get("name"), + "pid": 100, + "launch_state": "window_ready", + "windows": [ + { + "window_id": 1, + "pid": 100, + "app_name": "Mock App", + "title": "Mock UI", + "z_index": 0, + "is_on_screen": True, + "minimized": False, + } + ], + "mock": True, + } if method == Methods.APP_ACTIVATE: - return {"ok": True, "bundle_id": p.get("bundle_id"), "mock": True} + return {"ok": True, "pid": p.get("pid"), "window_id": p.get("window_id"), "mock": True} + if method == Methods.APP_LIST: + return { + "apps": [ + { + "name": "Mock App", + "bundle_id": "com.mock.app", + "pid": 100, + "running": True, + "active": True, + "kind": "desktop", + } + ], + "mock": True, + } if method == Methods.AX_TREE: return { - "root": {"role": "mock_window", "title": "Mock UI", "children": []}, + "pid": p.get("pid", 100), + "window_id": p.get("window_id", 1), + "snapshot_id": "s00000001", + "element_count": 4, + "returned_element_count": 4, + "total_element_count": 4, + "elements_complete": True, + "elements": [ + { + "element_index": 0, + "element_token": "s00000001:0", + "role": "Button", + "label": "Save", + "enabled": True, + "frame": {"x": 10, "y": 10, "w": 80, "h": 24}, + "depth": 1, + }, + { + "element_index": 1, + "element_token": "s00000001:1", + "role": "Button", + "label": "Cancel", + "enabled": True, + "frame": {"x": 100, "y": 10, "w": 80, "h": 24}, + "depth": 1, + }, + { + # Unlabeled Edit without frame — a real driver variant. + "element_index": 2, + "element_token": "s00000001:2", + "role": "Edit", + "enabled": True, + "value": "hello", + "depth": 2, + }, + { + "element_index": 3, + "element_token": "s00000001:3", + "role": "TabItem", + "label": "Home", + "enabled": True, + "selected": True, + "parent_index": 2, + "depth": 3, + }, + ], + "tree_markdown": ( + '- Window "Mock UI"\n' + ' - [0] Button "Save"\n' + ' - [1] Button "Cancel"\n' + ' - [2] Edit\n' + ' - [3] TabItem "Home"\n' + ), + "mock": True, + } + if method == Methods.SCREEN_CAPTURE_FRAME: + return { + "ok": True, + "screenshot_file_path": p.get("screenshot_out_file", ""), + "mock": True, + } + if method == Methods.AX_LIST: + return { + "windows": [ + { + "window_id": 1, + "pid": 100, + "app_name": "Mock App", + "title": "Mock UI", + "bounds": {"x": 0, "y": 0, "width": 800, "height": 600}, + "z_index": 0, + "is_on_screen": True, + "minimized": False, + } + ], "mock": True, } if method == Methods.AX_PERFORM: return {"ok": True, "mock": True, "action": p} + if method == Methods.AX_SCROLL: + return { + "ok": True, + "mock": True, + "direction": p.get("direction", ""), + "amount": p.get("amount", 3), + } if method == Methods.FS_SUBSCRIBE: return {"subscription_id": str(uuid.uuid4()), "mock": True} if method == Methods.RECORDING_START: diff --git a/src/leapflow/platform/protocol.py b/src/leapflow/platform/protocol.py index d3c4d59..2053a93 100644 --- a/src/leapflow/platform/protocol.py +++ b/src/leapflow/platform/protocol.py @@ -62,6 +62,7 @@ class Methods: AX_TREE = "ax.tree" AX_PERFORM = "ax.perform" AX_SCROLL = "ax.scroll" + AX_LIST = "ax.list" APP_LAUNCH = "app.launch" APP_ACTIVATE = "app.activate" diff --git a/src/leapflow/skills/action_policy.py b/src/leapflow/skills/action_policy.py index 6ba316e..0150150 100644 --- a/src/leapflow/skills/action_policy.py +++ b/src/leapflow/skills/action_policy.py @@ -45,6 +45,10 @@ class PolicyContext: skill_name: str = "" iteration: int = 0 history: List[str] = field(default_factory=list) + # Human-readable description of the call's resolved target (e.g. the + # clicked element's role + label), filled by ToolBridge before rule + # evaluation. Element-index params carry no semantics by themselves. + target_description: str = "" @runtime_checkable @@ -94,13 +98,13 @@ class SendActionRule: def check(self, call: ToolCall, context: PolicyContext) -> Optional[PolicyDecision]: if call.name == "click": - selector = call.params.get("selector", "") - if _SEND_LABELS.search(selector): + target = context.target_description + if target and _SEND_LABELS.search(target): return PolicyDecision( verdict=Verdict.ASK, reason="send_action", tool_name=call.name, - summary=f"Click: {selector}", + summary=f"Click: {target}", ) if call.name == "shortcut": diff --git a/src/leapflow/skills/bridge_factory.py b/src/leapflow/skills/bridge_factory.py index abc712d..f860de6 100644 --- a/src/leapflow/skills/bridge_factory.py +++ b/src/leapflow/skills/bridge_factory.py @@ -45,28 +45,39 @@ def build_tool_bridge( adapter = SemanticAdapter(perception=perception, execution=execution) + bridge.register( + "list_windows", + "List all top-level windows with pid, window_id, title, and per-window state " + "(minimized, on-screen). Call this first to pick the pid and window_id that " + "observe_ui and other window tools require.", + {}, + adapter.list_windows, + ) bridge.register( "observe_ui", - "Observe current app UI. Returns a list of interactive elements with selectors.", + "Snapshot one window's actionable UI elements, each tagged with an element_index " + "for click/right_click/read_text. Re-observe after actions — indices belong to one " + "snapshot. Requires the window's pid and window_id from list_windows.", { - "app_id": "string (optional) — target app bundle ID, empty = frontmost", - "focus_area": "string (optional) — focus hint (e.g. 'toolbar', 'sidebar')", + "pid": "int (required) — target process ID from list_windows", + "window_id": "int (required) — target window ID from list_windows", + "query": "string (optional) — case-insensitive filter over roles/labels to shrink large windows", }, adapter.observe_ui, ) bridge.register( "click", - "Click a UI element by its selector (from observe_ui results)", - {"selector": "string (required) — element selector, e.g. 'AXButton[label=Send]'"}, + "Click a UI element by its element_index (from the latest observe_ui snapshot)", + {"element_index": "int (required) — element_index from observe_ui"}, adapter.click, mutates_state=True, + describer=adapter.describe_element, ) bridge.register( "type_text", "Type text into the currently focused element", { "text": "string (required) — text to type", - "method": "string (optional) — 'paste' (default, best for CJK) or 'keystroke'", }, adapter.type_text, mutates_state=True, @@ -119,8 +130,8 @@ def build_tool_bridge( ) bridge.register( "read_text", - "Read the text content of a specific UI element", - {"selector": "string (required) — element selector"}, + "Read the text content of a specific UI element from the latest snapshot", + {"element_index": "int (required) — element_index from observe_ui"}, adapter.read_text, ) bridge.register( @@ -136,7 +147,8 @@ def build_tool_bridge( "Wait until a UI condition is met (polls UI tree). Returns elements when found or on timeout.", { "condition": "string (required) — what to wait for (e.g. 'Send button', '发送')", - "app_id": "string (optional) — app to observe", + "pid": "int (optional) — window's process ID, default = last observed window", + "window_id": "int (optional) — window ID, default = last observed window", "timeout": "number (optional, default=30) — max seconds to wait", "poll_interval": "number (optional, default=2) — seconds between polls", }, @@ -150,7 +162,8 @@ def build_tool_bridge( { "timeout": "number (optional, default=30) — max seconds to wait", "poll_interval": "number (optional, default=2) — seconds between polls", - "app_id": "string (optional) — app to observe", + "pid": "int (optional) — window's process ID, default = last observed window", + "window_id": "int (optional) — window ID, default = last observed window", }, adapter.wait_until_stable, mutates_state=True, @@ -158,21 +171,23 @@ def build_tool_bridge( ) bridge.register( "scroll", - "Scroll a scrollable area. Use after observe_ui if target content is not visible.", + "Scroll a scrollable area of a window. Omit element_index to scroll the window's " + "focused/page scroller; pass one to scroll an exact element from the latest snapshot.", { - "selector": "string (optional) — scroll area selector, empty = first scrollable", + "element_index": "int (optional) — scroll target from observe_ui, omit for focused scroller", "direction": "string (optional, default='down') — up/down/left/right", "amount": "number (optional, default=3) — scroll units (1-20)", + "pid": "int (optional) — window's process ID, default = last observed window", + "window_id": "int (optional) — window ID, default = last observed window", }, adapter.scroll, mutates_state=True, ) bridge.register( "select_text", - "Select text in a UI element (for subsequent cmd+c copy)", + "Select all text in a UI element (focus + select-all, for subsequent copy)", { - "selector": "string (required) — element containing text to select", - "method": "string (optional, default='all') — 'all' (select all) or 'word' (double-click)", + "element_index": "int (required) — element containing text, from observe_ui", }, adapter.select_text, mutates_state=True, @@ -181,18 +196,20 @@ def build_tool_bridge( "right_click", "Right-click a UI element to open its context menu. Returns visible menu items.", { - "selector": "string (required) — element to right-click", + "element_index": "int (required) — element to right-click, from observe_ui", }, adapter.right_click, mutates_state=True, + describer=adapter.describe_element, ) bridge.register( "screenshot", - "Capture a screenshot for visual verification. " - "When app_id is provided, captures only that app's window (works across all displays).", + "Capture a screenshot for visual verification. With pid + window_id captures that " + "window (works across all displays); defaults to the last observed window, or the " + "full desktop when no window has been observed.", { - "app_id": "string (optional) — target app bundle ID for window-level capture", - "region": "string (optional) — empty for full screen", + "pid": "int (optional) — window's process ID from list_windows", + "window_id": "int (optional) — window ID from list_windows", }, adapter.screenshot, ) diff --git a/src/leapflow/skills/semantic_adapter.py b/src/leapflow/skills/semantic_adapter.py index 228372a..1b2fbd1 100644 --- a/src/leapflow/skills/semantic_adapter.py +++ b/src/leapflow/skills/semantic_adapter.py @@ -8,11 +8,14 @@ Architecture: LLM ToolCall → SemanticAdapter → ExecutionPort / PerceptionPort → RPC → OS -Responsibilities: - - UI tree summarization (raw AX tree → LLM-friendly element list) - - Selector resolution (human-readable selector → node_id via cache) - - Composite operations (switch_app = launch + activate + verify) - - Input strategy selection (type_text via paste vs keystroke) +Addressing model (mirrors cua-driver): + - list_windows supplies (pid, window_id) window targets. + - observe_ui snapshots one window; every element carries an + element_index. The snapshot is superseded by the next observation + of the same window. + - Action tools address elements by element_index from the latest + snapshot; the adapter translates to the element_token the driver + validates for staleness. """ from __future__ import annotations @@ -21,23 +24,77 @@ import hashlib import logging import shlex -import time -from typing import Any, Dict, List, Optional, Protocol, runtime_checkable +from typing import Any, Dict, List, Optional, Protocol, Tuple, runtime_checkable -from leapflow.domain.events import UINode -from leapflow.skills.ui_selector import ( - resolve_selector_string, -) -from leapflow.skills.ui_summarizer import UIElement, UITreeSummarizer +from leapflow.domain.events import UIElement, UISnapshot logger = logging.getLogger(__name__) +def _window_target(params: Dict[str, Any]) -> Optional[Tuple[int, int]]: + """Parse a (pid, window_id) target from tool params, or None if absent.""" + pid = params.get("pid") + window_id = params.get("window_id") + if pid is None or window_id is None: + return None + try: + return int(pid), int(window_id) + except (TypeError, ValueError): + return None + + +def _launch_window_target(launch_result: Dict[str, Any]) -> Optional[Tuple[int, int]]: + """Extract (pid, window_id) from a launch_app response, or None. + + The driver's launch_app returns the launched app's pid and a windows + array (same record shape as list_windows) precisely so callers can skip + an extra discovery round-trip. + """ + pid = launch_result.get("pid") + windows = launch_result.get("windows") + if not isinstance(pid, int) or pid <= 0 or not isinstance(windows, list): + return None + for record in windows: + if isinstance(record, dict) and isinstance(record.get("window_id"), int): + return pid, record["window_id"] + return None + + +def _serialize_element(el: UIElement) -> Dict[str, Any]: + """Compact LLM-facing element record; silent defaults are omitted.""" + record: Dict[str, Any] = { + "element_index": el.element_index, + "role": el.role, + } + if el.label: + record["label"] = el.label + if el.value: + record["value"] = el.value + if not el.enabled: + record["enabled"] = False + if el.selected is not None: + record["selected"] = el.selected + return record + + +def _snapshot_digest(snapshot: UISnapshot) -> str: + """Fast content digest of a snapshot's elements for change detection.""" + content = "|".join( + f"{el.role}:{el.label}:{el.value}" for el in snapshot.elements + ) + return hashlib.md5(content.encode(), usedforsecurity=False).hexdigest()[:12] + + @runtime_checkable class PerceptionPort(Protocol): - async def read_ui_tree(self, app_id: Optional[str] = None) -> UINode: ... + async def read_window_state( + self, pid: int, window_id: int, query: str = "" + ) -> UISnapshot: ... + async def list_windows(self) -> Dict[str, Any]: ... async def get_clipboard(self) -> Dict[str, Any]: ... - async def capture_screenshot(self, region: str = "", app_id: str = "") -> Dict[str, Any]: ... + async def capture_screenshot( + self, pid: Optional[int] = None, window_id: Optional[int] = None + ) -> Dict[str, Any]: ... @runtime_checkable @@ -45,21 +102,33 @@ class ExecutionPort(Protocol): async def perform_ui_action( self, node_id: str, action: str, params: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: ... - async def launch_app(self, app_id: str) -> Dict[str, Any]: ... + async def launch_app( + self, app_id: str, urls: Optional[List[str]] = None + ) -> Dict[str, Any]: ... async def exec_shell(self, command: str) -> Dict[str, Any]: ... async def set_clipboard(self, text: str) -> Dict[str, Any]: ... - async def type_text(self, text: str, method: str = "paste") -> Dict[str, Any]: ... + async def type_text(self, text: str) -> Dict[str, Any]: ... async def send_shortcut(self, keys: str) -> Dict[str, Any]: ... - async def activate_app(self, app_id: str) -> Dict[str, Any]: ... - async def list_apps(self, filter: str = "", running_only: bool = False) -> Dict[str, Any]: ... - async def scroll(self, node_id: str, delta_x: int, delta_y: int) -> Dict[str, Any]: ... + async def activate_app( + self, pid: int, window_id: Optional[int] = None + ) -> Dict[str, Any]: ... + async def list_apps(self) -> Dict[str, Any]: ... + async def scroll( + self, + node_id: str, + direction: str, + amount: int = 3, + pid: Optional[int] = None, + window_id: Optional[int] = None, + ) -> Dict[str, Any]: ... class SemanticAdapter: """Translates LLM semantic tool calls into platform port operations. - Manages a short-lived selector→node_id cache that's populated on each - observe_ui() call and invalidated after a configurable TTL. + Keeps the latest window snapshot; action tools resolve element_index + against it and address the driver via element_token, whose staleness + the driver itself validates. """ def __init__( @@ -67,50 +136,54 @@ def __init__( perception: PerceptionPort, execution: ExecutionPort, *, - cache_ttl: float = 5.0, settle_delay: float = 0.3, - summarizer: Optional[UITreeSummarizer] = None, + max_observed_elements: int = 120, ) -> None: self._perception = perception self._execution = execution - self._cache_ttl = cache_ttl self._settle_delay = settle_delay - self._summarizer = summarizer or UITreeSummarizer() - self._selector_cache: Dict[str, str] = {} - self._last_tree: Optional[UINode] = None - self._cache_ts: float = 0.0 + self._max_observed_elements = max_observed_elements + self._last_snapshot: Optional[UISnapshot] = None # ═══════════════════════════════════════════════════════════════════ # Perception tools (read-only) # ═══════════════════════════════════════════════════════════════════ async def observe_ui(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Observe the current UI state, returning a summarized element list.""" - app_id = params.get("app_id", "") or None - focus_area = params.get("focus_area", "") - - tree = await self._perception.read_ui_tree(app_id) - elements = self._summarizer.summarize(tree, focus_area=focus_area) - - self._update_cache(elements, tree) - - serialized = [ - { - "selector": el.selector, - "role": el.role, - "label": el.label, - **({"value": el.value} if el.value else {}), - **({"actions": el.actions} if el.actions else {}), - **({"path": el.path} if el.path else {}), + """Snapshot one window's actionable elements (indexed for actions).""" + target = _window_target(params) + if target is None: + return { + "ok": False, + "error": "missing_window_target", + "suggestion": "call list_windows first and pass the target's pid and window_id", } - for el in elements - ] + query = str(params.get("query", "") or "") - return { + pid, window_id = target + snapshot = await self._perception.read_window_state(pid, window_id, query) + self._last_snapshot = snapshot + + elements = snapshot.elements[: self._max_observed_elements] + result: Dict[str, Any] = { "ok": True, - "element_count": len(serialized), - "elements": serialized, + "pid": pid, + "window_id": window_id, + "element_count": len(snapshot.elements), + "elements": [_serialize_element(el) for el in elements], } + if len(snapshot.elements) > len(elements): + result["truncated"] = True + result["suggestion"] = "pass query to filter elements of interest" + self._attach_coverage(result, snapshot) + return result + + async def list_windows(self, params: Dict[str, Any]) -> Dict[str, Any]: + """List top-level windows — the source of pid/window_id targets.""" + result = await self._perception.list_windows() + if isinstance(result, dict): + return {"ok": True, **result} + return {"ok": True, "windows": result} async def get_clipboard(self, params: Dict[str, Any]) -> Dict[str, Any]: """Read current clipboard text.""" @@ -118,86 +191,53 @@ async def get_clipboard(self, params: Dict[str, Any]) -> Dict[str, Any]: return {"ok": True, "text": result.get("text", ""), **result} async def read_text(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Read text content of a specific element.""" - selector_str = params.get("selector", "") - node_id = await self._resolve_selector(selector_str) - if not node_id: - return {"ok": False, "error": f"element_not_found: {selector_str}"} - - if self._last_tree: - node = self._find_node_by_id(self._last_tree, node_id) - if node: - return {"ok": True, "text": node.value, "label": node.label} - - return {"ok": True, "text": "", "note": "value not available"} + """Read the text content of an element from the latest snapshot.""" + element, error = self._resolve_element(params) + if element is None: + return error + return {"ok": True, "text": element.value, "label": element.label} # ═══════════════════════════════════════════════════════════════════ # Execution tools (state-changing) # ═══════════════════════════════════════════════════════════════════ - _CLICK_ACTIONS = ("AXPress", "AXShowDefaultUI") - async def click(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Click a UI element with fallback actions, returning post-action state hint.""" - selector_str = params.get("selector", "") - node_id = await self._resolve_selector(selector_str) - if not node_id: - return {"ok": False, "error": f"element_not_found: {selector_str}"} - - result: Dict[str, Any] = {"ok": False} - for action in self._CLICK_ACTIONS: - result = await self._execution.perform_ui_action(node_id, action) - if result.get("ok"): - break + """Click an element by index, returning a post-action state hint.""" + element, error = self._resolve_element(params) + if element is None: + return error - self._invalidate_cache() + result = await self._execution.perform_ui_action(element.target, "press") if not result.get("ok"): - node = self._find_node_by_id(self._last_tree, node_id) if self._last_tree else None - error_info: Dict[str, Any] = {"ok": False, "error": f"click_failed: {selector_str}"} - if node and node.frame: - error_info["frame"] = node.frame + error_info: Dict[str, Any] = { + "ok": False, + "error": f"click_failed: element {element.element_index} ({element.role} {element.label!r})", + } + if element.frame: + error_info["frame"] = element.frame error_info["suggestion"] = ( - "click failed — try keyboard interaction (shortcut, type_text) " - "or a different selector" + "click failed — re-observe_ui for a fresh snapshot, or try " + "keyboard interaction (shortcut, type_text)" ) return error_info - await asyncio.sleep(self._settle_delay) - try: - tree = await self._perception.read_ui_tree(None) - elements = self._summarizer.summarize(tree) - self._update_cache(elements, tree) - state_hint = [ - f"{el.role}[{el.label}]" for el in elements[:10] if el.label - ] - return { - **result, - "selector": selector_str, - "state_after": state_hint, - "element_count": len(elements), - } - except Exception: - return {**result, "selector": selector_str} + state = await self._refresh_after_action() + return {**result, "element_index": element.element_index, **state} async def type_text(self, params: Dict[str, Any]) -> Dict[str, Any]: """Type text into the currently focused element.""" text = params.get("text", "") - method = params.get("method", "paste") if not text: return {"ok": False, "error": "empty text"} - result = await self._execution.type_text(text, method) - self._invalidate_cache() - return result + return await self._execution.type_text(text) async def shortcut(self, params: Dict[str, Any]) -> Dict[str, Any]: """Execute a keyboard shortcut.""" keys = params.get("keys", "") if not keys: return {"ok": False, "error": "no keys specified"} - result = await self._execution.send_shortcut(keys) - self._invalidate_cache() - return result + return await self._execution.send_shortcut(keys) async def set_clipboard(self, params: Dict[str, Any]) -> Dict[str, Any]: """Set clipboard text content.""" @@ -205,7 +245,7 @@ async def set_clipboard(self, params: Dict[str, Any]) -> Dict[str, Any]: return await self._execution.set_clipboard(text) async def switch_app(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Switch to an application and verify it's in foreground.""" + """Switch to an application and verify its window is readable.""" app_id = params.get("app_id", "") if not app_id: return {"ok": False, "error": "app_id required"} @@ -219,26 +259,64 @@ async def switch_app(self, params: Dict[str, Any]) -> Dict[str, Any]: "suggestion": "Use list_apps(filter='...') to discover correct bundle_id", } - await self._execution.activate_app(app_id) + target = _launch_window_target(launch_result) + if target is None: + return { + "ok": True, + "app_id": app_id, + "verified": False, + "suggestion": "call list_windows to pick the app's pid/window_id, then observe_ui", + } + + pid, window_id = target + await self._execution.activate_app(pid, window_id) for _ in range(10): await asyncio.sleep(0.5) try: - tree = await self._perception.read_ui_tree(app_id) - if tree and (tree.children or tree.label): - self._invalidate_cache() - return {"ok": True, "app_id": app_id, "window_title": tree.label} + snapshot = await self._perception.read_window_state(pid, window_id) except Exception: continue + if snapshot.elements or not snapshot.degraded: + self._last_snapshot = snapshot + return { + "ok": True, + "app_id": app_id, + "pid": pid, + "window_id": window_id, + "element_count": len(snapshot.elements), + } - self._invalidate_cache() return {"ok": False, "error": "app_not_ready", "app_id": app_id} async def list_apps(self, params: Dict[str, Any]) -> Dict[str, Any]: - """List available applications on the system.""" - filter_str = params.get("filter", "") - running_only = params.get("running_only", False) - return await self._execution.list_apps(filter_str, running_only) + """List available applications, honoring the declared filter locally. + + The driver's list_apps takes no arguments; the tool's filter and + running_only params are applied to the returned records here. + """ + filter_str = str(params.get("filter", "") or "").lower() + running_only = bool(params.get("running_only", False)) + result = await self._execution.list_apps() + if not isinstance(result, dict): + return {"ok": True, "apps": result} + apps = result.get("apps") + if not isinstance(apps, list) or (not filter_str and not running_only): + return result + + def _keep(record: Dict[str, Any]) -> bool: + if running_only and not record.get("running"): + pid = record.get("pid") + if not (isinstance(pid, int) and pid > 0): + return False + if filter_str: + name = str(record.get("name", "")).lower() + bundle = str(record.get("bundle_id", "")).lower() + if filter_str not in name and filter_str not in bundle: + return False + return True + + return {**result, "apps": [r for r in apps if isinstance(r, dict) and _keep(r)]} async def open_url(self, params: Dict[str, Any]) -> Dict[str, Any]: """Open a URL in the default or specified browser.""" @@ -260,18 +338,21 @@ async def wait(self, params: Dict[str, Any]) -> Dict[str, Any]: return {"ok": True, "waited": seconds} async def wait_until(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Poll UI until a condition appears met, or timeout. - - Checks if the condition string matches any element label or selector. - Returns the current UI snapshot so the LLM can verify the condition. - """ + """Poll the window until an element matching the condition appears.""" condition = params.get("condition", "") - app_id = params.get("app_id", "") or None + target = _window_target(params) or self._current_target() timeout = min(max(float(params.get("timeout", 30)), 1.0), 180.0) poll_interval = min(max(float(params.get("poll_interval", 2)), 0.5), 10.0) if not condition: return {"ok": False, "error": "condition required"} + if target is None: + return { + "ok": False, + "error": "missing_window_target", + "suggestion": "pass pid and window_id (from list_windows) or call observe_ui first", + } + pid, window_id = target condition_lower = condition.lower() elapsed = 0.0 @@ -281,19 +362,14 @@ async def wait_until(self, params: Dict[str, Any]) -> Dict[str, Any]: await asyncio.sleep(poll_interval) elapsed += poll_interval - tree = await self._perception.read_ui_tree(app_id) - elements = self._summarizer.summarize(tree) - self._update_cache(elements, tree) - - serialized = [ - {"selector": el.selector, "role": el.role, "label": el.label} - for el in elements[:20] - ] + snapshot = await self._perception.read_window_state(pid, window_id) + self._last_snapshot = snapshot + serialized = [_serialize_element(el) for el in snapshot.elements[:20]] found = any( condition_lower in el.label.lower() - or condition_lower in el.selector.lower() - for el in elements + or condition_lower in el.role.lower() + for el in snapshot.elements ) if found: return { @@ -315,96 +391,98 @@ async def wait_until(self, params: Dict[str, Any]) -> Dict[str, Any]: # Extended interaction tools # ═══════════════════════════════════════════════════════════════════ - _SCROLL_DELTAS = { - "down": (0, -1), "up": (0, 1), - "left": (1, 0), "right": (-1, 0), - } + _SCROLL_DIRECTIONS = ("up", "down", "left", "right") async def scroll(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Scroll a scrollable area in the given direction.""" - selector_str = params.get("selector", "") + """Scroll an element (by index) or the target window's focused scroller.""" direction = params.get("direction", "down") amount = min(max(int(params.get("amount", 3)), 1), 20) - if direction not in self._SCROLL_DELTAS: + if direction not in self._SCROLL_DIRECTIONS: return {"ok": False, "error": f"invalid_direction: {direction} (use up/down/left/right)"} - node_id = await self._resolve_selector(selector_str) if selector_str else None - if not node_id: - if not self._last_tree: - tree = await self._perception.read_ui_tree(None) - elements = self._summarizer.summarize(tree) - self._update_cache(elements, tree) - node_id = self._find_first_scrollable(self._last_tree) if self._last_tree else None - - unit_dx, unit_dy = self._SCROLL_DELTAS[direction] - dx, dy = unit_dx * amount, unit_dy * amount + # The driver requires pid even on the targetless keystroke path. + window = _window_target(params) or self._current_target() + if window is None: + return { + "ok": False, + "error": "missing_window_target", + "suggestion": "call observe_ui(pid, window_id) first", + } + pid, window_id = window - await self._execution.scroll(node_id or "", dx, dy) - self._invalidate_cache() - await asyncio.sleep(self._settle_delay) + target = "" + if params.get("element_index") is not None: + element, error = self._resolve_element(params) + if element is None: + return error + target = element.target - tree = await self._perception.read_ui_tree(None) - elements = self._summarizer.summarize(tree) - self._update_cache(elements, tree) - return {"ok": True, "direction": direction, "amount": amount, "element_count": len(elements)} + await self._execution.scroll( + target, direction, amount, pid=pid, window_id=window_id + ) + state = await self._refresh_after_action() + return {"ok": True, "direction": direction, "amount": amount, **state} async def select_text(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Select text in a UI element for subsequent copy.""" - selector_str = params.get("selector", "") - method = params.get("method", "all") - - node_id = await self._resolve_selector(selector_str) - if not node_id: - return {"ok": False, "error": f"element_not_found: {selector_str}"} + """Select text in an element (focus it, then select all).""" + element, error = self._resolve_element(params) + if element is None: + return error - await self._execution.perform_ui_action(node_id, "AXPress") + await self._execution.perform_ui_action(element.target, "press") await asyncio.sleep(self._settle_delay) - - if method == "all": - await self._execution.send_shortcut("cmd+a") - else: - await self._execution.perform_ui_action(node_id, "AXPress") - - self._invalidate_cache() - return {"ok": True, "selector": selector_str, "method": method} + await self._execution.send_shortcut("cmd+a") + return {"ok": True, "element_index": element.element_index} async def right_click(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Right-click a UI element to open its context menu.""" - selector_str = params.get("selector", "") - node_id = await self._resolve_selector(selector_str) - if not node_id: - return {"ok": False, "error": f"element_not_found: {selector_str}"} - - result = await self._execution.perform_ui_action(node_id, "AXShowMenu") - self._invalidate_cache() - await asyncio.sleep(self._settle_delay) + """Right-click an element to open its context menu.""" + element, error = self._resolve_element(params) + if element is None: + return error - tree = await self._perception.read_ui_tree(None) - elements = self._summarizer.summarize(tree) - self._update_cache(elements, tree) - menu_items = [el for el in elements if "Menu" in el.role] + result = await self._execution.perform_ui_action(element.target, "show_menu") + await asyncio.sleep(self._settle_delay) + snapshot = await self._resnapshot() + if snapshot is None: + return {**result, "element_index": element.element_index, "menu_items": []} + menu_items = [ + _serialize_element(el) for el in snapshot.elements if "Menu" in el.role + ] return { **result, - "selector": selector_str, - "menu_items": [ - {"selector": el.selector, "label": el.label} for el in menu_items - ], + "element_index": element.element_index, + "menu_items": menu_items, } async def screenshot(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Capture a screenshot for visual state verification.""" - region = params.get("region", "") - app_id = params.get("app_id", "") - result = await self._perception.capture_screenshot(region=region, app_id=app_id) + """Capture a screenshot for visual state verification. + + With a pid/window_id target (explicit or remembered) captures that + window; otherwise captures the full desktop. + """ + target = _window_target(params) or self._current_target() + if target is None: + result = await self._perception.capture_screenshot() + else: + result = await self._perception.capture_screenshot( + pid=target[0], window_id=target[1] + ) return {"ok": True, "captured": True, **result} async def wait_until_stable(self, params: Dict[str, Any]) -> Dict[str, Any]: - """Wait until the UI stops changing (element digest stabilizes).""" + """Wait until the window stops changing (element digest stabilizes).""" timeout = min(max(float(params.get("timeout", 30)), 1.0), 180.0) poll_interval = min(max(float(params.get("poll_interval", 2)), 0.5), 10.0) - app_id = params.get("app_id", "") or None + target = _window_target(params) or self._current_target() + if target is None: + return { + "ok": False, + "error": "missing_window_target", + "suggestion": "pass pid and window_id (from list_windows) or call observe_ui first", + } + pid, window_id = target elapsed = 0.0 prev_digest = "" @@ -414,14 +492,13 @@ async def wait_until_stable(self, params: Dict[str, Any]) -> Dict[str, Any]: await asyncio.sleep(poll_interval) elapsed += poll_interval - tree = await self._perception.read_ui_tree(app_id) - elements = self._summarizer.summarize(tree) - digest = _elements_digest(elements) + snapshot = await self._perception.read_window_state(pid, window_id) + digest = _snapshot_digest(snapshot) if digest == prev_digest: stable_count += 1 if stable_count >= 2: - self._update_cache(elements, tree) + self._last_snapshot = snapshot return {"ok": True, "stable": True, "elapsed": round(elapsed, 1)} else: stable_count = 0 @@ -430,69 +507,110 @@ async def wait_until_stable(self, params: Dict[str, Any]) -> Dict[str, Any]: return {"ok": True, "stable": False, "elapsed": round(elapsed, 1), "timeout": True} # ═══════════════════════════════════════════════════════════════════ - # Cache management + # Snapshot management # ═══════════════════════════════════════════════════════════════════ - def _update_cache(self, elements: List[UIElement], tree: UINode) -> None: - """Rebuild selector→node_id cache from fresh observation.""" - self._selector_cache.clear() - for el in elements: - if el.node_id: - self._selector_cache[el.selector] = el.node_id - self._last_tree = tree - self._cache_ts = time.monotonic() - - def _invalidate_cache(self) -> None: - """Mark cache as stale (actions may have changed UI state).""" - self._cache_ts = 0.0 - - @property - def _cache_valid(self) -> bool: - return (time.monotonic() - self._cache_ts) < self._cache_ttl - - async def _resolve_selector(self, selector_str: str) -> Optional[str]: - """Resolve a selector string to a node_id, refreshing cache if needed.""" - if self._cache_valid and selector_str in self._selector_cache: - return self._selector_cache[selector_str] - - if not self._cache_valid: - tree = await self._perception.read_ui_tree(None) - elements = self._summarizer.summarize(tree) - self._update_cache(elements, tree) - - if selector_str in self._selector_cache: - return self._selector_cache[selector_str] - - if self._last_tree: - node_id = resolve_selector_string(self._last_tree, selector_str) - if node_id: - self._selector_cache[selector_str] = node_id - return node_id + def _current_target(self) -> Optional[Tuple[int, int]]: + if self._last_snapshot is None: + return None + return self._last_snapshot.pid, self._last_snapshot.window_id - return None + def describe_element(self, params: Dict[str, Any]) -> str: + """Human-readable description of the element a call targets. - def _find_node_by_id(self, node: UINode, target_id: str) -> Optional[UINode]: - """DFS search for a node by id.""" - if node.node_id == target_id: - return node - for child in node.children: - found = self._find_node_by_id(child, target_id) - if found: - return found - return None + Consumed by the policy gate: element_index params carry no + semantics, so safety rules (e.g. send-button detection) need the + resolved role + label. + """ + if self._last_snapshot is None: + return "" + try: + index = int(params.get("element_index")) + except (TypeError, ValueError): + return "" + element = self._last_snapshot.find(index) + if element is None: + return "" + parts = [element.role] + if element.label: + parts.append(element.label) + if element.value: + parts.append(f"value={element.value[:40]}") + return " ".join(parts) + + def _resolve_element( + self, params: Dict[str, Any] + ) -> Tuple[Optional[UIElement], Dict[str, Any]]: + """Resolve params['element_index'] against the latest snapshot. + + Returns (element, {}) on success or (None, structured_error). + """ + if self._last_snapshot is None: + return None, { + "ok": False, + "error": "no_snapshot", + "suggestion": "call observe_ui(pid, window_id) first to index elements", + } + raw = params.get("element_index") + try: + index = int(raw) + except (TypeError, ValueError): + return None, { + "ok": False, + "error": f"invalid_element_index: {raw!r}", + "suggestion": "pass the element_index of an element from observe_ui", + } + element = self._last_snapshot.find(index) + if element is None: + return None, { + "ok": False, + "error": f"element_not_found: {index}", + "suggestion": "the snapshot may be stale — call observe_ui again", + } + return element, {} - def _find_first_scrollable(self, node: UINode) -> Optional[str]: - """DFS for the first AXScrollArea node_id.""" - if node.role == "AXScrollArea": - return node.node_id - for child in node.children: - found = self._find_first_scrollable(child) - if found: - return found - return None + async def _resnapshot(self) -> Optional[UISnapshot]: + """Re-observe the current window; None when no target is known.""" + target = self._current_target() + if target is None: + return None + try: + snapshot = await self._perception.read_window_state(*target) + except Exception: + return None + self._last_snapshot = snapshot + return snapshot + async def _refresh_after_action(self) -> Dict[str, Any]: + """Settle, re-snapshot, and produce a compact post-action state hint.""" + await asyncio.sleep(self._settle_delay) + snapshot = await self._resnapshot() + if snapshot is None: + return {} + state_hint = [ + f"{el.role}[{el.label}]" for el in snapshot.elements[:10] if el.label + ] + state: Dict[str, Any] = { + "state_after": state_hint, + "element_count": len(snapshot.elements), + } + self._attach_coverage(state, snapshot) + return state -def _elements_digest(elements: List[UIElement]) -> str: - """Compute a fast content digest of the element list for change detection.""" - content = "|".join(f"{el.selector}:{el.label}" for el in elements) - return hashlib.md5(content.encode(), usedforsecurity=False).hexdigest()[:12] + @staticmethod + def _attach_coverage(result: Dict[str, Any], snapshot: UISnapshot) -> None: + """Surface the driver's blind-spot statements to the model. + + elements_complete=False or a coverage entry (e.g. browser page + content not observable in window scope) means the model must not + conclude an element is absent — it should fall back to screenshot + pixels or app-appropriate tools. + """ + if snapshot.degraded: + result["degraded"] = True + if snapshot.degraded_reason: + result["degraded_reason"] = snapshot.degraded_reason + if not snapshot.elements_complete: + result["elements_complete"] = False + if snapshot.coverage: + result["coverage"] = snapshot.coverage diff --git a/src/leapflow/skills/semantic_schema.py b/src/leapflow/skills/semantic_schema.py index 7c00d3c..89826a8 100644 --- a/src/leapflow/skills/semantic_schema.py +++ b/src/leapflow/skills/semantic_schema.py @@ -66,6 +66,7 @@ def handlers(self) -> Dict[str, Any]: ... "shortcut", "switch_app", "list_apps", + "list_windows", "open_url", "get_clipboard", "set_clipboard", @@ -80,7 +81,7 @@ def handlers(self) -> Dict[str, Any]: ... }) _OBSERVATION_TOOLS: FrozenSet[str] = frozenset({ - "observe_ui", "list_apps", "read_text", "get_clipboard", "screenshot", + "observe_ui", "list_apps", "list_windows", "read_text", "get_clipboard", "screenshot", }) _WAIT_TOOLS: FrozenSet[str] = frozenset({ "wait", "wait_until", "wait_until_stable", diff --git a/src/leapflow/skills/tool_executor.py b/src/leapflow/skills/tool_executor.py index 04bc8c7..54bf2fd 100644 --- a/src/leapflow/skills/tool_executor.py +++ b/src/leapflow/skills/tool_executor.py @@ -115,7 +115,9 @@ class ExecutionPort(Protocol): async def perform_file_op(self, op: str, params: Dict[str, Any]) -> Dict[str, Any]: ... async def exec_shell(self, command: str) -> Dict[str, Any]: ... - async def launch_app(self, app_id: str) -> Dict[str, Any]: ... + async def launch_app( + self, app_id: str, urls: Optional[List[str]] = None + ) -> Dict[str, Any]: ... async def perform_ui_action( self, node_id: str, action: str, params: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: ... @@ -129,15 +131,19 @@ async def perform_ui_action( class _ToolHandler: """A single tool's definition + dispatch logic.""" - __slots__ = ("definition", "handler") + __slots__ = ("definition", "handler", "describer") def __init__( self, definition: ToolDefinition, handler: Any, + describer: Any = None, ) -> None: self.definition = definition self.handler = handler + # Optional callable(params) -> str: human-readable description of + # the call's resolved target, consumed by the policy gate. + self.describer = describer class ToolBridge: @@ -200,9 +206,10 @@ def _register_defaults(self) -> None: lambda p: ex.exec_shell(p.get("command", "")), ) self._register( - "launch_app", "Launch an application by bundle ID", - {"app_id": "string (required) — application bundle ID"}, - lambda p: ex.launch_app(p.get("app_id", "")), + "launch_app", "Launch an application by bundle ID (backgrounded)", + {"app_id": "string (required) — application bundle ID", + "urls": "array (optional) — file paths/URLs the app opens on launch"}, + lambda p: ex.launch_app(p.get("app_id", ""), p.get("urls") or None), mutates_state=True, ) self._register( @@ -230,21 +237,31 @@ def _register( *, mutates_state: bool = False, counts_as_progress: bool | None = None, + describer: Any = None, ) -> None: defn = ToolDefinition( name=name, description=description, parameters=parameters, mutates_state=mutates_state, counts_as_progress=counts_as_progress, ) - self._handlers[name] = _ToolHandler(definition=defn, handler=handler) + self._handlers[name] = _ToolHandler( + definition=defn, handler=handler, describer=describer + ) def register( self, name: str, description: str, parameters: Dict[str, str], handler: Any, *, mutates_state: bool = False, counts_as_progress: bool | None = None, + describer: Any = None, ) -> None: - """Register a custom tool (extensibility point).""" + """Register a custom tool (extensibility point). + + ``describer`` optionally maps call params to a human-readable + description of the resolved target for the policy gate — required + for tools whose params are opaque handles (e.g. element_index). + """ self._register( name, description, parameters, handler, mutates_state=mutates_state, counts_as_progress=counts_as_progress, + describer=describer, ) @property @@ -283,6 +300,13 @@ async def dispatch_guarded( from leapflow.skills.action_policy import Verdict + entry = self._handlers.get(call.name) + if entry is not None and entry.describer is not None: + try: + context.target_description = str(entry.describer(call.params) or "") + except Exception: + context.target_description = "" + decision = await self.policy.evaluate(call, context) if decision.verdict == Verdict.ALLOW: diff --git a/src/leapflow/skills/ui_selector.py b/src/leapflow/skills/ui_selector.py deleted file mode 100644 index 54715ae..0000000 --- a/src/leapflow/skills/ui_selector.py +++ /dev/null @@ -1,201 +0,0 @@ -"""UI element selector — parse, match, and resolve accessibility elements. - -Provides a human-readable addressing scheme for UI elements that replaces -unstable internal node_id values. Selectors use role + label + index, -mirroring the anchor format from the Recording side. - -Selector syntax: - "AXButton[label=Send]" — exact match on role + label - "AXTextField[label~=search]" — contains match (case-insensitive) - "AXButton#2" — 2nd AXButton sibling (0-based) - "AXToolbar > AXButton[label=New]" — path-based (ancestor > target) -""" - -from __future__ import annotations - -import re -from dataclasses import dataclass, field -from typing import Dict, List, Optional, Tuple - -from leapflow.domain.events import UINode -from leapflow.domain.skill_types import AnchorCandidate - -_SELECTOR_RE = re.compile( - r"^(?P(?:[\w]+\s*>\s*)*)?" - r"(?P\w+)" - r"(?:\[label(?P[~]?)=(?P