[Fix] Fix cua tools

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,

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",
 ]

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]: ...
 

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",
 }
 
 

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]:

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",
         ],

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):