From 4e116dd3954b990abe41e45e05d77b7d5d67edf6 Mon Sep 17 00:00:00 2001 From: fanqiNO1 <1848839264@qq.com> Date: Wed, 12 Aug 2026 18:32:44 +0800 Subject: [PATCH 1/5] fix cua tools --- src/leapflow/domain/__init__.py | 5 +- src/leapflow/domain/events.py | 70 ++- src/leapflow/domain/ui_vocabulary.py | 42 +- src/leapflow/learning/codegen.py | 9 +- src/leapflow/perception/state_snapshot.py | 70 ++- src/leapflow/platform/adapters/darwin.py | 215 +++++--- src/leapflow/platform/adapters/mock.py | 138 +++-- src/leapflow/platform/cua_client.py | 282 ++++++++--- src/leapflow/platform/facade.py | 3 + src/leapflow/platform/mock.py | 118 ++++- src/leapflow/platform/protocol.py | 1 + src/leapflow/skills/bridge_factory.py | 53 +- src/leapflow/skills/semantic_adapter.py | 592 ++++++++++++---------- src/leapflow/skills/semantic_schema.py | 3 +- src/leapflow/skills/ui_selector.py | 201 -------- src/leapflow/skills/ui_summarizer.py | 159 ------ tests/test_cua_client_mapping.py | 202 +++++++- tests/test_darwin_adapter.py | 132 +++++ tests/test_platform_adapters.py | 175 +++++++ tests/test_semantic_adapter.py | 150 ++++++ tests/test_semantic_schema.py | 2 +- 21 files changed, 1691 insertions(+), 931 deletions(-) delete mode 100644 src/leapflow/skills/ui_selector.py delete mode 100644 src/leapflow/skills/ui_summarizer.py create mode 100644 tests/test_darwin_adapter.py create mode 100644 tests/test_platform_adapters.py create mode 100644 tests/test_semantic_adapter.py 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..7e9d781 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]: ... @@ -78,7 +122,9 @@ async def perform_ui_action( async def launch_app(self, app_id: str) -> 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 +134,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..fa737e9 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 # ═══════════════════════════════════════════════════════════════════════════ diff --git a/src/leapflow/learning/codegen.py b/src/leapflow/learning/codegen.py index 496663f..0d60261 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]` @@ -752,7 +754,8 @@ 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", diff --git a/src/leapflow/perception/state_snapshot.py b/src/leapflow/perception/state_snapshot.py index 8d39ea1..b29a05b 100644 --- a/src/leapflow/perception/state_snapshot.py +++ b/src/leapflow/perception/state_snapshot.py @@ -13,7 +13,6 @@ from enum import IntEnum from typing import Any, 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..3ca6b7f 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: @@ -129,7 +152,8 @@ async def perform_ui_action( ) async def launch_app(self, app_id: str) -> Dict[str, Any]: - return await self._rpc.call(Methods.APP_LAUNCH, {"bundle_id": app_id}) + result = await self._rpc.call(Methods.APP_LAUNCH, {"bundle_id": app_id}) + 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 +162,77 @@ 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 scroll( + self, node_id: str, direction: str, amount: int = 3 + ) -> Dict[str, Any]: + params: Dict[str, Any] = {"direction": direction, "amount": amount} + if node_id: + params["node_id"] = node_id + return await self._rpc.call(Methods.AX_SCROLL, params) - 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]: 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]]: @@ -237,28 +290,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 [] - - frame = raw.get("frame") - frame_dict = ( - {"x": frame["x"], "y": frame["y"], "w": frame["w"], "h": frame["h"]} - if isinstance(frame, dict) - else None - ) +def _snapshot_from_payload( + payload: Dict[str, Any], *, pid: int, window_id: int +) -> UISnapshot: + """Parse a get_window_state payload into a flat UISnapshot. - 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..bbae9fe 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: @@ -118,23 +159,54 @@ async def perform_ui_action( async def launch_app(self, app_id: str) -> Dict[str, Any]: record = {"type": "launch_app", "app_id": app_id} 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 +239,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 +249,10 @@ 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 + ) -> Dict[str, Any]: + record = {"type": "scroll", "node_id": node_id, "direction": direction, "amount": amount} 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..028ca03 100644 --- a/src/leapflow/platform/cua_client.py +++ b/src/leapflow/platform/cua_client.py @@ -21,6 +21,7 @@ import logging import os import platform +import re import shutil import subprocess import sys @@ -552,65 +553,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", {}), +} + - cua-driver splits AX actions into discrete tools: click, type_text, - set_value. We infer the target from the `action` param. +def _element_target_args(params: Dict[str, Any]) -> Dict[str, Any]: + """Extract cua-driver element/pixel target args from neutral params. + + ``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 +875,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: @@ -850,37 +918,61 @@ def _map_to_cua_tool(self, method: str, params: Dict[str, Any]) -> Tuple[str, Di 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 +1061,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,8 +1104,13 @@ 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 {"text": _clipboard_get(), "change_count": 0, "change_ts": time.time()} def _local_clipboard_set(params: Dict[str, Any]) -> None: @@ -1004,8 +1118,8 @@ def _local_clipboard_set(params: Dict[str, Any]) -> None: 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 {"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/bridge_factory.py b/src/leapflow/skills/bridge_factory.py index abc712d..bd53a6b 100644 --- a/src/leapflow/skills/bridge_factory.py +++ b/src/leapflow/skills/bridge_factory.py @@ -45,19 +45,30 @@ 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, ) @@ -66,7 +77,6 @@ def build_tool_bridge( "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 +129,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 +146,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 +161,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,9 +170,10 @@ 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. Omit element_index to scroll the 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)", }, @@ -169,10 +182,9 @@ def build_tool_bridge( ) 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 +193,19 @@ 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, ) 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..886f255 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 @@ -48,18 +105,21 @@ async def perform_ui_action( async def launch_app(self, app_id: str) -> 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) -> 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 +127,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 +182,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}"} + """Click an element by index, returning a post-action state hint.""" + element, error = self._resolve_element(params) + if element is None: + return error - 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 - - 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 +236,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 +250,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 +329,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 +353,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 +382,88 @@ 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 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 + target = "" + if params.get("element_index") is not None: + element, error = self._resolve_element(params) + if element is None: + return error + target = element.target + # Without a target the driver's keystroke path drives the focused + # scroller — no scrollable-role guessing needed. - unit_dx, unit_dy = self._SCROLL_DELTAS[direction] - dx, dy = unit_dx * amount, unit_dy * amount - - await self._execution.scroll(node_id or "", dx, dy) - self._invalidate_cache() - await asyncio.sleep(self._settle_delay) - - 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) + 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") + """Select text in an element (focus it, then select all).""" + element, error = self._resolve_element(params) + if element is None: + return error - node_id = await self._resolve_selector(selector_str) - if not node_id: - return {"ok": False, "error": f"element_not_found: {selector_str}"} - - 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 +473,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 +488,87 @@ 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 _resolve_element( + self, params: Dict[str, Any] + ) -> Tuple[Optional[UIElement], Dict[str, Any]]: + """Resolve params['element_index'] against the latest snapshot. - 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 + 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/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