Add ax_events: reactive UIA focus-change wait

WHATS_NEW.md

@@ -1,5 +1,11 @@
 # What's New — AutoControl
 
+## What's new (2026-06-25) — Reactive UIA Event Wait (focus change)
+
+Wait until focus lands on the dialog — a real, zero-latency UIA event, not polling. Full reference: [`docs/source/Eng/doc/new_features/v202_features_doc.rst`](docs/source/Eng/doc/new_features/v202_features_doc.rst).
+
+- **`wait_for_focus_change`** (`AC_wait_for_focus_change`): the accessibility recorder *polls* focus every ~250 ms, so it can miss a fast transition and reacts late. This blocks on the native `AddFocusChangedEventHandler` and returns the moment focus moves — the zero-latency, miss-free "wait until focus lands on the dialog" primitive, the accessibility-tree analogue of `wait_for_window` / `wait_for_image`. Returns the newly-focused element (or `None` on timeout). The real event subscription is registered/unregistered under a lock on the calling thread; dispatched through the injectable accessibility backend seam (headless-testable via a fake backend; real UIA in the Windows backend). No `PySide6`.
+
 ## What's new (2026-06-25) — Container Selection + View Switching (Selection / MultipleView)
 
 Read what's selected in a listbox/grid, and switch Explorer-style views. Full reference: [`docs/source/Eng/doc/new_features/v201_features_doc.rst`](docs/source/Eng/doc/new_features/v201_features_doc.rst).

docs/source/Eng/doc/new_features/v202_features_doc.rst

@@ -0,0 +1,37 @@
+Reactive UIA Event Wait — Focus Change
+======================================
+
+The accessibility recorder *polls* the focused element every ~250 ms, so it can
+miss a fast focus transition and reacts a quarter-second late. UIA exposes real
+events: ``wait_for_focus_change`` blocks on the native
+``AddFocusChangedEventHandler`` and returns the moment focus moves — the
+zero-latency, miss-free "wait until focus lands on the dialog" primitive, the
+accessibility-tree analogue of ``wait_for_window`` / ``wait_for_image``.
+
+It is a thin dispatch onto the injectable ``accessibility.backends.get_backend()``
+seam — headless-testable on any platform by injecting a fake backend; the real
+event subscription (registered / unregistered under a lock, on the calling
+thread) lives in the Windows backend. Imports no ``PySide6``.
+
+Headless API
+------------
+
+.. code-block:: python
+
+    from je_auto_control import wait_for_focus_change, click_text
+
+    click_text("Settings")
+    focused = wait_for_focus_change(timeout=3)
+    # {"name": "Search", "role": "ControlType_50004", "app_name": "app.exe", ...}
+    if focused is not None:
+        ...   # focus has moved — the dialog / next field is ready
+
+Returns the newly-focused element as ``{name, role, app_name, bounds, …}``, or
+``None`` if no focus change occurs within ``timeout`` seconds (default ``5``).
+
+Executor commands
+-----------------
+
+``AC_wait_for_focus_change`` (``timeout``) returns ``{changed, element}``. It is
+exposed as the read-only ``ac_wait_for_focus_change`` MCP tool and as a Script
+Builder command under **Native UI**.

docs/source/Zh/doc/new_features/v202_features_doc.rst

@@ -0,0 +1,34 @@
+反應式 UIA 事件等待——焦點變化
+==============================
+
+無障礙錄製器每約 250 毫秒*輪詢*一次焦點元素,因此可能錯過快速的焦點轉換,且慢上四分之一秒才反應。
+UIA 提供真正的事件:``wait_for_focus_change`` 阻塞於原生的 ``AddFocusChangedEventHandler``,並在
+焦點移動的當下回傳——這是零延遲、不漏失的「等到焦點落在對話框上」原語,是 ``wait_for_window`` /
+``wait_for_image`` 在無障礙樹上的對應。
+
+它是對可注入的 ``accessibility.backends.get_backend()`` 接縫的薄分派——可在任何平台透過注入 fake
+backend 進行無頭測試;真正的事件訂閱(在呼叫執行緒上、以鎖註冊 / 取消註冊)位於 Windows 後端。
+不匯入 ``PySide6``。
+
+無頭 API
+--------
+
+.. code-block:: python
+
+    from je_auto_control import wait_for_focus_change, click_text
+
+    click_text("Settings")
+    focused = wait_for_focus_change(timeout=3)
+    # {"name": "Search", "role": "ControlType_50004", "app_name": "app.exe", ...}
+    if focused is not None:
+        ...   # 焦點已移動——對話框 / 下一個欄位已就緒
+
+回傳新獲得焦點的元素 ``{name, role, app_name, bounds, …}``,若在 ``timeout`` 秒內(預設 ``5``)
+沒有焦點變化則回傳 ``None``。
+
+執行器指令
+----------
+
+``AC_wait_for_focus_change``(``timeout``)回傳 ``{changed, element}``。以唯讀
+``ac_wait_for_focus_change`` MCP 工具及 Script Builder 指令(位於 **Native UI** 分類下)
+形式提供。

