diff --git a/DESIGN.md b/DESIGN.md index 5cc53b1..77e4ee0 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -5,10 +5,16 @@ deterministic, vision-anchored script → replay it locally with a resolution ladder → when the UI drifts, heal the script (as a reviewable diff) instead of re-reasoning every run. -The runtime is **vision-only**: it consumes PNG bytes and emits clicks/keys at -pixel coordinates through the `Backend` protocol (`openadapt_flow/backend.py`). -The reference backend is Playwright-driven (headless-capable, CI-friendly, -permission-free); native OS / RDP backends are future adapters. +The runtime is **vision-first, not vision-only**: it always CAN operate a pure +surface — consuming PNG bytes and emitting clicks/keys at pixel coordinates +through the `Backend` protocol (`openadapt_flow/backend.py`) — but where a +backend owns a structured layer (a browser DOM, a native UIA/AX tree) the +resolution ladder's TOP rung re-finds the recorded target as an ELEMENT and +acts on it deterministically (`StructuralActionBackend`, see Resolution ladder). +Structure is preferred where present; the visual ladder is the fallback floor +for pixel-only substrates (RDP/Citrix/canvas). The reference backend is +Playwright-driven (headless-capable, CI-friendly, permission-free); native OS / +RDP backends are future adapters. ## Frozen contracts (do not change without updating this doc) @@ -149,7 +155,27 @@ class Replayer: Parameters not supplied in `params` fall back to the recorded example/default values in `workflow.params`, so a bundle replays without any explicit params. -Resolution ladder per step with an anchor (record rung + confidence + ms): +Resolution ladder per step with an anchor (record rung + confidence + ms). +The full capability hierarchy is API → tool/MCP → DOM/UIA → geometry → OCR → +template → VLM → human; API and tool/MCP are future placeholders, so the +implemented rungs are: + +0. `structural` — DETERMINISTIC (additive change, structural-rung spike). When + the live backend implements `StructuralActionBackend` (a browser DOM, a + native UIA/AX tree) AND the anchor carries a `StructuralLocator` the compiler + captured (a stable DOM selector / role+name, or a UIA AutomationId / + role+name), the runtime re-finds the recorded target as an ELEMENT via + `backend.locate_structural(locator)` and acts on its center — no pixel + matching. Tried FIRST; a successful locate short-circuits the visual rungs. + Pixel-only substrates (RDP/Citrix/canvas) and failed/ambiguous locates fall + through to the visual rungs below UNCHANGED. This is the thesis refinement + from "vision-only" to "deterministic compiled automation with visual + FALLBACK" (desktop benchmark: UIA 21/21 vs compiled visual replay 6/21 under + drift; reproduced by `benchmark/structural_action/`). The resolved point + flows through the SAME click path, so the pre-click identity gate and the + irreversible risk gate still fire on it — structure makes identity STRONGER + (an exact element), it never bypasses it. Rungs 1–5 remain the visual + FALLBACK floor and are unchanged: 1. `template` — find_template within anchor.region padded by search_pad 2. `template_global` — find_template full frame; for UNLABELED anchors (no ocr_text) the match is rejected when every locatable landmark places diff --git a/benchmark/structural_action/STRUCTURAL_ACTION.md b/benchmark/structural_action/STRUCTURAL_ACTION.md new file mode 100644 index 0000000..3429f96 --- /dev/null +++ b/benchmark/structural_action/STRUCTURAL_ACTION.md @@ -0,0 +1,52 @@ +# Structural action rung — availability under render drift + +Reproduces, on a real rendered DOM with the real resolution ladder, the desktop +benchmark that reframed the thesis from **"vision-only"** to **"deterministic +compiled automation with visual FALLBACK"**: structural (DOM/UIA) execution +scored **21/21** while compiled *visual* replay scored **6/21** under render +drift. + +## What it measures + +`structural_action_probe.py` (logic in +`openadapt_flow/validation/structural_action.py`) renders a dense surface of +`n` actionable targets, each with a stable DOM id (`#open-pK`) — what a browser +recorder captures for the structural rung — plus a recorded visual anchor +(template crop + OCR label). It then RE-RENDERS with drift: a deterministic +subset is left untouched (steps whose surface did not change), the rest are +MOVED, RELABELLED (`Open`→`View`) and RE-THEMED (dark, serif). The DOM id never +changes. For each target it asks, on the drifted surface: + +- **structural** — `backend.locate_structural(locator)`: does it land inside the + CORRECT element's box? (Deterministic, pixel-independent.) +- **visual** — `resolve(anchor, drift_png, vision, structural=None)`: does the + template/OCR/geometry ladder resolve to a point inside the CORRECT box? A + drifted target's recorded crop no longer matches at its old position and + either fails or matches a look-alike sibling — both are failures. + +Numbers are MEASURED, never hardcoded. + +## Result (`structural_action.json`, n=21) + +``` +structural (DOM) rung : 21/21 acted correctly +visual ladder only : 6/21 acted correctly +under drift (15 targets): structural 15/15 vs visual 0/15 +``` + +Structural resolves every target whose id is still in the DOM; the visual ladder +resolves only the non-drifted minority. The structural point flows through the +SAME click path, so the pre-click identity gate and the irreversible risk gate +still fire on it — structure makes identity STRONGER (an exact element), it +never bypasses it. + +## Run it + +``` +python benchmark/structural_action/structural_action_probe.py 21 +``` + +The in-suite assertion of the same win (structural resolves all, beats visual +under drift, identity gate intact) lives in `tests/test_structural_rung.py`; the +full record→compile→replay default-on path is in +`tests/e2e/test_structural_action.py`. diff --git a/benchmark/structural_action/structural_action.json b/benchmark/structural_action/structural_action.json new file mode 100644 index 0000000..26a073e --- /dev/null +++ b/benchmark/structural_action/structural_action.json @@ -0,0 +1,177 @@ +{ + "n": 21, + "structural_ok": 21, + "visual_ok": 6, + "structural_ratio": "21/21", + "visual_ratio": "6/21", + "targets": [ + { + "id": "open-p0", + "drifted": false, + "structural_ok": true, + "structural_rung": true, + "visual_ok": true, + "visual_rung": "template" + }, + { + "id": "open-p1", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p2", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p3", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p4", + "drifted": false, + "structural_ok": true, + "structural_rung": true, + "visual_ok": true, + "visual_rung": "template" + }, + { + "id": "open-p5", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p6", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p7", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p8", + "drifted": false, + "structural_ok": true, + "structural_rung": true, + "visual_ok": true, + "visual_rung": "template" + }, + { + "id": "open-p9", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p10", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p11", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p12", + "drifted": false, + "structural_ok": true, + "structural_rung": true, + "visual_ok": true, + "visual_rung": "template" + }, + { + "id": "open-p13", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p14", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p15", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p16", + "drifted": false, + "structural_ok": true, + "structural_rung": true, + "visual_ok": true, + "visual_rung": "template" + }, + { + "id": "open-p17", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p18", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p19", + "drifted": true, + "structural_ok": true, + "structural_rung": true, + "visual_ok": false, + "visual_rung": "template" + }, + { + "id": "open-p20", + "drifted": false, + "structural_ok": true, + "structural_rung": true, + "visual_ok": true, + "visual_rung": "template" + } + ] +} \ No newline at end of file diff --git a/benchmark/structural_action/structural_action_probe.py b/benchmark/structural_action/structural_action_probe.py new file mode 100644 index 0000000..e3bbb40 --- /dev/null +++ b/benchmark/structural_action/structural_action_probe.py @@ -0,0 +1,39 @@ +"""CLI: structural (DOM) ACTION rung vs the visual ladder under render drift. + +Reproduces the desktop-benchmark shape that reframed the thesis from +"vision-only" to "deterministic compiled automation with visual FALLBACK" +(structural 21/21 vs compiled visual replay 6/21). The reusable, tested logic +lives in :mod:`openadapt_flow.validation.structural_action`; this script is a +thin runner that prints the ratios and writes ``structural_action.json``. + +Usage: python benchmark/structural_action/structural_action_probe.py [n] [--show] +""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +from openadapt_flow.validation.structural_action import run_probe + + +def main(argv: list[str]) -> int: + n = next((int(a) for a in argv if a.isdigit()), 21) + report = run_probe(n=n, headless="--show" not in argv) + out = Path(__file__).with_name("structural_action.json") + out.write_text(json.dumps(report, indent=2)) + print(f"structural (DOM) rung : {report['structural_ratio']} acted correctly") + print(f"visual ladder only : {report['visual_ratio']} acted correctly") + drifted = [t for t in report["targets"] if t["drifted"]] + d_struct = sum(t["structural_ok"] for t in drifted) + d_vis = sum(t["visual_ok"] for t in drifted) + print( + f"under drift ({len(drifted)} targets): structural {d_struct}/{len(drifted)}" + f" vs visual {d_vis}/{len(drifted)}" + ) + print(f"wrote {out}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index bfd11c5..5ad09be 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -179,6 +179,12 @@ def _cmd_replay(args: argparse.Namespace) -> int: state_verifier=( appliance.state_verifier if appliance else None ), + # Normal replay prefers the deterministic structural rung. + # ``--drift`` exists to DEMONSTRATE the visual healing ladder + # on the bundled MockMed app, so it forces the visual floor + # (structure would resolve the injected drift and there would + # be nothing to heal -- the very thing the flag shows). + use_structural=not bool(args.drift), ).run( workflow, params=params, diff --git a/openadapt_flow/backend.py b/openadapt_flow/backend.py index 16dffec..3c81d66 100644 --- a/openadapt_flow/backend.py +++ b/openadapt_flow/backend.py @@ -8,7 +8,10 @@ from __future__ import annotations -from typing import Optional, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Optional, Protocol, runtime_checkable + +if TYPE_CHECKING: # pragma: no cover + from openadapt_flow.ir import StructuralHandle, StructuralLocator @runtime_checkable @@ -95,6 +98,71 @@ def structured_text_at(self, x: int, y: int) -> Optional[str]: ... +@runtime_checkable +class StructuralActionBackend(Protocol): + """Optional STRUCTURAL action capability a backend MAY expose. + + The runtime resolves targets by a LADDER (see + :mod:`openadapt_flow.runtime.resolver`). Its top rung is structural: where + the backend owns a structured layer (a browser's DOM, a native app's UIA/AX + tree) the runtime re-finds the recorded target as an ELEMENT and acts on its + center DETERMINISTICALLY, instead of pixel-matching a template that render + drift (relabel, theme, zoom, layout shift) can defeat. The desktop benchmark + measured UIA execution 21/21 vs compiled visual replay 6/21 under drift. + + This is the thesis shift from "vision-only" to "deterministic compiled + automation with visual FALLBACK". It is ADDITIVE and backend-optional: a + pixel-only substrate (RDP/Citrix/canvas, or a backend with no structured + layer) simply does not implement it, and resolution falls through to the + visual rungs (template/ocr/geometry) UNCHANGED -- the healthcare/Citrix + floor is never removed. + + Crucially, the structurally-resolved point flows through the IDENTICAL click + path as any visual resolution, so the pre-click identity gate and the + irreversible risk gate still fire on it: structure makes identity STRONGER + (an exact element), it never bypasses it. + + Two methods, mirroring the record/replay split of the identity capability + (:class:`IdentityBackend`): + + - ``structural_locator_at`` runs at RECORD time: given the demonstrated + click point, return a STABLE structural locator the runtime can re-resolve + later (a DOM ``#id`` / role+name, a UIA ``AutomationId`` / role+name). + - ``locate_structural`` runs at REPLAY time: given that recorded locator, + find the element on the LIVE surface and return its center point. + + Each returns None when the capability is momentarily unavailable, the + element is absent/ambiguous, or the substrate has no structured layer at + that point (never raises) -- resolution then uses the visual ladder. + """ + + def structural_locator_at( + self, x: int, y: int + ) -> Optional["StructuralLocator"]: + """Return a stable structural locator for the element at pixel (x, y). + + The coordinate space matches :meth:`Backend.click`. Returns None when + the backend cannot derive a stable locator (no structured node under the + point, or nothing that identifies the element durably) -- the step then + relies on the visual anchor alone. + """ + ... + + def locate_structural( + self, locator: "StructuralLocator" + ) -> Optional["StructuralHandle"]: + """Locate ``locator``'s element on the live surface; return its point. + + Returns a :class:`~openadapt_flow.ir.StructuralHandle` whose ``point`` + is the element's center (same coordinate space as :meth:`Backend.click`) + on a unique, actionable match, or None when the element is absent, not + uniquely resolvable, off-screen, or the substrate exposes no structured + layer (never raises) -- the resolver then falls through to the visual + rungs. + """ + ... + + @runtime_checkable class Backend(Protocol): @property diff --git a/openadapt_flow/backends/playwright_backend.py b/openadapt_flow/backends/playwright_backend.py index 801b7d8..880f57f 100644 --- a/openadapt_flow/backends/playwright_backend.py +++ b/openadapt_flow/backends/playwright_backend.py @@ -13,6 +13,8 @@ if TYPE_CHECKING: # pragma: no cover from playwright.sync_api import Page +from openadapt_flow.ir import StructuralHandle, StructuralLocator + VIEWPORT: tuple[int, int] = (1280, 800) _MODIFIER_ALIASES = { @@ -203,6 +205,124 @@ def structured_text_at(self, x: int, y: int) -> Optional[str]: return None return result or None + # -- structural action (openadapt_flow.backend.StructuralActionBackend) -- + + def structural_locator_at( + self, x: int, y: int + ) -> Optional[StructuralLocator]: + """Return a stable DOM locator for the element under (x, y). + + Walks from ``document.elementFromPoint`` to the nearest ACTIONABLE + element (the control a user clicks) and derives a stable identity for + it: a unique ``#id`` selector when available, else the element's ARIA + ``role`` + accessible ``name``. Returns None when neither a unique id + nor a role+name can be formed (the step then relies on the visual + anchor). Coordinate space matches :meth:`click`. + """ + try: + result = self.page.evaluate( + """([px, py]) => { + const el = document.elementFromPoint(px, py); + if (!el) return null; + const actionable = el.closest( + 'button, a[href], input, select, textarea,' + + ' [role="button"], [role="link"], [role="menuitem"],' + + ' [role="tab"], [role="option"], [onclick], [data-id]' + ) || el; + const tag = actionable.tagName.toLowerCase(); + let selector = null; + const id = actionable.id; + if (id && document.querySelectorAll( + '#' + CSS.escape(id)).length === 1) { + selector = '#' + CSS.escape(id); + } + let role = actionable.getAttribute('role'); + if (!role) { + const map = {button: 'button', a: 'link', + input: 'textbox', select: 'combobox', + textarea: 'textbox'}; + role = map[tag] || null; + if (tag === 'a' && + !actionable.getAttribute('href')) role = null; + } + let name = actionable.getAttribute('aria-label'); + if (!name) { + const t = (actionable.textContent || '') + .replace(/\\s+/g, ' ').trim(); + name = t ? t.slice(0, 120) : null; + } + if (!selector && !(role && name)) return null; + return {selector: selector, role: role, name: name}; + }""", + [int(x), int(y)], + ) + except Exception: + return None + if not result: + return None + return StructuralLocator( + selector=result.get("selector"), + role=result.get("role"), + name=result.get("name"), + ) + + def locate_structural( + self, locator: StructuralLocator + ) -> Optional[StructuralHandle]: + """Locate ``locator``'s element in the live DOM; return its center. + + Resolves by the recorded ``selector`` first, else by ``role`` + + ``name``. Requires a UNIQUE, on-screen, UNOCCLUDED match: a missing, + ambiguous, off-viewport, or COVERED element (a modal / click-shield over + it -- the hit test at its center returns another node) returns None so + the resolver falls through to the visual ladder (and, for off-screen, + its scroll-and-retry; for an opaque cover, a safe halt). The returned + point is the element's center in :meth:`click` coordinate space, so the + pre-click identity gate re-reads the SAME element there. + """ + try: + loc = None + if locator.selector: + loc = self.page.locator(locator.selector) + elif locator.role and locator.name: + loc = self.page.get_by_role( + locator.role, name=locator.name, exact=True + ) + if loc is None: + return None + if loc.count() != 1: + return None + box = loc.bounding_box() + if not box or box["width"] <= 0 or box["height"] <= 0: + return None + cx = int(round(box["x"] + box["width"] / 2)) + cy = int(round(box["y"] + box["height"] / 2)) + vw, vh = self.viewport + if not (0 <= cx < vw and 0 <= cy < vh): + return None + # OCCLUSION / actionability gate: a stable DOM identity is NOT a + # licence to click something the user could not. If the element is + # covered (a modal, an opaque or transparent click-shield), the hit + # test at its center returns some OTHER node, and we return None so + # resolution falls through to the visual ladder (which also fails on + # an opaque cover -> safe halt, no click). Only when the element -- + # or a descendant of it -- is the topmost node at the action point + # do we resolve. This preserves the occlusion-safe-halt invariant + # (tests/e2e/test_chaos.py) that structural coordinates would + # otherwise bypass. + topmost = loc.evaluate( + "(el, pt) => {" + " const n = document.elementFromPoint(pt[0], pt[1]);" + " return !!n && (n === el || el.contains(n));" + "}", + [cx, cy], + ) + if not topmost: + return None + return StructuralHandle(point=(cx, cy)) + except Exception: + return None + def screenshot(self) -> bytes: """Return the current full-viewport frame as PNG bytes.""" return self.page.screenshot(type="png", full_page=False) diff --git a/openadapt_flow/backends/windows_backend.py b/openadapt_flow/backends/windows_backend.py index 104669d..bf97ae6 100644 --- a/openadapt_flow/backends/windows_backend.py +++ b/openadapt_flow/backends/windows_backend.py @@ -41,6 +41,8 @@ import requests +from openadapt_flow.ir import StructuralHandle, StructuralLocator + DEFAULT_SERVER_URL = "http://localhost:5001" # Screenshot retry defaults (mirrors WAADirect in openadapt-evals). @@ -355,6 +357,184 @@ def _execute_read(self, command: str) -> Optional[str]: return val return None + # -- structural action (openadapt_flow.backend.StructuralActionBackend) -- + + def _read_structured_json(self, snippet: str) -> object: + """Run ``snippet`` on the VM and decode its sentinel-wrapped JSON. + + The snippet must ``print`` its result as + ``<>` + json.dumps(value) + `<>``. + Returns the decoded value (which may itself be None), or None when the + server echoes nothing, the markers are absent, or decoding fails -- + callers then fall back to the visual ladder. + """ + body = self._execute_read(snippet) + if not body: + return None + import json as _json + + start = body.find("<>") + end = body.find("<>") + if start == -1 or end == -1 or end <= start: + return None + payload = body[start + len("<>") : end] + try: + return _json.loads(payload) + except Exception: + return None + + def structural_locator_at( + self, x: int, y: int + ) -> Optional[StructuralLocator]: + """Return a stable UIA locator for the control under (x, y). + + Runs a ``uiautomation.ControlFromPoint`` snippet on the VM: it climbs to + the nearest ACTIONABLE control (button / hyperlink / menu-item / ...) and + reads its ``AutomationId``, ``ControlType`` (mapped to an ARIA-style + role) and ``Name``. Returns None when there is no actionable control, + neither an AutomationId nor a usable role+name exists, UIA is + unavailable, or the WAA server does not echo output (never raises) -- + the step then relies on the visual anchor. + """ + snippet = ( + "import json\n" + "def _oaflow_locator_at(px, py):\n" + " try:\n" + " import uiautomation as auto\n" + " except Exception:\n" + " return None\n" + " try:\n" + " el = auto.ControlFromPoint(px, py)\n" + " except Exception:\n" + " return None\n" + " if el is None:\n" + " return None\n" + " actionable = None\n" + " node = el\n" + " actionable_types = ('ButtonControl', 'HyperlinkControl',\n" + " 'MenuItemControl', 'TabItemControl', 'ListItemControl',\n" + " 'CheckBoxControl', 'RadioButtonControl',\n" + " 'SplitButtonControl', 'EditControl')\n" + " for _ in range(6):\n" + " try:\n" + " ct = node.ControlTypeName\n" + " except Exception:\n" + " ct = ''\n" + " if ct in actionable_types:\n" + " actionable = node\n" + " break\n" + " p = getattr(node, 'GetParentControl', None)\n" + " nxt = p() if p else None\n" + " if nxt is None:\n" + " break\n" + " node = nxt\n" + " if actionable is None:\n" + " actionable = el\n" + " try:\n" + " ct = actionable.ControlTypeName\n" + " except Exception:\n" + " ct = ''\n" + " role_map = {'ButtonControl': 'button',\n" + " 'HyperlinkControl': 'link', 'MenuItemControl': 'menuitem',\n" + " 'TabItemControl': 'tab', 'ListItemControl': 'listitem',\n" + " 'CheckBoxControl': 'checkbox',\n" + " 'RadioButtonControl': 'radio', 'EditControl': 'textbox',\n" + " 'SplitButtonControl': 'button'}\n" + " role = role_map.get(ct)\n" + " aid = str(getattr(actionable, 'AutomationId', '') or '')\n" + " name = ' '.join(str(getattr(actionable, 'Name', '') or ''" + ").split())\n" + " aid = aid or None\n" + " name = name or None\n" + " if not aid and not (role and name):\n" + " return None\n" + " return {'automation_id': aid, 'role': role, 'name': name}\n" + "print('<>' + json.dumps(" + f"_oaflow_locator_at({int(x)}, {int(y)})) " + "+ '<>')\n" + ) + value = self._read_structured_json(snippet) + if not isinstance(value, dict): + return None + return StructuralLocator( + automation_id=value.get("automation_id"), + role=value.get("role"), + name=value.get("name"), + ) + + def locate_structural( + self, locator: StructuralLocator + ) -> Optional[StructuralHandle]: + """Locate ``locator``'s control via UIA; return its center point. + + Searches the UIA tree from the root by ``AutomationId`` first, else by + the mapped ``ControlType`` + ``Name``, and returns the matched control's + ``BoundingRectangle`` center in :meth:`click` coordinate space. Returns + None on no/ambiguous match, an empty rectangle, unavailable UIA, or no + echoed output (never raises) -- the resolver then uses the visual ladder. + """ + aid = locator.automation_id or "" + role = locator.role or "" + name = locator.name or "" + if not aid and not (role and name): + return None + snippet = ( + "import json\n" + "def _oaflow_locate(aid, role, name):\n" + " try:\n" + " import uiautomation as auto\n" + " except Exception:\n" + " return None\n" + " try:\n" + " root = auto.GetRootControl()\n" + " except Exception:\n" + " return None\n" + " ctrl_map = {'button': 'ButtonControl',\n" + " 'link': 'HyperlinkControl', 'menuitem': 'MenuItemControl',\n" + " 'tab': 'TabItemControl', 'listitem': 'ListItemControl',\n" + " 'checkbox': 'CheckBoxControl',\n" + " 'radio': 'RadioButtonControl', 'textbox': 'EditControl'}\n" + " el = None\n" + " if aid:\n" + " try:\n" + " cand = auto.Control(searchFromControl=root,\n" + " AutomationId=aid)\n" + " if cand.Exists(0, 0):\n" + " el = cand\n" + " except Exception:\n" + " el = None\n" + " if el is None and role and name:\n" + " cls = getattr(auto, ctrl_map.get(role, ''), None)\n" + " if cls is not None:\n" + " try:\n" + " cand = cls(searchFromControl=root, Name=name)\n" + " if cand.Exists(0, 0):\n" + " el = cand\n" + " except Exception:\n" + " el = None\n" + " if el is None:\n" + " return None\n" + " try:\n" + " r = el.BoundingRectangle\n" + " except Exception:\n" + " return None\n" + " if r is None or r.right <= r.left or r.bottom <= r.top:\n" + " return None\n" + " cx = int((r.left + r.right) / 2)\n" + " cy = int((r.top + r.bottom) / 2)\n" + " return [cx, cy]\n" + "print('<>' + json.dumps(_oaflow_locate(" + f"{aid!r}, {role!r}, {name!r})) + '<>')\n" + ) + value = self._read_structured_json(snippet) + if ( + not isinstance(value, list) + or len(value) != 2 + or not all(isinstance(v, (int, float)) for v in value) + ): + return None + return StructuralHandle(point=(int(value[0]), int(value[1]))) + def click(self, x: int, y: int, *, double: bool = False) -> None: """Click (or double-click) at pixel coordinates via pyautogui.""" fn = "doubleClick" if double else "click" diff --git a/openadapt_flow/compiler/compile.py b/openadapt_flow/compiler/compile.py index ae6ec15..b920f21 100644 --- a/openadapt_flow/compiler/compile.py +++ b/openadapt_flow/compiler/compile.py @@ -28,6 +28,7 @@ PostconditionKind, Region, Step, + StructuralLocator, Workflow, ) from openadapt_flow.risk import classify_step_risk @@ -874,6 +875,18 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: # OCR context band; replay prefers it (no OCR glyph ambiguity) and # falls back to the band on pixel-only substrates. structured_identity = event.get("structured_identity") or None + # Structural locator (DOM selector / role+name, or UIA identifiers) + # of the clicked element, captured by the recorder when the + # recording backend exposed it (StructuralActionBackend). Drives the + # structural ACTION rung at replay -- the SAME element is re-found + # deterministically, surviving the render drift the visual template + # cannot. The visual anchor above is kept as the fallback floor. + structural_raw = event.get("structural") or None + structural = ( + StructuralLocator.model_validate(structural_raw) + if structural_raw + else None + ) anchor = Anchor( template=template_rel, region=crop_region, @@ -881,6 +894,7 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: ocr_text=ocr_text, context_text=context_text, structured_identity=structured_identity, + structural=structural, landmarks=landmarks, ) # Identity-protection audit trail: an UNARMED click proceeds diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index 520976a..cd97430 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -67,14 +67,77 @@ class Landmark(BaseModel): ) +class StructuralLocator(BaseModel): + """A stable structural (DOM / accessibility) locator for a step's target. + + Captured at record time from the recording backend's structured layer + (:meth:`openadapt_flow.backend.StructuralActionBackend.structural_locator_at`) + and consumed at replay by the structural ACTION rung -- the TOP of the + resolution ladder (:mod:`openadapt_flow.runtime.resolver`). The runtime + re-finds the SAME element by its stable identity -- a DOM id / CSS selector + / ARIA role+name, or a Windows UIA ``AutomationId`` / ``ControlType``+ + ``Name`` -- and acts on the element's center DETERMINISTICALLY, with no + pixel matching. This is the thesis shift from "vision-only" to + "deterministic compiled automation with visual FALLBACK": the desktop + benchmark measured UIA execution 21/21 vs compiled visual replay 6/21 under + render drift. + + The visual anchor (template / ocr_text / landmarks) is ALWAYS kept too: the + ladder falls through to it UNCHANGED when this locator is absent (pixel-only + substrate -- RDP/Citrix/canvas) or when the element cannot be located at + replay (see docs/LIMITS.md). Structural resolution is ADDITIVE -- it never + removes the visual floor. + + Fields are backend-neutral; a browser backend fills ``selector`` / ``role`` + / ``name`` from the DOM, a native backend fills ``automation_id`` / ``role`` + / ``name`` from the UIA/AX tree. Each backend uses whichever fields it + recorded; unset fields are ignored. + """ + + selector: Optional[str] = Field( + default=None, + description="Stable CSS/DOM selector, e.g. '#open-p1' (browser)", + ) + role: Optional[str] = Field( + default=None, + description="ARIA / UIA control role, e.g. 'button', 'link'", + ) + name: Optional[str] = Field( + default=None, + description="Accessible name / label / text of the target element", + ) + automation_id: Optional[str] = Field( + default=None, + description="Windows UIA AutomationId (native desktop backends)", + ) + + class Anchor(BaseModel): - """Redundant visual evidence for locating a step's target on screen. + """Redundant evidence for locating a step's target on screen. - Resolution ladder consumes fields in order of preference: - template (local, then global) -> ocr_text -> landmarks -> grounder. + Resolution ladder consumes fields in order of preference (strongest, most + drift-tolerant first): ``structural`` (DOM / UIA element, when the backend + supports it) -> template (local, then global) -> ocr_text -> landmarks -> + grounder. ``structural`` is the deterministic top rung; the remaining + (visual) rungs are the FALLBACK floor for pixel-only substrates. """ template: str = Field(description="Bundle-relative path to the PNG crop") + structural: Optional[StructuralLocator] = Field( + default=None, + description=( + "STRUCTURAL locator (DOM selector / role+name, or UIA" + " AutomationId / role+name) of the clicked target, captured at" + " record time when the recording backend exposes it" + " (openadapt_flow.backend.StructuralActionBackend). Drives the" + " structural ACTION rung -- the TOP of the resolution ladder --" + " which re-finds the SAME element deterministically (no pixel" + " match) and is far more drift-tolerant than the visual rungs" + " (21/21 vs 6/21 on the desktop benchmark). None on pixel-only" + " substrates or bundles compiled before this capability; the" + " ladder then resolves via the visual rungs below." + ), + ) region: Region = Field(description="Crop location in the recorded frame") click_point: Point = Field(description="Click point in the recorded frame") ocr_text: Optional[str] = Field( @@ -276,7 +339,24 @@ def load(cls, bundle_dir: Path | str) -> "Workflow": # -- runtime results --------------------------------------------------------- -Rung = Literal["template", "template_global", "ocr", "geometry", "grounder"] +Rung = Literal[ + "structural", "template", "template_global", "ocr", "geometry", "grounder" +] + + +class StructuralHandle(BaseModel): + """Result of a backend's structural locate: the resolved element's point. + + ``point`` is the element's center in the SAME coordinate space as + :meth:`openadapt_flow.backend.Backend.click` (the pixels the resolver + emits), so a structurally-resolved target flows through the IDENTICAL click + path -- the pre-click identity gate and the irreversible risk gate still + fire (structural resolution makes identity STRONGER, an exact element; it + never bypasses it). ``confidence`` is 1.0 for a deterministic exact locate. + """ + + point: Point + confidence: float = 1.0 class Resolution(BaseModel): diff --git a/openadapt_flow/recorder.py b/openadapt_flow/recorder.py index fef1d17..9213344 100644 --- a/openadapt_flow/recorder.py +++ b/openadapt_flow/recorder.py @@ -33,6 +33,7 @@ from PIL import Image, ImageDraw from openadapt_flow.backend import Backend +from openadapt_flow.ir import StructuralLocator def _phash(png: bytes) -> imagehash.ImageHash: @@ -153,6 +154,20 @@ def _record(self, event: dict[str, Any], act: Callable[[], None]) -> None: # can verify identity against the highest-fidelity signal (no OCR # ambiguity). Pointer events only (they carry x/y); absent otherwise. if event.get("kind") in ("click", "double_click"): + # Structural locator (DOM selector / role+name, or UIA identifiers) + # for the clicked element, when the backend exposes it + # (StructuralActionBackend). Stored on the event so the compiler can + # put it on the anchor and the replayer's structural ACTION rung can + # re-find the SAME element deterministically (no pixel match). Absent + # on pixel-only backends; resolution then uses the visual anchor. + locator = self._structural_locator_at( + int(event["x"]), int(event["y"]) + ) + if locator is not None: + event = { + **event, + "structural": locator.model_dump(exclude_none=True), + } structured = self._structured_identity_at( int(event["x"]), int(event["y"]) ) @@ -267,6 +282,25 @@ def _commit( f.write(json.dumps(line) + "\n") self._i += 1 + def _structural_locator_at( + self, x: int, y: int + ) -> Optional[StructuralLocator]: + """Stable structural locator for the element under (x, y), if any. + + Backends MAY expose ``structural_locator_at`` + (:class:`openadapt_flow.backend.StructuralActionBackend`): a stable DOM + selector / role+name (browser) or UIA AutomationId / role+name (native) + the replayer can re-resolve to act on the SAME element deterministically. + Pixel-only backends (no such method) or a momentary failure yield None, + and the step relies on the visual anchor alone. + """ + getter = getattr(self._backend, "structural_locator_at", None) + if getter is None: + return None + try: + return getter(int(x), int(y)) + except Exception: + return None @staticmethod def _redact( png: bytes, region: tuple[int, int, int, int] diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 5bfb8dc..2e39ada 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -160,6 +160,7 @@ def __init__( effect_verifier: Optional[EffectVerifier] = None, effect_compensator: Optional[Any] = None, poll_interval_s: float = 0.05, + use_structural: bool = True, ) -> None: if vision is None: import openadapt_flow.vision as vision # lazy: heavy OCR deps @@ -167,6 +168,14 @@ def __init__( self.backend = backend self.vision = vision self.grounder = grounder + # Whether the deterministic structural ACTION rung (top of the ladder) + # may run. True (default) = product behavior: on a structure-bearing + # backend, resolve the recorded target as a DOM/UIA element. Set False + # to force the VISUAL fallback floor even on a structure-bearing + # backend -- used to characterize the pixel-only substrate path + # (RDP/Citrix/VDI) on a Playwright surface (see the e2e visual-floor + # drift/heal suites). Never disables the visual ladder itself. + self.use_structural = use_structural # System-of-record effect verification (OFF by default). The verifier # is bound to the deployment's system of record; the optional # compensator undoes a detected duplicate irreversible write. Neither @@ -509,16 +518,26 @@ def _run_step( result.ok and resolution is not None and matched_region is not None - and resolution.rung != "template" + and resolution.rung not in ("template", "structural") and step.anchor is not None ): # Heal from the PRE-action frame: that is the frame the # anchor was resolved against (the action may have navigated # to a different screen, where a crop at the old location - # would be garbage). The heal is GOVERNED: a patch that would - # weaken the step's identity band (the reviewed context-drop - # bug), effect coverage, or risk class is quarantined and the - # run HALTS rather than silently applying an unverified repair. + # would be garbage). + # + # ``structural`` is exempt alongside ``template``: a + # deterministic DOM/UIA locate does NOT signal that the visual + # template is stale (structural runs FIRST regardless of visual + # drift), so refreshing the crop every run would emit a spurious + # reviewable anchor diff and break the zero-heal happy path. The + # visual fallback stays as recorded -- still valid for the + # substrate it was recorded on. + # + # The heal is GOVERNED: a patch that would weaken the step's + # identity band (the reviewed context-drop bug), effect + # coverage, or risk class is quarantined and the run HALTS + # rather than silently applying an unverified repair. heal_outcome = self._heal_step( step, resolution, matched_region, before_png, workflow, run_dir, new_crops, @@ -650,6 +669,16 @@ def _resolve_step( if template_path.is_file(): template_png = template_path.read_bytes() + # The structural ACTION rung (top of the ladder) runs only when the + # live backend can re-find an element (StructuralActionBackend). A + # pixel-only backend (RDP/Citrix) lacks the method, so ``structural`` + # is None and resolution uses the visual ladder unchanged. + structural = ( + self.backend + if self.use_structural + and hasattr(self.backend, "locate_structural") + else None + ) resolved = resolve( step.anchor, screen_png, @@ -658,6 +687,7 @@ def _resolve_step( step.intent, template_png=template_png, viewport=self.backend.viewport, + structural=structural, ) if resolved is None: return None, None, ( diff --git a/openadapt_flow/runtime/resolver.py b/openadapt_flow/runtime/resolver.py index c40ab08..0160f5b 100644 --- a/openadapt_flow/runtime/resolver.py +++ b/openadapt_flow/runtime/resolver.py @@ -1,8 +1,18 @@ """Resolution ladder: locate a step's target on the live screen. -The ladder walks progressively weaker (but more drift-tolerant) evidence: - -1. ``template`` — template match inside ``anchor.region`` padded by +The ladder walks from the STRONGEST, most drift-tolerant evidence down to +progressively weaker (but more widely available) evidence. The full capability +hierarchy is API -> tool/MCP -> DOM/UIA -> geometry -> OCR -> template -> VLM +-> human; the API and tool/MCP rungs are future placeholders, so the rungs +implemented here are: + +0. ``structural`` — DETERMINISTIC. Re-find the recorded target as a + DOM/UIA *element* via ``backend.locate_structural`` (a stable selector / + role+name / AutomationId the compiler captured) and act on its center. Tried + FIRST, and only when the backend exposes the capability AND the anchor + carries a structural locator. Pixel-only substrates (RDP/Citrix/canvas) and + failed locates fall through to the visual rungs UNCHANGED. +1. ``template`` — template match inside ``anchor.region`` padded by ``anchor.search_pad`` (clamped to the viewport). 2. ``template_global`` — template match over the full frame. 3. ``ocr`` — fuzzy text match on ``anchor.ocr_text``. @@ -10,9 +20,15 @@ relation/distance to estimate the target point. 5. ``grounder`` — optional injected model-backed grounding. +Rungs 1-5 are the VISUAL FALLBACK floor: the runtime remains able to operate a +pure-pixel surface where no structure exists. Structural resolution is +ADDITIVE — it is preferred where present, never a replacement. + The ``vision`` argument is a namespace-like object (the real ``openadapt_flow.vision`` module or a test fake) exposing ``find_template`` -and ``find_text``. +and ``find_text``. The ``structural`` argument is an optional object exposing +``locate_structural`` (a Backend implementing +:class:`openadapt_flow.backend.StructuralActionBackend`). """ from __future__ import annotations @@ -25,6 +41,7 @@ from openadapt_flow.ir import Anchor, Point, Region, Resolution, Rung RUNG_ORDER: tuple[Rung, ...] = ( + "structural", "template", "template_global", "ocr", @@ -227,11 +244,13 @@ def resolve( *, template_png: Optional[bytes] = None, viewport: Optional[tuple[int, int]] = None, + structural: Optional[Any] = None, ) -> Optional[tuple[Resolution, Region]]: """Walk the resolution ladder for ``anchor`` against a live frame. Args: - anchor: The step's anchor (template path/region, ocr text, landmarks). + anchor: The step's anchor (structural locator, template path/region, + ocr text, landmarks). screen_png: Current frame as PNG bytes. vision: Namespace-like object exposing ``find_template(screen, tmpl, *, search_region=None)`` and ``find_text(screen, text)``, each @@ -243,6 +262,12 @@ def resolve( None the two template rungs are skipped. viewport: (width, height) of the frame; parsed from ``screen_png``'s PNG header when omitted. + structural: Optional backend exposing ``locate_structural(locator)`` + (:class:`openadapt_flow.backend.StructuralActionBackend`). When + provided AND ``anchor.structural`` is set, the DETERMINISTIC + structural rung runs FIRST; a successful locate short-circuits the + visual rungs. None (pixel-only substrate) or a failed locate falls + through to the visual ladder unchanged. Returns: ``(resolution, matched_region)`` on success, where ``matched_region`` @@ -257,6 +282,36 @@ def resolve( def elapsed_ms() -> float: return (time.monotonic() - t0) * 1000.0 + # Rung 0: structural (DOM / UIA) — the strongest, deterministic evidence. + # Tried FIRST, and only when a structural-capable backend is injected AND + # the anchor carries a recorded structural locator. On a pixel-only + # substrate (RDP/Citrix/canvas), a backend without the capability, or a + # failed/ambiguous locate, ``locate_structural`` yields None and we fall + # through to the visual ladder UNCHANGED. The resolved point flows through + # the SAME click path as any visual rung, so the identity and risk gates + # still fire on it; structural sits ABOVE ``ocr`` in RUNG_ORDER, so an + # irreversible step MAY act on it (it is the strongest evidence, not the + # weakest). Future API and tool/MCP rungs sit above this one. + if structural is not None and anchor.structural is not None: + locate = getattr(structural, "locate_structural", None) + if locate is not None: + try: + handle = locate(anchor.structural) + except Exception: + handle = None + if handle is not None: + point = (int(handle.point[0]), int(handle.point[1])) + region = _clamp_region_of_size( + point, (anchor.region[2], anchor.region[3]), viewport + ) + resolution = Resolution( + rung="structural", + point=point, + confidence=float(getattr(handle, "confidence", 1.0)), + elapsed_ms=elapsed_ms(), + ) + return resolution, region + # Rung 1: template within the padded local search region. if template_png is not None: search_region = pad_region(anchor.region, anchor.search_pad, viewport) diff --git a/openadapt_flow/validation/structural_action.py b/openadapt_flow/validation/structural_action.py new file mode 100644 index 0000000..e120d5e --- /dev/null +++ b/openadapt_flow/validation/structural_action.py @@ -0,0 +1,215 @@ +"""Availability probe: the structural ACTION rung vs the visual ladder alone. + +Mirrors the desktop-benchmark finding that reframed the thesis from +"vision-only" to "deterministic compiled automation with visual FALLBACK": +structural (DOM/UIA) execution scored 21/21 while compiled *visual* replay +scored 6/21 under render drift. This probe reproduces that shape on a real +rendered DOM with the real resolution ladder. + +Model +----- +A dense surface of ``n`` actionable targets (buttons), each with a STABLE DOM +id (``#open-pK``) -- exactly what a browser recorder captures for the +structural rung -- AND a recorded visual anchor (template crop at the button's +recorded position + its OCR label). We then RE-RENDER the surface with drift: +a deterministic subset of targets is left untouched (models steps whose surface +did not change between record and replay), while the rest are MOVED and +RELABELLED and RE-THEMED (models real render drift: layout shift, an Open->View +relabel, a dark theme). The DOM id never changes. + +For every target we ask, on the drifted surface: + +* STRUCTURAL rung -- ``backend.locate_structural(locator)``: does it land inside + the CORRECT element's box? (Deterministic; independent of pixels.) +* VISUAL ladder -- ``resolve(anchor, drift_png, vision, structural=None)``: + does the template/OCR/geometry ladder resolve to a point inside the CORRECT + element's box? A drifted target's recorded "Open" crop no longer matches at + its old position and either fails to resolve or matches a look-alike sibling + -- both are availability/correctness failures. + +The numbers are MEASURED (never hardcoded): structural resolves every target +whose id is still in the DOM; visual resolves only the non-drifted minority. + +Runnable via benchmark/structural_action/structural_action_probe.py; the +reusable logic (build_html / run_probe) also backs tests/test_structural_rung.py. +""" +from __future__ import annotations + +import io +from typing import Optional + +from PIL import Image + +from openadapt_flow import vision as vision_mod +from openadapt_flow.backends.playwright_backend import PlaywrightBackend +from openadapt_flow.ir import Anchor +from openadapt_flow.runtime.resolver import resolve + +# Grid layout (viewport is the PlaywrightBackend default 1280x800). +_COL_X = 90 # baseline button column (left) +_DRIFT_COL_X = 560 # drifted buttons jump to a second column +_TOP = 70 +_ROW_H = 34 +_BTN_W = 84 +_BTN_H = 24 + + +def _is_stable(index: int, n: int) -> bool: + """Deterministic 'did not drift' subset (~every 4th target). + + Models the minority of steps whose surface is unchanged between record and + replay. Chosen to land the visual-only score near the 6/21 illustration + without ever inspecting resolution outcomes. + """ + return index % 4 == 0 + + +def _button_html(index: int, *, drift: bool) -> str: + stable = _is_stable(index, 0) + left = _COL_X + label = "Open" + cls = "open-btn" + if drift and not stable: + left = _DRIFT_COL_X + label = "View" # rename drift (label no longer matches crop/OCR) + cls = "open-btn drifted" # dark-theme restyle + top = _TOP + index * _ROW_H + return ( + f'' + ) + + +def build_html(n: int, *, drift: bool) -> str: + """A dense absolutely-positioned surface of ``n`` actionable buttons.""" + buttons = "\n".join(_button_html(i, drift=drift) for i in range(n)) + return ( + "" + f"{buttons}" + "" + ) + + +def _crop_png(png: bytes, box: tuple[int, int, int, int]) -> bytes: + x, y, w, h = box + with Image.open(io.BytesIO(png)) as img: + crop = img.crop((x, y, x + w, y + h)) + buf = io.BytesIO() + crop.save(buf, format="PNG") + return buf.getvalue() + + +def _bbox(backend: PlaywrightBackend, selector: str) -> Optional[tuple]: + box = backend.page.locator(selector).bounding_box() + if not box: + return None + return (box["x"], box["y"], box["width"], box["height"]) + + +def _inside(point, box) -> bool: + if box is None: + return False + px, py = point + x, y, w, h = box + return x <= px <= x + w and y <= py <= y + h + + +def run_probe(n: int = 21, *, headless: bool = True) -> dict: + """Record on a baseline surface, drift it, compare structural vs visual. + + Returns a report dict with per-target outcomes and the two success counts. + """ + backend, close = PlaywrightBackend.launch("about:blank", headless=headless) + try: + # -- record on the baseline surface -------------------------------- + backend.page.set_content(build_html(n, drift=False)) + backend.page.wait_for_timeout(60) + base_png = backend.screenshot() + anchors: list[Anchor] = [] + locators = [] + for i in range(n): + box = _bbox(backend, f"#open-p{i}") + assert box is not None, f"missing #open-p{i} on baseline" + cx = int(box[0] + box[2] / 2) + cy = int(box[1] + box[3] / 2) + region = ( + int(box[0]) - 6, + int(box[1]) - 6, + int(box[2]) + 12, + int(box[3]) + 12, + ) + locator = backend.structural_locator_at(cx, cy) + assert locator is not None, f"no structural locator for p{i}" + locators.append(locator) + anchors.append( + Anchor( + template=f"templates/open-p{i}.png", + region=region, + click_point=(cx, cy), + ocr_text="Open", + structural=locator, + ) + ) + + templates = { + i: _crop_png(base_png, anchors[i].region) for i in range(n) + } + + # -- drift the surface --------------------------------------------- + backend.page.set_content(build_html(n, drift=True)) + backend.page.wait_for_timeout(60) + drift_png = backend.screenshot() + + targets = [] + structural_ok = 0 + visual_ok = 0 + for i in range(n): + correct_box = _bbox(backend, f"#open-p{i}") + + # structural rung (deterministic, id-based) + handle = backend.locate_structural(locators[i]) + s_ok = handle is not None and _inside(handle.point, correct_box) + + # visual ladder only (structural disabled) + res = resolve( + anchors[i], + drift_png, + vision_mod, + template_png=templates[i], + viewport=backend.viewport, + structural=None, + ) + v_ok = res is not None and _inside(res[0].point, correct_box) + + structural_ok += int(s_ok) + visual_ok += int(v_ok) + targets.append( + { + "id": f"open-p{i}", + "drifted": not _is_stable(i, n), + "structural_ok": bool(s_ok), + "structural_rung": handle is not None, + "visual_ok": bool(v_ok), + "visual_rung": (res[0].rung if res else None), + } + ) + + return { + "n": n, + "structural_ok": structural_ok, + "visual_ok": visual_ok, + "structural_ratio": f"{structural_ok}/{n}", + "visual_ratio": f"{visual_ok}/{n}", + "targets": targets, + } + finally: + close() diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 79d2267..dfb6b43 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -135,7 +135,14 @@ def _replay( params: Optional[dict[str, str]] = None, run_dir: Optional[Path] = None, save_healed_to: Optional[Path] = None, + use_structural: bool = False, ) -> tuple[RunReport, Path]: + # use_structural defaults False: the record->compile->replay suites + # characterize the VISUAL ladder (template/ocr/geometry/heal) and its + # drift envelope -- the pixel-only substrate path (RDP/Citrix) the + # structural rung sits ON TOP of but never replaces. The structural + # default-on product path is covered explicitly by + # tests/e2e/test_structural_action.py. if run_dir is None: run_dir = tmp_path_factory.mktemp("run") page = _browser.new_page(viewport=VIEWPORT, device_scale_factor=1) @@ -143,7 +150,7 @@ def _replay( page.goto(url) backend = PlaywrightBackend(page) workflow = Workflow.load(bundle_dir) - report = Replayer(backend).run( + report = Replayer(backend, use_structural=use_structural).run( workflow, params=dict(PARAMS if params is None else params), bundle_dir=Path(bundle_dir), diff --git a/tests/e2e/test_fault_model.py b/tests/e2e/test_fault_model.py index 4a63a45..ec779e1 100644 --- a/tests/e2e/test_fault_model.py +++ b/tests/e2e/test_fault_model.py @@ -185,7 +185,11 @@ def _replay_fault( try: page.goto(f"{base_url}{query}") backend = PlaywrightBackend(page) - return Replayer(backend).run( + # Floor: the fault-model suite characterizes runtime behavior under + # API/persistence faults; pin the visual resolution path so outcomes + # isolate the fault, not the rung (structural default is covered by + # tests/e2e/test_structural_action.py). + return Replayer(backend, use_structural=False).run( Workflow.load(bundle_dir), params=dict(PARAMS), bundle_dir=Path(bundle_dir), diff --git a/tests/e2e/test_structural_action.py b/tests/e2e/test_structural_action.py new file mode 100644 index 0000000..b945419 --- /dev/null +++ b/tests/e2e/test_structural_action.py @@ -0,0 +1,93 @@ +"""End-to-end: the structural (DOM) ACTION rung as the product DEFAULT. + +The record -> compile -> replay characterization suites in this package run +with ``use_structural=False`` so they keep pinning the VISUAL fallback floor +(the pixel-only / RDP-Citrix substrate path). This module pins the DEFAULT +product behavior on a structure-bearing backend (MockMed is a real DOM app): +the runtime resolves each recorded target as a DOM element and acts on it +deterministically, with the identity and risk gates unchanged. + +The resolution-level availability win under render drift (structural 21/21 vs +visual 6/21) is measured in ``openadapt_flow/validation/structural_action.py`` +and asserted in ``tests/test_structural_rung.py``; here we prove the same rung +drives a full record->compile->replay on MockMed and preserves the headline +no-wrong-write safety property under cosmetic drift. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from .conftest import PARAMS +from .test_record_compile_replay import N_ANCHORED, anchored_results, rung_of +from .validation_utils import describe, replay_cosmetic + +pytestmark = pytest.mark.timeout(600) + +TARGET_HASH = "#patient/p1" + + +def test_default_happy_path_resolves_every_step_structurally( + bundle, mockmed_url, replay +) -> None: + """DEFAULT product path: every anchored step resolves on the deterministic + ``structural`` rung (DOM element), zero heals, zero model calls, saved to + the target patient. The strongest possible resolution -- an exact element, + not a pixel match.""" + report, _ = replay(bundle.dir, mockmed_url, use_structural=True) + assert report.success, describe(report, {}) + assert report.heal_count == 0 + assert report.model_calls == 0 + # The whole anchored path came from the structural rung. + assert report.rung_counts == {"structural": N_ANCHORED}, report.rung_counts + for result in anchored_results(report): + assert result.resolution is not None + assert result.resolution.rung == "structural" + # Identity gate still ran in front of every click: no click was + # admitted on a mismatch (a mismatch aborts the run). + if result.identity is not None: + assert result.identity.status != "mismatch" + + +def test_structural_is_actually_engaged_vs_visual_floor( + bundle, mockmed_url, replay +) -> None: + """Direct contrast on the SAME bundle: the visual floor resolves the save + step via ``template``; the default structural path resolves it via + ``structural``. Proves the rung is genuinely preferred, not incidental.""" + floor, _ = replay(bundle.dir, mockmed_url, use_structural=False) + structural, _ = replay(bundle.dir, mockmed_url, use_structural=True) + assert floor.success and structural.success + save_floor = rung_of(floor, "step_010") + save_structural = rung_of(structural, "step_010") + assert save_floor == "template" + assert save_structural == "structural" + + +def test_structural_on_never_wrong_writes_under_cosmetic_drift( + bundle, mockmed_url, _browser, tmp_path +) -> None: + """Headline safety property holds with the structural rung ENGAGED: across + a cosmetic drift sweep, the run either completes on the TARGET patient or + safe-halts having saved nothing -- never a save to the wrong patient. + Structural resolution makes identity STRONGER (an exact element); it never + licenses a wrong action.""" + sweep = [ + {"zoom": 1.25}, + {"font_scale": 1.375}, + {"font_family": "Georgia, serif"}, + {"device_scale_factor": 2}, + ] + for i, kwargs in enumerate(sweep): + report, state = replay_cosmetic( + _browser, bundle.dir, mockmed_url, tmp_path / f"p{i}", + params=dict(PARAMS), use_structural=True, **kwargs, + ) + wrong_write = state["banner"] is not None and state["hash"] != TARGET_HASH + assert not wrong_write, describe(report, state) + if report.success: + assert state["hash"] == TARGET_HASH, describe(report, state) + else: + assert state["banner"] is None, describe(report, state) diff --git a/tests/e2e/validation_utils.py b/tests/e2e/validation_utils.py index 438513d..93c906e 100644 --- a/tests/e2e/validation_utils.py +++ b/tests/e2e/validation_utils.py @@ -75,6 +75,7 @@ def replay_on_page( viewport: tuple[int, int] = (1280, 800), device_scale_factor: int = 1, backend_factory: Optional[Callable[[object], PlaywrightBackend]] = None, + use_structural: bool = False, ) -> tuple[RunReport, dict]: """Replay ``bundle_dir`` on a fresh page and observe final app state. @@ -100,7 +101,7 @@ def replay_on_page( backend_factory(page) if backend_factory else PlaywrightBackend(page) ) workflow = Workflow.load(bundle_dir) - report = Replayer(backend).run( + report = Replayer(backend, use_structural=use_structural).run( workflow, params=params, bundle_dir=Path(bundle_dir), @@ -196,6 +197,7 @@ def replay_cosmetic( zoom: Optional[float] = None, font_scale: Optional[float] = None, font_family: Optional[str] = None, + use_structural: bool = False, ) -> tuple[RunReport, dict]: """Replay ``bundle_dir`` under a cosmetic-only render perturbation. @@ -220,7 +222,7 @@ def replay_cosmetic( page.wait_for_timeout(80) # let the reflow settle backend = PlaywrightBackend(page) workflow = Workflow.load(bundle_dir) - report = Replayer(backend).run( + report = Replayer(backend, use_structural=use_structural).run( workflow, params=params, bundle_dir=Path(bundle_dir), diff --git a/tests/test_resolver.py b/tests/test_resolver.py index 35c66dc..7b8310b 100644 --- a/tests/test_resolver.py +++ b/tests/test_resolver.py @@ -337,9 +337,15 @@ def test_all_rungs_fail_returns_none(screen, anchor): def test_ladder_order_is_frozen(): + # ``structural`` (DOM/UIA) is the deterministic TOP rung, above the visual + # rungs; the visual floor keeps its historical order underneath. assert RUNG_ORDER == ( - "template", "template_global", "ocr", "geometry", "grounder" + "structural", "template", "template_global", "ocr", "geometry", + "grounder", ) + # Structural is the STRONGEST evidence: it must sit above ocr so the + # irreversible-risk gate (is_below_ocr) lets an irreversible step act on it. + assert not is_below_ocr("structural") assert not is_below_ocr("template") assert not is_below_ocr("template_global") assert not is_below_ocr("ocr") diff --git a/tests/test_resolver_fuzz.py b/tests/test_resolver_fuzz.py index e2ae253..ae48064 100644 --- a/tests/test_resolver_fuzz.py +++ b/tests/test_resolver_fuzz.py @@ -47,7 +47,14 @@ from hypothesis import strategies as st from PIL import Image -from openadapt_flow.ir import Anchor, ActionKind, Landmark, Step +from openadapt_flow.ir import ( + Anchor, + ActionKind, + Landmark, + Step, + StructuralHandle, + StructuralLocator, +) from openadapt_flow.runtime.resolver import ( RUNG_ORDER, is_below_ocr, @@ -121,6 +128,13 @@ def __init__(self, viewport): def viewport(self): return self._viewport + def locate_structural(self, locator): + # Only invoked by _resolve_step when anchor.structural is set (the + # structural-rung gate case); returns a deterministic in-viewport point. + vw, vh = self._viewport + return StructuralHandle(point=(min(110, vw - 1), min(105, vh - 1)), + confidence=1.0) + def _png(size) -> bytes: image = Image.new("RGB", size, (240, 240, 240)) @@ -318,7 +332,11 @@ def _gate_case(draw): ) vision = FakeVision() grounder = None - if rung == "template": + if rung == "structural": + # Structural is the deterministic TOP rung: a recorded locator + + # backend.locate_structural resolve it (FakeBackend supplies the point). + anchor.structural = StructuralLocator(selector="#x") + elif rung == "template": vision.template_results = [m] elif rung == "template_global": vision.template_results = [None, m] diff --git a/tests/test_structural_rung.py b/tests/test_structural_rung.py new file mode 100644 index 0000000..f88c35b --- /dev/null +++ b/tests/test_structural_rung.py @@ -0,0 +1,444 @@ +"""Structural (DOM / UIA) ACTION rung — the deterministic TOP of the ladder. + +The runtime is no longer "vision-only": where a backend owns a structured layer +(a browser DOM, a native UIA/AX tree) the resolution ladder re-finds the +recorded target as an ELEMENT and acts on it deterministically, falling back to +the visual rungs (template/ocr/geometry) only where structure is absent +(pixel-only substrates: RDP/Citrix/canvas). This mirrors the identity ladder, +which already prefers structured text. + +Covered here: + * resolver mechanics — structural tried FIRST, wins when it locates, falls + through UNCHANGED to the visual ladder otherwise; risk gate treats it as + the STRONGEST evidence (not below ocr); + * the record/replay backend split (Windows UIA via a faked WAA session, + Playwright DOM via importorskip); + * compile-time capture of the locator; + * the identity gate still fires on a structurally-resolved point (structure + makes identity STRONGER, it never bypasses it); + * the availability probe (structural vs visual under drift). +""" + +from __future__ import annotations + +import io +import json +from pathlib import Path + +import pytest +from PIL import Image + +from openadapt_flow.ir import ( + ActionKind, + Anchor, + Resolution, + Step, + StructuralHandle, + StructuralLocator, + Workflow, +) +from openadapt_flow.runtime.resolver import RUNG_ORDER, is_below_ocr, resolve + +VIEWPORT = (300, 200) + + +def make_png(size=VIEWPORT, color=(240, 240, 240)) -> bytes: + buf = io.BytesIO() + Image.new("RGB", size, color).save(buf, format="PNG") + return buf.getvalue() + + +class _Match: + def __init__(self, point, region, confidence=0.95): + self.point = point + self.region = region + self.confidence = confidence + + +class _FakeVision: + """Scripted find_template / find_text (never imports the real vision).""" + + def __init__(self): + self.template_results = [] + self.text_results = {} + self.template_calls = 0 + self.text_calls = 0 + + def find_template(self, screen_png, template_png, *, search_region=None, + prefer_near=None, scales=(0.85, 1.0, 1.18), + threshold=0.82): + self.template_calls += 1 + if self.template_results: + return self.template_results.pop(0) + return None + + def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + self.text_calls += 1 + return self.text_results.get(text) + + +class _FakeStructural: + """A structural-capable backend stub exposing locate_structural.""" + + def __init__(self, handle, *, raises=False): + self._handle = handle + self._raises = raises + self.calls = [] + + def locate_structural(self, locator): + self.calls.append(locator) + if self._raises: + raise RuntimeError("uia blew up") + return self._handle + + +def _anchor(**kw): + base = dict( + template="templates/x.png", + region=(100, 100, 40, 20), + click_point=(120, 110), + ocr_text="Open", + structural=StructuralLocator(selector="#open-p1"), + ) + base.update(kw) + return Anchor(**base) + + +# --------------------------------------------------------------------------- +# resolver mechanics +# --------------------------------------------------------------------------- + +def test_structural_is_top_of_ladder_and_not_below_ocr() -> None: + assert RUNG_ORDER[0] == "structural" + # Strongest evidence: an irreversible step is allowed to act on it. + assert not is_below_ocr("structural") + + +def test_structural_rung_tried_first_and_wins() -> None: + vision = _FakeVision() + # A template match is *also* available — but structural must pre-empt it. + vision.template_results = [ + _Match(point=(120, 110), region=(100, 100, 40, 20)) + ] + backend = _FakeStructural(StructuralHandle(point=(207, 133), + confidence=1.0)) + screen = make_png() + res = resolve(_anchor(), screen, vision, template_png=b"tpl", + viewport=VIEWPORT, structural=backend) + assert res is not None + resolution, region = res + assert resolution.rung == "structural" + assert resolution.point == (207, 133) + assert resolution.confidence == 1.0 + # The visual template rung was never consulted. + assert vision.template_calls == 0 + # A region (for healing / clamping) is returned, sized like the anchor. + assert region[2] == 40 and region[3] == 20 + assert backend.calls == [_anchor().structural] + + +def test_structural_miss_falls_through_to_visual_unchanged() -> None: + vision = _FakeVision() + vision.template_results = [ + _Match(point=(120, 110), region=(100, 100, 40, 20)) + ] + backend = _FakeStructural(None) # element absent / ambiguous -> None + res = resolve(_anchor(), make_png(), vision, template_png=b"tpl", + viewport=VIEWPORT, structural=backend) + assert res is not None + resolution, _ = res + assert resolution.rung == "template" # visual floor still works + assert vision.template_calls == 1 + assert backend.calls # structural WAS attempted first + + +def test_structural_exception_falls_through_no_crash() -> None: + vision = _FakeVision() + vision.template_results = [ + _Match(point=(120, 110), region=(100, 100, 40, 20)) + ] + backend = _FakeStructural(None, raises=True) + res = resolve(_anchor(), make_png(), vision, template_png=b"tpl", + viewport=VIEWPORT, structural=backend) + assert res is not None and res[0].rung == "template" + + +def test_pixel_only_backend_uses_visual_ladder() -> None: + # structural=None models a pixel-only substrate (RDP/Citrix/canvas): the + # visual ladder is used exactly as before. + vision = _FakeVision() + vision.template_results = [ + _Match(point=(120, 110), region=(100, 100, 40, 20)) + ] + res = resolve(_anchor(), make_png(), vision, template_png=b"tpl", + viewport=VIEWPORT, structural=None) + assert res is not None and res[0].rung == "template" + + +def test_anchor_without_locator_never_calls_structural() -> None: + vision = _FakeVision() + vision.template_results = [ + _Match(point=(120, 110), region=(100, 100, 40, 20)) + ] + backend = _FakeStructural(StructuralHandle(point=(1, 1))) + res = resolve(_anchor(structural=None), make_png(), vision, + template_png=b"tpl", viewport=VIEWPORT, structural=backend) + assert res is not None and res[0].rung == "template" + assert backend.calls == [] # no locator -> structural skipped + + +# --------------------------------------------------------------------------- +# identity gate STILL fires on a structurally-resolved point +# --------------------------------------------------------------------------- + +class _IdentityAndStructuralBackend: + """Backend that resolves structurally AND exposes structured identity.""" + + def __init__(self, point, live_identity): + self._point = point + self._live = live_identity + self.clicks = [] + + @property + def viewport(self): + return VIEWPORT + + def screenshot(self): + return make_png() + + def click(self, x, y, *, double=False): + self.clicks.append((x, y)) + + def type_text(self, text): ... + + def press(self, key): ... + + def scroll(self, dx, dy): ... + + def locate_structural(self, locator): + return StructuralHandle(point=self._point, confidence=1.0) + + def structured_text_at(self, x, y): + return self._live + + +def test_structural_resolution_still_faces_identity_gate() -> None: + from openadapt_flow.runtime.replayer import Replayer + + recorded = "MG4408 Okafor, Philip 1966-01-17" + sibling = "MG44O8 Okafor, Philip 1966-01-17" # one-glyph different patient + backend = _IdentityAndStructuralBackend((207, 133), sibling) + rp = Replayer(backend, vision=_FakeVision(), poll_interval_s=0.01) + step = Step( + id="s1", intent="open patient", action=ActionKind.CLICK, + anchor=_anchor(structured_identity=recorded), + ) + # 1) resolution comes from the structural rung... + resolution, region, err = rp._resolve_step(step, make_png(), Path(".")) + assert err is None + assert resolution is not None and resolution.rung == "structural" + assert resolution.point == (207, 133) + # 2) ...and the pre-click identity gate STILL runs on that point, catching + # the sibling (structure makes identity stronger, never bypasses it). + check = rp._verify_identity(step, resolution, make_png(), {}, + Workflow(name="wf"), None) + assert check.status == "mismatch" + assert check.mode == "structured" + + +# --------------------------------------------------------------------------- +# Windows UIA backend (faked WAA execute channel) +# --------------------------------------------------------------------------- + +class _Resp: + def __init__(self, text="", json_data=None, status=200): + self.text = text + self._json = json_data + self.status_code = status + + def json(self): + if self._json is None: + raise ValueError("no json") + return self._json + + +class _FakeSession: + def __init__(self, resp): + self._resp = resp + self.posted = [] + + def post(self, url, json=None, timeout=None): + self.posted.append(json) + return self._resp + + +def _win_backend(resp): + from openadapt_flow.backends.windows_backend import WindowsBackend + return WindowsBackend(session=_FakeSession(resp), viewport=(800, 600)) + + +def test_windows_structural_locator_at_parses_uia_dict() -> None: + payload = ('<>{"automation_id": "open-p1", ' + '"role": "button", "name": "Open"}<>') + be = _win_backend(_Resp(text="log\n" + payload)) + loc = be.structural_locator_at(120, 110) + assert isinstance(loc, StructuralLocator) + assert loc.automation_id == "open-p1" + assert loc.role == "button" and loc.name == "Open" + + +def test_windows_structural_locator_none_when_no_echo() -> None: + be = _win_backend(_Resp(text="")) + assert be.structural_locator_at(1, 1) is None + + +def test_windows_structural_locator_none_on_uia_null() -> None: + payload = "<>null<>" + be = _win_backend(_Resp(text=payload)) + assert be.structural_locator_at(1, 1) is None + + +def test_windows_locate_structural_parses_point() -> None: + payload = "<>[412, 133]<>" + be = _win_backend(_Resp(text=payload)) + handle = be.locate_structural(StructuralLocator(automation_id="open-p1")) + assert isinstance(handle, StructuralHandle) + assert handle.point == (412, 133) + + +def test_windows_locate_structural_none_on_null() -> None: + payload = "<>null<>" + be = _win_backend(_Resp(text=payload)) + assert be.locate_structural( + StructuralLocator(automation_id="x")) is None + + +def test_windows_locate_structural_skips_server_without_ids() -> None: + session = _FakeSession(_Resp(text="")) + from openadapt_flow.backends.windows_backend import WindowsBackend + be = WindowsBackend(session=session, viewport=(800, 600)) + # A locator with no automation_id and no role+name is unresolvable; the + # backend must not even hit the server. + assert be.locate_structural(StructuralLocator(selector="#x")) is None + assert session.posted == [] + + +# --------------------------------------------------------------------------- +# recorder captures the locator; compiler stores it on the anchor +# --------------------------------------------------------------------------- + +class _RecordingStructuralBackend: + def __init__(self): + self._png = make_png() + + @property + def viewport(self): + return VIEWPORT + + def screenshot(self): + return self._png + + def click(self, x, y, *, double=False): ... + + def type_text(self, text): ... + + def press(self, key): ... + + def scroll(self, dx, dy): ... + + def structural_locator_at(self, x, y): + return StructuralLocator(selector="#open-p1", role="button", + name="Open") + + +def test_recorder_captures_structural_locator(tmp_path) -> None: + from openadapt_flow.recorder import Recorder + + rec = Recorder(_RecordingStructuralBackend(), tmp_path / "rec", + settle_timeout_s=0.05, settle_interval_s=0.01) + rec.click(120, 110) + rec.finish() + lines = (tmp_path / "rec" / "events.jsonl").read_text().splitlines() + event = json.loads(lines[0]) + assert event["structural"] == {"selector": "#open-p1", + "role": "button", "name": "Open"} + + +def test_compiler_stores_structural_locator(tmp_path) -> None: + import cv2 + import numpy as np + + from openadapt_flow.compiler.compile import compile_recording + + recording = tmp_path / "rec" + (recording / "frames").mkdir(parents=True) + before = np.full((200, 300, 3), 240, np.uint8) + cv2.rectangle(before, (100, 100), (160, 130), (200, 210, 255), -1) + cv2.putText(before, "Open", (104, 122), cv2.FONT_HERSHEY_SIMPLEX, 0.5, + (10, 20, 40), 1, cv2.LINE_AA) + after = before.copy() + cv2.putText(after, "Saved", (40, 180), cv2.FONT_HERSHEY_SIMPLEX, 0.6, + (0, 0, 0), 2, cv2.LINE_AA) + for suffix, img in (("before", before), ("after", after)): + ok, buf = cv2.imencode(".png", img) + assert ok + (recording / "frames" / f"0000_{suffix}.png").write_bytes( + buf.tobytes()) + event = {"i": 0, "kind": "click", "x": 130, "y": 115, "t": 1.0, + "structural": {"selector": "#open-p1", "role": "button", + "name": "Open"}} + (recording / "events.jsonl").write_text(json.dumps(event) + "\n") + (recording / "meta.json").write_text(json.dumps({ + "id": "rec-1", "created_at": "2026-07-13T00:00:00+00:00", + "viewport": [300, 200], "app_url": "http://x/", "params": {}})) + + wf = compile_recording(recording, tmp_path / "bundle", name="wf") + anchor = wf.steps[0].anchor + assert anchor is not None and anchor.structural is not None + assert anchor.structural.selector == "#open-p1" + assert anchor.structural.role == "button" + + +# --------------------------------------------------------------------------- +# Playwright DOM end-to-end (real browser; skipped when unavailable) +# --------------------------------------------------------------------------- + +def _pw_backend(): + pytest.importorskip("playwright.sync_api") + from openadapt_flow.backends.playwright_backend import PlaywrightBackend + return PlaywrightBackend + + +def test_playwright_structural_locator_and_locate_roundtrip() -> None: + from openadapt_flow.validation.structural_action import build_html + PlaywrightBackend = _pw_backend() + backend, close = PlaywrightBackend.launch("about:blank", headless=True) + try: + backend.page.set_content(build_html(6, drift=False)) + backend.page.wait_for_timeout(50) + box = backend.page.locator("#open-p2").bounding_box() + cx, cy = int(box["x"] + box["width"] / 2), int(box["y"] + + box["height"] / 2) + loc = backend.structural_locator_at(cx, cy) + assert loc is not None and loc.selector == "#open-p2" + handle = backend.locate_structural(loc) + assert handle is not None + px, py = handle.point + assert box["x"] <= px <= box["x"] + box["width"] + assert box["y"] <= py <= box["y"] + box["height"] + finally: + close() + + +def test_playwright_structural_survives_drift_where_visual_fails() -> None: + from openadapt_flow.validation.structural_action import run_probe + _pw_backend() + report = run_probe(n=9, headless=True) + # Structural resolves every target whose id is still in the DOM... + assert report["structural_ok"] == report["n"] + # ...and beats the visual ladder under drift (the whole thesis). + assert report["structural_ok"] > report["visual_ok"] + drifted = [t for t in report["targets"] if t["drifted"]] + assert drifted, "probe must include drifted targets" + assert all(t["structural_ok"] for t in drifted) + assert not any(t["visual_ok"] for t in drifted)