From 66852e046d534e4d1e5aaa1fd6fdbd29ab9db438 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 13 Jul 2026 10:58:25 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20structural=20(DOM/UIA)=20action=20r?= =?UTF-8?q?ung=20=E2=80=94=20vision-first,=20not=20vision-only?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make structural (DOM/accessibility) evidence a first-class ACTION rung — the deterministic top of the resolution ladder — not just an identity signal. On structure-bearing backends the runtime re-finds the recorded target as a DOM/UIA element and acts on its center deterministically, falling back to the visual ladder (template/ocr/geometry/grounder) only where structure is absent (pixel-only substrates: RDP/Citrix/canvas). Two external reviews + the desktop benchmark converge here: UIA execution 21/21 vs compiled visual replay 6/21. Ladder: API → tool/MCP → [structural DOM/UIA] → template → template_global → ocr → geometry → grounder(VLM) → human. `structural` is rung 0, above `ocr`, so an irreversible step may act on it (strongest evidence). The visual rungs are unchanged — the fallback floor for pixel-only substrates. - ir: StructuralLocator (selector / role+name / UIA AutomationId) on Anchor.structural; StructuralHandle; "structural" added to Rung. - backend: optional StructuralActionBackend protocol (structural_locator_at + locate_structural). - resolver: structural rung first; falls through unchanged on miss/pixel-only. - playwright/windows backends: DOM (#id / role+name, with an occlusion hit-test) and UIA (AutomationId / role+name) locate. - recorder/compiler: capture the locator at record time; keep the visual anchor. - replayer: structural resolution flows through the SAME click path, so the identity gate + risk gate still fire; exempt from healing (deterministic locate ≠ stale template). New use_structural flag (default True) lets the visual-floor characterization suites exercise the pixel-only path. Availability measured in benchmark/structural_action (21/21 vs 6/21). Identity gate proven to still abort a sibling on a structurally-resolved point. Occlusion safe-halt preserved. New coverage in tests/test_structural_rung.py and tests/e2e/test_structural_action.py. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CKrVJJy5jWVCkXAqgUqtqZ --- DESIGN.md | 36 +- .../structural_action/STRUCTURAL_ACTION.md | 52 ++ .../structural_action/structural_action.json | 177 +++++++ .../structural_action_probe.py | 39 ++ openadapt_flow/__main__.py | 6 + openadapt_flow/backend.py | 70 ++- openadapt_flow/backends/playwright_backend.py | 120 +++++ openadapt_flow/backends/windows_backend.py | 180 +++++++ openadapt_flow/compiler/compile.py | 14 + openadapt_flow/ir.py | 88 +++- openadapt_flow/recorder.py | 35 ++ openadapt_flow/runtime/replayer.py | 30 +- openadapt_flow/runtime/resolver.py | 65 ++- .../validation/structural_action.py | 215 +++++++++ tests/e2e/conftest.py | 9 +- tests/e2e/test_fault_model.py | 6 +- tests/e2e/test_structural_action.py | 93 ++++ tests/e2e/validation_utils.py | 6 +- tests/test_resolver.py | 8 +- tests/test_resolver_fuzz.py | 22 +- tests/test_structural_rung.py | 444 ++++++++++++++++++ 21 files changed, 1692 insertions(+), 23 deletions(-) create mode 100644 benchmark/structural_action/STRUCTURAL_ACTION.md create mode 100644 benchmark/structural_action/structural_action.json create mode 100644 benchmark/structural_action/structural_action_probe.py create mode 100644 openadapt_flow/validation/structural_action.py create mode 100644 tests/e2e/test_structural_action.py create mode 100644 tests/test_structural_rung.py 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 3b49e89..c7de23e 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -153,6 +153,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 ef59ae5..857d865 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.runtime.identity import band_region, context_from_lines, coverage @@ -868,6 +869,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, @@ -875,6 +888,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 cf6318a..f7448c1 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -61,14 +61,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( @@ -238,7 +301,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 a6fbeb4..b1cc7f0 100644 --- a/openadapt_flow/recorder.py +++ b/openadapt_flow/recorder.py @@ -33,6 +33,7 @@ from PIL import Image 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"]) ) @@ -171,6 +186,26 @@ def _record(self, event: dict[str, Any], act: Callable[[], None]) -> None: 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 + def _structured_identity_at(self, x: int, y: int) -> Optional[str]: """Structured (DOM / a11y) identity text under (x, y), if available. diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 1a97565..a17b6a2 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -112,6 +112,7 @@ def __init__( identity_vlm: Optional[Any] = None, state_verifier: 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 @@ -119,6 +120,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 # Optional local-VLM identity veto (tier 3 of the identity ladder), # OFF by default like the grounder: an IdentityVLM verifier or None. # When None the ladder runs structured-text + pixel-compare + OCR + @@ -420,13 +429,21 @@ 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). + # + # ``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. result.heal = self._heal_step( step, resolution, matched_region, before_png, workflow, run_dir, new_crops, @@ -466,6 +483,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, @@ -474,6 +501,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 e17c255..f1a57fc 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -107,7 +107,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) @@ -115,7 +122,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) From 0fb9c2a33423ddfb3b9773cf804be77200b432f7 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Mon, 13 Jul 2026 12:00:37 -0400 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20API/tool=20actuator=20tier=20?= =?UTF-8?q?=E2=80=94=20perform=20writes=20via=20API=20when=20available,=20?= =?UTF-8?q?GUI=20as=20fallback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The EXECUTE half of the capability ladder (the reviews' 'where a real API exists, GUI-driving it is the wrong tool'). When a step carries a reachable ApiBinding, actuate the write via the API deterministically, confirm it with the EffectVerifier (non-CONFIRMED -> HALT), and skip GUI actuation; otherwise fall through to the structural->visual ladder unchanged. Fail-safe: an attempted-but-unknown API outcome HALTS rather than risk a double-write. Additive (no binding -> replays as today). $0 / zero model calls. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CKrVJJy5jWVCkXAqgUqtqZ --- openadapt_flow/ir.py | 100 +++- openadapt_flow/runtime/actuators/__init__.py | 38 ++ openadapt_flow/runtime/actuators/api.py | 278 +++++++++++ openadapt_flow/runtime/replayer.py | 179 ++++++- tests/test_replayer_api_actuator.py | 487 +++++++++++++++++++ 5 files changed, 1071 insertions(+), 11 deletions(-) create mode 100644 openadapt_flow/runtime/actuators/__init__.py create mode 100644 openadapt_flow/runtime/actuators/api.py create mode 100644 tests/test_replayer_api_actuator.py diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index 6751ec8..c6e5dd5 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -20,7 +20,7 @@ from datetime import datetime, timezone from enum import Enum from pathlib import Path -from typing import TYPE_CHECKING, Literal, Optional +from typing import TYPE_CHECKING, Any, Literal, Optional from pydantic import BaseModel, Field @@ -233,6 +233,88 @@ class Postcondition(BaseModel): ) +class ApiBinding(BaseModel): + """A declarative API/tool call that performs a step's write WITHOUT the GUI. + + The TOP of the capability ladder (RFC ``docs/design/WORKFLOW_PROGRAM_IR.md`` + section 4, the ``api`` implementation of a ``TransitionContract``): where the + target app exposes a real API, driving its GUI to make the same write is the + wrong tool. When a step carries an ``ApiBinding`` AND the run configures an + :class:`~openadapt_flow.runtime.actuators.ApiActuator`, the runtime performs + the write by CALLING the API deterministically (``$0``, zero model calls), + confirms it with the same + :class:`~openadapt_flow.runtime.effects.EffectVerifier` that gates a GUI + write, and SKIPS the GUI resolution/act for that step. This is the + ``api`` leaf of the same contract the structural rung realizes as ``dom_uia`` + and the visual ladder realizes as ``vision_rdp`` -- one semantic effect, + backend-specific implementation. + + ADDITIVE and back-compatible: the field is optional and defaults absent, so a + bundle carrying no binding replays EXACTLY as today (GUI actuation). A binding + present with no actuator configured also falls through to the GUI ladder -- + the API tier is an OPTIMIZATION whose safe fallback is the GUI, never a gate + that can block a runnable step. + + Fields are REST/JSON-first but shaped so a FHIR / MCP / tool binding fits the + same model (``kind`` selects the substrate; a FHIR resource POST, an MCP tool + invocation, and a plain REST write all reduce to method + endpoint + body + + the expected effect). Placeholders ``{param}`` in the URL / query / body are + substituted from the run's typed params (``Workflow.params`` overlaid by the + caller's values) at actuation time. + """ + + kind: Literal["rest", "fhir", "mcp", "tool"] = Field( + default="rest", + description="Substrate: 'rest'/'fhir' HTTP, or an 'mcp'/'tool' call", + ) + method: str = Field( + default="POST", + description="HTTP verb (REST/FHIR) or logical operation name (mcp/tool)", + ) + url_template: str = Field( + description=( + "Endpoint template; absolute (http...) or relative to the" + " actuator's base_url. `{param}` placeholders are substituted from" + " the run's typed params." + ), + ) + body_template: dict[str, Any] = Field( + default_factory=dict, + description=( + "JSON request body template; string leaves may carry `{param}`" + " placeholders substituted from the run's params." + ), + ) + query: dict[str, str] = Field( + default_factory=dict, + description="Query-string params; values may carry `{param}` too", + ) + headers: dict[str, str] = Field( + default_factory=dict, + description="Extra request headers; values may carry `{param}`", + ) + expected_status: list[int] = Field( + default_factory=list, + description=( + "Explicit acceptable HTTP status codes; empty means any 2xx is" + " success (anything else is treated as an attempted write of" + " unknown effect and HALTs -- never GUI-retried)." + ), + ) + timeout_s: float = Field( + default=5.0, description="Per-request timeout in seconds" + ) + effects: list["Effect"] = Field( + default_factory=list, + description=( + "The system-of-record effect(s) this call is expected to produce." + " Used to CONFIRM the API write via the run's EffectVerifier when" + " the step itself declares no `effects` (an API write must be" + " confirmable, exactly as a GUI write with declared effects is)." + ), + ) + + class Step(BaseModel): id: str intent: str = Field(description="Human-readable purpose of the step") @@ -256,6 +338,16 @@ class Step(BaseModel): # before, and a declared effect with no verifier configured is a # deployment error that HALTS (never a silent unverifiable write). effects: list["Effect"] = Field(default_factory=list) + # API/tool binding (RFC section 4, the `api` implementation of the + # transition contract): a declarative description of the API call that + # performs THIS step's write. When present AND the run configures an + # ApiActuator, the runtime performs the write via the API (deterministic, + # $0, no model), confirms it with the EffectVerifier, and SKIPS the GUI + # resolve/act for this step (see openadapt_flow.runtime.replayer). Additive + # and back-compatible: None (default) means the step actuates through the + # GUI resolution ladder EXACTLY as today; a binding present with no actuator + # configured also falls through to the GUI (the API tier's safe fallback). + api_binding: Optional[ApiBinding] = None risk: Literal["reversible", "irreversible"] = "reversible" timeout_s: float = 10.0 # Identity-protection audit trail (clicks and anchored TYPE steps): @@ -416,6 +508,11 @@ class StepResult(BaseModel): # calls on this path — effect verification reads the system of record. effect_verified: Optional[bool] = None effect_results: list[str] = Field(default_factory=list) + # How this step's write was PERFORMED: "api" when actuated via an + # ApiBinding (GUI resolve/act skipped), None when it went through the GUI + # resolution ladder (the default). Diagnostic/audit — lets an operator see + # which steps ran on the deterministic API tier vs the visual floor. + actuation: Optional[str] = None # Drift-oracle: postconditions that deterministically FAILED but were # confirmed by the optional on-prem VLM state-verifier under render drift # (recorded for audit; empty unless an appliance is configured). @@ -477,5 +574,6 @@ def save(self, run_dir: Path | str) -> Path: # the forward reference; bundles with no effects are unaffected. from openadapt_flow.runtime.effects.effect import Effect # noqa: E402,F401 +ApiBinding.model_rebuild() Step.model_rebuild() Workflow.model_rebuild() diff --git a/openadapt_flow/runtime/actuators/__init__.py b/openadapt_flow/runtime/actuators/__init__.py new file mode 100644 index 0000000..1b04fc0 --- /dev/null +++ b/openadapt_flow/runtime/actuators/__init__.py @@ -0,0 +1,38 @@ +"""Actuators: PERFORM a step's write through a non-GUI channel. + +The runtime's default actuator is the GUI: it resolves the recorded target on +the live screen (:mod:`openadapt_flow.runtime.resolver`) and clicks / types +through the :class:`~openadapt_flow.backend.Backend`. That is the FLOOR -- it +works on any pixel surface (RDP/Citrix/canvas) -- but it is also the *weakest* +and most expensive way to effect a change: where the target app exposes a real +API, driving its GUI to make the same write is the wrong tool. + +This package adds the TOP of the capability ladder (RFC +``docs/design/WORKFLOW_PROGRAM_IR.md`` section 4, the ``api`` implementation of +a ``TransitionContract``): when a step carries an +:class:`~openadapt_flow.ir.ApiBinding`, perform the write by CALLING the API +deterministically -- $0, zero model calls -- and confirm it with the same +:class:`~openadapt_flow.runtime.effects.EffectVerifier` that gates a GUI write. +The GUI resolution ladder is then SKIPPED for that step. A step with no binding +(or with no actuator configured) behaves EXACTLY as before -- the API tier is +additive and falls through to the structural -> visual ladder. + +Public surface: + +- :class:`ApiActuator` -- the REST/JSON actuator (and the shape a FHIR/MCP/tool + actuator slots into). +- :class:`ApiActuationResult`, :class:`ActuationStatus` -- the fail-safe + outcome of an actuation attempt (the no-double-write contract). +""" + +from openadapt_flow.runtime.actuators.api import ( # noqa: F401 + ActuationStatus, + ApiActuationResult, + ApiActuator, +) + +__all__ = [ + "ApiActuator", + "ApiActuationResult", + "ActuationStatus", +] diff --git a/openadapt_flow/runtime/actuators/api.py b/openadapt_flow/runtime/actuators/api.py new file mode 100644 index 0000000..f45ce99 --- /dev/null +++ b/openadapt_flow/runtime/actuators/api.py @@ -0,0 +1,278 @@ +"""REST/JSON :class:`ApiActuator` -- perform a step's write via its API binding. + +This is the ``api`` implementation tier of the RFC's transition contract +(``docs/design/WORKFLOW_PROGRAM_IR.md`` section 4: "call the app's API / DB +write; effect probed against the system of record"). Given a step's +:class:`~openadapt_flow.ir.ApiBinding` and the run's typed params, it +substitutes the params into the endpoint / query / body templates and issues a +single deterministic HTTP request -- no pixel matching, no model call, ``$0``. + +The hard requirement is **never double-write the same effect**. A naive +"try the API, and on any failure fall back to GUI-clicking the same save" +would, on a request that the server actually PROCESSED (a read-timeout after +commit, a 5xx after a partial write), perform the write TWICE. So the actuator +classifies every attempt into exactly one of three fail-safe outcomes +(:class:`ActuationStatus`), keyed on *whether the request could have reached +the server*: + +- :attr:`ActuationStatus.UNAVAILABLE` -- the request was **never sent** (the + TCP connection was never established: connection refused, DNS failure, + connect-timeout) or the binding could not even be built (a param the URL/body + needs was not supplied). Nothing was written, so it is SAFE for the caller to + fall through to the GUI ladder for this step. This is the "reachable + ApiBinding" gate: an unreachable endpoint simply is not actuated. +- :attr:`ActuationStatus.ACTUATED` -- the request was sent and the server + returned success (2xx, or an explicitly-allowed status). The write was + performed; the caller MUST now confirm it with the EffectVerifier and MUST + skip the GUI (never re-do the write). +- :attr:`ActuationStatus.HALT` -- the request WAS sent but its outcome is + unknown or a rejection (read-timeout after the bytes went out, a non-2xx + response, any post-send transport error). The write MAY have landed, so the + caller must NEITHER accept it as success NOR GUI-write it again -- it HALTs + (the same refuse-rather-than-guess posture as the EffectVerifier's + INDETERMINATE verdict). + +The connect-phase / read-phase split is exact in ``requests``: +``ConnectTimeout`` subclasses ``ConnectionError`` (nothing sent -> UNAVAILABLE) +while ``ReadTimeout`` does not (bytes sent -> HALT), so catching +``ConnectionError`` before ``Timeout`` gives the right classification. + +Import-light: ``requests`` is imported lazily so importing this module (and the +runtime package) stays cheap and model-free. +""" + +from __future__ import annotations + +from enum import Enum +from typing import Any, Optional + +from pydantic import BaseModel + +from openadapt_flow.ir import ApiBinding + + +class ActuationStatus(str, Enum): + """The fail-safe outcome of an API actuation attempt (no-double-write).""" + + #: The write was performed and the server acknowledged success -> the + #: caller confirms it with the EffectVerifier and SKIPS the GUI. + ACTUATED = "actuated" + #: The request was never sent (endpoint unreachable, or the binding could + #: not be built) -> nothing was written; SAFE to fall through to the GUI + #: ladder for this step. + UNAVAILABLE = "unavailable" + #: The request WAS sent but its outcome is unknown or a rejection -> the + #: write may have landed; HALT (never accept, never GUI-write it again). + HALT = "halt" + + +class ApiActuationResult(BaseModel): + """Outcome of one :meth:`ApiActuator.actuate` call.""" + + status: ActuationStatus + substrate: str = "rest" + #: Human-readable reason (audit trail / error surface). Contains the + #: UNSUBSTITUTED method + endpoint template only -- never the substituted + #: values -- so it is safe to log without leaking PHI-bearing params. + reason: str = "" + #: HTTP status code when a response was received; None otherwise. + http_status: Optional[int] = None + #: ``"METHOD url_template"`` (unsubstituted) for the audit line. + request_summary: str = "" + + @property + def actuated(self) -> bool: + return self.status is ActuationStatus.ACTUATED + + @property + def should_fall_through(self) -> bool: + """True when the caller may safely fall through to the GUI ladder + (the request was never sent, so nothing was written).""" + return self.status is ActuationStatus.UNAVAILABLE + + @property + def should_halt(self) -> bool: + return self.status is ActuationStatus.HALT + + +class _MissingParam(KeyError): + """A template referenced a param the run did not supply.""" + + +class _StrictMap(dict): + def __missing__(self, key: str) -> Any: # noqa: D401 + raise _MissingParam(key) + + +def _fill(template: str, params: dict[str, str]) -> str: + """Substitute ``{param}`` placeholders in ``template`` from ``params``. + + Raises :class:`_MissingParam` when the template references a key that is + not in ``params`` -- the binding cannot be built, so the actuator reports + UNAVAILABLE (a before-send problem: nothing is written, GUI fallback is + safe) rather than sending a half-formed request. + """ + return template.format_map(_StrictMap(params)) + + +def _fill_body(node: Any, params: dict[str, str]) -> Any: + """Recursively substitute ``{param}`` in every string leaf of a JSON body.""" + if isinstance(node, str): + return _fill(node, params) + if isinstance(node, dict): + return {k: _fill_body(v, params) for k, v in node.items()} + if isinstance(node, list): + return [_fill_body(v, params) for v in node] + return node + + +class ApiActuator: + """Perform a step's write via its :class:`~openadapt_flow.ir.ApiBinding`. + + Bound to a deployment's API base URL (and an optional injected session for + auth headers / tests), mirroring + :class:`~openadapt_flow.runtime.effects.rest.RestRecordVerifier`. The + binding's ``url_template`` may be absolute (``http...``) or relative to + ``base_url``; params from the run substitute into the URL, query, and body + templates. A single request is issued and classified into the fail-safe + :class:`ActuationStatus` outcome. Makes ZERO model calls. + + Args: + base_url: Deployment API base URL (used for relative ``url_template``s; + trailing slash optional). May be empty when every binding is + absolute. + session: Optional ``requests``-style session (auth headers / test + injection); a module-level default is created lazily when omitted. + timeout_s: Default per-request timeout when the binding sets none. + """ + + substrate = "rest" + + def __init__( + self, + base_url: str = "", + *, + session: Any = None, + timeout_s: float = 5.0, + ) -> None: + self.base_url = base_url.rstrip("/") + self.default_timeout_s = timeout_s + self._session = session + + def _get_session(self) -> Any: + if self._session is None: + import requests # lazy: keep module import light and model-free + + self._session = requests.Session() + return self._session + + def _resolve_url(self, url_template: str) -> str: + if url_template.startswith(("http://", "https://")): + return url_template + return f"{self.base_url}{url_template}" + + def actuate( + self, binding: ApiBinding, params: dict[str, str] + ) -> ApiActuationResult: + """Perform ``binding``'s write, substituting ``params``; classify safely. + + Returns an :class:`ApiActuationResult` whose :attr:`status` tells the + caller exactly one safe next move: confirm-and-skip-GUI (ACTUATED), + fall-through-to-GUI (UNAVAILABLE, nothing was written), or HALT + (attempted, outcome unknown -- never double-write). Never raises. + """ + summary = f"{binding.method} {binding.url_template}" + + # -- build the request (a before-send problem is UNAVAILABLE) --------- + try: + url = self._resolve_url(_fill(binding.url_template, params)) + query = {k: _fill(v, params) for k, v in binding.query.items()} + body = _fill_body(binding.body_template, params) + headers = {k: _fill(v, params) for k, v in binding.headers.items()} + except _MissingParam as exc: + return ApiActuationResult( + status=ActuationStatus.UNAVAILABLE, + substrate=self.substrate, + reason=( + f"binding for {summary} references param {exc} not supplied " + "by the run -- API tier unavailable, falling through to GUI" + ), + request_summary=summary, + ) + + import requests # lazy; hierarchy: ConnectTimeout is a ConnectionError + + timeout = binding.timeout_s or self.default_timeout_s + try: + resp = self._get_session().request( + binding.method.upper(), + url, + params=query or None, + json=body if body else None, + headers=headers or None, + timeout=timeout, + ) + except requests.exceptions.ConnectionError as exc: + # Connection never established (refused / DNS / connect-timeout): + # the request was NEVER sent, so nothing was written -> it is safe + # to fall through to the GUI ladder for this step. + return ApiActuationResult( + status=ActuationStatus.UNAVAILABLE, + substrate=self.substrate, + reason=( + f"endpoint unreachable ({type(exc).__name__}) -- request " + "not sent, API tier unavailable, falling through to GUI" + ), + request_summary=summary, + ) + except requests.exceptions.Timeout as exc: + # Read-timeout: the bytes WENT OUT and the server may have processed + # the write. Outcome unknown -> HALT (never GUI-write it again). + return ApiActuationResult( + status=ActuationStatus.HALT, + substrate=self.substrate, + reason=( + f"request sent but timed out awaiting the response " + f"({type(exc).__name__}) -- the write may have landed; HALT " + "(never double-write via the GUI)" + ), + request_summary=summary, + ) + except Exception as exc: # noqa: BLE001 + # Any other transport error after the request left the client is of + # unknown effect on the server -> HALT rather than risk a duplicate. + return ApiActuationResult( + status=ActuationStatus.HALT, + substrate=self.substrate, + reason=( + f"request failed after being sent ({type(exc).__name__}) -- " + "outcome unknown; HALT (never double-write via the GUI)" + ), + request_summary=summary, + ) + + allowed = binding.expected_status or list(range(200, 300)) + ok = resp.status_code in allowed or ( + not binding.expected_status and resp.status_code // 100 == 2 + ) + if ok: + return ApiActuationResult( + status=ActuationStatus.ACTUATED, + substrate=self.substrate, + reason=f"{summary} -> {resp.status_code}", + http_status=resp.status_code, + request_summary=summary, + ) + # Non-success response: the request was PROCESSED by the server. Even a + # clean rejection is ambiguous about what (if anything) persisted, and + # re-driving the same write through the GUI risks a duplicate -> HALT. + return ApiActuationResult( + status=ActuationStatus.HALT, + substrate=self.substrate, + reason=( + f"{summary} returned {resp.status_code} (not success) -- the " + "write was attempted; HALT (never double-write via the GUI)" + ), + http_status=resp.status_code, + request_summary=summary, + ) diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 14f0102..45727f4 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -56,6 +56,7 @@ from openadapt_flow.runtime import heal as heal_mod from openadapt_flow.runtime import identity as identity_mod from openadapt_flow.runtime.effects import ( + Effect, EffectState, EffectVerifier, reconcile_or_escalate, @@ -133,6 +134,15 @@ class Replayer: is REFUTED as a duplicate — deletes the extras and re-verifies (RECONCILED continues, else ESCALATE halts). None (default) means every refuted/indeterminate effect simply halts. + api_actuator: Optional ``ApiActuator`` (``runtime.actuators``) bound to + the deployment's API. It is the TOP of the capability ladder: when a + step carries an ``ir.Step.api_binding`` and this actuator is set, + the step's write is PERFORMED via the API (deterministic, $0, no + model), CONFIRMED by the ``effect_verifier``, and the GUI + resolve/act is SKIPPED. None (default) disables the API tier -- a + step with a binding then actuates through the GUI ladder exactly as + today (the API tier is an optimization whose safe fallback IS the + GUI). Never makes a model call. poll_interval_s: Postcondition polling interval in seconds. """ @@ -146,6 +156,7 @@ def __init__( state_verifier: Optional[Any] = None, effect_verifier: Optional[EffectVerifier] = None, effect_compensator: Optional[Any] = None, + api_actuator: Optional[Any] = None, poll_interval_s: float = 0.05, use_structural: bool = True, ) -> None: @@ -161,6 +172,13 @@ def __init__( # makes a model call — the $0 runtime guarantee is preserved. self.effect_verifier = effect_verifier self.effect_compensator = effect_compensator + # API/tool actuator -- the TOP of the capability ladder (RFC section 4 + # `api` tier). When set, a step carrying an `api_binding` has its write + # performed via the API and confirmed by the effect_verifier, SKIPPING + # the GUI resolve/act. None (default) = the API tier is off and such a + # step falls through to the GUI ladder unchanged. Deterministic; makes + # no model call (the $0 runtime guarantee holds on this path too). + self.api_actuator = api_actuator # 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 @@ -250,6 +268,12 @@ def run( report.rung_counts[rung] = report.rung_counts.get(rung, 0) + 1 if rung == "grounder": report.model_calls += 1 + elif result.ok and result.actuation == "api": + # API-tier actuation has no visual resolution rung; count it + # under "api" so the run report shows the deterministic top of + # the ladder in the same place as the visual rungs. Zero model + # calls (the $0 guarantee holds on the API path too). + report.rung_counts["api"] = report.rung_counts.get("api", 0) + 1 # Drift-oracle state-verifier calls are model calls too (honest # accounting: a rescued run is not a zero-model run). report.model_calls += result.drift_oracle_calls @@ -326,6 +350,20 @@ def _run_step( t0 = time.monotonic() result = StepResult(step_id=step.id, intent=step.intent, ok=False) + # API/tool tier -- the TOP of the capability ladder. When the step + # carries an api_binding and an ApiActuator is configured, PERFORM the + # write via the API (deterministic, $0, no GUI), CONFIRM it with the + # EffectVerifier, and SKIP the GUI resolve/act entirely. Returns True + # when the API tier took responsibility (actuated+verified, or HALTed); + # returns False (API tier unavailable -- endpoint unreachable / no + # binding param) to fall through to the GUI ladder below with NO write + # yet performed (the no-double-write contract). See + # openadapt_flow.runtime.actuators. + if self.api_actuator is not None and step.api_binding is not None: + if self._try_api_tier(step, params, result): + result.elapsed_ms = (time.monotonic() - t0) * 1000.0 + return result + # Settle before the pre-action screenshot. before_png = self.vision.wait_settled(self.backend) result.before_png = self._save_step_png(run_dir, step.id, "before", before_png) @@ -532,6 +570,115 @@ def _run_step( result.elapsed_ms = (time.monotonic() - t0) * 1000.0 return result + # -- API/tool actuation tier (top of the capability ladder) ----------------- + + def _try_api_tier( + self, + step: Step, + params: dict[str, str], + result: StepResult, + ) -> bool: + """Perform ``step.api_binding``'s write via the API and confirm it. + + The TOP of the capability ladder (RFC section 4 ``api`` tier): a + deterministic, ``$0``, model-free write, gated by the SAME + ``EffectVerifier`` as a GUI write. For an API write the target + "identity" is the explicit API parameter -- stronger than a resolved + pixel band -- so no identity gate is weakened by skipping the GUI. + + Fail-safe ordering (the no-double-write contract): + + - refuse BEFORE any request when the write could not be confirmed (no + effect contract, or no verifier) -- an unverifiable consequential + write never proceeds; + - snapshot the system of record, then actuate ONCE; + - UNAVAILABLE (request never sent) -> return False so the caller falls + through to the GUI ladder; nothing was written, so no double-write; + - HALT (request sent, outcome unknown / rejected) -> stop the run; the + write may have landed, so it is NEVER re-done through the GUI; + - ACTUATED (2xx) -> CONFIRM with the EffectVerifier; a non-CONFIRMED + verdict HALTs exactly as it would for a GUI write. + + Returns True when the API tier took responsibility for the step (the + result is final -- actuated+verified, HALTed, or a config-error HALT), + False when the API tier is UNAVAILABLE and the caller must fall through + to the GUI resolution ladder. + """ + binding = step.api_binding + assert binding is not None # guaranteed by the caller + + # An API write MUST be confirmable against the system of record -- + # exactly as a GUI write that declares effects must be. The binding may + # carry its own effect contract; the step's own effects take precedence. + effects = step.effects or binding.effects + if not effects: + result.effect_verified = False + result.effect_results.append( + "API binding declares no effect to confirm the write (neither " + "step.effects nor api_binding.effects) -- refusing an " + "unverifiable API write (fail-safe HALT)" + ) + result.ok = False + result.error = ( + f"Step '{step.id}' ({step.intent}) has an API binding but no " + "system-of-record effect to CONFIRM the write -- an API write " + "must be verifiable; refusing to actuate " + "(deployment/configuration error); run aborted" + ) + return True + if self.effect_verifier is None: + result.effect_verified = False + result.effect_results.append( + "API binding present but no EffectVerifier is configured to " + "confirm the write (fail-safe HALT)" + ) + result.ok = False + result.error = ( + f"Step '{step.id}' ({step.intent}) would actuate via the API " + "but no EffectVerifier is configured to confirm the write -- " + "refusing an unverifiable consequential write; run aborted" + ) + return True + + # Snapshot the system of record BEFORE the write so the verifier counts + # only what THIS actuation wrote (delta / at-most-once / collateral + # loss), then actuate exactly once. + before = self.effect_verifier.capture_pre_state() + outcome = self.api_actuator.actuate(binding, params) + + from openadapt_flow.runtime.actuators import ActuationStatus + + if outcome.status == ActuationStatus.UNAVAILABLE: + # The request was NEVER sent -- nothing was written. Fall through to + # the GUI ladder for this step (no double-write risk). The GUI path + # populates the result; leave only an audit breadcrumb here. + result.effect_results.append(f"[api] {outcome.reason}") + return False + + # From here the request WAS attempted -- this step is API-tier and is + # NEVER also GUI-written (the no-double-write contract). + result.actuation = "api" + + if outcome.status == ActuationStatus.HALT: + result.effect_verified = False + result.effect_results.append(f"[api] {outcome.reason}") + result.ok = False + result.error = ( + f"API actuation HALTED step '{step.id}' ({step.intent}): " + f"{outcome.reason} -- run aborted" + ) + return True + + # ACTUATED (2xx): confirm the write against the system of record with + # the same EffectVerifier that gates a GUI write. A non-CONFIRMED + # verdict HALTs. The GUI resolve/act (and screen postconditions) are + # SKIPPED -- the record, not the screen, is the oracle for an API write. + result.effect_results.append(f"[api] actuated {outcome.reason}") + error = self._verify_effects(step, before, result, effects=effects) + result.ok = error is None + result.error = error + return True + # -- system-of-record effect verification ----------------------------------- def _verify_effects( @@ -539,25 +686,37 @@ def _verify_effects( step: Step, before: EffectState, result: StepResult, + effects: Optional[list["Effect"]] = None, ) -> Optional[str]: """Verify each declared Effect against the system of record; HALT on any non-CONFIRMED verdict. - Runs AFTER the action and its screen postconditions. For each - ``step.effects`` entry the configured verifier rules CONFIRMED / - REFUTED / INDETERMINATE against the REAL record (an API/DB read, never - the screen). A CONFIRMED effect proceeds; a non-CONFIRMED effect halts, - except that an IRREVERSIBLE effect is first routed through - ``reconcile_or_escalate`` -- a duplicate the configured compensator can - undo is RECONCILED (proceed), everything else ESCALATEs (halt). Every - verdict is recorded on ``result.effect_results`` for the audit trail. - Makes ZERO model calls (the $0 runtime guarantee). + Runs AFTER the action and its screen postconditions (for a GUI write), + or immediately after an API actuation. For each effect the configured + verifier rules CONFIRMED / REFUTED / INDETERMINATE against the REAL + record (an API/DB read, never the screen). A CONFIRMED effect proceeds; + a non-CONFIRMED effect halts, except that an IRREVERSIBLE effect is + first routed through ``reconcile_or_escalate`` -- a duplicate the + configured compensator can undo is RECONCILED (proceed), everything else + ESCALATEs (halt). Every verdict is recorded on ``result.effect_results`` + for the audit trail. Makes ZERO model calls (the $0 runtime guarantee). + + Args: + step: The step being verified (for the audit/error text). + before: The pre-action snapshot of the system of record. + result: The step result to record verdicts on. + effects: The effects to verify; defaults to ``step.effects`` (the + GUI path). The API tier passes ``step.effects or + api_binding.effects`` so a self-contained binding carries its + own confirmation contract. Returns an error string (HALT) or None (all effects confirmed or reconciled). """ assert self.effect_verifier is not None # guaranteed by the caller - for effect in step.effects: + if effects is None: + effects = step.effects + for effect in effects: verdict = self.effect_verifier.verify(effect, before) if verdict.confirmed: result.effect_results.append( diff --git a/tests/test_replayer_api_actuator.py b/tests/test_replayer_api_actuator.py new file mode 100644 index 0000000..24cd66a --- /dev/null +++ b/tests/test_replayer_api_actuator.py @@ -0,0 +1,487 @@ +"""Live-local tests for the API/tool actuator tier (top of the capability ladder). + +These drive the REAL Replayer wired with a REAL +:class:`~openadapt_flow.runtime.actuators.ApiActuator` against a REAL system of +record (the in-process MockMed ``fault_server``), confirmed by the same +:class:`~openadapt_flow.runtime.effects.RestRecordVerifier` the GUI-write +effects tests use. No model calls, no network beyond localhost -- runs in CI. + +The theses these pin (RFC ``docs/design/WORKFLOW_PROGRAM_IR.md`` section 4, the +``api`` implementation of a transition contract): + +- a step with a REACHABLE ``ApiBinding`` performs its write via the API, the + EffectVerifier CONFIRMS it against the record, and the GUI actuation is + SKIPPED entirely -- ``$0``, zero model calls; +- an UNREACHABLE API falls through to the GUI ladder CLEANLY, with NO + double-write (the request never left the client, so nothing was written); +- a step with NO binding replays byte-identically to today (back-compat); +- a REFUTED effect after an API write HALTS (the record, not the screen, is the + oracle); +- an API write whose outcome is unknown / rejected HALTs (never GUI-retried). +""" + +from __future__ import annotations + +import requests +from openadapt_flow.runtime.actuators import ActuationStatus, ApiActuator +from openadapt_flow.runtime.effects import ( + Effect, + EffectKind, + RestRecordVerifier, +) + +# Reuse the scripted fakes from the main replayer unit tests (pytest's prepend +# import mode puts tests/ on sys.path). +from test_replayer import FakeBackend, FakeVision, Match + +from openadapt_flow.ir import ( + ActionKind, + ApiBinding, + Postcondition, + PostconditionKind, + Step, + Workflow, +) +from openadapt_flow.mockmed.fault_server import serve as fault_serve +from openadapt_flow.runtime.replayer import Replayer + +TARGET = {"patient_id": "p1", "type": "Triage"} + + +class GuiWritingBackend(FakeBackend): + """A GUI backend whose ``press`` writes to the system of record. + + Models the CURRENT (GUI) path: the consequential keypress makes the app + POST an encounter. Used to prove the API tier SKIPS the GUI (no press + lands) on the actuated path, and to prove the fall-through path DOES + GUI-write when the API tier is unavailable. ``record_presses`` records + every ``press`` so a test can assert the GUI was or was not driven. + """ + + def __init__(self, sor_url, *, viewport=(300, 200)): + super().__init__(viewport=viewport) + self.sor_url = sor_url.rstrip("/") + + def press(self, key): + super().press(key) + requests.post( + f"{self.sor_url}/api/encounter", + json={"patient_id": "p1", "type": "Triage", "note": "gui"}, + timeout=5, + ) + + +def _fault_server(): + url, db, stop = fault_serve() + return url.rstrip("/"), db, stop + + +def _api_save_workflow( + *, url_template="/api/encounter", effects, risk="reversible", effects_on="step" +): + """A one-step workflow: press Enter (the GUI action), but carrying an + ApiBinding so the API tier performs the write instead. The screen + postcondition PASSES -- the point is that the API tier bypasses it. + + ``effects_on`` places the effect contract on the ``"step"`` (the canonical + location, verified on BOTH the API tier and a GUI fall-through) or on the + ``"binding"`` (the self-contained-binding case, used by the API tier when + the step declares none).""" + step_effects = effects if effects_on == "step" else [] + binding_effects = effects if effects_on == "binding" else [] + return Workflow( + name="api-save", + steps=[ + Step( + id="save", + intent="save encounter", + action=ActionKind.KEY, + key="Enter", + expect=[ + Postcondition( + kind=PostconditionKind.TEXT_PRESENT, + text="Saved", + timeout_s=0.2, + ) + ], + risk=risk, + effects=step_effects, + api_binding=ApiBinding( + method="POST", + url_template=url_template, + body_template={ + "patient_id": "p1", + "type": "Triage", + "note": "{note}", + }, + effects=binding_effects, + timeout_s=2.0, + ), + ) + ], + params={"note": "charted via API"}, + ) + + +def _vision_that_confirms_saved(): + vision = FakeVision() + vision.text_results = { + "Saved": Match(point=(50, 10), region=(30, 5, 40, 10), confidence=0.9) + } + return vision + + +def _dirs(tmp_path): + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + return bundle, tmp_path / "run" + + +def _record_written(**over): + kw = dict(kind=EffectKind.RECORD_WRITTEN, match=TARGET, expected_count=1, + timeout_s=2.0) + kw.update(over) + return Effect(**kw) + + +# -- ACTUATED + CONFIRMED: API performs the write, GUI is skipped ----------- + + +def test_api_binding_actuates_and_confirms_skipping_gui(tmp_path): + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + # The effect contract rides on the BINDING itself (self-contained) to + # prove the API tier confirms even when the step declares no effects. + workflow = _api_save_workflow( + effects=[_record_written()], effects_on="binding" + ) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is True + r = report.results[0] + # The write landed via the API and was CONFIRMED against the record. + assert r.actuation == "api" + assert r.effect_verified is True + assert any("CONFIRMED" in line for line in r.effect_results) + assert any("actuated" in line for line in r.effect_results) + # The GUI was SKIPPED: no click/type/press was ever issued. + assert backend.actions == [] + # Exactly one record, written by the API with the run's param value. + records = db.snapshot()["records"] + assert len(records) == 1 + assert records[0]["note"] == "charted via API" + # $0 guarantee: the API path makes no model calls. + assert report.model_calls == 0 + assert report.est_model_cost_usd == 0.0 + # Audit: the run report counts the deterministic top-of-ladder tier. + assert report.rung_counts.get("api") == 1 + finally: + stop() + + +# -- UNREACHABLE API falls through to the GUI ladder cleanly (no double-write) -- + + +def test_unreachable_api_falls_through_to_gui_no_double_write(tmp_path): + url, db, stop = _fault_server() + try: + # The actuator points at a DEAD endpoint (connection refused): the + # request is never sent, so the API tier is UNAVAILABLE and the step + # falls through to the GUI, which performs the write exactly once. + dead = "http://127.0.0.1:1" # nothing listens here -> ConnectionError + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow(effects=[_record_written()]) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(dead, timeout_s=1.0), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is True + r = report.results[0] + # Fell through to the GUI: actuation is NOT "api"; the keypress fired. + assert r.actuation is None + assert ("press", "Enter") in backend.actions + # An audit breadcrumb records the API tier was unavailable. + assert any("unavailable" in line.lower() for line in r.effect_results) + # The write happened EXACTLY ONCE (no double-write): the GUI wrote it, + # the dead API did not, and the effect check CONFIRMS the single row. + assert len(db.snapshot()["records"]) == 1 + assert r.effect_verified is True + assert report.model_calls == 0 + finally: + stop() + + +# -- REFUTED effect after an API write HALTS --------------------------------- + + +def test_api_write_refuted_by_record_halts(tmp_path): + url, db, stop = _fault_server() + try: + # The API write lands ONE row, but the effect asserts a record for a + # DIFFERENT patient (p2) -- the record refutes it (0 found, expected 1) + # -> HALT, even though the API returned 2xx. + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow( + effects=[_record_written(match={"patient_id": "p2", "type": "Triage"})] + ) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + r = report.results[0] + assert r.actuation == "api" + assert r.effect_verified is False + assert r.ok is False + assert "refuted" in (r.error or "").lower() + assert "system of record" in (r.error or "") + # The API DID write one (p1) row -- the write was performed; it is the + # RECORD check that refused it, and the GUI never ran (no double-write). + assert backend.actions == [] + assert len(db.snapshot()["records"]) == 1 + finally: + stop() + + +# -- ATTEMPTED-but-rejected API write HALTS (never GUI-retried) -------------- + + +def test_api_non_2xx_halts_never_double_writes(tmp_path): + url, db, stop = _fault_server() + try: + # ?fault=session makes /api/encounter return 401: the request WAS sent + # (nothing persisted here, but that is not knowable in general), so the + # actuator must HALT rather than fall through and GUI-write a possible + # duplicate. + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow( + url_template="/api/encounter?fault=session", effects=[_record_written()] + ) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + r = report.results[0] + assert r.actuation == "api" + assert r.ok is False + assert "halted" in (r.error or "").lower() + # The GUI was NEVER driven -- the attempted API write is not re-done. + assert backend.actions == [] + assert db.snapshot()["records"] == [] + finally: + stop() + + +# -- FAIL-SAFE: an API binding with no effect to confirm the write -> HALT ---- + + +def test_api_binding_without_effects_is_config_error_halt(tmp_path): + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + # An ApiBinding with NO effects (neither on the step nor the binding): + # the write could not be confirmed, so it must be refused BEFORE any + # request is sent. + workflow = _api_save_workflow(effects=[]) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + r = report.results[0] + assert r.ok is False + assert "must be verifiable" in (r.error or "") + # Nothing was written and the GUI never ran -- refused before actuating. + assert backend.actions == [] + assert db.snapshot()["records"] == [] + finally: + stop() + + +# -- FAIL-SAFE: an API binding but no EffectVerifier configured -> HALT ------- + + +def test_api_binding_without_verifier_halts(tmp_path): + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow(effects=[_record_written()]) + bundle, run_dir = _dirs(tmp_path) + # An ApiActuator but NO EffectVerifier: an API write we cannot confirm. + replayer = Replayer( + backend, + vision=vision, + api_actuator=ApiActuator(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is False + r = report.results[0] + assert r.ok is False + assert "no EffectVerifier" in (r.error or "") + assert backend.actions == [] + assert db.snapshot()["records"] == [] + finally: + stop() + + +# -- BACK-COMPAT: a binding present but NO actuator configured -> GUI path ---- + + +def test_binding_present_but_no_actuator_uses_gui_unchanged(tmp_path): + url, db, stop = _fault_server() + try: + backend = GuiWritingBackend(url) + vision = _vision_that_confirms_saved() + workflow = _api_save_workflow(effects=[_record_written()]) + bundle, run_dir = _dirs(tmp_path) + # NO api_actuator -> the API tier is OFF; the step actuates via the GUI + # exactly as today (the binding is inert, its declared effects are still + # verified by the normal GUI effect path). + replayer = Replayer( + backend, + vision=vision, + effect_verifier=RestRecordVerifier(url), + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is True + r = report.results[0] + assert r.actuation is None + assert ("press", "Enter") in backend.actions + assert r.effect_verified is True + assert len(db.snapshot()["records"]) == 1 + finally: + stop() + + +# -- BACK-COMPAT: a no-binding, no-effects bundle replays byte-identically ---- + + +def test_no_binding_bundle_replays_unchanged(tmp_path): + # A plain step with no api_binding and no effects: no API machinery engages, + # the GUI runs, and the result carries no actuation marker. + backend = FakeBackend() + vision = _vision_that_confirms_saved() + workflow = Workflow( + name="plain", + steps=[ + Step( + id="save", + intent="save", + action=ActionKind.KEY, + key="Enter", + expect=[ + Postcondition( + kind=PostconditionKind.TEXT_PRESENT, text="Saved", + timeout_s=0.2, + ) + ], + ) + ], + ) + bundle, run_dir = _dirs(tmp_path) + replayer = Replayer( + backend, + vision=vision, + api_actuator=ApiActuator("http://127.0.0.1:1"), # present but unused + poll_interval_s=0.01, + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + + assert report.success is True + r = report.results[0] + assert r.actuation is None + assert r.effect_verified is None + assert r.effect_results == [] + assert ("press", "Enter") in backend.actions + + +# -- Unit: the actuator's fail-safe classification (no double-write contract) -- + + +def test_actuator_unavailable_on_connection_refused(): + binding = ApiBinding(url_template="http://127.0.0.1:1/api/encounter", + body_template={"patient_id": "p1"}, timeout_s=1.0) + res = ApiActuator().actuate(binding, {}) + assert res.status is ActuationStatus.UNAVAILABLE + assert res.should_fall_through is True + + +def test_actuator_unavailable_on_missing_param(): + # A URL/body that references a param the run did not supply cannot be built + # -> UNAVAILABLE (before-send, nothing written, safe to fall through). + binding = ApiBinding(url_template="/api/encounter", + body_template={"note": "{missing}"}) + res = ApiActuator("http://127.0.0.1:9").actuate(binding, {}) + assert res.status is ActuationStatus.UNAVAILABLE + assert "missing" in res.reason + + +def test_actuator_actuated_on_2xx(): + url, db, stop = _fault_server() + try: + binding = ApiBinding(url_template="/api/encounter", + body_template={"patient_id": "p1", "type": "Triage", + "note": "n"}, timeout_s=2.0) + res = ApiActuator(url).actuate(binding, {}) + assert res.status is ActuationStatus.ACTUATED + assert res.http_status == 200 + assert len(db.snapshot()["records"]) == 1 + finally: + stop() + + +def test_actuator_halts_on_non_2xx(): + url, _db, stop = _fault_server() + try: + binding = ApiBinding(url_template="/api/encounter?fault=session", + body_template={"patient_id": "p1"}, timeout_s=2.0) + res = ApiActuator(url).actuate(binding, {}) + assert res.status is ActuationStatus.HALT + assert res.http_status == 401 + assert res.should_halt is True + finally: + stop()