From 2875722b029deb2a339e12d9b32a70d5e572e6e4 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 24 Oct 2026 20:02:13 -0400 Subject: [PATCH 1/7] feat(flow): count look-alike rows on native and remote surfaces PR #1504 halts a row pick before a write when the surface cannot count the rows that read the same (identity_uncounted). This gives the native and remote-display backends a census, so those workflows run again without failing open. - identity_census.tree_row_census: one rule for accessibility trees. Find the target's record row, read every sibling row at the same child path the way structured_text_at reads it, and report the matches with their record ids. A container that reports more rows than it exposes, or a row that isn't built yet, is incomplete. Position-only ids are dropped. - macOS: QuartzMacAXClient.row_census over AXTable/AXOutline rows and AXList items; record ids are AXIdentifier and AXURL. The backend converts row offsets to captured pixels. - Linux: AtspiLinuxClient.row_census over list items, table rows and tree items, reading rows exactly as structured_text does; record ids are AT-SPI accessible ids. - Windows: the guest agent serves /uia/identity-candidates-at (a stdlib mirror of the rule) over DataItem/ListItem/TreeItem/TableRow siblings; record ids are non-positional AutomationIds. A Grid pattern RowCount above the exposed rows, or a scrollable ItemContainer list, is incomplete. - Remote display: a pixel identity with no OCR band now counts the places the identifier crop verifies (vision.match.template_peaks plus the same pixel verifier), instead of halting as uncounted. Tests: unit tests over fake trees for each backend (identical rows halt, distinct ids pick the row, a unique row runs, positional ids, incomplete containers) and replayer-level native cases. New opt-in real-app tests: a WPF ListBox over UI Automation (must run on the new Windows CI leg) and a Swift NSTableView over AX (skips when the runner has no Accessibility access). Co-Authored-By: Claude Opus 5.5 --- .github/workflows/packages.yml | 58 ++ llms-full.txt | 25 + packages/flow/docs/LIMITS.md | 25 + .../openadapt_flow/backends/linux_backend.py | 110 +++- .../openadapt_flow/backends/macos_backend.py | 154 +++++ .../backends/win_agent/server.py | 189 +++++- .../backends/windows_backend.py | 18 +- .../openadapt_flow/runtime/identity_census.py | 185 +++++- .../flow/openadapt_flow/runtime/replayer.py | 62 +- packages/flow/openadapt_flow/vision/match.py | 23 + .../flow/tests/native_apps/Worklist.swift | 33 ++ .../test_identity_evidence_capability.py | 26 +- .../tests/test_native_census_real_apps.py | 301 ++++++++++ .../flow/tests/test_native_identity_census.py | 560 ++++++++++++++++++ 14 files changed, 1752 insertions(+), 17 deletions(-) create mode 100644 packages/flow/tests/native_apps/Worklist.swift create mode 100644 packages/flow/tests/test_native_census_real_apps.py create mode 100644 packages/flow/tests/test_native_identity_census.py diff --git a/.github/workflows/packages.yml b/.github/workflows/packages.yml index 20556c037..baedac0d2 100644 --- a/.github/workflows/packages.yml +++ b/.github/workflows/packages.yml @@ -476,6 +476,63 @@ jobs: sys.exit(1 if bad else 0) PY + # Real native list apps read through the row census (identical rows halt, + # record ids pick the row, a unique row runs): a WPF ListBox over UI + # Automation on Windows, which must run with no skips, and a Swift + # NSTableView over AX on macOS, which skips when the hosted runner gives the + # job no Accessibility access (the step summary says so). + flow-native-census: + name: flow native row census (${{ matrix.os }}) + needs: scope + if: needs.scope.outputs.flow == 'true' + runs-on: ${{ matrix.os }} + timeout-minutes: 25 + strategy: + fail-fast: false + matrix: + os: [windows-latest, macos-latest] + defaults: + run: + working-directory: packages/flow + shell: bash + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: "3.12" + cache: pip + - name: Install + run: | + pip install -e ../types -e . pytest pytest-timeout + if [ "$RUNNER_OS" = "Windows" ]; then pip install uiautomation; fi + if [ "$RUNNER_OS" = "macOS" ]; then + pip install pyobjc-framework-ApplicationServices pyobjc-framework-Cocoa pyobjc-framework-Quartz + fi + - name: Census unit tests and real-app tests + env: + OPENADAPT_TEST_WINDOWS_UIA: "1" + OPENADAPT_TEST_MACOS_AX: "1" + run: | + mkdir -p runs + pytest -v -rs tests/test_native_identity_census.py \ + tests/test_native_census_real_apps.py --junitxml=runs/native-junit.xml + - name: The Windows real-app tests ran + if: runner.os == 'Windows' + run: | + python - <<'PY' + import sys + import xml.etree.ElementTree as ET + cases = [ + c for c in ET.parse("runs/native-junit.xml").getroot().iter("testcase") + if c.get("name", "").startswith("test_windows_wpf_") + ] + skipped = [c.get("name") for c in cases if c.find("skipped") is not None] + print(f"{len(cases)} WPF tests, skipped: {skipped}") + sys.exit(1 if not cases or skipped else 0) + PY + capture-linux: name: capture (linux, py${{ matrix.python-version }}) needs: scope @@ -926,6 +983,7 @@ jobs: - flow-lint - flow-unit - flow-critical + - flow-native-census - capture-linux - capture-macos - capture-windows diff --git a/llms-full.txt b/llms-full.txt index 83e688986..bfed0cd0c 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -1335,6 +1335,31 @@ For an armed step: run's value into the recorded band and requires the whole band to match. A band that rests on a glyph-confusable identifier keeps the pixel comparison. +A verified identity says the row reads like the recorded row, not that no +other row reads the same. Before an armed step acts, the runtime counts the +rows that show the same identity (`runtime/identity_census.py`). When more +than one does, it acts only on the row with the record id recorded at teach +time, or halts. How each surface counts: + +- browser: every shown row in the page, with app-owned `data-*` ids and row + links as record ids; +- Windows (UI Automation agent): the sibling rows of the clicked row, with + non-positional AutomationIds as record ids. A grid that reports more rows + than it exposes, or a scrollable list that builds items on demand, cannot be + counted, and a row pick before a write halts; +- macOS (AX): the rows of the `AXTable`, `AXOutline` or `AXList` around the + element, with the AX identifier and URL as record ids. A table that reports + more `AXRows` than it exposes cannot be counted; +- Linux (AT-SPI): the sibling list items, table rows or tree items of the + clicked row, with AT-SPI accessible ids as record ids. GTK 3 list boxes and + tree views expose no row identity under the structured reading (one child + per row, or cells without a row), so the identity check there rests on OCR; +- pixel-only displays: the OCR rows of the frame, or, with no OCR band, the + places the identifier crop verifies. + +A surface that cannot count the rows halts any row pick that leads into a +write. + On a pixel-only display, the OCR tier also stops the correct record whenever its own identifier contains a look-alike glyph (`O`/`0`, `l`/`1`, `B`/`8`). A deployment can remove most of that cost by giving the runtime a complete diff --git a/packages/flow/docs/LIMITS.md b/packages/flow/docs/LIMITS.md index 4211a7e91..499753072 100644 --- a/packages/flow/docs/LIMITS.md +++ b/packages/flow/docs/LIMITS.md @@ -185,6 +185,31 @@ For an armed step: run's value into the recorded band and requires the whole band to match. A band that rests on a glyph-confusable identifier keeps the pixel comparison. +A verified identity says the row reads like the recorded row, not that no +other row reads the same. Before an armed step acts, the runtime counts the +rows that show the same identity (`runtime/identity_census.py`). When more +than one does, it acts only on the row with the record id recorded at teach +time, or halts. How each surface counts: + +- browser: every shown row in the page, with app-owned `data-*` ids and row + links as record ids; +- Windows (UI Automation agent): the sibling rows of the clicked row, with + non-positional AutomationIds as record ids. A grid that reports more rows + than it exposes, or a scrollable list that builds items on demand, cannot be + counted, and a row pick before a write halts; +- macOS (AX): the rows of the `AXTable`, `AXOutline` or `AXList` around the + element, with the AX identifier and URL as record ids. A table that reports + more `AXRows` than it exposes cannot be counted; +- Linux (AT-SPI): the sibling list items, table rows or tree items of the + clicked row, with AT-SPI accessible ids as record ids. GTK 3 list boxes and + tree views expose no row identity under the structured reading (one child + per row, or cells without a row), so the identity check there rests on OCR; +- pixel-only displays: the OCR rows of the frame, or, with no OCR band, the + places the identifier crop verifies. + +A surface that cannot count the rows halts any row pick that leads into a +write. + On a pixel-only display, the OCR tier also stops the correct record whenever its own identifier contains a look-alike glyph (`O`/`0`, `l`/`1`, `B`/`8`). A deployment can remove most of that cost by giving the runtime a complete diff --git a/packages/flow/openadapt_flow/backends/linux_backend.py b/packages/flow/openadapt_flow/backends/linux_backend.py index fcebd00a7..0a62bfb7d 100644 --- a/packages/flow/openadapt_flow/backends/linux_backend.py +++ b/packages/flow/openadapt_flow/backends/linux_backend.py @@ -585,6 +585,25 @@ def locate_structural( supported_operations=list(candidate.supported_operations), ) + def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the row at (x, y). + + See :class:`openadapt_flow.backend.IdentityCandidatesBackend`. Uses the + client's optional ``row_census``; a client without it cannot count + rows (None). Offsets are global pixels, the same scale as clicks. + """ + census_at = getattr(self._client, "row_census", None) + if not callable(census_at): + return None + try: + window, global_x, global_y = self._global_point( + int(x), int(y), require_active=False + ) + census = census_at(window, global_x, global_y, limit=2000) + except Exception: + return None + return census if isinstance(census, dict) else None + def displayed_text_at(self, x: int, y: int) -> Optional[str]: """The text of the one AT-SPI element under (x, y), for a read-back. @@ -1358,12 +1377,7 @@ def structured_text(self, element: LinuxElement) -> Optional[str]: own = element.native node = own row = None - row_roles = { - "list item", - "table cell", - "table row", - "tree item", - } + row_roles = self._ROW_ROLES for _ in range(8): if self._raw_role(node) in row_roles: row = node @@ -1384,6 +1398,13 @@ def structured_text(self, element: LinuxElement) -> Optional[str]: break branch = parent + return self._row_text(row, branch) + + _ROW_ROLES = frozenset({"list item", "table cell", "table row", "tree item"}) + + def _row_text(self, row: Any, branch: Any) -> Optional[str]: + """A row's identity: its children's text except the target's branch.""" + parts: list[str] = [] for child in self._children(row): if child is branch: @@ -1403,6 +1424,45 @@ def structured_text(self, element: LinuxElement) -> Optional[str]: parts.append(text) return _clean_text(" ".join(parts)) + def row_census( + self, window: LinuxWindow, x: int, y: int, *, limit: int + ) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the row at (x, y). + + Mirrors :meth:`structured_text`: the record row is the nearest list + item, table row or cell, tree item; its identity is its children's + text except the target's branch; every sibling row is read the same + way. Record ids are AT-SPI accessible ids on the row and the target's + branch. A table whose Table interface reports more rows than it + exposes, or a row whose children are not built, is ``incomplete``. + Offsets are in global screen pixels. None on any failure. + """ + try: + from openadapt_flow.runtime.identity_census import tree_row_census + + element = self.element_at_point(window, x, y) + if element is None: + return None + client = self + + def identity(row: Any, path: list[int]) -> Optional[str]: + if not path: + return None + children = client._children(row) + if path[0] >= len(children): + return None + return client._row_text(row, children[path[0]]) + + return tree_row_census( + element.native, + _AtspiTreeView(self), + row_roles=self._ROW_ROLES, + identity=identity, + max_rows=limit, + ) + except Exception: + return None + def perform_native(self, element: LinuxElement, operation: str) -> str: if operation == "focus": component = self._call( @@ -1526,3 +1586,41 @@ def physical_scroll(self, dx: int, dy: int) -> bool: # not provide a reliable pixel-delta contract. Refuse rather than # fabricate delivery evidence. return False + + +class _AtspiTreeView: + """``identity_census.TreeView`` over live AT-SPI nodes (screen pixels).""" + + def __init__(self, client: AtspiLinuxClient) -> None: + self._client = client + + def parent(self, node: Any) -> Any: + return self._client._call(node, "get_parent", default=None) + + def children(self, node: Any) -> list[Any]: + return self._client._children(node) + + def role(self, node: Any) -> Optional[str]: + return self._client._raw_role(node) or None + + def origin(self, node: Any) -> Optional[tuple[float, float]]: + bounds = self._client._bounds(node) + return (float(bounds[0]), float(bounds[1])) if bounds is not None else None + + def id_attributes(self, node: Any) -> list[tuple[str, str]]: + value = self._client._accessible_id(node) + return [("atspi-id", value)] if value else [] + + def row_count(self, container: Any) -> Optional[int]: + table = self._client._call( + container, "get_table_iface", "queryTable", default=None + ) + if table is None: + return None + count = self._client._call(table, "get_n_rows", default=None) + if count is None: + count = getattr(table, "nRows", None) + try: + return int(count) if count is not None else None + except (TypeError, ValueError): + return None diff --git a/packages/flow/openadapt_flow/backends/macos_backend.py b/packages/flow/openadapt_flow/backends/macos_backend.py index 48f031175..1b561a322 100644 --- a/packages/flow/openadapt_flow/backends/macos_backend.py +++ b/packages/flow/openadapt_flow/backends/macos_backend.py @@ -272,6 +272,14 @@ def structured_text(self, element: MacElement) -> Optional[str]: """Exact AX text of ``element`` (value/title/description), or None.""" ... + # Optional (IdentityCandidatesBackend): a client without it cannot count + # look-alike rows, and the runtime halts a row pick before a write. + # + # def row_census(self, pid: int, window_title: str, x: float, y: float, *, + # limit: int) -> Optional[dict[str, Any]]: + # """identity_candidates_at's census at screen point (x, y), with row + # offsets in screen points.""" + def _clean_text(value: object) -> Optional[str]: """Collapse whitespace to a single-spaced string, or None if empty.""" @@ -529,6 +537,118 @@ def structured_text(self, element: MacElement) -> Optional[str]: except Exception: # noqa: BLE001 - a momentary AX failure is not fatal return None + def row_census( + self, pid: int, window_title: str, x: float, y: float, *, limit: int + ) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the element at (x, y). + + The element's identity is its own AX text (as ``structured_text``). + Its record row is the enclosing ``AXRow``, or the item of an + ``AXList``; every sibling row is read at the same child path. Record + ids are the AX identifier and URL of the row and of the target's + node. A table that reports more ``AXRows`` than it exposes as + children, or a row whose cells are not built, is ``incomplete``. + Offsets are in screen points. None on any AX failure. + """ + try: + from ApplicationServices import ( + AXUIElementCopyAttributeValue, + AXUIElementCopyElementAtPosition, + AXUIElementCreateApplication, + kAXTitleAttribute, + kAXTopLevelUIElementAttribute, + ) + + from openadapt_flow.runtime.identity_census import tree_row_census + + app = AXUIElementCreateApplication(int(pid)) + hit_error, element = AXUIElementCopyElementAtPosition( + app, float(x), float(y), None + ) + if hit_error != 0 or element is None: + return None + top_error, top_level = AXUIElementCopyAttributeValue( + element, kAXTopLevelUIElementAttribute, None + ) + if top_error != 0 or top_level is None: + return None + title_error, title = AXUIElementCopyAttributeValue( + top_level, kAXTitleAttribute, None + ) + if title_error != 0 or str(title or "") != window_title: + return None + + def identity(row: Any, path: list[int]) -> Optional[str]: + node = row + for index in path: + children = _AXTreeView(self).children(node) + if index >= len(children): + return None + node = children[index] + return _clean_text(self._text(node)) + + return tree_row_census( + element, + _AXTreeView(self), + row_roles=frozenset({"AXRow"}), + container_roles=frozenset({"AXList"}), + identity=identity, + max_rows=limit, + ) + except Exception: # noqa: BLE001 - unobservable AX is an ordinary miss + return None + + +class _AXTreeView: + """``identity_census.TreeView`` over live AX elements (screen points).""" + + def __init__(self, client: QuartzMacAXClient) -> None: + self._client = client + + def parent(self, node: Any) -> Any: + from ApplicationServices import kAXParentAttribute + + return self._client._attr(node, kAXParentAttribute) + + def children(self, node: Any) -> list[Any]: + from ApplicationServices import kAXChildrenAttribute + + return list(self._client._attr(node, kAXChildrenAttribute) or []) + + def role(self, node: Any) -> Optional[str]: + from ApplicationServices import kAXRoleAttribute + + raw = self._client._attr(node, kAXRoleAttribute) + return str(raw) if raw is not None else None + + def origin(self, node: Any) -> Optional[tuple[float, float]]: + bounds = self._client._bounds(node) + return (bounds[0], bounds[1]) if bounds is not None else None + + def id_attributes(self, node: Any) -> list[tuple[str, str]]: + from ApplicationServices import kAXIdentifierAttribute, kAXURLAttribute + + found = [] + for name, attribute in ( + ("ax-identifier", kAXIdentifierAttribute), + ("ax-url", kAXURLAttribute), + ): + value = self._client._attr(node, attribute) + if value is not None and str(value).strip(): + found.append((name, str(value))) + return found + + def row_count(self, container: Any) -> Optional[int]: + from ApplicationServices import kAXRowsAttribute + + rows = self._client._attr(container, kAXRowsAttribute) + if rows is None: + return None + try: + return len(rows) + except TypeError: + return None + _NATIVE_MODIFIER_FLAGS = { "ctrl": "control", @@ -1222,6 +1342,40 @@ def locate_structural( supported_operations=list(candidate.supported_operations), ) + def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the element at (x, y). + + See :class:`openadapt_flow.backend.IdentityCandidatesBackend`. Uses the + AX client's optional ``row_census``; a client without it cannot count + rows (None). Row offsets are converted to captured pixels. + """ + census_at = getattr(self._ax_client, "row_census", None) + if not callable(census_at): + return None + try: + window = self._ensure_captured() + screen_x, screen_y = self._pixel_to_screen(window, int(x), int(y)) + census = census_at( + window.pid, + window.title, + screen_x, + screen_y, + limit=_AX_ENUMERATION_LIMIT, + ) + except Exception: # noqa: BLE001 - a momentary AX failure is not fatal + return None + if not isinstance(census, dict): + return None + scale_x = self._scale_x or 1.0 + scale_y = self._scale_y or 1.0 + for candidate in census.get("candidates") or []: + dx, dy = candidate.get("dx"), candidate.get("dy") + if isinstance(dx, (int, float)) and isinstance(dy, (int, float)): + candidate["dx"], candidate["dy"] = dx * scale_x, dy * scale_y + else: + candidate["dx"] = candidate["dy"] = None + return census + 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 a56b996a0..229b94e15 100644 --- a/packages/flow/openadapt_flow/backends/win_agent/server.py +++ b/packages/flow/openadapt_flow/backends/win_agent/server.py @@ -12,6 +12,7 @@ serialized request POST /uia/locator-at -> stable UIA locator at a demonstrated point POST /uia/text-at-point -> structured row text for identity verification + POST /uia/identity-candidates-at -> rows that read the same as that row POST /uia/find -> zero / unique / ambiguous exact candidates POST /uia/act -> unique native UIA action + delivery receipt POST /uia/read -> read-only text of one unique control (value or @@ -82,6 +83,7 @@ import io import json import os +import re import secrets import subprocess import sys @@ -734,6 +736,183 @@ def _structured_text_at(auto: Any, x: int, y: int) -> Optional[str]: return value or None +_CENSUS_ROW_TYPES = frozenset( + {"DataItemControl", "ListItemControl", "TreeItemControl", "TableRowControl"} +) +_MAX_CENSUS_ROWS = 2000 +_POSITIONAL_ID = re.compile(r"^\D*(\d+)\D*$") + + +def _row_identity(row: object, excluded: Optional[int]) -> Optional[str]: + """A row's identity as ``_structured_text_at`` reads it, by child index.""" + + parts: list[str] = [] + for index, child in enumerate(_children(row)): + if excluded is not None and index == excluded: + continue + name = str(_control_value(child, "Name", "") or "") + if name: + parts.append(name) + if not parts: + name = str(_control_value(row, "Name", "") or "") + if name: + parts.append(name) + value = " ".join(" ".join(parts).split()) + return value or None + + +def _census_record_key( + row: object, branch: Optional[object], index: int +) -> Optional[str]: + """The row's AutomationIds (row and target branch), minus positional ones.""" + + keys: set[str] = set() + for control in (row, branch): + if control is None: + continue + value = str(_control_value(control, "AutomationId", "") or "").strip() + if not value or len(value) > 256: + continue + match = _POSITIONAL_ID.match(value) + if match and int(match.group(1)) in (index, index + 1): + continue + keys.add(f"automation-id={value}") + return "\n".join(sorted(keys)) or None + + +def _identity_candidates_at(auto: Any, x: int, y: int) -> Optional[dict[str, Any]]: + """Census of the shown rows that read the same as the row at (x, y). + + Mirrors ``_structured_text_at`` (and ``identity_census.tree_row_census`` + in the openadapt_flow package, which this guest copy cannot import): the + record row is the nearest data, list, tree or table-row item; its identity + is its children's names except the clicked cell; every sibling row of the + same control type is read at the same child index. A container whose Grid + pattern reports more rows than it exposes, or a scrollable container that + realizes items on demand (ItemContainer pattern), is ``incomplete``: its + off-screen rows are not in the tree. ``{"row": False}`` when the point is + in no row; None when UIA fails. + """ + try: + element = auto.ControlFromPoint(x, y) + except Exception: # noqa: BLE001 + return None + if element is None: + return None + row: Optional[object] = element + for _ in range(6): + if str(_control_value(row, "ControlTypeName", "")) in _CENSUS_ROW_TYPES: + break + row = _parent(row) + if row is None: + break + else: + row = None + if row is None or ( + str(_control_value(row, "ControlTypeName", "")) not in _CENSUS_ROW_TYPES + ): + return {"row": False} + own: Optional[object] = element + for _ in range(6): + parent = _parent(own) + if parent is None: + own = None + break + if parent is row: + break + own = parent + row_children = _children(row) + excluded = next( + (i for i, child in enumerate(row_children) if own is not None and child is own), + None, + ) + branch = row_children[excluded] if excluded is not None else None + target_identity = _row_identity(row, excluded) + if not target_identity: + return None + container = _parent(row) + row_type = str(_control_value(row, "ControlTypeName", "")) + siblings = _children(container) if container is not None else [row] + rows = [ + c + for c in siblings + if c is row or str(_control_value(c, "ControlTypeName", "")) == row_type + ] + + # The target row's own wrapper may differ from the sibling's; locate it by + # runtime id when the provider exposes one, else by bounds. + def same(a: object, b: object) -> bool: + if a is b: + return True + compare = getattr(auto, "ControlsAreSame", None) + if callable(compare): + try: + return bool(compare(a, b)) + except Exception: # noqa: BLE001 + pass + return _bounds(a) is not None and _bounds(a) == _bounds(b) + + row_index = next((i for i, c in enumerate(rows) if same(c, row)), None) + incomplete = len(rows) > _MAX_CENSUS_ROWS or row_index is None + if container is not None and not incomplete: + grid = _control_value(container, "GetGridPattern", None) + try: + declared = int(grid().RowCount) if callable(grid) else None + except Exception: # noqa: BLE001 + declared = None + if declared is not None and declared > len(rows): + incomplete = True + scroll = _control_value(container, "GetScrollPattern", None) + items = _control_value(container, "GetItemContainerPattern", None) + try: + scrollable = ( + bool(scroll().VerticallyScrollable) if callable(scroll) else False + ) + realizes = callable(items) and items() is not None + except Exception: # noqa: BLE001 + scrollable, realizes = False, False + if scrollable and realizes: + incomplete = True + result: dict[str, Any] = { + "record_key": _census_record_key(row, branch, row_index or 0), + "incomplete": incomplete, + "top_level": True, + "candidates": [], + } + if incomplete: + return result + origin = _bounds(row) + for index, candidate in enumerate(rows): + is_self = index == row_index + if is_self: + observed: Optional[str] = target_identity + candidate_branch = branch + else: + observed = _row_identity(candidate, excluded) + kids = _children(candidate) + candidate_branch = ( + kids[excluded] + if excluded is not None and excluded < len(kids) + else None + ) + if observed != target_identity: + continue + here = _bounds(candidate) + result["candidates"].append( + { + "self": is_self, + "record_key": ( + result["record_key"] + if is_self + else _census_record_key(candidate, candidate_branch, index) + ), + "dx": here[0] - origin[0] if here and origin else None, + "dy": here[1] - origin[1] if here and origin else None, + } + ) + return result + + _ROW_TYPES = frozenset( { "DataItemControl", @@ -844,12 +1023,19 @@ def _perform_uia_initialized( auto: Any, operation: str, payload: dict[str, Any] ) -> dict[str, Any]: """Perform one typed UIA operation in an initialized COM apartment.""" - if operation in {"locator-at", "text-at-point", "focused-at-point"}: + if operation in { + "locator-at", + "text-at-point", + "focused-at-point", + "identity-candidates-at", + }: data = _exact_object(payload, required=frozenset({"x", "y"}), label=operation) x = _bounded_int(data["x"], "x") y = _bounded_int(data["y"], "y") if operation == "text-at-point": return {"status": "ok", "text": _structured_text_at(auto, x, y)} + if operation == "identity-candidates-at": + return {"status": "ok", "census": _identity_candidates_at(auto, x, y)} try: element = auto.ControlFromPoint(x, y) except Exception as exc: # noqa: BLE001 @@ -1432,6 +1618,7 @@ def do_POST(self) -> None: # noqa: N802 - stdlib naming "/context/identity": ("context", "identity"), "/uia/locator-at": ("uia", "locator-at"), "/uia/text-at-point": ("uia", "text-at-point"), + "/uia/identity-candidates-at": ("uia", "identity-candidates-at"), "/uia/find": ("uia", "find"), "/uia/act": ("uia", "act"), "/uia/read": ("uia", "read"), diff --git a/packages/flow/openadapt_flow/backends/windows_backend.py b/packages/flow/openadapt_flow/backends/windows_backend.py index ecb96191d..ecbc51834 100644 --- a/packages/flow/openadapt_flow/backends/windows_backend.py +++ b/packages/flow/openadapt_flow/backends/windows_backend.py @@ -40,7 +40,7 @@ import struct import warnings from dataclasses import dataclass -from typing import Callable, Optional +from typing import Any, Callable, Optional from urllib.parse import urlparse import requests @@ -687,6 +687,22 @@ def structured_text_at(self, x: int, y: int) -> Optional[str]: return None return str(value) + def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]: + """Census of the rows that read the same as the row under (x, y). + + See :class:`openadapt_flow.backend.IdentityCandidatesBackend`. Served by + the agent's typed ``/uia/identity-candidates-at``; an agent without it, + or the legacy exec path, cannot count rows (None), and the runtime + halts a row pick before a write. + """ + typed = self._post_typed_read( + "/uia/identity-candidates-at", {"x": int(x), "y": int(y)} + ) + if not typed.available or typed.payload is None: + return None + census = typed.payload.get("census") + return census if isinstance(census, dict) else None + def _execute_read(self, command: str) -> Optional[str]: """POST bare Python to WAA and return the response body if any. diff --git a/packages/flow/openadapt_flow/runtime/identity_census.py b/packages/flow/openadapt_flow/runtime/identity_census.py index c0839bc43..ee2d23bb9 100644 --- a/packages/flow/openadapt_flow/runtime/identity_census.py +++ b/packages/flow/openadapt_flow/runtime/identity_census.py @@ -33,7 +33,8 @@ from __future__ import annotations -from typing import Any, Callable, Literal, Mapping, NamedTuple, Optional +import re +from typing import Any, Callable, Literal, Mapping, NamedTuple, Optional, Protocol CensusVerdict = Literal[ "unique", @@ -138,3 +139,185 @@ def count_matching_rows( if abs(y - target_y) >= same_row_px and text.strip() and verifies(text) ) return others + 1 + + +# --------------------------------------------------------------------------- +# Row census over an accessibility tree (macOS AX, Linux AT-SPI) +# --------------------------------------------------------------------------- +# +# The browser backend counts look-alike rows in page JavaScript. A native +# accessibility tree is walked here instead, through a small adapter, so the +# rules are the same on every backend: find the record row that holds the +# target, read the same identity from every sibling row, and report each row +# that reads the same with its stable record id. The Windows agent runs in the +# guest without this package and carries a mirror of this function +# (backends/win_agent/server.py, ``_identity_candidates_at``). + +#: Rows read before a census gives up and reports ``incomplete``. +TREE_CENSUS_ROW_CAP = 2000 + +_POSITIONAL_ID = re.compile(r"^\D*(\d+)\D*$") + + +class TreeView(Protocol): + """Read-only access to one accessibility tree, for :func:`tree_row_census`. + + Nodes are opaque. ``origin`` is the node's top-left in the backend's click + coordinates, or None when it cannot be mapped. ``id_attributes`` lists the + node's candidate record ids as ``(name, value)`` pairs (an AX identifier, + an AT-SPI ``id`` attribute, a URL). ``row_count`` is the number of rows the + container says it has, when it says (None otherwise). + """ + + def parent(self, node: Any) -> Any: ... + + def children(self, node: Any) -> list[Any]: ... + + def role(self, node: Any) -> Optional[str]: ... + + def origin(self, node: Any) -> Optional[tuple[float, float]]: ... + + def id_attributes(self, node: Any) -> list[tuple[str, str]]: ... + + def row_count(self, container: Any) -> Optional[int]: ... + + +def _record_key(view: TreeView, row: Any, path: list[int], index: int) -> Optional[str]: + """The stable record id of ``row``: ids on the row and on the path to the + target, minus values that are only the row's position.""" + + parts: set[str] = set() + node = row + nodes = [row] + for step in path: + children = view.children(node) + if step >= len(children): + break + node = children[step] + nodes.append(node) + for item in nodes: + for name, value in view.id_attributes(item): + value = (value or "").strip() + if not value or len(value) > 256: + continue + match = _POSITIONAL_ID.match(value) + if match and int(match.group(1)) in (index, index + 1): + continue + parts.add(f"{name}={value}") + return "\n".join(sorted(parts)) or None + + +def tree_row_census( + target: Any, + view: TreeView, + *, + row_roles: frozenset[str], + identity: Callable[[Any, list[int]], Optional[str]], + container_roles: frozenset[str] = frozenset(), + max_depth: int = 8, + max_rows: int = TREE_CENSUS_ROW_CAP, +) -> Optional[dict[str, Any]]: + """Count the sibling rows that read the same identity as ``target``'s row. + + Args: + target: The node under the point. + view: The tree adapter. + row_roles: Roles that make a node a record row. + identity: ``identity(row, path)`` reads a row's identity the way the + backend's ``structured_text_at`` does, where ``path`` is the child + index path from the row to the target's place in it. None when the + row has nothing to read there. + container_roles: Roles of a list or table whose direct children are + its rows even when they carry no row role (an ``AXList`` item). + max_depth: How far above ``target`` to look for its row. + max_rows: More sibling rows than this reports ``incomplete``. + + Returns: + The census shape of ``IdentityCandidatesBackend.identity_candidates_at``, + ``{"row": False}`` when the target is in no row, or None when the + target's own identity cannot be read. + """ + row = None + previous = None + node = target + for _ in range(max_depth + 1): + if node is None: + break + role = view.role(node) + if role in row_roles: + row = node + break + if role in container_roles and previous is not None: + row = previous + break + previous, node = node, view.parent(node) + if row is None: + return {"row": False} + path: list[int] = [] + node = target + while node is not row: + parent = view.parent(node) + if parent is None: + return None + siblings = view.children(parent) + try: + path.insert(0, next(i for i, c in enumerate(siblings) if c == node)) + except StopIteration: + return None + node = parent + own_identity = identity(row, path) + if not own_identity: + return None + container = view.parent(row) + siblings = view.children(container) if container is not None else [row] + row_role = view.role(row) + rows = [c for c in siblings if c == row or view.role(c) == row_role] + declared = view.row_count(container) if container is not None else None + incomplete = len(rows) > max_rows or (declared is not None and declared > len(rows)) + row_index = next((i for i, c in enumerate(rows) if c == row), 0) + result: dict[str, Any] = { + "record_key": _record_key(view, row, path, row_index), + "incomplete": incomplete, + "top_level": True, + "candidates": [], + } + if incomplete: + return result + origin = view.origin(row) + for index, candidate in enumerate(rows): + if candidate == row: + observed: Optional[str] = own_identity + else: + node = candidate + for step in path: + children = view.children(node) + if step >= len(children): + # The row is not populated where the target sits (a lazily + # built row): its identity cannot be read, so not every + # row was counted. + result["incomplete"] = True + result["candidates"] = [] + return result + node = children[step] + observed = identity(candidate, path) + if observed != own_identity: + continue + here = view.origin(candidate) + offset = ( + (here[0] - origin[0], here[1] - origin[1]) + if here is not None and origin is not None + else (None, None) + ) + result["candidates"].append( + { + "self": candidate == row, + "record_key": ( + result["record_key"] + if candidate == row + else _record_key(view, candidate, path, index) + ), + "dx": offset[0], + "dy": offset[1], + } + ) + return result diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index ac227ace6..35f67669c 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -15507,7 +15507,7 @@ def ocr_tier() -> Optional[IdentityCheck]: [structured_tier, pixel_tier, vlm_tier, ocr_tier] ) return self._apply_candidate_census( - step, resolution, check, before_png, params, workflow + step, resolution, check, before_png, params, workflow, bundle_dir ) def _structured_census(self, step: Step, point: Point) -> census_mod.CensusDecision: @@ -15626,6 +15626,7 @@ def _apply_candidate_census( before_png: bytes, params: Mapping[str, RuntimeParamScalar], workflow: Workflow, + bundle_dir: Optional[Path] = None, ) -> IdentityCheck: """Keep ``verified`` only when the identity picks one shown record. @@ -15647,6 +15648,12 @@ def _apply_candidate_census( decision = self._structured_census(step, resolution.point) else: decision = self._ocr_census(step, resolution, before_png, params, workflow) + if decision.verdict == "cannot_count" and check.mode in ("pixel", "vlm"): + # No OCR band to read: count the places the identifier crop + # itself verifies instead. + decision = self._pixel_census( + step, resolution, before_png, workflow, bundle_dir + ) tmpl = step.anchor.identity_template if step.anchor is not None else None if ( decision.verdict == "cannot_count" @@ -15689,6 +15696,59 @@ def _apply_candidate_census( } ) + def _pixel_census( + self, + step: Step, + resolution: Resolution, + before_png: bytes, + workflow: Workflow, + bundle_dir: Optional[Path], + ) -> census_mod.CensusDecision: + """Count the places on the frame where the identifier crop verifies. + + For a pixel identity with no OCR band (a remote display): every + place the recorded identifier crop could match is cut from the frame + and checked with the same pixel verifier the ladder used. More than + one verifying place means another row shows the same identifier. + """ + from openadapt_flow.vision.match import template_peaks + + anchor = step.anchor + assert anchor is not None + recorded, live = self._identifier_crops( + anchor, resolution, before_png, bundle_dir, workflow=workflow + ) + if recorded is None or live is None or anchor.identifier_region is None: + return census_mod.CensusDecision("cannot_count") + try: + peaks = template_peaks(before_png, recorded) + except Exception: + return census_mod.CensusDecision("cannot_count") + if peaks is None: + return census_mod.CensusDecision("uncounted") + rx, ry, rw, rh = anchor.identifier_region + dx, dy = self._identifier_displacement( + anchor, resolution, before_png, bundle_dir, workflow=workflow + ) + target = (rx + dx, ry + dy) + count = 1 + for x, y in peaks: + if abs(x - target[0]) < max(1, rw // 2) and abs(y - target[1]) < max( + 1, rh // 2 + ): + continue # the target itself + crop = identity_mod.crop_region(before_png, (x, y, rw, rh)) + if crop is None: + continue + verdict = identity_mod.verify_pixel_identity( + recorded, crop, enable_verify=self.pixel_verify_enabled + ) + if verdict is not None and verdict.status == "verified": + count += 1 + if count > 1: + return census_mod.CensusDecision("ambiguous", count) + return census_mod.CensusDecision("unique", count) + def _leads_into_consequential( self, step: Step, workflow: Optional[Workflow] ) -> bool: diff --git a/packages/flow/openadapt_flow/vision/match.py b/packages/flow/openadapt_flow/vision/match.py index 251e769a5..a61721a07 100644 --- a/packages/flow/openadapt_flow/vision/match.py +++ b/packages/flow/openadapt_flow/vision/match.py @@ -309,6 +309,29 @@ def find_template( ) +def template_peaks( + screen_png: bytes, template_png: bytes, *, threshold: float = 0.6 +) -> Optional[list[tuple[int, int]]]: + """Every spatially distinct place ``template`` matches, at its own scale. + + Returns the top-left ``(x, y)`` of each non-max-suppressed peak at or above + ``threshold``, strongest first, for a caller that must know every place a + crop could be (the pixel identity census). None when the frame holds at + least :data:`MAX_PEAKS` peaks, so the list may be incomplete, or when the + template does not fit the frame. + """ + screen = _decode_gray(screen_png) + template = _decode_gray(template_png) + th, tw = template.shape[:2] + if th > screen.shape[0] or tw > screen.shape[1]: + return None + result = imgops.match_template_ccoeff_normed(screen, template) + peaks = _peaks_above(result, threshold, tw, th) + if len(peaks) >= MAX_PEAKS: + return None + return [(x, y) for _score, x, y in peaks] + + def find_structural_template( screen_png: bytes, template_png: bytes, diff --git a/packages/flow/tests/native_apps/Worklist.swift b/packages/flow/tests/native_apps/Worklist.swift new file mode 100644 index 000000000..362a33114 --- /dev/null +++ b/packages/flow/tests/native_apps/Worklist.swift @@ -0,0 +1,33 @@ +import AppKit +import Foundation + +struct Row: Decodable { let text: String; let id: String } +let rows = try! JSONDecoder().decode([Row].self, from: CommandLine.arguments[1].data(using: .utf8)!) + +final class Source: NSObject, NSTableViewDataSource, NSTableViewDelegate { + func numberOfRows(in tableView: NSTableView) -> Int { rows.count } + func tableView(_ tableView: NSTableView, viewFor tableColumn: NSTableColumn?, row: Int) -> NSView? { + let field = NSTextField(labelWithString: rows[row].text) + if !rows[row].id.isEmpty { field.identifier = NSUserInterfaceItemIdentifier(rows[row].id) } + return field + } +} + +let app = NSApplication.shared +app.setActivationPolicy(.regular) +let window = NSWindow(contentRect: NSRect(x: 120, y: 120, width: 480, height: 320), + styleMask: [.titled, .closable, .resizable], backing: .buffered, defer: false) +window.title = CommandLine.arguments[2] +let table = NSTableView(frame: NSRect(x: 0, y: 0, width: 480, height: 320)) +let column = NSTableColumn(identifier: NSUserInterfaceItemIdentifier("record")) +column.width = 460 +table.addTableColumn(column) +let source = Source() +table.dataSource = source +table.delegate = source +let scroll = NSScrollView(frame: NSRect(x: 0, y: 0, width: 480, height: 320)) +scroll.documentView = table +window.contentView = scroll +window.makeKeyAndOrderFront(nil) +app.activate(ignoringOtherApps: true) +app.run() diff --git a/packages/flow/tests/test_identity_evidence_capability.py b/packages/flow/tests/test_identity_evidence_capability.py index 75d01b6f6..2beddbf2d 100644 --- a/packages/flow/tests/test_identity_evidence_capability.py +++ b/packages/flow/tests/test_identity_evidence_capability.py @@ -116,21 +116,33 @@ def test_no_surrounding_text_valid_crop_verifies_where_crops_are_verifiable( assert error is None assert check is not None assert check.status == "verified" and check.mode == "pixel" - # A crop alone gives no row text to count look-alikes with; nothing - # consequential follows this step, so it runs and records that. - assert check.census_status == "cannot_count" + # No row text to count look-alikes with: the crop itself is counted. It + # verifies in one place only. + assert check.candidate_count == 1 -def test_a_verified_crop_with_no_row_count_halts_before_a_write( +def test_a_crop_found_once_on_the_frame_runs_before_a_write( tmp_path: Path, ) -> None: - """The crop matches, but nothing can say no other row shows the same.""" + """No OCR band: the census counts where the crop verifies. Once: it runs.""" step, res = _crop_only_step(tmp_path, _button()) error, check = _gate(tmp_path, step, res, _button(), pixel_verify_enabled=True) - assert error is not None and "identity_uncounted" in error + assert error is None + assert check is not None and check.status == "verified" + assert check.candidate_count == 1 + + +def test_a_crop_found_twice_on_the_frame_halts(tmp_path: Path) -> None: + """Another row shows the same identifier pixels: the step stops.""" + + step, res = _crop_only_step(tmp_path, _button()) + frame = np.vstack([_button(), np.full((8, _W), 255, np.uint8), _button()]) + error, check = _gate(tmp_path, step, res, frame, pixel_verify_enabled=True) + assert error is not None and "2 shown rows" in error assert check is not None - assert check.status == "abstain" and check.ambiguity_halt == "identity_uncounted" + assert check.status == "abstain" and check.ambiguity_halt == "identical_candidates" + assert check.candidate_count == 2 def test_crop_that_does_not_match_halts(tmp_path: Path) -> None: diff --git a/packages/flow/tests/test_native_census_real_apps.py b/packages/flow/tests/test_native_census_real_apps.py new file mode 100644 index 000000000..87d82419c --- /dev/null +++ b/packages/flow/tests/test_native_census_real_apps.py @@ -0,0 +1,301 @@ +"""Row census against real native list apps (opt-in, one per platform). + +Each test starts a small native app showing a worklist, reads it through the +backend's real accessibility code, and checks the census: identical rows are +ambiguous, distinct record ids pick the recorded row, a unique row runs. + +* Windows: a WPF ``ListBox`` whose items carry ``AutomationProperties + .AutomationId``, read through the guest agent's UI Automation census. + Set ``OPENADAPT_TEST_WINDOWS_UIA=1``; needs the ``uiautomation`` package. +* macOS: a view-based ``NSTableView`` (tests/native_apps/Worklist.swift, built + with ``swiftc``) whose cells carry an identifier, read through + ``QuartzMacAXClient.row_census``. Set ``OPENADAPT_TEST_MACOS_AX=1``; needs + PyObjC and Accessibility access for the test process. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from pathlib import Path +from typing import Any, Optional + +import pytest + +from openadapt_flow.runtime import identity_template as itmpl +from openadapt_flow.runtime.identity_census import CensusDecision, decide + +pytestmark = pytest.mark.timeout(180) + +TITLE = "OA Census Worklist" +SAME = "John Smith Follow-up Dr Lee" +OTHER = "Maria Garcia Intake Dr Lee" + + +def _decide(census: Optional[dict], recorded_key: Optional[str] = None): + tmpl = itmpl.build_identity_template( + None, structured_identity="x", record_key=recorded_key + ) + assert tmpl is not None + return decide( + census, + has_recorded_key=tmpl.record_key is not None, + key_matches=lambda key: itmpl.record_key_matches(tmpl, key), + ) + + +def _wait(predicate, timeout_s: float = 30.0) -> Any: + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + value = predicate() + if value: + return value + time.sleep(0.5) + raise AssertionError("the native test app did not appear") + + +# -- Windows (WPF + UI Automation) --------------------------------------------- + +_WPF_APP = r""" +param([string]$RowsJson) +Add-Type -AssemblyName PresentationFramework +$rows = $RowsJson | ConvertFrom-Json +$window = New-Object System.Windows.Window +$window.Title = '__TITLE__' +$window.Width = 520 +$window.Height = 360 +$window.Left = 40 +$window.Top = 40 +$list = New-Object System.Windows.Controls.ListBox +foreach ($row in $rows) { + $item = New-Object System.Windows.Controls.ListBoxItem + $panel = New-Object System.Windows.Controls.StackPanel + $panel.Orientation = 'Horizontal' + foreach ($cell in $row.cells) { + $text = New-Object System.Windows.Controls.TextBlock + $text.Text = $cell + $text.Margin = '0,0,24,0' + [void]$panel.Children.Add($text) + } + $item.Content = $panel + if ($row.id) { + [System.Windows.Automation.AutomationProperties]::SetAutomationId($item, $row.id) + } + [void]$list.Items.Add($item) +} +$window.Content = $list +[void]$window.ShowDialog() +""".replace("__TITLE__", TITLE) + + +@pytest.fixture +def wpf_worklist(tmp_path: Path): + if sys.platform != "win32" or os.environ.get("OPENADAPT_TEST_WINDOWS_UIA") != "1": + pytest.skip("set OPENADAPT_TEST_WINDOWS_UIA=1 on Windows") + auto = pytest.importorskip("uiautomation") + script = tmp_path / "worklist.ps1" + script.write_text(_WPF_APP, encoding="utf-8") + procs: list[subprocess.Popen] = [] + + def start(rows: list[dict[str, Any]]) -> list[Any]: + proc = subprocess.Popen( + [ + "powershell", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-File", + str(script), + "-RowsJson", + json.dumps(rows), + ] + ) + procs.append(proc) + window = _wait( + lambda: ( + auto.WindowControl(searchDepth=1, Name=TITLE) + if auto.WindowControl(searchDepth=1, Name=TITLE).Exists(0, 0) + else None + ) + ) + listing = window.ListControl(searchDepth=3) + return _wait( + lambda: ( + [ + c + for c in listing.GetChildren() + if c.ControlTypeName == "ListItemControl" + ] + if listing.Exists(0, 0) + else None + ) + ) + + yield auto, start + for proc in procs: + proc.kill() + + +def _first_cell_center(item: Any) -> tuple[int, int]: + cell = item.GetChildren()[0] + rect = cell.BoundingRectangle + return (rect.left + rect.right) // 2, (rect.top + rect.bottom) // 2 + + +def test_windows_wpf_identical_rows_are_ambiguous(wpf_worklist) -> None: + from openadapt_flow.backends.win_agent import server + + auto, start = wpf_worklist + items = start( + [ + {"cells": SAME.split(" ", 2), "id": ""}, + {"cells": SAME.split(" ", 2), "id": ""}, + {"cells": OTHER.split(" ", 2), "id": ""}, + ] + ) + with auto.UIAutomationInitializerInThread(): + census = server._identity_candidates_at(auto, *_first_cell_center(items[1])) + assert _decide(census) == CensusDecision("ambiguous", 2), census + + +def test_windows_wpf_record_ids_pick_the_recorded_row(wpf_worklist) -> None: + from openadapt_flow.backends.win_agent import server + + auto, start = wpf_worklist + items = start( + [ + {"cells": SAME.split(" ", 2), "id": "chart-1002"}, + {"cells": SAME.split(" ", 2), "id": "chart-1001"}, + ] + ) + with auto.UIAutomationInitializerInThread(): + census = server._identity_candidates_at(auto, *_first_cell_center(items[0])) + unique = server._identity_candidates_at(auto, *_first_cell_center(items[1])) + decision = _decide(census, "automation-id=chart-1001") + assert decision.verdict == "retarget", census + assert decision.offset is not None and decision.offset[1] > 0 + assert _decide(unique, "automation-id=chart-1001").verdict == "unique" + + +def test_windows_wpf_unique_row_runs(wpf_worklist) -> None: + from openadapt_flow.backends.win_agent import server + + auto, start = wpf_worklist + items = start( + [ + {"cells": SAME.split(" ", 2), "id": ""}, + {"cells": OTHER.split(" ", 2), "id": ""}, + ] + ) + with auto.UIAutomationInitializerInThread(): + point = _first_cell_center(items[1]) + census = server._identity_candidates_at(auto, *point) + text = server._structured_text_at(auto, *point) + assert text, "the identity check reads no row text from this list" + assert _decide(census) == CensusDecision("unique", 1), census + + +# -- macOS (NSTableView + AX) -------------------------------------------------- + +_SWIFT_APP = Path(__file__).parent / "native_apps" / "Worklist.swift" + + +def _on_screen(pid: int) -> bool: + import Quartz + + windows = Quartz.CGWindowListCopyWindowInfo( + Quartz.kCGWindowListOptionOnScreenOnly, Quartz.kCGNullWindowID + ) + return any( + w.get("kCGWindowOwnerPID") == pid and w.get("kCGWindowName") == TITLE + for w in windows + ) + + +@pytest.fixture +def cocoa_worklist(tmp_path: Path): + if sys.platform != "darwin" or os.environ.get("OPENADAPT_TEST_MACOS_AX") != "1": + pytest.skip("set OPENADAPT_TEST_MACOS_AX=1 on macOS") + services = pytest.importorskip("ApplicationServices") + pytest.importorskip("Quartz") + if not services.AXIsProcessTrusted(): + pytest.skip("this process has no Accessibility access") + from openadapt_flow.backends.macos_backend import QuartzMacAXClient, _AXTreeView + + binary = tmp_path / "worklist" + built = subprocess.run( + ["swiftc", "-O", "-o", str(binary), str(_SWIFT_APP)], + capture_output=True, + text=True, + timeout=170, + ) + if built.returncode != 0: + pytest.skip(f"swiftc could not build the table app: {built.stderr[-400:]}") + procs: list[subprocess.Popen] = [] + client = QuartzMacAXClient() + + def start(rows: list[dict[str, str]]) -> tuple[int, list[tuple[float, float]]]: + proc = subprocess.Popen([str(binary), json.dumps(rows), TITLE]) + procs.append(proc) + view = _AXTreeView(client) + + def row_centers() -> Optional[list[tuple[float, float]]]: + window = client._window_element(proc.pid, TITLE) + if window is None: + return None + stack, found = [window], [] + while stack: + node = stack.pop() + if view.role(node) == "AXRow": + bounds = client._bounds(node) + if bounds is not None: + found.append((bounds[0] + 20, bounds[1] + bounds[3] / 2)) + continue + stack.extend(view.children(node)) + return ( + sorted(found, key=lambda p: p[1]) if len(found) == len(rows) else None + ) + + try: + return proc.pid, _wait(row_centers) + except AssertionError: + if _on_screen(proc.pid): + # The window is on screen but AX lists no window for it: this + # session's Accessibility access is degraded (AX answers with + # the application element), so nothing can be read. + pytest.skip("AX lists no windows in this session; cannot read the app") + raise + + yield client, start + for proc in procs: + proc.kill() + + +def test_macos_table_identical_rows_are_ambiguous(cocoa_worklist) -> None: + client, start = cocoa_worklist + pid, centers = start( + [{"text": SAME, "id": ""}, {"text": SAME, "id": ""}, {"text": OTHER, "id": ""}] + ) + census = client.row_census(pid, TITLE, *centers[1], limit=4000) + assert _decide(census) == CensusDecision("ambiguous", 2), census + + +def test_macos_table_record_ids_pick_the_recorded_row(cocoa_worklist) -> None: + client, start = cocoa_worklist + pid, centers = start( + [{"text": SAME, "id": "chart-1002"}, {"text": SAME, "id": "chart-1001"}] + ) + census = client.row_census(pid, TITLE, *centers[0], limit=4000) + decision = _decide(census, "ax-identifier=chart-1001") + assert decision.verdict == "retarget", census + assert decision.offset is not None and decision.offset[1] > 0 + + +def test_macos_table_unique_row_runs(cocoa_worklist) -> None: + client, start = cocoa_worklist + pid, centers = start([{"text": SAME, "id": ""}, {"text": OTHER, "id": ""}]) + census = client.row_census(pid, TITLE, *centers[1], limit=4000) + assert _decide(census) == CensusDecision("unique", 1), census diff --git a/packages/flow/tests/test_native_identity_census.py b/packages/flow/tests/test_native_identity_census.py new file mode 100644 index 000000000..1fa6c4f14 --- /dev/null +++ b/packages/flow/tests/test_native_identity_census.py @@ -0,0 +1,560 @@ +"""Row census on native accessibility trees (macOS AX, Linux AT-SPI, Windows UIA). + +A worklist of two rows that read "John Smith Follow-up Dr Lee" must halt +before it opens either chart, distinct record ids must pick the recorded row, +and a unique row must still run. Each backend is exercised through its real +census code over a fake accessibility tree; the browser census has its own +real-browser tests (tests/test_realism_browser.py). +""" + +from __future__ import annotations + +import types +from typing import Any, Optional + +import pytest + +from openadapt_flow.backends import linux_backend +from openadapt_flow.backends.win_agent import server +from openadapt_flow.ir import ( + ActionKind, + Anchor, + IdentityCheck, + Resolution, + Step, + Workflow, +) +from openadapt_flow.runtime import identity_template as itmpl +from openadapt_flow.runtime.identity_census import ( + CensusDecision, + decide, + tree_row_census, +) +from openadapt_flow.runtime.replayer import Replayer + +ROW = ("John Smith", "Follow-up", "Dr Lee") +OTHER = ("Maria Garcia", "Intake", "Dr Lee") + + +def _decide(census: Optional[dict], recorded_key: Optional[str] = None): + tmpl = itmpl.build_identity_template( + None, structured_identity="x", record_key=recorded_key + ) + assert tmpl is not None + return decide( + census, + has_recorded_key=tmpl.record_key is not None, + key_matches=lambda key: itmpl.record_key_matches(tmpl, key), + ) + + +# -- generic tree census ------------------------------------------------------- + + +class Node: + def __init__(self, role: str, text: str = "", *, ids=(), y: float = 0.0): + self.role = role + self.text = text + self.ids = list(ids) + self.y = y + self.parent: Optional[Node] = None + self.kids: list[Node] = [] + self.declared_rows: Optional[int] = None + + def add(self, *children: "Node") -> "Node": + for child in children: + child.parent = self + self.kids.append(child) + return self + + +class View: + def parent(self, node: Node) -> Any: + return node.parent + + def children(self, node: Node) -> list[Any]: + return list(node.kids) + + def role(self, node: Node) -> Optional[str]: + return node.role + + def origin(self, node: Node) -> Optional[tuple[float, float]]: + return (0.0, node.y) + + def id_attributes(self, node: Node) -> list[tuple[str, str]]: + return [("id", value) for value in node.ids] + + def row_count(self, container: Node) -> Optional[int]: + return container.declared_rows + + +def _table(*rows: tuple[tuple[str, ...], tuple[str, ...]]) -> Node: + """A table of rows ``(cells, row ids)``, 40 px apart.""" + + table = Node("table") + for index, (cells, ids) in enumerate(rows): + row = Node("row", ids=ids, y=40.0 * index) + row.add(*(Node("cell", text) for text in cells)) + table.add(row) + return table + + +def _row_identity(row: Node, path: list[int]) -> Optional[str]: + """Row text except the target's cell (the Windows/Linux reading).""" + + return " ".join(c.text for i, c in enumerate(row.kids) if i != path[0]) or None + + +def _census(target: Node) -> Optional[dict]: + return tree_row_census( + target, View(), row_roles=frozenset({"row"}), identity=_row_identity + ) + + +def test_two_identical_rows_are_ambiguous() -> None: + table = _table((ROW, ()), (ROW, ()), (OTHER, ())) + census = _census(table.kids[1].kids[0]) + assert census is not None and len(census["candidates"]) == 2 + assert _decide(census) == CensusDecision("ambiguous", 2) + + +def test_three_identical_rows_are_ambiguous() -> None: + table = _table((ROW, ()), (OTHER, ()), (ROW, ()), (ROW, ())) + assert _decide(_census(table.kids[0].kids[0])) == CensusDecision("ambiguous", 3) + + +def test_distinct_record_ids_pick_the_recorded_row() -> None: + table = _table((ROW, ("chart-1002",)), (ROW, ("chart-1001",))) + census = _census(table.kids[0].kids[0]) # resolver landed on 1002 + assert _decide(census, "id=chart-1001") == CensusDecision( + "retarget", 2, (0.0, 40.0) + ) + assert _decide(_census(table.kids[1].kids[0]), "id=chart-1001") == ( + CensusDecision("unique", 2) + ) + + +def test_a_unique_row_runs() -> None: + table = _table((ROW, ()), (OTHER, ())) + assert _decide(_census(table.kids[1].kids[2])) == CensusDecision("unique", 1) + + +def test_a_control_outside_any_row_is_not_a_row() -> None: + window = Node("window").add(Node("button", "Sign visit note")) + assert _census(window.kids[0]) == {"row": False} + + +def test_positional_ids_never_tell_rows_apart() -> None: + table = _table((ROW, ("row-0",)), (ROW, ("row-1",))) + census = _census(table.kids[0].kids[0]) + assert census is not None + assert [c["record_key"] for c in census["candidates"]] == [None, None] + assert _decide(census).verdict == "ambiguous" + + +def test_a_table_with_more_rows_than_it_exposes_is_incomplete() -> None: + table = _table((ROW, ()), (OTHER, ())) + table.declared_rows = 50 # a virtualized list: rows built on scroll + census = _census(table.kids[1].kids[0]) + assert census is not None and census["incomplete"] is True + assert _decide(census).verdict == "uncounted" + + +def test_a_row_whose_cells_are_not_built_is_incomplete() -> None: + table = _table((ROW, ()), (OTHER, ())) + table.add(Node("row", y=80.0)) # placeholder row, no cells yet + census = _census(table.kids[0].kids[2]) + assert census is not None and census["incomplete"] is True + + +def test_list_items_without_a_row_role_are_rows_of_their_list() -> None: + """macOS AXList: the list's direct children are its rows.""" + + listing = Node("AXList") + for index, name in enumerate(("John Smith", "John Smith", "Maria Garcia")): + item = Node("AXGroup", y=30.0 * index).add(Node("AXStaticText", name)) + listing.add(item) + + def own_text(row: Node, path: list[int]) -> Optional[str]: + node = row + for step in path: + node = node.kids[step] + return node.text or None + + census = tree_row_census( + listing.kids[2].kids[0], + View(), + row_roles=frozenset({"AXRow"}), + container_roles=frozenset({"AXList"}), + identity=own_text, + ) + assert _decide(census) == CensusDecision("unique", 1) + census = tree_row_census( + listing.kids[0].kids[0], + View(), + row_roles=frozenset({"AXRow"}), + container_roles=frozenset({"AXList"}), + identity=own_text, + ) + assert _decide(census) == CensusDecision("ambiguous", 2) + + +# -- Linux AT-SPI client ------------------------------------------------------- + + +class _States: + def __init__(self, names: set[str]): + self._names = names + + def contains(self, state: str) -> bool: + return state in self._names + + +class _Extents: + def __init__(self, x: int, y: int, w: int, h: int): + self.x, self.y, self.width, self.height = x, y, w, h + + +class AtspiNode: + def __init__(self, role: str, name: str = "", *, bounds=(0, 0, 10, 10), ident=""): + self._role = role + self._name = name + self._bounds = bounds + self._ident = ident + self._parent: Optional[AtspiNode] = None + self._kids: list[AtspiNode] = [] + + def add(self, *kids: "AtspiNode") -> "AtspiNode": + for kid in kids: + kid._parent = self + self._kids.append(kid) + return self + + def get_role_name(self) -> str: + return self._role + + def get_name(self) -> str: + return self._name + + def get_parent(self): + return self._parent + + def get_child_count(self) -> int: + return len(self._kids) + + def get_child_at_index(self, index: int): + return self._kids[index] + + def get_state_set(self) -> _States: + return _States({"VISIBLE", "SHOWING", "ENABLED"}) + + def get_component_iface(self): + node = self + + class _Component: + def get_extents(self, _coord): + return _Extents(*node._bounds) + + return _Component() + + def get_accessible_id(self) -> str: + return self._ident + + +def _atspi_client() -> linux_backend.AtspiLinuxClient: + client = object.__new__(linux_backend.AtspiLinuxClient) + client._atspi = types.SimpleNamespace( + StateType=types.SimpleNamespace( + VISIBLE="VISIBLE", SHOWING="SHOWING", ENABLED="ENABLED", FOCUSED="FOCUSED" + ), + CoordType=types.SimpleNamespace(SCREEN="SCREEN"), + ) + return client + + +def _atspi_window(*rows: tuple[tuple[str, ...], str]): + listing = AtspiNode("list", bounds=(0, 0, 400, 300)) + for index, (cells, ident) in enumerate(rows): + top = 40 * index + row = AtspiNode("list item", bounds=(0, top, 400, 40), ident=ident) + for column, text in enumerate(cells): + row.add(AtspiNode("label", text, bounds=(100 * column, top, 100, 40))) + listing.add(row) + frame = AtspiNode("frame", "Worklist", bounds=(0, 0, 400, 300)).add(listing) + return linux_backend.LinuxWindow( + "w1", "app", "Worklist", 1, (0, 0, 400, 300), frame + ) + + +def test_linux_identical_rows_are_ambiguous() -> None: + window = _atspi_window((ROW, ""), (ROW, ""), (OTHER, "")) + census = _atspi_client().row_census(window, 150, 50, limit=100) + assert _decide(census) == CensusDecision("ambiguous", 2) + + +def test_linux_record_ids_pick_the_recorded_row() -> None: + window = _atspi_window((ROW, "chart-1002"), (ROW, "chart-1001")) + census = _atspi_client().row_census(window, 150, 10, limit=100) + assert _decide(census, "atspi-id=chart-1001") == CensusDecision( + "retarget", 2, (0.0, 40.0) + ) + + +def test_linux_unique_row_runs_and_matches_structured_text() -> None: + client = _atspi_client() + window = _atspi_window((ROW, ""), (OTHER, "")) + element = client.element_at_point(window, 150, 50) + assert element is not None + # The census reads rows exactly as the identity check does. + assert client.structured_text(element) == "Maria Garcia Dr Lee" + census = client.row_census(window, 150, 50, limit=100) + assert _decide(census) == CensusDecision("unique", 1) + + +# -- Windows UIA agent --------------------------------------------------------- + + +class _Rect: + def __init__(self, left: int, top: int, right: int, bottom: int): + self.left, self.top, self.right, self.bottom = left, top, right, bottom + + +class UiaControl: + def __init__(self, control_type: str, name: str = "", *, aid: str = "", top=0): + self.ControlTypeName = control_type + self.Name = name + self.AutomationId = aid + self.BoundingRectangle = _Rect(0, top, 400, top + 40) + self.parent: Optional[UiaControl] = None + self.children: list[UiaControl] = [] + self.grid_rows: Optional[int] = None + self.virtual = False + + def add(self, *kids: "UiaControl") -> "UiaControl": + for kid in kids: + kid.parent = self + self.children.append(kid) + return self + + def GetParentControl(self): + return self.parent + + def GetChildren(self): + return list(self.children) + + def GetGridPattern(self): + if self.grid_rows is None: + raise RuntimeError("no grid pattern") + return types.SimpleNamespace(RowCount=self.grid_rows) + + def GetScrollPattern(self): + return types.SimpleNamespace(VerticallyScrollable=self.virtual) + + def GetItemContainerPattern(self): + return object() if self.virtual else None + + +def _uia_list(*rows: tuple[tuple[str, ...], str]): + listing = UiaControl("ListControl") + for index, (cells, aid) in enumerate(rows): + item = UiaControl("DataItemControl", aid=aid, top=40 * index) + item.add(*(UiaControl("TextControl", text, top=40 * index) for text in cells)) + listing.add(item) + return listing + + +def _auto(target: UiaControl): + return types.SimpleNamespace(ControlFromPoint=lambda _x, _y: target) + + +def test_windows_identical_rows_are_ambiguous() -> None: + listing = _uia_list((ROW, ""), (ROW, ""), (ROW, "")) + census = server._identity_candidates_at( + _auto(listing.children[1].children[0]), 5, 45 + ) + assert _decide(census) == CensusDecision("ambiguous", 3) + + +def test_windows_automation_ids_pick_the_recorded_row() -> None: + listing = _uia_list((ROW, "chart-1002"), (ROW, "chart-1001")) + census = server._identity_candidates_at( + _auto(listing.children[0].children[0]), 5, 5 + ) + assert _decide(census, "automation-id=chart-1001") == CensusDecision( + "retarget", 2, (0.0, 40.0) + ) + + +def test_windows_positional_automation_ids_do_not_tell_rows_apart() -> None: + listing = _uia_list((ROW, "ListViewItem-0"), (ROW, "ListViewItem-1")) + census = server._identity_candidates_at( + _auto(listing.children[0].children[0]), 5, 5 + ) + assert _decide(census).verdict == "ambiguous" + + +def test_windows_unique_row_runs_and_matches_structured_text() -> None: + listing = _uia_list((ROW, ""), (OTHER, "")) + target = listing.children[1].children[0] + auto = _auto(target) + census = server._identity_candidates_at(auto, 5, 45) + assert _decide(census) == CensusDecision("unique", 1) + # The census reads rows exactly as the identity check does: the clicked + # cell is left out. + assert server._structured_text_at(auto, 5, 45) == "Intake Dr Lee" + + +def test_windows_virtualized_or_short_grids_are_incomplete() -> None: + listing = _uia_list((ROW, ""), (OTHER, "")) + listing.grid_rows = 300 + census = server._identity_candidates_at( + _auto(listing.children[1].children[0]), 5, 45 + ) + assert _decide(census).verdict == "uncounted" + listing.grid_rows = None + listing.virtual = True + census = server._identity_candidates_at( + _auto(listing.children[1].children[0]), 5, 45 + ) + assert _decide(census).verdict == "uncounted" + + +def test_windows_standalone_control_is_not_a_row() -> None: + window = UiaControl("WindowControl").add(UiaControl("ButtonControl", "Sign")) + census = server._identity_candidates_at(_auto(window.children[0]), 5, 5) + assert _decide(census).verdict == "not_a_row" + + +def test_windows_agent_serves_the_census_operation() -> None: + listing = _uia_list((ROW, ""), (ROW, "")) + reply = server._perform_uia_initialized( + _auto(listing.children[0].children[0]), + "identity-candidates-at", + {"x": 5, "y": 5}, + ) + assert reply["status"] == "ok" and len(reply["census"]["candidates"]) == 2 + + +# -- macOS backend ------------------------------------------------------------- + + +def test_macos_backend_converts_row_offsets_to_pixels() -> None: + from tests.test_macos_structural import FakeMacAXClient, backend + + class CensusAX(FakeMacAXClient): + def row_census(self, pid, window_title, x, y, *, limit): + return { + "record_key": None, + "incomplete": False, + "top_level": True, + "candidates": [ + {"self": True, "record_key": None, "dx": 0.0, "dy": 0.0}, + {"self": False, "record_key": None, "dx": 0.0, "dy": 22.0}, + ], + } + + target = backend(ax_client=CensusAX()) + target.screenshot() + census = target.identity_candidates_at(10, 10) + assert census is not None + assert census["candidates"][1]["dy"] == pytest.approx(22.0 * (target._scale_y or 1)) + assert _decide(census).verdict == "ambiguous" + + +def test_macos_client_without_a_census_cannot_count() -> None: + from tests.test_macos_structural import FakeMacAXClient, backend + + assert backend(ax_client=FakeMacAXClient()).identity_candidates_at(10, 10) is None + + +# -- the replayer on a native backend ------------------------------------------ + + +class NativeBackend: + """Structured text and a row census over the fake tree, like a native backend.""" + + viewport = (400, 300) + + def __init__(self, table: Node, at: dict[tuple[int, int], Node]): + self._table = table + self._at = at + + def screenshot(self) -> bytes: + return b"" + + def structured_text_at(self, x: int, y: int) -> Optional[str]: + node = self._at.get((x, y)) + if node is None: + return None + row = node.parent + return _row_identity(row, [row.kids.index(node)]) + + def identity_candidates_at(self, x: int, y: int) -> Optional[dict]: + node = self._at.get((x, y)) + return _census(node) if node is not None else None + + +def _step(recorded_key: Optional[str]) -> Step: + template = itmpl.build_identity_template( + None, + structured_identity="Follow-up Dr Lee", + record_key=recorded_key, + structured_row=True, + ) + return Step( + id="open", + intent="click 'John Smith'", + action=ActionKind.CLICK, + identity_armed=True, + anchor=Anchor( + template="templates/open.png", + region=(0, 0, 100, 40), + click_point=(50, 20), + identity_template=template, + ), + ) + + +def _sign() -> Step: + return Step(id="sign", intent="click 'Sign visit note'", action=ActionKind.CLICK) + + +def _verify(backend: NativeBackend, step: Step, point) -> IdentityCheck: + replayer = Replayer(backend, vision=object(), poll_interval_s=0.01) + resolution = Resolution(rung="template", point=point, confidence=1.0, elapsed_ms=1) + return replayer._verify_identity( + step, resolution, b"", {}, Workflow(name="wf", steps=[step, _sign()]) + ) + + +def _native(rows) -> tuple[NativeBackend, Node]: + table = _table(*rows) + at = {(50, 20 + 40 * i): row.kids[0] for i, row in enumerate(table.kids)} + return NativeBackend(table, at), table + + +def test_replay_halts_on_identical_native_rows() -> None: + backend, _ = _native([(ROW, ()), (ROW, ())]) + check = _verify(backend, _step(None), (50, 20)) + assert check.status == "abstain" + assert check.ambiguity_halt == "identical_candidates" + assert check.candidate_count == 2 + + +def test_replay_moves_to_the_native_row_with_the_recorded_id() -> None: + backend, _ = _native([(ROW, ("chart-1002",)), (ROW, ("chart-1001",))]) + step = _step("id=chart-1001") + replayer = Replayer(backend, vision=object(), poll_interval_s=0.01) + resolution = Resolution( + rung="template", point=(50, 20), confidence=1.0, elapsed_ms=1 + ) + moved, _region = replayer._retarget_by_record_key(step, resolution, (0, 0, 100, 40)) + assert moved.point == (50, 60) + check = _verify(backend, step, moved.point) + assert check.status == "verified" and check.candidate_count == 2 + + +def test_replay_runs_a_unique_native_row() -> None: + backend, _ = _native([(ROW, ()), (OTHER, ())]) + check = _verify(backend, _step(None), (50, 20)) + assert check.status == "verified" and check.candidate_count == 1 From 2bfeb15534c9cf36ee3c1bd75d7c8f0e1ca981b4 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 24 Oct 2026 21:16:40 -0400 Subject: [PATCH 2/7] fix(flow): compare AX elements with CFEqual in the row census On the macOS runner, the real-app census returned None. AX hands back a new wrapper for the same element on every read, so == never matched while the census walked from the hit element up to its row. Tree views can now define same(); the AX view uses CFEqual. A failing real-app test now prints the AX roles from the hit element up. The LIMITS.md path is now package-relative, which the docs consistency check requires. Co-Authored-By: Claude Opus 5.5 --- llms-full.txt | 2 +- packages/flow/docs/LIMITS.md | 2 +- .../openadapt_flow/backends/macos_backend.py | 7 ++++ .../openadapt_flow/runtime/identity_census.py | 34 ++++++++++++++---- .../tests/test_native_census_real_apps.py | 36 +++++++++++++++++-- 5 files changed, 69 insertions(+), 12 deletions(-) diff --git a/llms-full.txt b/llms-full.txt index bfed0cd0c..a046c909a 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -1337,7 +1337,7 @@ For an armed step: A verified identity says the row reads like the recorded row, not that no other row reads the same. Before an armed step acts, the runtime counts the -rows that show the same identity (`runtime/identity_census.py`). When more +rows that show the same identity (`openadapt_flow/runtime/identity_census.py`). When more than one does, it acts only on the row with the record id recorded at teach time, or halts. How each surface counts: diff --git a/packages/flow/docs/LIMITS.md b/packages/flow/docs/LIMITS.md index 499753072..2d9a49182 100644 --- a/packages/flow/docs/LIMITS.md +++ b/packages/flow/docs/LIMITS.md @@ -187,7 +187,7 @@ For an armed step: A verified identity says the row reads like the recorded row, not that no other row reads the same. Before an armed step acts, the runtime counts the -rows that show the same identity (`runtime/identity_census.py`). When more +rows that show the same identity (`openadapt_flow/runtime/identity_census.py`). When more than one does, it acts only on the row with the record id recorded at teach time, or halts. How each surface counts: diff --git a/packages/flow/openadapt_flow/backends/macos_backend.py b/packages/flow/openadapt_flow/backends/macos_backend.py index 1b561a322..258d7ca3e 100644 --- a/packages/flow/openadapt_flow/backends/macos_backend.py +++ b/packages/flow/openadapt_flow/backends/macos_backend.py @@ -605,6 +605,13 @@ class _AXTreeView: def __init__(self, client: QuartzMacAXClient) -> None: self._client = client + @staticmethod + def same(a: Any, b: Any) -> bool: + # Every AX read returns a new wrapper; CFEqual compares the elements. + from CoreFoundation import CFEqual + + return bool(CFEqual(a, b)) + def parent(self, node: Any) -> Any: from ApplicationServices import kAXParentAttribute diff --git a/packages/flow/openadapt_flow/runtime/identity_census.py b/packages/flow/openadapt_flow/runtime/identity_census.py index ee2d23bb9..d50e6e71a 100644 --- a/packages/flow/openadapt_flow/runtime/identity_census.py +++ b/packages/flow/openadapt_flow/runtime/identity_census.py @@ -182,6 +182,23 @@ def id_attributes(self, node: Any) -> list[tuple[str, str]]: ... def row_count(self, container: Any) -> Optional[int]: ... +def _same(view: TreeView, a: Any, b: Any) -> bool: + """Whether two node handles are the same node. + + A view MAY define ``same(a, b)``: AX hands back a fresh wrapper for the + same element on every read, so equality there is ``CFEqual``. + """ + if a is b: + return True + same = getattr(view, "same", None) + if callable(same): + try: + return bool(same(a, b)) + except Exception: + return False + return bool(a == b) + + def _record_key(view: TreeView, row: Any, path: list[int], index: int) -> Optional[str]: """The stable record id of ``row``: ids on the row and on the path to the target, minus values that are only the row's position.""" @@ -255,13 +272,15 @@ def tree_row_census( return {"row": False} path: list[int] = [] node = target - while node is not row: + while not _same(view, node, row): parent = view.parent(node) if parent is None: return None siblings = view.children(parent) try: - path.insert(0, next(i for i, c in enumerate(siblings) if c == node)) + path.insert( + 0, next(i for i, c in enumerate(siblings) if _same(view, c, node)) + ) except StopIteration: return None node = parent @@ -271,10 +290,10 @@ def tree_row_census( container = view.parent(row) siblings = view.children(container) if container is not None else [row] row_role = view.role(row) - rows = [c for c in siblings if c == row or view.role(c) == row_role] + rows = [c for c in siblings if _same(view, c, row) or view.role(c) == row_role] declared = view.row_count(container) if container is not None else None incomplete = len(rows) > max_rows or (declared is not None and declared > len(rows)) - row_index = next((i for i, c in enumerate(rows) if c == row), 0) + row_index = next((i for i, c in enumerate(rows) if _same(view, c, row)), 0) result: dict[str, Any] = { "record_key": _record_key(view, row, path, row_index), "incomplete": incomplete, @@ -285,7 +304,8 @@ def tree_row_census( return result origin = view.origin(row) for index, candidate in enumerate(rows): - if candidate == row: + is_row = _same(view, candidate, row) + if is_row: observed: Optional[str] = own_identity else: node = candidate @@ -310,10 +330,10 @@ def tree_row_census( ) result["candidates"].append( { - "self": candidate == row, + "self": is_row, "record_key": ( result["record_key"] - if candidate == row + if is_row else _record_key(view, candidate, path, index) ), "dx": offset[0], diff --git a/packages/flow/tests/test_native_census_real_apps.py b/packages/flow/tests/test_native_census_real_apps.py index 87d82419c..e14b76900 100644 --- a/packages/flow/tests/test_native_census_real_apps.py +++ b/packages/flow/tests/test_native_census_real_apps.py @@ -274,13 +274,40 @@ def row_centers() -> Optional[list[tuple[float, float]]]: proc.kill() +def _ax_trail(client: Any, pid: int, x: float, y: float) -> list[Any]: + """Roles and text from the element at (x, y) up to its window, for a + failure message.""" + + from ApplicationServices import ( + AXUIElementCopyElementAtPosition, + AXUIElementCreateApplication, + ) + + from openadapt_flow.backends.macos_backend import _AXTreeView + + view = _AXTreeView(client) + _err, node = AXUIElementCopyElementAtPosition( + AXUIElementCreateApplication(pid), float(x), float(y), None + ) + trail = [] + for _ in range(10): + if node is None: + break + trail.append((view.role(node), client._text(node), view.id_attributes(node))) + node = view.parent(node) + return trail + + def test_macos_table_identical_rows_are_ambiguous(cocoa_worklist) -> None: client, start = cocoa_worklist pid, centers = start( [{"text": SAME, "id": ""}, {"text": SAME, "id": ""}, {"text": OTHER, "id": ""}] ) census = client.row_census(pid, TITLE, *centers[1], limit=4000) - assert _decide(census) == CensusDecision("ambiguous", 2), census + assert _decide(census) == CensusDecision("ambiguous", 2), ( + census, + _ax_trail(client, pid, *centers[1]), + ) def test_macos_table_record_ids_pick_the_recorded_row(cocoa_worklist) -> None: @@ -290,7 +317,7 @@ def test_macos_table_record_ids_pick_the_recorded_row(cocoa_worklist) -> None: ) census = client.row_census(pid, TITLE, *centers[0], limit=4000) decision = _decide(census, "ax-identifier=chart-1001") - assert decision.verdict == "retarget", census + assert decision.verdict == "retarget", (census, _ax_trail(client, pid, *centers[0])) assert decision.offset is not None and decision.offset[1] > 0 @@ -298,4 +325,7 @@ def test_macos_table_unique_row_runs(cocoa_worklist) -> None: client, start = cocoa_worklist pid, centers = start([{"text": SAME, "id": ""}, {"text": OTHER, "id": ""}]) census = client.row_census(pid, TITLE, *centers[1], limit=4000) - assert _decide(census) == CensusDecision("unique", 1), census + assert _decide(census) == CensusDecision("unique", 1), ( + census, + _ax_trail(client, pid, *centers[1]), + ) From 06bb5200c19cfa89113b01f3edbf1f6ea0f70dc7 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 24 Oct 2026 21:32:46 -0400 Subject: [PATCH 3/7] fix(flow): close the census fail-opens found in review - Pixel census: only a clear pixel mismatch rules a place out. A verify, an abstain, no verdict, or verify turned off counts the place as a look-alike. Before this, a VLM-verified remote step read as unique. A place that cannot be cut, or a target place missing from the peaks (another render scale), is uncounted. Peaks are found at the matcher's look-alike score. - Tree census: a container with no row count is incomplete unless it is known not to scroll (macOS AXScrollArea with a vertical scroll bar, Linux scroll pane with a scroll bar, or unknown). - Windows agent: a pattern provider that raises makes the census incomplete, and so does a sibling row with no cells. - Record ids: positional_id drops ids that name a position (row-3, r3c2, row-3-col-1, item_12, absolute row numbers, or several numbers) in the tree census, the Windows agent and the browser. - CI: both real-app legs must run every test, and each outcome goes to the job summary. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/packages.yml | 26 ++- llms-full.txt | 15 +- packages/flow/docs/LIMITS.md | 15 +- .../openadapt_flow/backends/linux_backend.py | 13 ++ .../openadapt_flow/backends/macos_backend.py | 15 ++ .../backends/playwright_backend.py | 4 + .../backends/win_agent/server.py | 51 +++-- .../openadapt_flow/runtime/identity_census.py | 49 +++- .../flow/openadapt_flow/runtime/replayer.py | 43 ++-- .../flow/tests/test_native_identity_census.py | 211 +++++++++++++++++- 10 files changed, 390 insertions(+), 52 deletions(-) diff --git a/.github/workflows/packages.yml b/.github/workflows/packages.yml index baedac0d2..7e48ff345 100644 --- a/.github/workflows/packages.yml +++ b/.github/workflows/packages.yml @@ -478,9 +478,8 @@ jobs: # Real native list apps read through the row census (identical rows halt, # record ids pick the row, a unique row runs): a WPF ListBox over UI - # Automation on Windows, which must run with no skips, and a Swift - # NSTableView over AX on macOS, which skips when the hosted runner gives the - # job no Accessibility access (the step summary says so). + # Automation on Windows and a Swift NSTableView over AX on macOS. Both legs + # must run every real-app test; the step summary lists each outcome. flow-native-census: name: flow native row census (${{ matrix.os }}) needs: scope @@ -518,18 +517,31 @@ jobs: mkdir -p runs pytest -v -rs tests/test_native_identity_census.py \ tests/test_native_census_real_apps.py --junitxml=runs/native-junit.xml - - name: The Windows real-app tests ran - if: runner.os == 'Windows' + - name: The real-app tests ran (no skips) run: | python - <<'PY' + import os import sys import xml.etree.ElementTree as ET + prefix = {"Windows": "test_windows_wpf_", "macOS": "test_macos_table_"}[ + os.environ["RUNNER_OS"] + ] cases = [ c for c in ET.parse("runs/native-junit.xml").getroot().iter("testcase") - if c.get("name", "").startswith("test_windows_wpf_") + if c.get("name", "").startswith(prefix) ] + lines = [f"### Native row census, real app ({os.environ['RUNNER_OS']})", ""] + for case in cases: + skip = case.find("skipped") + failed = case.find("failure") is not None or case.find("error") is not None + state = "skipped: " + (skip.get("message") or "") if skip is not None else ( + "failed" if failed else "passed" + ) + lines.append(f"- `{case.get('name')}`: {state}") + with open(os.environ["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as out: + out.write("\n".join(lines) + "\n") + print("\n".join(lines)) skipped = [c.get("name") for c in cases if c.find("skipped") is not None] - print(f"{len(cases)} WPF tests, skipped: {skipped}") sys.exit(1 if not cases or skipped else 0) PY diff --git a/llms-full.txt b/llms-full.txt index a046c909a..de16d1220 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -1349,13 +1349,20 @@ time, or halts. How each surface counts: counted, and a row pick before a write halts; - macOS (AX): the rows of the `AXTable`, `AXOutline` or `AXList` around the element, with the AX identifier and URL as record ids. A table that reports - more `AXRows` than it exposes cannot be counted; + more `AXRows` than it exposes, or a list with no row count inside a scroll + area with a scroll bar, cannot be counted; - Linux (AT-SPI): the sibling list items, table rows or tree items of the - clicked row, with AT-SPI accessible ids as record ids. GTK 3 list boxes and + clicked row, with AT-SPI accessible ids as record ids. A list with no Table + row count inside a scroll pane with a scroll bar cannot be counted. GTK 3 + list boxes and tree views expose no row identity under the structured reading (one child per row, or cells without a row), so the identity check there rests on OCR; -- pixel-only displays: the OCR rows of the frame, or, with no OCR band, the - places the identifier crop verifies. +- pixel-only displays: the OCR rows of the frame, or, with no OCR band, every + place the identifier crop matches. Only a clear pixel mismatch rules a place + out; a frame at another scale than the recording cannot be counted. + +Ids that name a position (`row-3`, `r3c2`, `item_12`, the row's own index) +never tell rows apart. A surface that cannot count the rows halts any row pick that leads into a write. diff --git a/packages/flow/docs/LIMITS.md b/packages/flow/docs/LIMITS.md index 2d9a49182..6baf20a49 100644 --- a/packages/flow/docs/LIMITS.md +++ b/packages/flow/docs/LIMITS.md @@ -199,13 +199,20 @@ time, or halts. How each surface counts: counted, and a row pick before a write halts; - macOS (AX): the rows of the `AXTable`, `AXOutline` or `AXList` around the element, with the AX identifier and URL as record ids. A table that reports - more `AXRows` than it exposes cannot be counted; + more `AXRows` than it exposes, or a list with no row count inside a scroll + area with a scroll bar, cannot be counted; - Linux (AT-SPI): the sibling list items, table rows or tree items of the - clicked row, with AT-SPI accessible ids as record ids. GTK 3 list boxes and + clicked row, with AT-SPI accessible ids as record ids. A list with no Table + row count inside a scroll pane with a scroll bar cannot be counted. GTK 3 + list boxes and tree views expose no row identity under the structured reading (one child per row, or cells without a row), so the identity check there rests on OCR; -- pixel-only displays: the OCR rows of the frame, or, with no OCR band, the - places the identifier crop verifies. +- pixel-only displays: the OCR rows of the frame, or, with no OCR band, every + place the identifier crop matches. Only a clear pixel mismatch rules a place + out; a frame at another scale than the recording cannot be counted. + +Ids that name a position (`row-3`, `r3c2`, `item_12`, the row's own index) +never tell rows apart. A surface that cannot count the rows halts any row pick that leads into a write. diff --git a/packages/flow/openadapt_flow/backends/linux_backend.py b/packages/flow/openadapt_flow/backends/linux_backend.py index 0a62bfb7d..f79aa24bc 100644 --- a/packages/flow/openadapt_flow/backends/linux_backend.py +++ b/packages/flow/openadapt_flow/backends/linux_backend.py @@ -1624,3 +1624,16 @@ def row_count(self, container: Any) -> Optional[int]: return int(count) if count is not None else None except (TypeError, ValueError): return None + + def scrollable(self, container: Any) -> Optional[bool]: + """True when an enclosing scroll pane holds a scroll bar.""" + node = container + for _ in range(3): + if node is None: + return False + if self.role(node) == "scroll pane": + return any( + self.role(child) == "scroll bar" for child in self.children(node) + ) + node = self.parent(node) + return False diff --git a/packages/flow/openadapt_flow/backends/macos_backend.py b/packages/flow/openadapt_flow/backends/macos_backend.py index 258d7ca3e..e52bc9372 100644 --- a/packages/flow/openadapt_flow/backends/macos_backend.py +++ b/packages/flow/openadapt_flow/backends/macos_backend.py @@ -656,6 +656,21 @@ def row_count(self, container: Any) -> Optional[int]: except TypeError: return None + def scrollable(self, container: Any) -> Optional[bool]: + """True when an enclosing AXScrollArea has a vertical scroll bar.""" + from ApplicationServices import kAXVerticalScrollBarAttribute + + node = container + for _ in range(3): + if node is None: + return False + if self.role(node) == "AXScrollArea": + return ( + self._client._attr(node, kAXVerticalScrollBarAttribute) is not None + ) + node = self.parent(node) + return False + _NATIVE_MODIFIER_FLAGS = { "ctrl": "control", diff --git a/packages/flow/openadapt_flow/backends/playwright_backend.py b/packages/flow/openadapt_flow/backends/playwright_backend.py index e0864d5d4..bb5f0ad29 100644 --- a/packages/flow/openadapt_flow/backends/playwright_backend.py +++ b/packages/flow/openadapt_flow/backends/playwright_backend.py @@ -711,7 +711,11 @@ class _FramePoint: const POSITIONAL = /(?:index|idx|position|pos|order|rank|seq)/i; const parent = row.parentElement; const index = parent ? Array.prototype.indexOf.call(parent.children, row) : -1; + // Mirror of identity_census.positional_id: an id that names a position + // (row-3, r3c2, item_12, the row's own index) moves with the row. const positional = (value) => { + if ((value.match(/\d+/g) || []).length > 1) return true; + if (/(?:^|[^a-z])(?:row|r|item|idx|index|cell|col|c|line|pos|position|listviewitem|node|child)[-_ :#]?\d+/i.test(value)) return true; const m = /^\D*(\d+)\D*$/.exec(value); if (!m) return false; const n = Number(m[1]); diff --git a/packages/flow/openadapt_flow/backends/win_agent/server.py b/packages/flow/openadapt_flow/backends/win_agent/server.py index 229b94e15..490a6c28f 100644 --- a/packages/flow/openadapt_flow/backends/win_agent/server.py +++ b/packages/flow/openadapt_flow/backends/win_agent/server.py @@ -740,7 +740,24 @@ def _structured_text_at(auto: Any, x: int, y: int) -> Optional[str]: {"DataItemControl", "ListItemControl", "TreeItemControl", "TableRowControl"} ) _MAX_CENSUS_ROWS = 2000 -_POSITIONAL_ID = re.compile(r"^\D*(\d+)\D*$") +_SINGLE_NUMBER = re.compile(r"^\D*(\d+)\D*$") +# Mirror of identity_census.positional_id. +_POSITION_WORD = re.compile( + r"(?:^|[^a-z])(?:row|r|item|idx|index|cell|col|c|line|pos|position|" + r"listviewitem|node|child)[-_ :#]?\d+", + re.IGNORECASE, +) + + +def _positional_id(value: str, index: int) -> bool: + """Whether a row id names a position (``row-3``, ``r3c2``) not a record.""" + + if len(re.findall(r"\d+", value)) > 1: + return True + if _POSITION_WORD.search(value): + return True + match = _SINGLE_NUMBER.match(value) + return bool(match and int(match.group(1)) in (index, index + 1)) def _row_identity(row: object, excluded: Optional[int]) -> Optional[str]: @@ -773,8 +790,7 @@ def _census_record_key( value = str(_control_value(control, "AutomationId", "") or "").strip() if not value or len(value) > 256: continue - match = _POSITIONAL_ID.match(value) - if match and int(match.group(1)) in (index, index + 1): + if _positional_id(value, index): continue keys.add(f"automation-id={value}") return "\n".join(sorted(keys)) or None @@ -855,22 +871,22 @@ def same(a: object, b: object) -> bool: row_index = next((i for i, c in enumerate(rows) if same(c, row)), None) incomplete = len(rows) > _MAX_CENSUS_ROWS or row_index is None if container is not None and not incomplete: - grid = _control_value(container, "GetGridPattern", None) - try: - declared = int(grid().RowCount) if callable(grid) else None - except Exception: # noqa: BLE001 - declared = None - if declared is not None and declared > len(rows): - incomplete = True - scroll = _control_value(container, "GetScrollPattern", None) - items = _control_value(container, "GetItemContainerPattern", None) + # A provider that fails while answering is not a complete answer. try: + grid_get = _control_value(container, "GetGridPattern", None) + grid = grid_get() if callable(grid_get) else None + declared = int(grid.RowCount) if grid is not None else None + scroll_get = _control_value(container, "GetScrollPattern", None) + scroll = scroll_get() if callable(scroll_get) else None scrollable = ( - bool(scroll().VerticallyScrollable) if callable(scroll) else False + bool(scroll.VerticallyScrollable) if scroll is not None else False ) - realizes = callable(items) and items() is not None + items_get = _control_value(container, "GetItemContainerPattern", None) + realizes = callable(items_get) and items_get() is not None except Exception: # noqa: BLE001 - scrollable, realizes = False, False + declared, scrollable, realizes, incomplete = None, True, True, True + if declared is not None and declared > len(rows): + incomplete = True if scrollable and realizes: incomplete = True result: dict[str, Any] = { @@ -890,6 +906,11 @@ def same(a: object, b: object) -> bool: else: observed = _row_identity(candidate, excluded) kids = _children(candidate) + if row_children and not kids: + # A row whose cells are not built yet: it cannot be read. + result["incomplete"] = True + result["candidates"] = [] + return result candidate_branch = ( kids[excluded] if excluded is not None and excluded < len(kids) diff --git a/packages/flow/openadapt_flow/runtime/identity_census.py b/packages/flow/openadapt_flow/runtime/identity_census.py index d50e6e71a..340de3721 100644 --- a/packages/flow/openadapt_flow/runtime/identity_census.py +++ b/packages/flow/openadapt_flow/runtime/identity_census.py @@ -156,7 +156,30 @@ def count_matching_rows( #: Rows read before a census gives up and reports ``incomplete``. TREE_CENSUS_ROW_CAP = 2000 -_POSITIONAL_ID = re.compile(r"^\D*(\d+)\D*$") +_SINGLE_NUMBER = re.compile(r"^\D*(\d+)\D*$") +_POSITION_WORD = re.compile( + r"(?:^|[^a-z])(?:row|r|item|idx|index|cell|col|c|line|pos|position|" + r"listviewitem|node|child)[-_ :#]?\d+", + re.IGNORECASE, +) + + +def positional_id(value: str, index: int) -> bool: + """Whether a row id names a position rather than a record. + + Such an id moves with the row (``row-3``, ``r3c2``, ``item_12``, a bare + ``3``): after the rows reorder or scroll it points at whichever record now + sits there, so it never tells rows apart. True for an id whose only number + is the row's index (0- or 1-based), for an id that spells a position + (``row``, ``item``, ``cell``, ``col`` ... followed by a number), and for an + id with more than one number (grid coordinates). + """ + if len(re.findall(r"\d+", value)) > 1: + return True + if _POSITION_WORD.search(value): + return True + match = _SINGLE_NUMBER.match(value) + return bool(match and int(match.group(1)) in (index, index + 1)) class TreeView(Protocol): @@ -166,7 +189,9 @@ class TreeView(Protocol): coordinates, or None when it cannot be mapped. ``id_attributes`` lists the node's candidate record ids as ``(name, value)`` pairs (an AX identifier, an AT-SPI ``id`` attribute, a URL). ``row_count`` is the number of rows the - container says it has, when it says (None otherwise). + container says it has, when it says (None otherwise). ``scrollable`` says + whether a container without a row count can scroll rows out of view; + anything but a definite False makes the census ``incomplete``. """ def parent(self, node: Any) -> Any: ... @@ -181,6 +206,10 @@ def id_attributes(self, node: Any) -> list[tuple[str, str]]: ... def row_count(self, container: Any) -> Optional[int]: ... + def scrollable(self, container: Any) -> Optional[bool]: + """Whether the container can scroll rows out of view; None unknown.""" + ... + def _same(view: TreeView, a: Any, b: Any) -> bool: """Whether two node handles are the same node. @@ -217,8 +246,7 @@ def _record_key(view: TreeView, row: Any, path: list[int], index: int) -> Option value = (value or "").strip() if not value or len(value) > 256: continue - match = _POSITIONAL_ID.match(value) - if match and int(match.group(1)) in (index, index + 1): + if positional_id(value, index): continue parts.add(f"{name}={value}") return "\n".join(sorted(parts)) or None @@ -292,7 +320,18 @@ def tree_row_census( row_role = view.role(row) rows = [c for c in siblings if _same(view, c, row) or view.role(c) == row_role] declared = view.row_count(container) if container is not None else None - incomplete = len(rows) > max_rows or (declared is not None and declared > len(rows)) + if declared is not None: + hidden_rows = declared > len(rows) + elif container is None: + hidden_rows = False + else: + # No row count to check against: only a container that cannot + # scroll shows every row it has. Unknown scrollability is incomplete. + try: + hidden_rows = view.scrollable(container) is not False + except Exception: + hidden_rows = True + incomplete = len(rows) > max_rows or hidden_rows row_index = next((i for i, c in enumerate(rows) if _same(view, c, row)), 0) result: dict[str, Any] = { "record_key": _record_key(view, row, path, row_index), diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 35f67669c..e40a3aa17 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -15704,14 +15704,21 @@ def _pixel_census( workflow: Workflow, bundle_dir: Optional[Path], ) -> census_mod.CensusDecision: - """Count the places on the frame where the identifier crop verifies. - - For a pixel identity with no OCR band (a remote display): every - place the recorded identifier crop could match is cut from the frame - and checked with the same pixel verifier the ladder used. More than - one verifying place means another row shows the same identifier. + """Count the places on the frame that could show the identifier crop. + + For a pixel identity with no OCR band (a remote display): every place + the recorded identifier crop matches is cut from the frame and checked + with the ladder's pixel verifier. Only a clear ``mismatch`` rules a + place out; a verify, an abstain (drift, the uncertain band, verify + turned off) or no verdict counts it as a look-alike. The census is + ``uncounted`` when the frame has too many places, when a place cannot + be cut, or when the target's own place is not among them (the frame + renders at another scale than the recorded crop). """ - from openadapt_flow.vision.match import template_peaks + from openadapt_flow.vision.match import ( + AMBIGUITY_SUSPICION_SCORE, + template_peaks, + ) anchor = step.anchor assert anchor is not None @@ -15721,7 +15728,9 @@ def _pixel_census( if recorded is None or live is None or anchor.identifier_region is None: return census_mod.CensusDecision("cannot_count") try: - peaks = template_peaks(before_png, recorded) + peaks = template_peaks( + before_png, recorded, threshold=AMBIGUITY_SUSPICION_SCORE + ) except Exception: return census_mod.CensusDecision("cannot_count") if peaks is None: @@ -15731,19 +15740,25 @@ def _pixel_census( anchor, resolution, before_png, bundle_dir, workflow=workflow ) target = (rx + dx, ry + dy) + + def is_target(x: int, y: int) -> bool: + return abs(x - target[0]) < max(1, rw // 2) and abs(y - target[1]) < max( + 1, rh // 2 + ) + + if not any(is_target(x, y) for x, y in peaks): + return census_mod.CensusDecision("uncounted") count = 1 for x, y in peaks: - if abs(x - target[0]) < max(1, rw // 2) and abs(y - target[1]) < max( - 1, rh // 2 - ): - continue # the target itself + if is_target(x, y): + continue crop = identity_mod.crop_region(before_png, (x, y, rw, rh)) if crop is None: - continue + return census_mod.CensusDecision("uncounted") verdict = identity_mod.verify_pixel_identity( recorded, crop, enable_verify=self.pixel_verify_enabled ) - if verdict is not None and verdict.status == "verified": + if verdict is None or verdict.status != "mismatch": count += 1 if count > 1: return census_mod.CensusDecision("ambiguous", count) diff --git a/packages/flow/tests/test_native_identity_census.py b/packages/flow/tests/test_native_identity_census.py index 1fa6c4f14..5fece9673 100644 --- a/packages/flow/tests/test_native_identity_census.py +++ b/packages/flow/tests/test_native_identity_census.py @@ -60,6 +60,7 @@ def __init__(self, role: str, text: str = "", *, ids=(), y: float = 0.0): self.parent: Optional[Node] = None self.kids: list[Node] = [] self.declared_rows: Optional[int] = None + self.scrolls: Optional[bool] = False def add(self, *children: "Node") -> "Node": for child in children: @@ -87,6 +88,9 @@ def id_attributes(self, node: Node) -> list[tuple[str, str]]: def row_count(self, container: Node) -> Optional[int]: return container.declared_rows + def scrollable(self, container: Node) -> Optional[bool]: + return container.scrolls + def _table(*rows: tuple[tuple[str, ...], tuple[str, ...]]) -> Node: """A table of rows ``(cells, row ids)``, 40 px apart.""" @@ -144,14 +148,40 @@ def test_a_control_outside_any_row_is_not_a_row() -> None: assert _census(window.kids[0]) == {"row": False} -def test_positional_ids_never_tell_rows_apart() -> None: - table = _table((ROW, ("row-0",)), (ROW, ("row-1",))) +@pytest.mark.parametrize( + "ids", + [ + ("row-0", "row-1"), + ("r1c2", "r2c2"), + ("row-3-col-1", "row-4-col-1"), + ("row-53", "row-54"), # absolute positions in a longer list + ("item_12", "item_13"), + ("0", "1"), + ], +) +def test_positional_ids_never_tell_rows_apart(ids) -> None: + table = _table((ROW, (ids[0],)), (ROW, (ids[1],))) census = _census(table.kids[0].kids[0]) assert census is not None assert [c["record_key"] for c in census["candidates"]] == [None, None] assert _decide(census).verdict == "ambiguous" +def test_record_ids_survive_the_positional_filter() -> None: + from openadapt_flow.runtime.identity_census import positional_id + + for value in ("chart-1001", "patient_1001", "MRN-A1234", "rec-77"): + assert not positional_id(value, 0), value + + +@pytest.mark.parametrize("scrolls", [True, None]) +def test_a_list_without_a_row_count_that_may_scroll_is_incomplete(scrolls) -> None: + table = _table((ROW, ()), (OTHER, ())) + table.scrolls = scrolls # no row count, and rows may be out of view + census = _census(table.kids[1].kids[0]) + assert census is not None and census["incomplete"] is True + + def test_a_table_with_more_rows_than_it_exposes_is_incomplete() -> None: table = _table((ROW, ()), (OTHER, ())) table.declared_rows = 50 # a virtualized list: rows built on scroll @@ -329,6 +359,7 @@ def __init__(self, control_type: str, name: str = "", *, aid: str = "", top=0): self.children: list[UiaControl] = [] self.grid_rows: Optional[int] = None self.virtual = False + self.broken = False def add(self, *kids: "UiaControl") -> "UiaControl": for kid in kids: @@ -343,8 +374,10 @@ def GetChildren(self): return list(self.children) def GetGridPattern(self): + if self.broken: + raise RuntimeError("provider failed") if self.grid_rows is None: - raise RuntimeError("no grid pattern") + return None # unsupported, as uiautomation reports it return types.SimpleNamespace(RowCount=self.grid_rows) def GetScrollPattern(self): @@ -419,6 +452,24 @@ def test_windows_virtualized_or_short_grids_are_incomplete() -> None: assert _decide(census).verdict == "uncounted" +def test_windows_a_failing_provider_is_incomplete() -> None: + listing = _uia_list((ROW, ""), (OTHER, "")) + listing.broken = True + census = server._identity_candidates_at( + _auto(listing.children[1].children[0]), 5, 45 + ) + assert _decide(census).verdict == "uncounted" + + +def test_windows_a_row_without_cells_is_incomplete() -> None: + listing = _uia_list((ROW, ""), (OTHER, "")) + listing.add(UiaControl("DataItemControl", top=80)) # not built yet + census = server._identity_candidates_at( + _auto(listing.children[0].children[0]), 5, 5 + ) + assert _decide(census).verdict == "uncounted" + + def test_windows_standalone_control_is_not_a_row() -> None: window = UiaControl("WindowControl").add(UiaControl("ButtonControl", "Sign")) census = server._identity_candidates_at(_auto(window.children[0]), 5, 5) @@ -558,3 +609,157 @@ def test_replay_runs_a_unique_native_row() -> None: backend, _ = _native([(ROW, ()), (OTHER, ())]) check = _verify(backend, _step(None), (50, 20)) assert check.status == "verified" and check.candidate_count == 1 + + +# -- pixel census (remote display, no OCR band) --------------------------------- + + +def _crop_png(bar: int = 60, height: int = 48, width: int = 240) -> Any: + import numpy as np + + img = np.full((height, width), 255, np.uint8) + img[8:40, bar : bar + 6] = 0 + img[8:40, bar + 40 : bar + 44] = 0 + return img + + +def _png(arr: Any) -> bytes: + import cv2 + + ok, buf = cv2.imencode(".png", arr) + assert ok + return buf.tobytes() + + +def test_template_peaks_lists_every_place_and_refuses_a_crowded_frame() -> None: + import numpy as np + + from openadapt_flow.vision.match import MAX_PEAKS, template_peaks + + crop = _crop_png() + gap = np.full((10, crop.shape[1]), 255, np.uint8) + frame = np.vstack([crop, gap, crop, gap, _crop_png(bar=150)]) + peaks = template_peaks(_png(frame), _png(crop), threshold=0.9) + assert peaks is not None and sorted(y for _x, y in peaks) == [0, 58] + crowded = np.vstack([np.vstack([crop, gap])] * (MAX_PEAKS + 1)) + assert template_peaks(_png(crowded), _png(crop), threshold=0.9) is None + assert template_peaks(_png(crop[:10, :10]), _png(crop)) is None # too small + + +def _pixel_step(tmp_path, recorded: bytes) -> Step: + (tmp_path / "templates").mkdir(parents=True, exist_ok=True) + (tmp_path / "templates" / "id.png").write_bytes(recorded) + (tmp_path / "templates" / "t.png").write_bytes(recorded) + return Step( + id="open", + intent="click 'Open referral'", + action=ActionKind.CLICK, + identity_armed=True, + anchor=Anchor( + template="templates/t.png", + region=(0, 0, 240, 48), + click_point=(0, 0), + identifier_crop="templates/id.png", + identifier_region=(0, 0, 240, 48), + ), + ) + + +class _PixelBackend: + viewport = (240, 400) + + def screenshot(self) -> bytes: + return b"" + + +def _pixel_census( + tmp_path, frame: Any, *, verify: bool, monkeypatch=None, verdict=None +): + from openadapt_flow.runtime import identity as identity_mod + + if verdict is not None: + assert monkeypatch is not None + monkeypatch.setattr( + identity_mod, "verify_pixel_identity", lambda *a, **k: verdict + ) + replayer = Replayer( + _PixelBackend(), + vision=object(), + poll_interval_s=0.01, + pixel_verify_enabled=verify, + ) + step = _pixel_step(tmp_path, _png(_crop_png())) + resolution = Resolution(rung="template", point=(0, 0), confidence=1.0, elapsed_ms=1) + return replayer._pixel_census( + step, resolution, _png(frame), Workflow(name="wf"), tmp_path + ) + + +def _two_copies() -> Any: + import numpy as np + + gap = np.full((10, 240), 255, np.uint8) + return np.vstack([_crop_png(), gap, _crop_png()]) + + +def test_pixel_census_counts_a_second_copy(tmp_path) -> None: + decision = _pixel_census(tmp_path, _two_copies(), verify=True) + assert decision == CensusDecision("ambiguous", 2) + + +def test_pixel_census_counts_a_copy_when_verify_is_off(tmp_path) -> None: + """Verify disabled never yields 'verified'; the copy must still count.""" + + decision = _pixel_census(tmp_path, _two_copies(), verify=False) + assert decision == CensusDecision("ambiguous", 2) + + +@pytest.mark.parametrize( + "verdict", + [None, IdentityCheck(status="abstain", mode="pixel")], + ids=["no-verdict", "abstain"], +) +def test_pixel_census_counts_anything_short_of_a_mismatch( + tmp_path, monkeypatch, verdict +) -> None: + decision = _pixel_census( + tmp_path, _two_copies(), verify=True, monkeypatch=monkeypatch, verdict=verdict + ) + assert decision.verdict == "ambiguous" + + +def test_pixel_census_rules_out_a_clear_mismatch(tmp_path, monkeypatch) -> None: + decision = _pixel_census( + tmp_path, + _two_copies(), + verify=True, + monkeypatch=monkeypatch, + verdict=IdentityCheck(status="mismatch", mode="pixel"), + ) + assert decision == CensusDecision("unique", 1) + + +def test_pixel_census_is_uncounted_at_another_scale(tmp_path) -> None: + """The target's own place is not found at the recorded scale.""" + + import cv2 + + frame = cv2.resize(_two_copies(), None, fx=1.3, fy=1.3) + decision = _pixel_census(tmp_path, frame, verify=True) + assert decision.verdict == "uncounted" + + +def test_pixel_census_is_uncounted_when_a_place_cannot_be_cut( + tmp_path, monkeypatch +) -> None: + from openadapt_flow.runtime import identity as identity_mod + + real = identity_mod.crop_region + # The target's own crop cuts; the second place does not. + monkeypatch.setattr( + identity_mod, + "crop_region", + lambda png, region: real(png, region) if region[1] == 0 else None, + ) + decision = _pixel_census(tmp_path, _two_copies(), verify=True) + assert decision.verdict == "uncounted" From a08534d6f964771173cf65972a1b83215b4de82b Mon Sep 17 00:00:00 2001 From: abrichr Date: Sat, 24 Oct 2026 23:32:27 -0400 Subject: [PATCH 4/7] fix(flow): keep real record ids; signing out a patient is a write positional_id treated any id with more than one number as a position, and a bare "c" as a grid column. That dropped 100% of UUIDs, ObjectIds, ULIDs, encounter UUIDs and composite ids like ENC2024-000123, so rows that carried them could not be told apart. It now splits the id into alphanumeric tokens. An id is positional when a token is a position word with its number (row3, r3c2), when a position word is followed by a number token (row-3, item_12, row-3-col-1), when every token is a short number (3-1), or when its only number is the row's index. Measured over 12 families of 1000 realistic ids each, none is dropped; every positional example still is. The browser and the Windows agent use the same rule, with parity tests for both. Risk: "Sign out patient", "Log out " and "Discharge" are consequential writes. A bare "Sign out" / "Log out", or one that names the session ("of OpenAdapt", "now", "all devices"), stays session chrome. step_text joins its labels with " | ", so the intent and label of a session "Sign out" button no longer run together into "sign out Sign". Co-Authored-By: Claude Opus 5.5 --- .../backends/playwright_backend.py | 11 ++++- .../backends/win_agent/server.py | 16 ++++--- packages/flow/openadapt_flow/risk.py | 17 +++++-- .../openadapt_flow/runtime/identity_census.py | 29 +++++++---- .../flow/tests/test_native_identity_census.py | 48 +++++++++++++++++-- packages/flow/tests/test_realism_browser.py | 39 +++++++++++++++ packages/flow/tests/test_risk.py | 27 +++++++++++ 7 files changed, 163 insertions(+), 24 deletions(-) diff --git a/packages/flow/openadapt_flow/backends/playwright_backend.py b/packages/flow/openadapt_flow/backends/playwright_backend.py index bb5f0ad29..4f2d50b83 100644 --- a/packages/flow/openadapt_flow/backends/playwright_backend.py +++ b/packages/flow/openadapt_flow/backends/playwright_backend.py @@ -713,9 +713,16 @@ class _FramePoint: const index = parent ? Array.prototype.indexOf.call(parent.children, row) : -1; // Mirror of identity_census.positional_id: an id that names a position // (row-3, r3c2, item_12, the row's own index) moves with the row. + const WORDS = 'row|item|idx|index|cell|col|column|line|pos|position|listviewitem|node|child'; + const POSITION_TOKEN = new RegExp('^(?:(?:' + WORDS + ')\\d+|r\\d+c\\d+)$', 'i'); + const POSITION_WORD = new RegExp('^(?:' + WORDS + '|r|c)$', 'i'); const positional = (value) => { - if ((value.match(/\d+/g) || []).length > 1) return true; - if (/(?:^|[^a-z])(?:row|r|item|idx|index|cell|col|c|line|pos|position|listviewitem|node|child)[-_ :#]?\d+/i.test(value)) return true; + const tokens = value.split(/[^A-Za-z0-9]+/).filter(Boolean); + if (tokens.some((t) => POSITION_TOKEN.test(t))) return true; + for (let i = 0; i + 1 < tokens.length; i += 1) { + if (POSITION_WORD.test(tokens[i]) && /^\d+$/.test(tokens[i + 1])) return true; + } + if (tokens.length > 1 && tokens.every((t) => /^\d{1,3}$/.test(t))) return true; const m = /^\D*(\d+)\D*$/.exec(value); if (!m) return false; const n = Number(m[1]); diff --git a/packages/flow/openadapt_flow/backends/win_agent/server.py b/packages/flow/openadapt_flow/backends/win_agent/server.py index 490a6c28f..1ba38b2aa 100644 --- a/packages/flow/openadapt_flow/backends/win_agent/server.py +++ b/packages/flow/openadapt_flow/backends/win_agent/server.py @@ -742,19 +742,23 @@ def _structured_text_at(auto: Any, x: int, y: int) -> Optional[str]: _MAX_CENSUS_ROWS = 2000 _SINGLE_NUMBER = re.compile(r"^\D*(\d+)\D*$") # Mirror of identity_census.positional_id. -_POSITION_WORD = re.compile( - r"(?:^|[^a-z])(?:row|r|item|idx|index|cell|col|c|line|pos|position|" - r"listviewitem|node|child)[-_ :#]?\d+", - re.IGNORECASE, +_POSITION_WORDS = ( + "row|item|idx|index|cell|col|column|line|pos|position|listviewitem|node|child" ) +_POSITION_TOKEN = re.compile(rf"^(?:(?:{_POSITION_WORDS})\d+|r\d+c\d+)$", re.IGNORECASE) +_POSITION_WORD_TOKEN = re.compile(rf"^(?:{_POSITION_WORDS}|r|c)$", re.IGNORECASE) def _positional_id(value: str, index: int) -> bool: """Whether a row id names a position (``row-3``, ``r3c2``) not a record.""" - if len(re.findall(r"\d+", value)) > 1: + tokens = [t for t in re.split(r"[^A-Za-z0-9]+", value) if t] + if any(_POSITION_TOKEN.match(t) for t in tokens): return True - if _POSITION_WORD.search(value): + for word, following in zip(tokens, tokens[1:]): + if _POSITION_WORD_TOKEN.match(word) and following.isdigit(): + return True + if len(tokens) > 1 and all(t.isdigit() and len(t) <= 3 for t in tokens): return True match = _SINGLE_NUMBER.match(value) return bool(match and int(match.group(1)) in (index, index + 1)) diff --git a/packages/flow/openadapt_flow/risk.py b/packages/flow/openadapt_flow/risk.py index 8a61623f1..687d5c8b0 100644 --- a/packages/flow/openadapt_flow/risk.py +++ b/packages/flow/openadapt_flow/risk.py @@ -300,8 +300,9 @@ def string_tuple(field: str) -> tuple[str, ...]: # "postal". Ordered roughly by how common they are on real controls. # # "sign" counts in its writing sense ("Sign visit note", "Co-sign", "Signed"), -# the clinical attestation that locks a note, but never as "sign in", "sign -# on" or "sign out", which are session chrome. +# the clinical attestation that locks a note, but never as "sign in" or "sign +# on", and "sign out" / "log out" only with a clinical object ("Sign out +# patient"), never as the session's own "Sign out". # # Deliberately EXCLUDED to avoid noisy false positives on navigation chrome: # bare "post"/"order"/"book"/"complete" (collide with "Posts"/"Orders"/ @@ -323,6 +324,14 @@ def string_tuple(field: str) -> tuple[str, ...]: r"sign[\s\-_]?up", r"signup", r"(?:co[\-_]?|e[\-_]?)?sign(?:s|ed|ing)?(?![\s\-_]*(?:in|on|out)\b)", + # "Sign out patient", "Log out Jane Doe": signing out a record (a + # discharge or check-out) is a write. Only a bare "sign out" / "log out", + # or one that names the session ("sign out of OpenAdapt", "sign out now"), + # is session chrome. + r"(?:sign|log)(?:s|ged|ging|ed|ing)?[\s\-_]*out[\s\-_]+" + r"(?!(?:of|now|here|and|from|all|everywhere|completely|securely|safely)\b)" + r"[a-z]+", + r"discharg(?:e|es|ed|ing)", r"regist(?:er|ers|ering|ration)", r"enroll(?:s|ed|ing|ment)?", r"add(?:s|ed|ing)?", @@ -605,7 +614,9 @@ def step_text(step: Step) -> str: drag_end_structural = drag_end.structural if drag_end is not None else None if drag_end_structural is not None and drag_end_structural.name: parts.append(drag_end_structural.name) - return " ".join(parts) + # " | " keeps a write phrase from spanning two labels: "Sign out" twice + # must not read as "sign out Sign". + return " | ".join(parts) def infer_step_risk( diff --git a/packages/flow/openadapt_flow/runtime/identity_census.py b/packages/flow/openadapt_flow/runtime/identity_census.py index 340de3721..3672fe8d9 100644 --- a/packages/flow/openadapt_flow/runtime/identity_census.py +++ b/packages/flow/openadapt_flow/runtime/identity_census.py @@ -157,11 +157,13 @@ def count_matching_rows( TREE_CENSUS_ROW_CAP = 2000 _SINGLE_NUMBER = re.compile(r"^\D*(\d+)\D*$") -_POSITION_WORD = re.compile( - r"(?:^|[^a-z])(?:row|r|item|idx|index|cell|col|c|line|pos|position|" - r"listviewitem|node|child)[-_ :#]?\d+", - re.IGNORECASE, +_POSITION_WORDS = ( + "row|item|idx|index|cell|col|column|line|pos|position|listviewitem|node|child" ) +# A whole token that is a position word with its number (row3, item_12 is +# split into item + 12) or a grid coordinate (r3c2). +_POSITION_TOKEN = re.compile(rf"^(?:(?:{_POSITION_WORDS})\d+|r\d+c\d+)$", re.IGNORECASE) +_POSITION_WORD_TOKEN = re.compile(rf"^(?:{_POSITION_WORDS}|r|c)$", re.IGNORECASE) def positional_id(value: str, index: int) -> bool: @@ -169,14 +171,21 @@ def positional_id(value: str, index: int) -> bool: Such an id moves with the row (``row-3``, ``r3c2``, ``item_12``, a bare ``3``): after the rows reorder or scroll it points at whichever record now - sits there, so it never tells rows apart. True for an id whose only number - is the row's index (0- or 1-based), for an id that spells a position - (``row``, ``item``, ``cell``, ``col`` ... followed by a number), and for an - id with more than one number (grid coordinates). + sits there, so it never tells rows apart. The id is split into + alphanumeric tokens. It is positional when a token is a position word with + its number (``row3``, ``r3c2``), when a position word token is followed by + a number token (``row-3``, ``item_12``, ``row-3-col-1``), when every token + is a short number (grid coordinates such as ``3-1``), or when its only + number is the row's index (0- or 1-based). UUIDs, ULIDs, MRNs and + composite record ids (``ENC2024-000123``) are not. """ - if len(re.findall(r"\d+", value)) > 1: + tokens = [t for t in re.split(r"[^A-Za-z0-9]+", value) if t] + if any(_POSITION_TOKEN.match(t) for t in tokens): return True - if _POSITION_WORD.search(value): + for word, following in zip(tokens, tokens[1:]): + if _POSITION_WORD_TOKEN.match(word) and following.isdigit(): + return True + if len(tokens) > 1 and all(t.isdigit() and len(t) <= 3 for t in tokens): return True match = _SINGLE_NUMBER.match(value) return bool(match and int(match.group(1)) in (index, index + 1)) diff --git a/packages/flow/tests/test_native_identity_census.py b/packages/flow/tests/test_native_identity_census.py index 5fece9673..e7b84a2dd 100644 --- a/packages/flow/tests/test_native_identity_census.py +++ b/packages/flow/tests/test_native_identity_census.py @@ -10,6 +10,7 @@ from __future__ import annotations import types +import uuid from typing import Any, Optional import pytest @@ -167,11 +168,52 @@ def test_positional_ids_never_tell_rows_apart(ids) -> None: assert _decide(census).verdict == "ambiguous" -def test_record_ids_survive_the_positional_filter() -> None: +_STABLE_ID_FAMILIES = { + "uuid4": lambda r: str(uuid.UUID(int=r.getrandbits(128), version=4)), + "uuid4 without dashes": lambda r: uuid.UUID(int=r.getrandbits(128)).hex, + "encounter uuid": lambda r: f"encounter-{uuid.UUID(int=r.getrandbits(128))}", + "MRN-A1234": lambda r: f"MRN-A{r.randint(1000, 99999)}", + "zero-padded MRN": lambda r: f"{r.randint(1, 99_999_999):08d}", + "chart-1001": lambda r: f"chart-{r.randint(1000, 999_999)}", + "patient_N": lambda r: f"patient_{r.randint(100, 999_999)}", + "ENC2024-000123": lambda r: ( + f"ENC{r.randint(2015, 2026)}-{r.randint(1, 999_999):06d}" + ), + "FHIR Patient/123": lambda r: f"Patient/{r.randint(1, 99_999)}", + "dated visit": lambda r: f"visit-2026-10-{r.randint(10, 28)}-{r.randint(1, 99)}", + "ObjectId": lambda r: "%024x" % r.getrandbits(96), + "ULID": lambda r: ( + "01H" + "".join(r.choice("0123456789ABCDEFGHJKMNPQRSTVWXYZ") for _ in range(23)) + ), +} + + +@pytest.mark.parametrize("family", sorted(_STABLE_ID_FAMILIES)) +def test_stable_record_ids_are_never_dropped_as_positional(family) -> None: + """Measured: 0 of 1000 ids per family read as a position.""" + + import random + + from openadapt_flow.runtime.identity_census import positional_id + + rng = random.Random(family) + values = [_STABLE_ID_FAMILIES[family](rng) for _ in range(1000)] + dropped = [v for i, v in enumerate(values) if positional_id(v, i % 50)] + assert dropped == [] + + +def test_the_windows_agent_drops_the_same_ids() -> None: + import random + from openadapt_flow.runtime.identity_census import positional_id - for value in ("chart-1001", "patient_1001", "MRN-A1234", "rec-77"): - assert not positional_id(value, 0), value + rng = random.Random(3) + values = [g(rng) for g in _STABLE_ID_FAMILIES.values() for _ in range(50)] + values += ["row-3", "r3c2", "item_12", "row-3-col-1", "3-1", "ListViewItem-0", "7"] + for index, value in enumerate(values): + assert server._positional_id(value, index % 9) == positional_id( + value, index % 9 + ), value @pytest.mark.parametrize("scrolls", [True, None]) diff --git a/packages/flow/tests/test_realism_browser.py b/packages/flow/tests/test_realism_browser.py index 8469bdae2..a43a16dd0 100644 --- a/packages/flow/tests/test_realism_browser.py +++ b/packages/flow/tests/test_realism_browser.py @@ -450,3 +450,42 @@ def test_a_backend_that_cannot_count_rows_still_runs_a_plain_form( assert page.locator("#status").inner_text() == "Saved Ada" finally: page.close() + + +def test_browser_record_ids_follow_the_same_positional_rule(browser) -> None: + """The page-side record-id filter drops exactly what positional_id drops.""" + + import uuid + + from openadapt_flow.backends.playwright_backend import _RECORD_KEY_JS + from openadapt_flow.runtime.identity_census import positional_id + + values = [ + str(uuid.uuid4()), + uuid.uuid4().hex, + f"encounter-{uuid.uuid4()}", + "MRN-A1234", + "00012345", + "chart-1001", + "ENC2024-000123", + "01HEHRGVBY0HPFKMYZV5Q9KR35", + "row-3", + "r3c2", + "item_12", + "row-3-col-1", + "3-1", + "ListViewItem-0", + "2", + ] + page = browser.new_page() + try: + items = "".join(f'
  • x
    • ' for v in values) + page.set_content(f"
      {items}
    ") + keys = page.evaluate( + f"() => Array.from(document.querySelectorAll('li'))" + f".map((li) => ({_RECORD_KEY_JS})(li, li))" + ) + finally: + page.close() + for index, (value, key) in enumerate(zip(values, keys)): + assert (key is None) == positional_id(value, index), (value, key) diff --git a/packages/flow/tests/test_risk.py b/packages/flow/tests/test_risk.py index 070dec4c8..61001c14a 100644 --- a/packages/flow/tests/test_risk.py +++ b/packages/flow/tests/test_risk.py @@ -61,6 +61,13 @@ class TestHeuristic: "Approve", "Pay now", "Sign up", + "Sign visit note", + "Co-sign", + "Sign out patient", + "Sign-out patient Jane Doe", + "Log out patient", + "Discharge patient", + "Check out patient", "Next", "Continue", "Cancel", @@ -74,6 +81,15 @@ def test_write_shaped_true(self, text: str) -> None: [ "Login", "Sign In", + "Sign in with SSO", + "Sign on", + "Sign out", + "Sign Out", + "Signing out", + "Sign out of OpenAdapt", + "Sign out now", + "Sign out all devices", + "Log out", "Search", "Belford, Phil", "Address book", # 'add' must not trip inside 'address' @@ -85,6 +101,17 @@ def test_write_shaped_true(self, text: str) -> None: def test_write_shaped_false(self, text: str) -> None: assert not is_write_shaped(text) + def test_the_session_sign_out_button_stays_reversible(self) -> None: + """Intent and label both read "Sign out": the labels must not join + into "sign out Sign" and look like a clinical sign-out.""" + + step = _click_step("click 'Sign out'", "Sign out") + assert classify_step_risk(step) == "reversible" + + def test_signing_out_a_patient_is_irreversible(self) -> None: + step = _click_step("click 'Sign out patient'", "Sign out patient") + assert classify_step_risk(step) == "irreversible" + def test_submit_click_is_irreversible(self) -> None: assert ( classify_step_risk(_click_step("click 'Save as new message'", "Save")) From 152d69a3d7c08f4dc778da26e14c0d1519bdb0f6 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 25 Oct 2026 06:35:11 -0400 Subject: [PATCH 5/7] fix(flow): sibling tabs that share chrome are told apart by their label The pixel census counted every place where the identifier crop matched at 0.7 or more and wasn't a clear pixel mismatch. On OpenEMR-like section toggles, where the chrome dominates the crop and the label is small, siblings such as "Who" and "Misc" get no pixel verdict, so the "Contact" toggle would halt as ambiguous. The threshold is unchanged. A place that the pixels don't rule out is now distinguishable only when OCR reads both crops and their text differs. Unreadable text on either crop, or the same text, still counts it as a look-alike. Tests use rendered OpenEMR-like sibling tabs: "Contact" among distinct labels is unique; with unreadable labels it halts; two "Contact" tabs halt. Co-Authored-By: Claude Opus 5.5 --- .../flow/openadapt_flow/runtime/replayer.py | 34 +++++- .../flow/tests/test_native_identity_census.py | 101 ++++++++++++++++++ 2 files changed, 130 insertions(+), 5 deletions(-) diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 6496a94d0..4f74d1977 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -15768,9 +15768,11 @@ def _pixel_census( For a pixel identity with no OCR band (a remote display): every place the recorded identifier crop matches is cut from the frame and checked - with the ladder's pixel verifier. Only a clear ``mismatch`` rules a - place out; a verify, an abstain (drift, the uncertain band, verify - turned off) or no verdict counts it as a look-alike. The census is + with the ladder's pixel verifier. A clear ``mismatch`` rules a place + out, and so does its own text when OCR reads both crops and the text + differs (sibling tabs share their chrome, not their label). A verify, + an abstain (drift, the uncertain band, verify turned off) or no verdict + with the same or unreadable text counts it as a look-alike. The census is ``uncounted`` when the frame has too many places, when a place cannot be cut, or when the target's own place is not among them (the frame renders at another scale than the recorded crop). @@ -15808,6 +15810,18 @@ def is_target(x: int, y: int) -> bool: if not any(is_target(x, y) for x, y in peaks): return census_mod.CensusDecision("uncounted") + + def crop_text(png: bytes) -> str: + """The crop's own text, squashed; empty when OCR reads nothing.""" + try: + lines = self.vision.ocr(png, region=None) + except Exception: + return "" + return identity_mod.squash( + " ".join(str(getattr(line, "text", "") or "") for line in lines) + ) + + target_text: Optional[str] = None count = 1 for x, y in peaks: if is_target(x, y): @@ -15818,8 +15832,18 @@ def is_target(x: int, y: int) -> bool: verdict = identity_mod.verify_pixel_identity( recorded, crop, enable_verify=self.pixel_verify_enabled ) - if verdict is None or verdict.status != "mismatch": - count += 1 + if verdict is not None and verdict.status == "mismatch": + continue + # The pixels do not rule this place out. Sibling controls that + # share their chrome (section tabs, toggle buttons) look alike; + # their own text still tells them apart. Only when OCR reads both + # crops and the text differs is the place distinguishable. + if target_text is None: + target_text = crop_text(live) + other_text = crop_text(crop) + if target_text and other_text and other_text != target_text: + continue + count += 1 if count > 1: return census_mod.CensusDecision("ambiguous", count) return census_mod.CensusDecision("unique", count) diff --git a/packages/flow/tests/test_native_identity_census.py b/packages/flow/tests/test_native_identity_census.py index e7b84a2dd..0869a578a 100644 --- a/packages/flow/tests/test_native_identity_census.py +++ b/packages/flow/tests/test_native_identity_census.py @@ -805,3 +805,104 @@ def test_pixel_census_is_uncounted_when_a_place_cannot_be_cut( ) decision = _pixel_census(tmp_path, _two_copies(), verify=True) assert decision.verdict == "uncounted" + + +# -- pixel census tie-break: sibling tabs share chrome, not their label --------- + +_TAB_W, _TAB_H = 300, 40 +_TABS = ["Who", "Contact", "Choices", "Employer", "Stats", "Misc"] + + +def _tab(label: str) -> Any: + """An OpenEMR-like section toggle: chrome dominates, the label is small.""" + + import cv2 + import numpy as np + + img = np.full((_TAB_H, _TAB_W), 245, np.uint8) + cv2.rectangle(img, (0, 0), (_TAB_W - 1, _TAB_H - 1), 120, 2) + cv2.rectangle(img, (0, 0), (_TAB_W - 1, 10), 200, -1) + cv2.circle(img, (16, 25), 6, 60, -1) + cv2.putText( + img, label, (30, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.45, 40, 1, cv2.LINE_AA + ) + return img + + +def _tabs(labels: list[str]) -> Any: + import numpy as np + + gap = np.full((12, _TAB_W), 255, np.uint8) + return np.vstack([part for label in labels for part in (_tab(label), gap)]) + + +class _TabOcr: + """OCR that reads a tab crop as its nearest rendered label (or nothing).""" + + def __init__(self, readable: bool = True): + self.readable = readable + + def ocr(self, png: bytes, *, region=None): + import cv2 + import numpy as np + + if not self.readable: + return [] + crop = cv2.imdecode(np.frombuffer(png, np.uint8), cv2.IMREAD_GRAYSCALE) + if crop is None or crop.shape != (_TAB_H, _TAB_W): + return [] + label = min( + _TABS + ["Contact"], + key=lambda name: float(np.abs(crop.astype(int) - _tab(name)).mean()), + ) + return [types.SimpleNamespace(text=label)] + + +def _tab_census(tmp_path, labels: list[str], target: int, *, readable: bool = True): + (tmp_path / "templates").mkdir(parents=True, exist_ok=True) + recorded = _png(_tab(labels[target])) + (tmp_path / "templates" / "id.png").write_bytes(recorded) + (tmp_path / "templates" / "t.png").write_bytes(recorded) + top = target * (_TAB_H + 12) + step = Step( + id="open_contact", + intent="click 'Contact'", + action=ActionKind.CLICK, + identity_armed=True, + anchor=Anchor( + template="templates/t.png", + region=(0, top, _TAB_W, _TAB_H), + click_point=(0, top), + identifier_crop="templates/id.png", + identifier_region=(0, top, _TAB_W, _TAB_H), + ), + ) + replayer = Replayer( + _PixelBackend(), + vision=_TabOcr(readable), + poll_interval_s=0.01, + pixel_verify_enabled=True, + ) + resolution = Resolution( + rung="template", point=(0, top), confidence=1.0, elapsed_ms=1 + ) + return replayer._pixel_census( + step, resolution, _png(_tabs(labels)), Workflow(name="wf"), tmp_path + ) + + +def test_a_section_tab_among_sibling_tabs_is_unique_by_its_label(tmp_path) -> None: + """The OpenEMR "Contact" toggle: its siblings share the chrome, and some + are not a clear pixel mismatch, but their labels read differently.""" + + assert _tab_census(tmp_path, _TABS, 1) == CensusDecision("unique", 1) + + +def test_sibling_tabs_with_unreadable_labels_still_halt(tmp_path) -> None: + decision = _tab_census(tmp_path, _TABS, 1, readable=False) + assert decision.verdict == "ambiguous" + + +def test_two_tabs_with_the_same_label_halt(tmp_path) -> None: + decision = _tab_census(tmp_path, ["Who", "Contact", "Misc", "Contact"], 1) + assert decision == CensusDecision("ambiguous", 2) From ee707b80d907d0e69a9d1643bd1759c6009cd35c Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 25 Oct 2026 09:00:06 -0400 Subject: [PATCH 6/7] fix(flow): a featureless identifier crop cannot count rows On OpenEMR, #1518 halted step_016 as identical_candidates with 12 candidates. Its recorded identifier crop is a dark background band with one light rule and no label. It matches itself shifted, so template matching found it at every offset along the band, and OCR read no text to break the tie. vision.match.crop_is_featureless flags a flat crop, or one whose quarter-shift self-correlation is at least 0.9 on either axis (real identifier crops measured at most ~0.65; this one measures 1.0). The pixel census returns cannot_count for it instead of counting. A teach-time control then verifies, as on main, and a row or a target with no role still halts as identity_uncounted. Checked offline on six soak-final-2 frames of step_016: cannot_count, then not_a_row, verified. Tests: a blank band on a button before a save verifies; with no role it halts; crop_is_featureless on bands, flat, text and tab crops. The census tests that used the full-height-bar helper now use a text crop. Co-Authored-By: Claude Opus 5.5 --- .../flow/openadapt_flow/runtime/replayer.py | 8 +++- packages/flow/openadapt_flow/vision/match.py | 34 +++++++++++++++++ packages/flow/tests/test_census_teach_role.py | 37 +++++++++++++++++-- .../test_identity_evidence_capability.py | 24 ++++++++---- .../flow/tests/test_native_identity_census.py | 16 ++++++++ 5 files changed, 108 insertions(+), 11 deletions(-) diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 4f74d1977..48bc241f7 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -15775,10 +15775,12 @@ def _pixel_census( with the same or unreadable text counts it as a look-alike. The census is ``uncounted`` when the frame has too many places, when a place cannot be cut, or when the target's own place is not among them (the frame - renders at another scale than the recorded crop). + renders at another scale than the recorded crop). A featureless crop + (a blank band, a ruled line) cannot count at all: ``cannot_count``. """ from openadapt_flow.vision.match import ( AMBIGUITY_SUSPICION_SCORE, + crop_is_featureless, template_peaks, ) @@ -15790,6 +15792,10 @@ def _pixel_census( if recorded is None or live is None or anchor.identifier_region is None: return census_mod.CensusDecision("cannot_count") try: + if crop_is_featureless(recorded): + # A blank band or a ruled line matches everywhere along it: it + # cannot count rows (cannot_count, not "ambiguous"). + return census_mod.CensusDecision("cannot_count") peaks = template_peaks( before_png, recorded, threshold=AMBIGUITY_SUSPICION_SCORE ) diff --git a/packages/flow/openadapt_flow/vision/match.py b/packages/flow/openadapt_flow/vision/match.py index a61721a07..2f96557a7 100644 --- a/packages/flow/openadapt_flow/vision/match.py +++ b/packages/flow/openadapt_flow/vision/match.py @@ -309,6 +309,40 @@ def find_template( ) +# A crop that matches itself shifted by a quarter of its width or height at or +# above this correlation (a blank band, a rule, a repeating pattern) matches +# along that axis anywhere: it cannot tell one place from another. Identifier +# crops of real records measured at most ~0.65. +SELF_SIMILAR_SHIFT_SCORE = 0.9 +# Below this grey-level spread a crop is flat. +FLAT_CROP_STD = 2.0 + + +def crop_is_featureless(template_png: bytes) -> bool: + """Whether a crop has too little structure to tell one place from another. + + True for a flat crop, and for one that matches itself shifted by a quarter + of its width or height (a background band, a ruled line, a repeat), so a + template match finds it at every offset along that axis. + """ + crop = _decode_gray(template_png).astype(np.float64) + if crop.size == 0 or float(crop.std()) < FLAT_CROP_STD: + return True + height, width = crop.shape[:2] + + def shifted_score(a: np.ndarray, b: np.ndarray) -> float: + a = a - a.mean() + b = b - b.mean() + denom = math.sqrt(float((a * a).sum()) * float((b * b).sum())) + return 1.0 if denom == 0 else float((a * b).sum()) / denom + + dx = max(1, width // 4) + dy = max(1, height // 4) + horizontal = shifted_score(crop[:, dx:], crop[:, :-dx]) if width > dx else 1.0 + vertical = shifted_score(crop[dy:, :], crop[:-dy, :]) if height > dy else 1.0 + return max(horizontal, vertical) >= SELF_SIMILAR_SHIFT_SCORE + + def template_peaks( screen_png: bytes, template_png: bytes, *, threshold: float = 0.6 ) -> Optional[list[tuple[int, int]]]: diff --git a/packages/flow/tests/test_census_teach_role.py b/packages/flow/tests/test_census_teach_role.py index 6aba12a11..8039043be 100644 --- a/packages/flow/tests/test_census_teach_role.py +++ b/packages/flow/tests/test_census_teach_role.py @@ -22,10 +22,12 @@ from openadapt_flow.runtime.replayer import Replayer from openadapt_flow.vision import Match from tests.test_identity_evidence_capability import ( + _H, _W, _button, _crop_only_step, _gate, + _id_crop, ) from tests.test_replayer import ( FakeBackend, @@ -226,13 +228,42 @@ def test_the_pixel_census_decides_for_a_control_when_it_can_count( """A census that can count always decides, a control role included: the crop found once verifies, the crop found twice halts.""" - step, res = _crop_only_step(tmp_path, _button()) + step, res = _crop_only_step(tmp_path, _id_crop()) _with_role(step, role) - error, check = _gate(tmp_path, step, res, _button(), pixel_verify_enabled=True) + error, check = _gate(tmp_path, step, res, _id_crop(), pixel_verify_enabled=True) assert error is None assert check is not None and check.candidate_count == 1 - twice = np.vstack([_button(), np.full((8, _W), 255, np.uint8), _button()]) + twice = np.vstack([_id_crop(), np.full((8, _W), 255, np.uint8), _id_crop()]) error, check = _gate(tmp_path, step, res, twice, pixel_verify_enabled=True) assert error is not None assert check is not None and check.ambiguity_halt == "identical_candidates" + + +def _blank_band() -> np.ndarray: + """OpenEMR step_016's recorded crop: a dark background band with one + light rule across it, and no label.""" + img = np.full((_H, _W), 40, np.uint8) + img[_H // 2 : _H // 2 + 2, :] = 220 + return img + + +def test_a_blank_band_crop_on_a_control_before_a_save_verifies(tmp_path: Path) -> None: + """The OpenEMR case on #1518: the crop matches all along its band, so the + pixel census cannot count; the recorded button role then applies.""" + + step, res = _crop_only_step(tmp_path, _blank_band()) + _with_role(step, "button") + frame = np.hstack([_blank_band()] * 3) + error, check = _gate(tmp_path, step, res, frame, pixel_verify_enabled=True) + assert error is None + assert check is not None and check.census_status == "not_a_row" + + +def test_a_blank_band_crop_with_no_role_still_halts(tmp_path: Path) -> None: + step, res = _crop_only_step(tmp_path, _blank_band()) + _with_role(step, None) + frame = np.hstack([_blank_band()] * 3) + error, check = _gate(tmp_path, step, res, frame, pixel_verify_enabled=True) + assert error is not None + assert check is not None and check.ambiguity_halt == "identity_uncounted" diff --git a/packages/flow/tests/test_identity_evidence_capability.py b/packages/flow/tests/test_identity_evidence_capability.py index 2beddbf2d..35c9c927c 100644 --- a/packages/flow/tests/test_identity_evidence_capability.py +++ b/packages/flow/tests/test_identity_evidence_capability.py @@ -40,6 +40,16 @@ def _button(val: int = 255) -> np.ndarray: return img +def _id_crop() -> np.ndarray: + """An identifier crop with real structure (text), unlike ``_button``'s + full-height bar, which matches itself shifted and so is featureless.""" + img = np.full((_H, _W), 255, np.uint8) + cv2.putText( + img, "MG4408 Okafor", (6, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.7, 0, 2, cv2.LINE_AA + ) + return img + + def _with_extra_bar(img: np.ndarray) -> np.ndarray: other = img.copy() other[:, 140:145] = 0 @@ -116,9 +126,9 @@ def test_no_surrounding_text_valid_crop_verifies_where_crops_are_verifiable( assert error is None assert check is not None assert check.status == "verified" and check.mode == "pixel" - # No row text to count look-alikes with: the crop itself is counted. It - # verifies in one place only. - assert check.candidate_count == 1 + # A full-height bar matches itself shifted, so it cannot count rows; with + # nothing consequential after the step it runs and records that. + assert check.census_status == "cannot_count" def test_a_crop_found_once_on_the_frame_runs_before_a_write( @@ -126,8 +136,8 @@ def test_a_crop_found_once_on_the_frame_runs_before_a_write( ) -> None: """No OCR band: the census counts where the crop verifies. Once: it runs.""" - step, res = _crop_only_step(tmp_path, _button()) - error, check = _gate(tmp_path, step, res, _button(), pixel_verify_enabled=True) + step, res = _crop_only_step(tmp_path, _id_crop()) + error, check = _gate(tmp_path, step, res, _id_crop(), pixel_verify_enabled=True) assert error is None assert check is not None and check.status == "verified" assert check.candidate_count == 1 @@ -136,8 +146,8 @@ def test_a_crop_found_once_on_the_frame_runs_before_a_write( def test_a_crop_found_twice_on_the_frame_halts(tmp_path: Path) -> None: """Another row shows the same identifier pixels: the step stops.""" - step, res = _crop_only_step(tmp_path, _button()) - frame = np.vstack([_button(), np.full((8, _W), 255, np.uint8), _button()]) + step, res = _crop_only_step(tmp_path, _id_crop()) + frame = np.vstack([_id_crop(), np.full((8, _W), 255, np.uint8), _id_crop()]) error, check = _gate(tmp_path, step, res, frame, pixel_verify_enabled=True) assert error is not None and "2 shown rows" in error assert check is not None diff --git a/packages/flow/tests/test_native_identity_census.py b/packages/flow/tests/test_native_identity_census.py index 0869a578a..e3ae7facf 100644 --- a/packages/flow/tests/test_native_identity_census.py +++ b/packages/flow/tests/test_native_identity_census.py @@ -906,3 +906,19 @@ def test_sibling_tabs_with_unreadable_labels_still_halt(tmp_path) -> None: def test_two_tabs_with_the_same_label_halt(tmp_path) -> None: decision = _tab_census(tmp_path, ["Who", "Contact", "Misc", "Contact"], 1) assert decision == CensusDecision("ambiguous", 2) + + +def test_featureless_crops_are_recognized() -> None: + import cv2 + import numpy as np + + from openadapt_flow.vision.match import crop_is_featureless + + band = np.full((64, 160), 40, np.uint8) + band[32:34, :] = 220 # OpenEMR step_016: a band with one rule + assert crop_is_featureless(_png(band)) + assert crop_is_featureless(_png(np.full((48, 240), 128, np.uint8))) + text = np.full((48, 240), 255, np.uint8) + cv2.putText(text, "MG4408 Okafor", (6, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.7, 0, 2) + assert not crop_is_featureless(_png(text)) + assert not crop_is_featureless(_png(_tab("Contact"))) From 68e79e73f8ef2b875d6473cb8ef7f21a130f1138 Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 25 Oct 2026 09:20:34 -0400 Subject: [PATCH 7/7] fix(flow): an OCR misread never tells identical rows apart Review of #1518 found two holes. - The pixel census's OCR tie-break treated any text difference as a different label. With real RapidOCR on two rendered "Smith, John" rows, 18 of 34 cases read unique ("Smith, 10 lohn" at JPEG q75, "Smith, S John" with a 1 px offset). Now a crop with any OCR line below 0.8 confidence reads as unreadable, and two texts differ only when neither contains the other and their similarity ratio is below 0.6 (SequenceMatcher, autojunk off). - crop_is_featureless scored 1.0 when only one shifted window was flat, so a short label at the edge of a wide cell counted as featureless. It scores 1.0 only when both windows are flat, and 0.0 when exactly one is. The fix is the reviewer's (rr-1518). With it, the rendered Smith grid is 0 of 288 unique, the OpenEMR "Contact" tab is still unique, and two Contact tabs halt. Six real soak-final-2 frames of step_016 still verify. Regression tests: a low-confidence misread and a confident near-identical reading both leave identical rows ambiguous, and an edge-label crop is not featureless. Co-Authored-By: Claude Opus 5.5 --- .../flow/openadapt_flow/runtime/replayer.py | 29 ++++++- packages/flow/openadapt_flow/vision/match.py | 11 ++- .../flow/tests/test_native_identity_census.py | 79 ++++++++++++++++++- 3 files changed, 115 insertions(+), 4 deletions(-) diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 48bc241f7..5aae38dde 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -736,6 +736,16 @@ class _WebFormApprovalView: ref: Any +# The pixel census's OCR tie-break: a place the pixels do not rule out is +# distinguishable only when OCR reads both crops with every line at or above +# this confidence, neither text contains the other, and the two texts are less +# similar than this ratio. One label read two ways measured 0.82-0.95 +# ("Smith, John" / "Smith, S John" / "Smith, 10 lohn"); sibling tab labels +# ("Contact" / "Stats") measure 0.5 or less. +PIXEL_CENSUS_OCR_MIN_CONFIDENCE = 0.8 +PIXEL_CENSUS_OCR_MAX_SIMILARITY = 0.6 + + class Replayer: """Replays a Workflow against a Backend using injected vision. @@ -15823,10 +15833,27 @@ def crop_text(png: bytes) -> str: lines = self.vision.ocr(png, region=None) except Exception: return "" + # A low-confidence fragment ("S", "10", "lohn") is a misread, not + # evidence: the whole crop then counts as unreadable. + if any( + float(getattr(line, "confidence", 0.0) or 0.0) + < PIXEL_CENSUS_OCR_MIN_CONFIDENCE + for line in lines + ): + return "" return identity_mod.squash( " ".join(str(getattr(line, "text", "") or "") for line in lines) ) + def texts_differ(a: str, b: str) -> bool: + """Clearly different labels, not one label read two ways.""" + import difflib + + if a in b or b in a: + return False # a dropped or clipped word, not another label + ratio = difflib.SequenceMatcher(None, a, b, autojunk=False).ratio() + return ratio < PIXEL_CENSUS_OCR_MAX_SIMILARITY + target_text: Optional[str] = None count = 1 for x, y in peaks: @@ -15847,7 +15874,7 @@ def crop_text(png: bytes) -> str: if target_text is None: target_text = crop_text(live) other_text = crop_text(crop) - if target_text and other_text and other_text != target_text: + if target_text and other_text and texts_differ(target_text, other_text): continue count += 1 if count > 1: diff --git a/packages/flow/openadapt_flow/vision/match.py b/packages/flow/openadapt_flow/vision/match.py index 2f96557a7..933d12c7d 100644 --- a/packages/flow/openadapt_flow/vision/match.py +++ b/packages/flow/openadapt_flow/vision/match.py @@ -333,8 +333,15 @@ def crop_is_featureless(template_png: bytes) -> bool: def shifted_score(a: np.ndarray, b: np.ndarray) -> float: a = a - a.mean() b = b - b.mean() - denom = math.sqrt(float((a * a).sum()) * float((b * b).sum())) - return 1.0 if denom == 0 else float((a * b).sum()) / denom + energy_a = float((a * a).sum()) + energy_b = float((b * b).sum()) + if energy_a == 0 and energy_b == 0: + return 1.0 # both windows flat: nothing tells places apart + if energy_a == 0 or energy_b == 0: + # Content confined to one edge quarter (a short label in a wide + # cell): the shifted windows are not alike. + return 0.0 + return float((a * b).sum()) / math.sqrt(energy_a * energy_b) dx = max(1, width // 4) dy = max(1, height // 4) diff --git a/packages/flow/tests/test_native_identity_census.py b/packages/flow/tests/test_native_identity_census.py index e3ae7facf..7f014b862 100644 --- a/packages/flow/tests/test_native_identity_census.py +++ b/packages/flow/tests/test_native_identity_census.py @@ -855,7 +855,7 @@ def ocr(self, png: bytes, *, region=None): _TABS + ["Contact"], key=lambda name: float(np.abs(crop.astype(int) - _tab(name)).mean()), ) - return [types.SimpleNamespace(text=label)] + return [types.SimpleNamespace(text=label, confidence=1.0)] def _tab_census(tmp_path, labels: list[str], target: int, *, readable: bool = True): @@ -922,3 +922,80 @@ def test_featureless_crops_are_recognized() -> None: cv2.putText(text, "MG4408 Okafor", (6, 32), cv2.FONT_HERSHEY_SIMPLEX, 0.7, 0, 2) assert not crop_is_featureless(_png(text)) assert not crop_is_featureless(_png(_tab("Contact"))) + + +class _ScriptedOcr: + """OCR that reads the target crop first, then every other place, as given.""" + + def __init__(self, target: tuple[str, float], others: tuple[str, float]): + self._target = target + self._others = others + self._calls = 0 + + def ocr(self, png: bytes, *, region=None): + self._calls += 1 + text, confidence = self._target if self._calls == 1 else self._others + return [types.SimpleNamespace(text=text, confidence=confidence)] + + +def _identical_tabs_census(tmp_path, ocr) -> CensusDecision: + """Two identical "Contact" tabs, read through ``ocr``.""" + + labels = ["Contact", "Contact"] + (tmp_path / "templates").mkdir(parents=True, exist_ok=True) + recorded = _png(_tab("Contact")) + (tmp_path / "templates" / "id.png").write_bytes(recorded) + (tmp_path / "templates" / "t.png").write_bytes(recorded) + step = Step( + id="open_contact", + intent="click 'Contact'", + action=ActionKind.CLICK, + identity_armed=True, + anchor=Anchor( + template="templates/t.png", + region=(0, 0, _TAB_W, _TAB_H), + click_point=(0, 0), + identifier_crop="templates/id.png", + identifier_region=(0, 0, _TAB_W, _TAB_H), + ), + ) + replayer = Replayer( + _PixelBackend(), vision=ocr, poll_interval_s=0.01, pixel_verify_enabled=True + ) + resolution = Resolution(rung="template", point=(0, 0), confidence=1.0, elapsed_ms=1) + return replayer._pixel_census( + step, resolution, _png(_tabs(labels)), Workflow(name="wf"), tmp_path + ) + + +def test_a_low_confidence_misread_never_tells_identical_rows_apart(tmp_path) -> None: + """JPEG noise read "Smith, John" as "Smith, 10 lohn" with low confidence: + that reading is unreadable, not a different label.""" + + ocr = _ScriptedOcr(("Smith, John", 0.97), ("Smith, 10 lohn", 0.55)) + assert _identical_tabs_census(tmp_path, ocr) == CensusDecision("ambiguous", 2) + + +def test_one_label_read_two_ways_is_still_the_same_label(tmp_path) -> None: + """A confident misread close to the target text ("Smith, S John") or one + that contains it is the same label, not a sibling.""" + + for other in ("Smith, S John", "Smith, John Jr", "John"): + ocr = _ScriptedOcr(("Smith, John", 0.97), (other, 0.95)) + assert _identical_tabs_census(tmp_path, ocr).verdict == "ambiguous", other + + +def test_an_edge_label_in_a_wide_crop_is_not_featureless() -> None: + """A short label at one edge of a wide cell ("Li, Al" in 200 px) leaves one + shifted window flat; that is structure, not a blank band.""" + + import cv2 + import numpy as np + + from openadapt_flow.vision.match import crop_is_featureless + + cell = np.full((32, 200), 255, np.uint8) + cv2.putText( + cell, "Li, Al", (2, 22), cv2.FONT_HERSHEY_SIMPLEX, 0.5, 0, 1, cv2.LINE_AA + ) + assert not crop_is_featureless(_png(cell))