je_auto_control/__init__.py

@@ -89,6 +89,8 @@
 from je_auto_control.utils.selection_view import (
     get_selection, list_views, set_view,
 )
+# Reactive UIA event waits (focus-changed)
+from je_auto_control.utils.ax_events import wait_for_focus_change
 # Rich clipboard formats — RTF + CSV/TSV codecs and Windows get / set
 from je_auto_control.utils.clipboard_rich_formats import (
     build_rtf, csv_to_rows, get_clipboard_csv, get_clipboard_rtf, rows_to_csv,
@@ -1697,6 +1699,7 @@ def start_autocontrol_gui(*args, **kwargs):
     "window_interaction_state",
     "legacy_info", "legacy_default_action",
     "get_selection", "list_views", "set_view",
+    "wait_for_focus_change",
     "build_rtf", "rtf_to_text", "rows_to_csv", "csv_to_rows",
     "set_clipboard_rtf", "get_clipboard_rtf",
     "set_clipboard_csv", "get_clipboard_csv",

je_auto_control/gui/script_builder/command_schema.py

@@ -1691,6 +1691,12 @@ def _add_native_control_specs(specs: List[CommandSpec]) -> None:
         fields=(FieldSpec("view", FieldType.STRING),) + fields,
         description="Switch a control to the named view (MultipleViewPattern).",
     ))
+    specs.append(CommandSpec(
+        "AC_wait_for_focus_change", "Native UI", "Wait for Focus Change",
+        fields=(FieldSpec("timeout", FieldType.FLOAT, optional=True,
+                          default=5.0),),
+        description="Block until keyboard focus moves (real UIA focus event).",
+    ))
     specs.append(CommandSpec(
         "AC_get_control_text", "Native UI", "Get Control Text",
         fields=fields,

je_auto_control/utils/accessibility/backends/base.py

@@ -281,6 +281,18 @@
         """Switch a control to the named view (MultipleViewPattern); True on success."""
         self._unsupported("set_view")
 
+    # --- reactive events (UIA event subscription) --------------------------
+
+    def wait_for_focus_change(self, timeout: float = 5.0,
+                              ) -> Optional[Dict[str, Any]]:
+        """Block until the keyboard focus moves, then return the newly-focused
+        element ``{name, role, app_name, ...}`` (or None on ``timeout``).
+
+        A zero-latency native wait (UIA AddFocusChangedEventHandler) — unlike the
+        polling recorder, it can't miss a fast focus transition.
+        """
+        self._unsupported("wait_for_focus_change")
+
     def _unsupported(self, operation: str):
         """Raise a clear error for an action this backend can't perform."""
         raise AccessibilityNotAvailableError(

je_auto_control/utils/accessibility/backends/windows_backend.py

@@ -62,9 +62,11 @@ class WindowsAccessibilityBackend(AccessibilityBackend):
     name = "windows-uia"
 
     def __init__(self) -> None:
+        import threading
         self.available = _is_available()
         self._automation = None
         self._uia_module = None
+        self._event_lock = threading.Lock()
 
     def _ensure_automation(self):
         if self._automation is not None:
@@ -420,6 +422,48 @@ def set_view(self, view="", name=None, role=None, app_name=None,
             return False
         return False
 
+    def _make_focus_handler(self, sink):
+        """Build a COM focus-changed event handler that puts events on ``sink``."""
+        try:
+            import comtypes
+            interface = self._uia_module.IUIAutomationFocusChangedEventHandler
+        except (ImportError, AttributeError):
+            return None
+
+        class _FocusHandler(comtypes.COMObject):
+            _com_interfaces_ = [interface]
+
+            def IUIAutomationFocusChangedEventHandler_HandleFocusChangedEvent(
+                    self, sender):  # noqa: N802  # reason: comtypes callback name
+                element = _convert_uia(sender)
+                sink.put(element.to_dict() if element else {"focused": True})
+                return 0
+
+        return _FocusHandler()
+
+    def wait_for_focus_change(self, timeout=5.0) -> Optional[Dict[str, Any]]:
+        import queue
+        automation = self._ensure_automation()
+        events: "queue.Queue" = queue.Queue()
+        handler = self._make_focus_handler(events)
+        if handler is None:
+            return None
+        try:
+            with self._event_lock:
+                automation.AddFocusChangedEventHandler(None, handler)
+        except (OSError, AttributeError):
+            return None
+        try:
+            return events.get(timeout=float(timeout))
+        except queue.Empty:
+            return None
+        finally:
+            with self._event_lock:
+                try:
+                    automation.RemoveFocusChangedEventHandler(handler)
+                except (OSError, AttributeError):
+                    pass
+
     def get_table_headers(self, name=None, role=None, app_name=None,
                           automation_id=None) -> Optional[Dict[str, Any]]:
         raw = self._find_raw(name, role, app_name, automation_id)

je_auto_control/utils/ax_events/__init__.py

@@ -0,0 +1,4 @@
+"""Reactive UIA event waits (focus-changed) via the accessibility backend seam."""
+from je_auto_control.utils.ax_events.ax_events import wait_for_focus_change
+
+__all__ = ["wait_for_focus_change"]

je_auto_control/utils/ax_events/ax_events.py

@@ -0,0 +1,25 @@
+"""Reactive UI Automation event waits (focus-changed).
+
+The accessibility recorder *polls* the focused element every ~250 ms, so it can
+miss a fast focus transition and reacts a quarter-second late. UIA exposes real
+events: ``wait_for_focus_change`` blocks on the native
+``AddFocusChangedEventHandler`` and returns the moment focus moves — the
+zero-latency, miss-free "wait until focus lands on the dialog" primitive, the
+accessibility-tree analogue of ``wait_for_window`` / ``wait_for_image``.
+
+It is a thin dispatch onto the injectable ``accessibility.backends.get_backend()``
+seam — headless-testable on any platform by injecting a fake backend; the real
+event subscription (registered / unregistered under a lock, on the calling
+thread) lives in the Windows backend. Imports no ``PySide6``.
+"""
+from typing import Any, Dict, Optional
+
+
+def wait_for_focus_change(*, timeout: float = 5.0) -> Optional[Dict[str, Any]]:
+    """Block until the keyboard focus moves, then return the newly-focused element.
+
+    Returns the focused element as ``{name, role, app_name, bounds, ...}``, or
+    ``None`` if no focus change occurs within ``timeout`` seconds.
+    """
+    from je_auto_control.utils.accessibility.backends import get_backend
+    return get_backend().wait_for_focus_change(float(timeout))

je_auto_control/utils/executor/action_executor.py

@@ -2602,6 +2602,13 @@ def _set_view(view: str, name: Optional[str] = None, role: Optional[str] = None,
                     automation_id=automation_id)
 
 
+def _wait_for_focus_change(timeout: Any = 5.0) -> Dict[str, Any]:
+    """Adapter: block until the keyboard focus moves (UIA focus event)."""
+    from je_auto_control.utils.ax_events import wait_for_focus_change
+    element = wait_for_focus_change(timeout=float(timeout))
+    return {"changed": element is not None, "element": element}
+
+
 def _get_control_text(name: Optional[str] = None, role: Optional[str] = None,
                       app_name: Optional[str] = None,
                       automation_id: Optional[str] = None) -> Dict[str, Any]:
@@ -6591,6 +6598,7 @@ def __init__(self):
             "AC_get_selection": _get_selection,
             "AC_list_views": _list_views,
             "AC_set_view": _set_view,
+            "AC_wait_for_focus_change": _wait_for_focus_change,
             "AC_get_control_text": _get_control_text,
             "AC_find_control_text": _find_control_text,
             "AC_select_control_text": _select_control_text,

je_auto_control/utils/mcp_server/tools/_factories.py

@@ -1310,6 +1310,16 @@ def a11y_control_tools() -> List[MCPTool]:
             handler=h.set_view,
             annotations=DESTRUCTIVE,
         ),
+        MCPTool(
+            name="ac_wait_for_focus_change",
+            description=("Block until the keyboard focus moves (real UIA focus "
+                         "event — zero-latency, miss-free, unlike polling). "
+                         "Returns {changed, element} or {changed: false} on "
+                         "'timeout' seconds (default 5)."),
+            input_schema=schema({"timeout": {"type": "number"}}),
+            handler=h.wait_for_focus_change,
+            annotations=READ_ONLY,
+        ),
         MCPTool(
             name="ac_get_control_text",
             description=("Read a control's full text via TextPattern: "

je_auto_control/utils/mcp_server/tools/_handlers.py

@@ -900,6 +900,12 @@ def set_view(view, name=None, role=None, app_name=None, automation_id=None):
     return _set_view(view, name, role, app_name, automation_id)
 
 
+def wait_for_focus_change(timeout=5.0):
+    from je_auto_control.utils.executor.action_executor import (
+        _wait_for_focus_change)
+    return _wait_for_focus_change(timeout)
+
+
 def get_selected_text(name=None, role=None, app_name=None, automation_id=None):
     from je_auto_control.utils.executor.action_executor import _get_selected_text
     return _get_selected_text(name, role, app_name, automation_id)

test/unit_test/headless/test_ax_events_batch.py

@@ -0,0 +1,77 @@
+"""Headless tests for reactive UIA focus-change waits (fake backend seam)."""
+import je_auto_control as ac
+from je_auto_control.utils.accessibility.backends import base as backend_base
+from je_auto_control.utils.ax_events import wait_for_focus_change
+
+_FOCUSED = {"name": "Username", "role": "ControlType_50004",
+            "app_name": "login.exe", "bounds": [10, 20, 200, 24]}
+
+
+class _FakeBackend(backend_base.AccessibilityBackend):
+    name = "fake"
+    available = True
+
+    def __init__(self, result=None):
+        self.result = result
+        self.timeouts = []
+
+    def wait_for_focus_change(self, timeout=5.0):
+        self.timeouts.append(timeout)
+        return self.result
+
+
+def _inject(monkeypatch, backend):
+    import je_auto_control.utils.accessibility.backends as backends
+    monkeypatch.setattr(backends, "_cached_backend", backend, raising=False)
+
+
+def test_focus_change_returns_element(monkeypatch):
+    fake = _FakeBackend(dict(_FOCUSED))
+    _inject(monkeypatch, fake)
+    assert wait_for_focus_change(timeout=2.0) == _FOCUSED
+    assert fake.timeouts == [2.0]
+
+
+def test_focus_change_timeout_returns_none(monkeypatch):
+    _inject(monkeypatch, _FakeBackend(None))
+    assert wait_for_focus_change(timeout=0.1) is None
+
+
+def test_unsupported_backend_raises(monkeypatch):
+    from je_auto_control.utils.accessibility.element import (
+        AccessibilityNotAvailableError)
+    _inject(monkeypatch, backend_base.AccessibilityBackend())  # all _unsupported
+    try:
+        wait_for_focus_change(timeout=0.1)
+        raised = False
+    except AccessibilityNotAvailableError:
+        raised = True
+    assert raised is True
+
+
+# --- wiring ---------------------------------------------------------------
+
+def test_executor_adapter_wraps_element(monkeypatch):
+    _inject(monkeypatch, _FakeBackend(dict(_FOCUSED)))
+    from je_auto_control.utils.executor.action_executor import (
+        _wait_for_focus_change)
+    out = _wait_for_focus_change(1.5)
+    assert out["changed"] is True and out["element"]["name"] == "Username"
+    _inject(monkeypatch, _FakeBackend(None))
+    assert _wait_for_focus_change(0.1)["changed"] is False
+
+
+def test_wiring():
+    known = set(ac.executor.known_commands())
+    assert "AC_wait_for_focus_change" in known
+    from je_auto_control.utils.mcp_server.tools import build_default_tool_registry
+    names = {t.name for t in build_default_tool_registry()}
+    assert "ac_wait_for_focus_change" in names
+    from je_auto_control.gui.script_builder.command_schema import _build_specs
+    specs = {s.command for s in _build_specs()}
+    assert "AC_wait_for_focus_change" in specs
+
+
+def test_facade_export():
+    assert hasattr(ac, "wait_for_focus_change")
+    assert "wait_for_focus_change" in ac.__all__