From 3b39e4dcd6c56fd7890ee0eec1e77dddfd6384cc Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 04:13:28 -0500 Subject: [PATCH 1/2] Native backends expose focused_field_label from the a11y tree MacOSBackend reads AXTitleUIElement then AXTitle/AXDescription; WindowsBackend reads UIA LabeledBy then Name via a new focused-field-label agent route; LinuxBackend reads AT-SPI LABELLED_BY then the accessible name. Unit tests cover the priority and fail-closed None path. Real Chart.swift / chart_gtk.py proofs still need the lab VM (Part of #1868). --- .../openadapt_flow/backends/linux_backend.py | 74 +++++++ .../openadapt_flow/backends/macos_backend.py | 54 +++++ .../backends/win_agent/server.py | 51 +++++ .../backends/windows_backend.py | 18 ++ .../tests/test_native_focused_field_label.py | 197 ++++++++++++++++++ 5 files changed, 394 insertions(+) create mode 100644 packages/flow/tests/test_native_focused_field_label.py diff --git a/packages/flow/openadapt_flow/backends/linux_backend.py b/packages/flow/openadapt_flow/backends/linux_backend.py index 3ac51c8ed1..918e0f017f 100644 --- a/packages/flow/openadapt_flow/backends/linux_backend.py +++ b/packages/flow/openadapt_flow/backends/linux_backend.py @@ -699,6 +699,36 @@ def text_value_at(self, x: int, y: int) -> Optional[str]: return None return value if isinstance(value, str) else None + def focused_field_label(self) -> Optional[str]: + """Return the focused field's AT-SPI label, or None (never raises). + + RECORD-TIME seam (:class:`openadapt_flow.backend.FieldLabelBackend`): + reads the focused element's ``LABELLED_BY`` relation, then its + accessible name. Never by position; None when there is no label so + compile-time OCR fallback still applies. + """ + + try: + window = ( + self._require_same_window(self._captured_window) + if self._captured_window is not None + else self._resolve_window() + ) + element = self._client.focused_element(window) + if element is None or not self._in_scope(element, window): + return None + reader = getattr(self._client, "field_label", None) + if callable(reader): + value = reader(element) + else: + value = element.name + except Exception: + return None + if isinstance(value, str): + text = " ".join(value.split()) + return text or None + return None + def focused_text_value(self) -> Optional[str]: """The exact text of the one focused editable field in this window.""" reader = getattr(self._client, "editable_text", None) @@ -1696,6 +1726,50 @@ def perform_native(self, element: LinuxElement, operation: str) -> str: #: Raw AT-SPI roles whose text is an editable value (not a password). _EDITABLE_ROLES = frozenset({"text", "entry", "editbar"}) + def field_label(self, element: LinuxElement) -> Optional[str]: + """LABELLED_BY relation name, else the accessible name; None if empty.""" + + node = element.native + if node is not None: + labelled = self._labelled_by_name(node) + if labelled: + return labelled + name = _clean_text(self._call(node, "get_name", default=None)) + if name: + return name + return _clean_text(element.name) + + def _labelled_by_name(self, node: Any) -> Optional[str]: + """Name of the first LABELLED_BY relation target, if any.""" + + relation_set = self._call(node, "get_relation_set", default=None) + if not relation_set: + return None + labelled_by = getattr(self._atspi.RelationType, "LABELLED_BY", None) + if labelled_by is None: + return None + try: + relations = list(relation_set) + except TypeError: + relations = relation_set if isinstance(relation_set, (list, tuple)) else [] + for relation in relations: + rel_type = self._call(relation, "get_relation_type", default=None) + if rel_type != labelled_by: + continue + target = None + get_target = getattr(relation, "get_target", None) + if callable(get_target): + try: + target = get_target(0) + except Exception: + target = None + if target is None: + continue + name = _clean_text(self._call(target, "get_name", default=None)) + if name: + return name + return None + def editable_text(self, element: LinuxElement) -> Optional[str]: """The full text of an editable text field; None for anything else.""" node = element.native diff --git a/packages/flow/openadapt_flow/backends/macos_backend.py b/packages/flow/openadapt_flow/backends/macos_backend.py index 8d3f351492..476a508506 100644 --- a/packages/flow/openadapt_flow/backends/macos_backend.py +++ b/packages/flow/openadapt_flow/backends/macos_backend.py @@ -345,6 +345,28 @@ def _fingerprint(element: MacElement) -> str: return hashlib.sha256(canonical.encode("utf-8")).hexdigest() +def _ax_field_label(element: Any, copy_attr) -> Optional[str]: + """Best human label for one focused AX element, or None. + + Prefers ``AXTitleUIElement`` (the associated label control), then the + element's own ``AXTitle`` / ``AXDescription``. Used by + :meth:`MacOSBackend.focused_field_label`; ``copy_attr`` is injectable so + unit tests can exercise the priority without ApplicationServices. + """ + + title_ui = copy_attr(element, "AXTitleUIElement") + if title_ui is not None: + for attribute in ("AXTitle", "AXDescription"): + text = _clean_text(copy_attr(title_ui, attribute)) + if text: + return text + for attribute in ("AXTitle", "AXDescription"): + text = _clean_text(copy_attr(element, attribute)) + if text: + return text + return None + + class QuartzMacAXClient: """Live :class:`MacAXClient` over the macOS Accessibility (AX) API. @@ -1887,6 +1909,38 @@ def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]: candidate["dx"] = candidate["dy"] = None return census + def focused_field_label(self) -> Optional[str]: + """Return the focused field's AX label, or None (never raises). + + RECORD-TIME seam (:class:`openadapt_flow.backend.FieldLabelBackend`): + reads ``AXTitleUIElement`` then ``AXTitle`` / ``AXDescription`` on the + one focused AX element in this window. Never by position; None when + there is no label so compile-time OCR fallback still applies. + """ + + try: + window = ( + self._captured_window + if self._captured_window is not None + else self._resolve_window() + ) + if window is None: + return None + token = self._mac_client.focused_element_token(window) + if token is None: + return None + from ApplicationServices import AXUIElementCopyAttributeValue + + def copy_attr(element: Any, attribute: str) -> Any: + error, value = AXUIElementCopyAttributeValue(element, attribute, None) + if error != 0: + return None + return value + + return _ax_field_label(token, copy_attr) + except Exception: # noqa: BLE001 - a momentary AX failure is not fatal + return None + def displayed_text_at(self, x: int, y: int) -> Optional[str]: """The text of the one AX element under (x, y), for a read-back. diff --git a/packages/flow/openadapt_flow/backends/win_agent/server.py b/packages/flow/openadapt_flow/backends/win_agent/server.py index 240518d94d..0515189aeb 100644 --- a/packages/flow/openadapt_flow/backends/win_agent/server.py +++ b/packages/flow/openadapt_flow/backends/win_agent/server.py @@ -756,6 +756,52 @@ def _target_fingerprint(control: object, candidate: dict[str, Any]) -> str: return hashlib.sha256(canonical.encode("utf-8")).hexdigest() +def _uia_labeled_by(control: object) -> Optional[object]: + """The UIA LabeledBy target for ``control``, or None.""" + + # uiautomation exposes LabeledBy as property id 30092; prefer a named + # constant when the package provides one. + try: + import uiautomation as auto # noqa: PLC0415 + except Exception: + auto = None + prop_id = 30092 + if auto is not None: + prop_id = int(getattr(auto, "LabeledByProperty", prop_id) or prop_id) + getter = getattr(control, "GetPropertyValue", None) + if not callable(getter): + return None + try: + labeled = getter(prop_id) + except Exception: # noqa: BLE001 + return None + return labeled if labeled is not None else None + + +def _uia_field_label(control: object) -> Optional[str]: + """Best human label for one focused UIA control: LabeledBy then Name.""" + + labeled = _uia_labeled_by(control) + if labeled is not None: + name = " ".join(str(_control_value(labeled, "Name", "") or "").split()) + if name: + return name + name = " ".join(str(_control_value(control, "Name", "") or "").split()) + return name or None + + +def _focused_field_label(auto: Any) -> Optional[str]: + """Label of the one focused UIA control, or None.""" + + try: + focused = auto.GetFocusedControl() + except Exception: # noqa: BLE001 + return None + if focused is None: + return None + return _uia_field_label(focused) + + def _candidate(control: object) -> Optional[dict[str, Any]]: locator = _locator_from_control(control) bounds = _bounds(control) @@ -1333,6 +1379,10 @@ def _perform_uia_initialized( locator = None return {"status": "ok", "locator": locator} + if operation == "focused-field-label": + _exact_object(payload, required=frozenset(), label=operation) + return {"status": "ok", "label": _focused_field_label(auto)} + if operation == "tree": data = _exact_object( payload, @@ -2634,6 +2684,7 @@ def do_POST(self) -> None: # noqa: N802 - stdlib naming "/window/root-at": ("uia", "window-root-at"), "/window/owned-over": ("uia", "owned-window-over"), "/uia/text-at-point": ("uia", "text-at-point"), + "/uia/focused-field-label": ("uia", "focused-field-label"), "/uia/identity-candidates-at": ("uia", "identity-candidates-at"), "/uia/find": ("uia", "find"), "/uia/act": ("uia", "act"), diff --git a/packages/flow/openadapt_flow/backends/windows_backend.py b/packages/flow/openadapt_flow/backends/windows_backend.py index 9b69963ed2..3334d07645 100644 --- a/packages/flow/openadapt_flow/backends/windows_backend.py +++ b/packages/flow/openadapt_flow/backends/windows_backend.py @@ -681,6 +681,24 @@ def workflow_state_identity(self) -> Optional[str]: # -- structured-text identity (openadapt_flow.backend.IdentityBackend) -- + def focused_field_label(self) -> Optional[str]: + """Return the focused field's UIA label, or None (never raises). + + RECORD-TIME seam (:class:`openadapt_flow.backend.FieldLabelBackend`): + reads UIA ``LabeledBy`` then ``Name`` on the focused control via the + in-guest agent. Never by position; None when there is no label so + compile-time OCR fallback still applies. + """ + + typed = self._post_typed_read("/uia/focused-field-label", {}) + if not typed.available or typed.payload is None: + return None + value = typed.payload.get("label") + if isinstance(value, str): + text = " ".join(value.split()) + return text or None + return None + def structured_text_at(self, x: int, y: int) -> Optional[str]: """Return the UI Automation text of the element/row under (x, y). diff --git a/packages/flow/tests/test_native_focused_field_label.py b/packages/flow/tests/test_native_focused_field_label.py new file mode 100644 index 0000000000..9d122e8bf9 --- /dev/null +++ b/packages/flow/tests/test_native_focused_field_label.py @@ -0,0 +1,197 @@ +"""Native backends expose focused_field_label from the accessibility tree (#1868). + +Unit coverage for the three substrates. Real-app proofs (Chart.swift / +chart_gtk.py) need the macOS lab VM and stay on the parent issue. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest + +from openadapt_flow.backend import FieldLabelBackend +from openadapt_flow.backends import linux_backend as linux_backend_module +from openadapt_flow.backends.linux_backend import LinuxElement +from openadapt_flow.backends.macos_backend import _ax_field_label +from openadapt_flow.backends.win_agent import server as win_agent_server +from openadapt_flow.backends.windows_backend import WindowsBackend, _TypedResponse +from tests.test_linux_backend import TEXT_ELEMENT, FakeLinuxClient +from tests.test_linux_backend import backend as linux_be + + +def test_ax_field_label_prefers_title_ui_element() -> None: + attrs = { + "field": { + "AXTitleUIElement": "label-el", + "AXTitle": "ignored-own-title", + "AXDescription": "ignored-own-desc", + }, + "label-el": { + "AXTitle": "Insurance No.", + "AXDescription": "unused", + }, + } + + def copy_attr(element: Any, attribute: str) -> Any: + return attrs.get(element, {}).get(attribute) + + assert _ax_field_label("field", copy_attr) == "Insurance No." + + +def test_ax_field_label_falls_back_to_own_title_then_description() -> None: + def copy_title(element: Any, attribute: str) -> Any: + if attribute == "AXTitleUIElement": + return None + if attribute == "AXTitle": + return " Policy # " + return None + + assert _ax_field_label("field", copy_title) == "Policy #" + + def copy_desc(element: Any, attribute: str) -> Any: + if attribute == "AXDescription": + return "member_id" + return None + + assert _ax_field_label("field", copy_desc) == "member_id" + assert _ax_field_label("field", lambda *_: None) is None + + +def test_linux_backend_focused_field_label_reads_client_then_name() -> None: + class LabelClient(FakeLinuxClient): + def field_label(self, element): + self.calls.append(("field-label", element.native_id)) + return "Insurance No." + + target = linux_be(LabelClient()) + assert isinstance(target, FieldLabelBackend) + assert target.focused_field_label() == "Insurance No." + + plain = linux_be(FakeLinuxClient()) + # TEXT_ELEMENT.name is "Document" + assert plain.focused_field_label() == TEXT_ELEMENT.name + + missing = FakeLinuxClient() + missing.focused = None + assert linux_be(missing).focused_field_label() is None + + +def test_linux_backend_focused_field_label_out_of_scope_is_none() -> None: + client = FakeLinuxClient() + client.focused = LinuxElement( + "0.1:9", + "body", + "textbox", + "Document", + "gedit", + "other.txt", + 9001, + (120, 260, 580, 380), + ("focus",), + ) + assert linux_be(client).focused_field_label() is None + + +def test_atspi_field_label_prefers_labelled_by() -> None: + client = object.__new__(linux_backend_module.AtspiLinuxClient) + labelled_by = object() + client._atspi = SimpleNamespace( + RelationType=SimpleNamespace(LABELLED_BY=labelled_by) + ) + + class Rel: + def __init__(self, rel_type, target): + self._type = rel_type + self._target = target + + def get_relation_type(self): + return self._type + + def get_target(self, index): + assert index == 0 + return self._target + + class Node: + def __init__(self, name, relations=None): + self._name = name + self._relations = relations or [] + + def get_name(self): + return self._name + + def get_relation_set(self): + return self._relations + + label_node = Node("Chart note") + field_node = Node( + "ignored-own-name", + relations=[Rel(labelled_by, label_node)], + ) + element = LinuxElement( + "0.1", + None, + "textbox", + "ignored-own-name", + "chart", + "Chart", + 1, + (1, 1, 10, 10), + ("focus",), + native=field_node, + ) + assert client.field_label(element) == "Chart note" + + bare = LinuxElement( + "0.2", + None, + "textbox", + "Own Name", + "chart", + "Chart", + 1, + (1, 1, 10, 10), + ("focus",), + native=Node("Own Name"), + ) + assert client.field_label(bare) == "Own Name" + + +def test_uia_field_label_prefers_labeled_by_then_name() -> None: + class Ctrl: + def __init__(self, name, labeled_by=None, prop_id=30092): + self.Name = name + self._labeled_by = labeled_by + self._prop_id = prop_id + + def GetPropertyValue(self, prop_id): + if prop_id == self._prop_id: + return self._labeled_by + return None + + label = Ctrl("Insurance No.") + field = Ctrl("ignored", labeled_by=label) + assert win_agent_server._uia_field_label(field) == "Insurance No." + assert win_agent_server._uia_field_label(Ctrl("Policy #")) == "Policy #" + assert win_agent_server._uia_field_label(Ctrl("")) is None + + +def test_windows_backend_focused_field_label_reads_typed_route( + monkeypatch: pytest.MonkeyPatch, +) -> None: + backend = WindowsBackend("http://127.0.0.1:9", timeout_s=0.5) + + def fake_post(path: str, payload: dict) -> _TypedResponse: + assert path == "/uia/focused-field-label" + assert payload == {} + return _TypedResponse(True, {"status": "ok", "label": " Policy # "}) + + monkeypatch.setattr(backend, "_post_typed_read", fake_post) + assert isinstance(backend, FieldLabelBackend) + assert backend.focused_field_label() == "Policy #" + + monkeypatch.setattr( + backend, "_post_typed_read", lambda *a, **k: _TypedResponse(False, None) + ) + assert backend.focused_field_label() is None From e3eb8850dadcec4399dc271a7b8f9d21cd0a48e3 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 04:19:23 -0500 Subject: [PATCH 2/2] fix(flow): type LABELLED_BY relations as list for mypy Keep the AT-SPI relation walk list-typed so mypy accepts the tuple fallback without an assignment conflict. --- packages/flow/openadapt_flow/backends/linux_backend.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/flow/openadapt_flow/backends/linux_backend.py b/packages/flow/openadapt_flow/backends/linux_backend.py index 918e0f017f..ad1a7d5fc5 100644 --- a/packages/flow/openadapt_flow/backends/linux_backend.py +++ b/packages/flow/openadapt_flow/backends/linux_backend.py @@ -1748,10 +1748,14 @@ def _labelled_by_name(self, node: Any) -> Optional[str]: labelled_by = getattr(self._atspi.RelationType, "LABELLED_BY", None) if labelled_by is None: return None + relations: list[Any] = [] try: relations = list(relation_set) except TypeError: - relations = relation_set if isinstance(relation_set, (list, tuple)) else [] + if isinstance(relation_set, list): + relations = relation_set + elif isinstance(relation_set, tuple): + relations = list(relation_set) for relation in relations: rel_type = self._call(relation, "get_relation_type", default=None) if rel_type != labelled_by: