From 87e14bcf0ace2e8608f7fb5930c690ed274c838e Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 21 Jul 2026 08:30:44 -0400 Subject: [PATCH 1/3] fix(resolver): refuse ambiguous OCR targets --- DESIGN.md | 22 +- benchmark/openemr_e2e/simulation.py | 2 + openadapt_flow/runtime/resolver.py | 66 ++++- openadapt_flow/vision/__init__.py | 2 + openadapt_flow/vision/ocr.py | 40 ++- tests/test_governed_healing.py | 11 +- tests/test_halt_learn_loop.py | 13 +- tests/test_heal.py | 11 +- tests/test_heal_fuzz.py | 11 +- tests/test_ocr_resolution_safety.py | 270 +++++++++++++++++++ tests/test_replayer.py | 11 +- tests/test_resolver.py | 29 +- tests/test_resolver_fuzz.py | 11 +- tests/test_resolver_global_landmark_guard.py | 11 +- tests/test_structural_rung.py | 11 +- 15 files changed, 481 insertions(+), 40 deletions(-) create mode 100644 tests/test_ocr_resolution_safety.py diff --git a/DESIGN.md b/DESIGN.md index 61075780..4aaedcd6 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -120,8 +120,11 @@ class OcrLine(BaseModel): text: str; region: Region; confidence: float def ocr(screen_png: bytes, *, region: Region | None = None) -> list[OcrLine] def find_text(screen_png: bytes, text: str, *, - region: Region | None = None, min_ratio: float = 0.8) -> Match | None - # normalized fuzzy match (difflib ratio on lowercased/stripped text) + region: Region | None = None, min_ratio: float = 0.8, + raise_on_ambiguity: bool = False) -> Match | None + # normalized fuzzy match (difflib ratio on lowercased/stripped text); + # resolution enables the typed ambiguity signal so duplicates halt instead + # of falling through, while presence/readiness callers retain best-match. # hashing.py def phash_png(png: bytes, region: Region | None = None) -> str @@ -202,13 +205,14 @@ implemented rungs are: 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 - the target more than 40px away (repeated-icon UIs: an identical glyph on - another card can outscore the true target when mutable content near the - target changed) — the ladder then continues to ocr/geometry -3. `ocr` — find_text(anchor.ocr_text) full frame +1. `template` — find_template within anchor.region padded by search_pad, with + recorded-position locality and repeated-widget uniqueness checks +2. `template_global` — find_template full frame; for labeled and unlabeled + anchors alike, a match is rejected when locatable landmarks contradict its + position, and repeated look-alikes with no expected-position candidate halt +3. `ocr` — require one qualifying `find_text(anchor.ocr_text)` candidate in the + padded local region; only a true miss searches globally. Multiple candidates + or contradictory landmarks halt instead of falling through to weaker rungs 4. `geometry` — landmarks: locate landmark text, offset by the exact `dx_px`/`dy_px` offsets when recorded, else by relation/distance 5. `grounder` — optional injected `Grounder.locate(png, intent) -> Match|None` diff --git a/benchmark/openemr_e2e/simulation.py b/benchmark/openemr_e2e/simulation.py index 689647f4..4f577ff6 100644 --- a/benchmark/openemr_e2e/simulation.py +++ b/benchmark/openemr_e2e/simulation.py @@ -321,7 +321,9 @@ def find_text( *, region: Any = None, min_ratio: float = 0.8, + raise_on_ambiguity: bool = False, ) -> Optional[_Match]: + del raise_on_ambiguity if text == "Saved": # The app always paints "Saved" after the commit key -- the whole # point of effect verification is that this can be true while the diff --git a/openadapt_flow/runtime/resolver.py b/openadapt_flow/runtime/resolver.py index 8a043b65..a68b7bc7 100644 --- a/openadapt_flow/runtime/resolver.py +++ b/openadapt_flow/runtime/resolver.py @@ -15,7 +15,8 @@ 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``. +3. ``ocr`` — unique fuzzy text match on ``anchor.ocr_text`` in + the anchor's padded local region, then globally only after a local miss. 4. ``geometry`` — locate landmark text and offset by relation/distance to estimate the target point. 5. ``grounder`` — optional injected model-backed grounding. @@ -40,6 +41,7 @@ from openadapt_flow.backend import StructuralResolutionRefused from openadapt_flow.ir import Anchor, Point, Region, Resolution, Rung +from openadapt_flow.vision.ocr import AmbiguousOcrMatchError RUNG_ORDER: tuple[Rung, ...] = ( "structural", @@ -218,7 +220,12 @@ def _landmarks_contradict( for landmark in anchor.landmarks: if landmark.dx_px is None or landmark.dy_px is None: continue - match = vision.find_text(screen_png, landmark.ocr_text, min_ratio=OCR_MIN_RATIO) + match = vision.find_text( + screen_png, + landmark.ocr_text, + min_ratio=OCR_MIN_RATIO, + raise_on_ambiguity=True, + ) if match is None: continue estimates.append( @@ -357,7 +364,11 @@ def elapsed_ms() -> float: ) if match is not None: point = _scaled_click_point(anchor, tuple(match.region)) - if not _landmarks_contradict(anchor, point, screen_png, vision): + try: + contradicted = _landmarks_contradict(anchor, point, screen_png, vision) + except AmbiguousOcrMatchError: + return None + if not contradicted: resolution = Resolution( rung="template_global", point=point, @@ -366,13 +377,42 @@ def elapsed_ms() -> float: ) return resolution, tuple(match.region) - # Rung 3: OCR text match. + # Rung 3: OCR text match. Search the same anchor-bounded padded region as + # the local template rung first. Only after a local miss may the resolver + # search the full frame. ``vision.find_text`` refuses multiple qualifying + # lines within either scope, so repeated labels never degrade to a + # first/best-match click. + # + # Landmarks are independent positional evidence. If they contradict an + # OCR candidate, halt immediately: falling through to geometry would turn + # the same contradictory landmarks into a blind offset and could silently + # act on a control that was never established to be present. if anchor.ocr_text: - match = vision.find_text(screen_png, anchor.ocr_text, min_ratio=OCR_MIN_RATIO) - if match is not None: + search_region = pad_region(anchor.region, anchor.search_pad, viewport) + ocr_regions: tuple[Region | None, ...] = (search_region, None) + for ocr_region in ocr_regions: + try: + match = vision.find_text( + screen_png, + anchor.ocr_text, + region=ocr_region, + min_ratio=OCR_MIN_RATIO, + raise_on_ambiguity=True, + ) + except AmbiguousOcrMatchError: + return None + if match is None: + continue + point = (int(match.point[0]), int(match.point[1])) + try: + contradicted = _landmarks_contradict(anchor, point, screen_png, vision) + except AmbiguousOcrMatchError: + return None + if contradicted: + return None resolution = Resolution( rung="ocr", - point=(int(match.point[0]), int(match.point[1])), + point=point, confidence=float(match.confidence), elapsed_ms=elapsed_ms(), ) @@ -382,9 +422,15 @@ def elapsed_ms() -> float: estimates: list[Point] = [] confidences: list[float] = [] for landmark in anchor.landmarks: - lm_match = vision.find_text( - screen_png, landmark.ocr_text, min_ratio=OCR_MIN_RATIO - ) + try: + lm_match = vision.find_text( + screen_png, + landmark.ocr_text, + min_ratio=OCR_MIN_RATIO, + raise_on_ambiguity=True, + ) + except AmbiguousOcrMatchError: + return None if lm_match is None: continue estimates.append( diff --git a/openadapt_flow/vision/__init__.py b/openadapt_flow/vision/__init__.py index 0c7bb31f..6965ff48 100644 --- a/openadapt_flow/vision/__init__.py +++ b/openadapt_flow/vision/__init__.py @@ -18,6 +18,7 @@ pixels_changed, ) from openadapt_flow.vision.ocr import ( + AmbiguousOcrMatchError, OcrLine, find_text, ocr, @@ -31,6 +32,7 @@ ) __all__ = [ + "AmbiguousOcrMatchError", "Match", "OcrLine", "SettleResult", diff --git a/openadapt_flow/vision/ocr.py b/openadapt_flow/vision/ocr.py index f7f5bd8e..8096880e 100644 --- a/openadapt_flow/vision/ocr.py +++ b/openadapt_flow/vision/ocr.py @@ -10,7 +10,7 @@ import difflib import threading -from typing import Any, Optional +from typing import Any import cv2 import numpy as np @@ -23,6 +23,16 @@ _engine_lock = threading.Lock() +class AmbiguousOcrMatchError(RuntimeError): + """Raised when target resolution sees more than one qualifying OCR line. + + Generic OCR consumers retain the historical best-match behavior. The + resolution ladder opts into this typed signal so it can distinguish + ambiguity (halt) from absence (continue to the next independent evidence + rung). + """ + + def _get_engine() -> Any: """Return the process-wide RapidOCR engine, creating it on first use.""" global _engine @@ -206,34 +216,46 @@ def find_text( *, region: Region | None = None, min_ratio: float = 0.8, + raise_on_ambiguity: bool = False, ) -> Match | None: """Locate a text label on screen via OCR plus fuzzy matching. Each OCR line is compared to ``text`` with ``difflib.SequenceMatcher.ratio()`` over normalized (lowercased, - whitespace-collapsed) strings; the best line at or above ``min_ratio`` - wins. + whitespace-collapsed) strings. Generic callers receive the best qualifying + line, preserving the historical presence/readiness behavior. Targeting + callers set ``raise_on_ambiguity`` so repeated labels become a typed refusal + instead of silently selecting the first or highest-scoring control. Args: screen_png: Full-frame screenshot as PNG bytes. text: Target text to find. region: Optional ``(x, y, w, h)`` sub-region to search within. min_ratio: Minimum similarity ratio in ``[0, 1]`` to accept. + raise_on_ambiguity: Raise :class:`AmbiguousOcrMatchError` instead of + selecting the best line when multiple lines qualify. Target + resolution enables this so ambiguity cannot be mistaken for a miss + and fall through to weaker evidence. Returns: - A :class:`Match` centered on the best-matching line's bounding box, + A :class:`Match` centered on the best qualifying line's bounding box, or ``None`` if no line is similar enough. """ target = normalize_text(text) if not target: return None - best: Optional[tuple[float, OcrLine]] = None + qualifying: list[tuple[float, OcrLine]] = [] for line in ocr(screen_png, region=region): ratio = difflib.SequenceMatcher(None, normalize_text(line.text), target).ratio() - if best is None or ratio > best[0]: - best = (ratio, line) - if best is None or best[0] < min_ratio: + if ratio >= min_ratio: + qualifying.append((ratio, line)) + if len(qualifying) > 1: + if raise_on_ambiguity: + raise AmbiguousOcrMatchError( + f"{len(qualifying)} OCR lines qualify for target text" + ) + if not qualifying: return None - ratio, line = best + ratio, line = max(qualifying, key=lambda item: item[0]) x, y, w, h = line.region return Match(point=(x + w // 2, y + h // 2), region=line.region, confidence=ratio) diff --git a/tests/test_governed_healing.py b/tests/test_governed_healing.py index 2f810362..95c778cd 100644 --- a/tests/test_governed_healing.py +++ b/tests/test_governed_healing.py @@ -93,7 +93,16 @@ def find_template( return self.template_results.pop(0) return None - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del raise_on_ambiguity result = self.text_results.get(text) if isinstance(result, list): return result.pop(0) if result else None diff --git a/tests/test_halt_learn_loop.py b/tests/test_halt_learn_loop.py index 173914be..458c9f81 100644 --- a/tests/test_halt_learn_loop.py +++ b/tests/test_halt_learn_loop.py @@ -28,7 +28,6 @@ from typing import Optional import pytest -from tests.test_replayer import FakeBackend, FakeVision, Match, make_png from openadapt_flow.ir import ( ActionKind, @@ -59,6 +58,7 @@ ) from openadapt_flow.runtime.replayer import Replayer from openadapt_flow.vision.ocr import OcrLine +from tests.test_replayer import FakeBackend, FakeVision, Match, make_png # -- the modal-once scenario -------------------------------------------------- @@ -163,7 +163,16 @@ def find_template( return Match(point=DISMISS_POINT, region=(90, 90, 20, 20), confidence=0.95) return None - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del raise_on_ambiguity self.text_calls.append(text) if self._modal_up() and text == self.modal_text: return Match(point=(50, 10), region=(30, 5, 160, 16), confidence=0.9) diff --git a/tests/test_heal.py b/tests/test_heal.py index 13de9a6a..9572bc3a 100644 --- a/tests/test_heal.py +++ b/tests/test_heal.py @@ -73,7 +73,16 @@ def find_template( return self.template_results.pop(0) return None - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del raise_on_ambiguity result = self.text_results.get(text) if isinstance(result, list): return result.pop(0) if result else None diff --git a/tests/test_heal_fuzz.py b/tests/test_heal_fuzz.py index 484372d0..4e9056b7 100644 --- a/tests/test_heal_fuzz.py +++ b/tests/test_heal_fuzz.py @@ -93,7 +93,16 @@ def find_template( return self.template_results.pop(0) return None - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del raise_on_ambiguity result = self.text_results.get(text) if isinstance(result, list): return result.pop(0) if result else None diff --git a/tests/test_ocr_resolution_safety.py b/tests/test_ocr_resolution_safety.py new file mode 100644 index 00000000..922c4988 --- /dev/null +++ b/tests/test_ocr_resolution_safety.py @@ -0,0 +1,270 @@ +"""Fail-closed OCR targeting for repeated labels and contradictory evidence. + +These are small synthetic mechanism tests. They carry no learned thresholds, +deployment data, or application-specific resolution recipes. +""" + +from __future__ import annotations + +import importlib +import io +from typing import Any + +import pytest +from PIL import Image + +from openadapt_flow.ir import Anchor, Landmark, Region +from openadapt_flow.runtime.resolver import resolve +from openadapt_flow.vision.match import Match +from openadapt_flow.vision.ocr import AmbiguousOcrMatchError, OcrLine, find_text + +VIEWPORT = (640, 480) +ocr_module = importlib.import_module("openadapt_flow.vision.ocr") +_AMBIGUOUS = object() + + +def _png() -> bytes: + buffer = io.BytesIO() + Image.new("RGB", VIEWPORT, (240, 240, 240)).save(buffer, format="PNG") + return buffer.getvalue() + + +def _match(point: tuple[int, int], region: Region) -> Match: + return Match(point=point, region=region, confidence=1.0) + + +def _anchor(*, landmarks: list[Landmark] | None = None) -> Anchor: + return Anchor( + template="templates/delete.png", + region=(80, 90, 90, 32), + click_point=(125, 106), + ocr_text="Delete", + landmarks=landmarks or [], + search_pad=40, + ) + + +class _Vision: + """Region-aware scripted OCR namespace; template matching always misses.""" + + def __init__(self, text_results: dict[tuple[str, Region | None], Any]): + self.text_results = text_results + self.text_calls: list[tuple[str, Region | None]] = [] + + @staticmethod + def find_template(*_args: Any, **_kwargs: Any) -> None: + return None + + def find_text( + self, + _screen_png: bytes, + text: str, + *, + region: Region | None = None, + min_ratio: float = 0.8, + raise_on_ambiguity: bool = False, + ) -> Match | None: + del min_ratio + self.text_calls.append((text, region)) + result = self.text_results.get((text, region)) + if result is _AMBIGUOUS: + if raise_on_ambiguity: + raise AmbiguousOcrMatchError("synthetic duplicate OCR lines") + return None + return result + + +class _Grounder: + def __init__(self) -> None: + self.calls = 0 + + def locate(self, *_args: Any, **_kwargs: Any) -> Match: + self.calls += 1 + return _match((600, 440), (580, 430, 40, 20)) + + +def test_find_text_preserves_best_match_for_non_targeting_callers(monkeypatch) -> None: + """Readiness/presence callers keep the historical best-match behavior.""" + lines = [ + OcrLine(text="Delete", region=(100, 100, 60, 20), confidence=0.99), + OcrLine(text="Delete", region=(100, 300, 60, 20), confidence=0.98), + ] + monkeypatch.setattr(ocr_module, "ocr", lambda *_args, **_kwargs: lines) + + result = find_text(b"synthetic", "Delete", min_ratio=0.9) + + assert result is not None + assert result.point == (130, 110) + + +def test_find_text_can_signal_duplicate_qualifying_lines(monkeypatch) -> None: + """Resolution callers can distinguish ambiguity from an ordinary miss.""" + lines = [ + OcrLine(text="Delete", region=(100, 100, 60, 20), confidence=0.99), + OcrLine(text="Delete", region=(100, 300, 60, 20), confidence=0.98), + ] + monkeypatch.setattr(ocr_module, "ocr", lambda *_args, **_kwargs: lines) + + with pytest.raises(AmbiguousOcrMatchError): + find_text( + b"synthetic", + "Delete", + min_ratio=0.9, + raise_on_ambiguity=True, + ) + + +def test_find_text_preserves_unique_success(monkeypatch) -> None: + """A unique qualifying line still resolves exactly as before.""" + lines = [ + OcrLine(text="Delete", region=(100, 100, 60, 20), confidence=0.99), + OcrLine(text="Unrelated", region=(100, 300, 80, 20), confidence=0.99), + ] + monkeypatch.setattr(ocr_module, "ocr", lambda *_args, **_kwargs: lines) + + result = find_text(b"synthetic", "Delete", min_ratio=0.9) + + assert result is not None + assert result.point == (130, 110) + assert result.region == (100, 100, 60, 20) + + +def test_resolver_uses_local_ocr_before_global_and_short_circuits() -> None: + """A unique local label wins without consulting the full frame.""" + local_region = (40, 50, 170, 112) + vision = _Vision({("Delete", local_region): _match((125, 106), (95, 96, 60, 20))}) + + result = resolve(_anchor(), _png(), vision, viewport=VIEWPORT) + + assert result is not None + assert result[0].rung == "ocr" + assert result[0].point == (125, 106) + assert vision.text_calls == [("Delete", local_region)] + + +def test_resolver_uses_global_ocr_only_after_local_miss() -> None: + """A uniquely moved label remains resolvable after local evidence misses.""" + local_region = (40, 50, 170, 112) + vision = _Vision( + { + ("Delete", local_region): None, + ("Delete", None): _match((400, 300), (370, 290, 60, 20)), + } + ) + + result = resolve(_anchor(), _png(), vision, viewport=VIEWPORT) + + assert result is not None + assert result[0].rung == "ocr" + assert result[0].point == (400, 300) + assert vision.text_calls == [("Delete", local_region), ("Delete", None)] + + +def test_local_ocr_ambiguity_halts_without_weaker_fallback() -> None: + """Duplicate local labels are a refusal, not a miss or grounding prompt.""" + local_region = (40, 50, 170, 112) + vision = _Vision({("Delete", local_region): _AMBIGUOUS}) + grounder = _Grounder() + + result = resolve(_anchor(), _png(), vision, grounder, viewport=VIEWPORT) + + assert result is None + assert vision.text_calls == [("Delete", local_region)] + assert grounder.calls == 0 + + +def test_global_ocr_ambiguity_halts_without_weaker_fallback() -> None: + """A local miss may search globally, but global duplicates must halt.""" + local_region = (40, 50, 170, 112) + vision = _Vision( + { + ("Delete", local_region): None, + ("Delete", None): _AMBIGUOUS, + } + ) + grounder = _Grounder() + + result = resolve(_anchor(), _png(), vision, grounder, viewport=VIEWPORT) + + assert result is None + assert vision.text_calls == [("Delete", local_region), ("Delete", None)] + assert grounder.calls == 0 + + +def test_ambiguous_landmark_halts_instead_of_blind_geometry() -> None: + """A repeated row landmark cannot be averaged into a coordinate action.""" + anchor = Anchor( + template="templates/icon.png", + region=(80, 90, 90, 32), + click_point=(125, 106), + landmarks=[ + Landmark( + relation="left_of", + ocr_text="Name:", + distance_px=80, + dx_px=80, + dy_px=0, + ) + ], + ) + vision = _Vision({("Name:", None): _AMBIGUOUS}) + grounder = _Grounder() + + result = resolve(anchor, _png(), vision, grounder, viewport=VIEWPORT) + + assert result is None + assert vision.text_calls == [("Name:", None)] + assert grounder.calls == 0 + + +def test_labeled_far_decoy_contradicted_by_landmark_halts() -> None: + """A far global label cannot outrank independent recorded-row geometry.""" + local_region = (40, 50, 170, 112) + landmark = Landmark( + relation="left_of", + ocr_text="Name:", + distance_px=200, + dx_px=200, + dy_px=0, + ) + vision = _Vision( + { + ("Delete", local_region): None, + ("Delete", None): _match((545, 416), (515, 406, 60, 20)), + ("Name:", None): _match((100, 106), (70, 96, 60, 20)), + } + ) + + result = resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) + + # The contradicted OCR result is a refusal. In particular, the ladder does + # not turn the same landmark into an unverified geometry action. + assert result is None + assert vision.text_calls == [ + ("Delete", local_region), + ("Delete", None), + ("Name:", None), + ] + + +def test_local_ocr_candidate_contradicted_by_landmark_halts() -> None: + """Landmark contradiction is enforced on the local OCR path too.""" + local_region = (40, 50, 170, 112) + landmark = Landmark( + relation="left_of", + ocr_text="Name:", + distance_px=80, + dx_px=80, + dy_px=0, + ) + vision = _Vision( + { + ("Delete", local_region): _match((190, 140), (160, 130, 60, 20)), + ("Name:", None): _match((50, 60), (20, 50, 60, 20)), + } + ) + + result = resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) + + assert result is None + assert vision.text_calls == [("Delete", local_region), ("Name:", None)] diff --git a/tests/test_replayer.py b/tests/test_replayer.py index 4beed068..f659fa97 100644 --- a/tests/test_replayer.py +++ b/tests/test_replayer.py @@ -101,7 +101,16 @@ def find_structural_template( return self.structural_template_results.pop(0) return None - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del raise_on_ambiguity self.text_calls.append(text) result = self.text_results.get(text) if isinstance(result, list): diff --git a/tests/test_resolver.py b/tests/test_resolver.py index 63bec1df..5fa7ea90 100644 --- a/tests/test_resolver.py +++ b/tests/test_resolver.py @@ -63,7 +63,16 @@ def find_template( return self.template_results.pop(0) return None - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del raise_on_ambiguity self.text_calls.append(text) result = self.text_results.get(text) if isinstance(result, list): @@ -400,9 +409,23 @@ def __init__(self): super().__init__() self.min_ratios: dict = {} - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): self.min_ratios[text] = min_ratio - return super().find_text(screen_png, text, region=region, min_ratio=min_ratio) + return super().find_text( + screen_png, + text, + region=region, + min_ratio=min_ratio, + raise_on_ambiguity=raise_on_ambiguity, + ) def test_ocr_rung_requires_strict_label_ratio(screen, anchor): diff --git a/tests/test_resolver_fuzz.py b/tests/test_resolver_fuzz.py index 4fd8f012..88385fc4 100644 --- a/tests/test_resolver_fuzz.py +++ b/tests/test_resolver_fuzz.py @@ -111,7 +111,16 @@ def find_template( return self.template_results.pop(0) return None - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del raise_on_ambiguity result = self.text_results.get(text) if isinstance(result, list): return result.pop(0) if result else None diff --git a/tests/test_resolver_global_landmark_guard.py b/tests/test_resolver_global_landmark_guard.py index 2708afd8..3f228c8c 100644 --- a/tests/test_resolver_global_landmark_guard.py +++ b/tests/test_resolver_global_landmark_guard.py @@ -56,7 +56,16 @@ def find_template( return self._template_results.pop(0) return None - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del raise_on_ambiguity return self._text_results.get(text) diff --git a/tests/test_structural_rung.py b/tests/test_structural_rung.py index 131e4f4b..6c6fe32c 100644 --- a/tests/test_structural_rung.py +++ b/tests/test_structural_rung.py @@ -80,7 +80,16 @@ def find_template( return self.template_results.pop(0) return None - def find_text(self, screen_png, text, *, region=None, min_ratio=0.8): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del raise_on_ambiguity self.text_calls += 1 return self.text_results.get(text) From 1145e6fb2aeb168ec7b11dd300f570c72e1acf96 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 21 Jul 2026 12:25:41 -0400 Subject: [PATCH 2/3] fix(resolver): preserve typed OCR refusals --- DESIGN.md | 5 + openadapt_flow/runtime/replayer.py | 15 ++- openadapt_flow/runtime/resolver.py | 85 +++++++++++--- openadapt_flow/vision/__init__.py | 4 + openadapt_flow/vision/ocr.py | 10 +- tests/test_ocr_resolution_safety.py | 175 ++++++++++++++++++++++++++-- tests/test_replayer.py | 35 ++++++ 7 files changed, 298 insertions(+), 31 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index 4aaedcd6..388ef6e2 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -118,6 +118,9 @@ def find_template(screen_png: bytes, template_png: bytes, *, # ocr.py (rapidocr_onnxruntime; instantiate the engine once, module-level lazy) class OcrLine(BaseModel): text: str; region: Region; confidence: float +class OcrResolutionRefused(RuntimeError): ... +class AmbiguousOcrMatchError(OcrResolutionRefused): ... +class ContradictoryOcrEvidenceError(OcrResolutionRefused): ... def ocr(screen_png: bytes, *, region: Region | None = None) -> list[OcrLine] def find_text(screen_png: bytes, text: str, *, region: Region | None = None, min_ratio: float = 0.8, @@ -125,6 +128,8 @@ def find_text(screen_png: bytes, text: str, *, # normalized fuzzy match (difflib ratio on lowercased/stripped text); # resolution enables the typed ambiguity signal so duplicates halt instead # of falling through, while presence/readiness callers retain best-match. + # The resolver also refuses contradictory unique landmark estimates rather + # than averaging them or delegating to a weaker model-backed rung. # hashing.py def phash_png(png: bytes, region: Region | None = None) -> str diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 5c1a461e..7e0220dc 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -96,6 +96,7 @@ reconcile_or_escalate, ) from openadapt_flow.runtime.resolver import is_below_ocr, pad_region, resolve +from openadapt_flow.vision.ocr import OcrResolutionRefused # REGION_STABLE template check: how far the expected content may shift from # the recorded region (real apps re-layout by a few pixels between runs), @@ -243,7 +244,9 @@ class Replayer: ``phash_distance``, and ``wait_settled``. Defaults to the real ``openadapt_flow.vision`` module, imported lazily so unit tests can inject a fake without - the OCR stack ever loading. + the OCR stack ever loading. Target-resolution providers implement + ``find_text(..., raise_on_ambiguity=True)`` so repeated candidates + produce a typed refusal instead of a first-match action. grounder: Optional Grounder used as the last resolution rung. identity_vlm: Optional IdentityVLM (see runtime.identity) used as the optional local-VLM veto tier of the pre-click identity ladder; @@ -2718,6 +2721,16 @@ def _run_step( result.ok = False error = heal_outcome.halt_reason result.error = error + except OcrResolutionRefused as exc: + # Repeated target/landmark OCR is a deterministic ambiguity, not a + # transient absence. Surface it as a typed operator-visible safety + # halt immediately; never retry into weaker evidence or actuation. + result.ok = False + result.safety_halt = True + result.error = ( + f"OCR safety refusal for step '{step.id}' " + f"({step.intent}): {exc} — no action was admitted" + ) except StructuralResolutionRefused as exc: # Ambiguity, disappearance, or a stale resolve->act fingerprint is # an intentional fail-closed refusal. Record it as a safety halt, diff --git a/openadapt_flow/runtime/resolver.py b/openadapt_flow/runtime/resolver.py index a68b7bc7..baacb40b 100644 --- a/openadapt_flow/runtime/resolver.py +++ b/openadapt_flow/runtime/resolver.py @@ -41,7 +41,10 @@ from openadapt_flow.backend import StructuralResolutionRefused from openadapt_flow.ir import Anchor, Point, Region, Resolution, Rung -from openadapt_flow.vision.ocr import AmbiguousOcrMatchError +from openadapt_flow.vision.ocr import ( + AmbiguousOcrMatchError, + ContradictoryOcrEvidenceError, +) RUNG_ORDER: tuple[Rung, ...] = ( "structural", @@ -220,12 +223,20 @@ def _landmarks_contradict( for landmark in anchor.landmarks: if landmark.dx_px is None or landmark.dy_px is None: continue - match = vision.find_text( - screen_png, - landmark.ocr_text, - min_ratio=OCR_MIN_RATIO, - raise_on_ambiguity=True, - ) + try: + match = vision.find_text( + screen_png, + landmark.ocr_text, + min_ratio=OCR_MIN_RATIO, + raise_on_ambiguity=True, + ) + except AmbiguousOcrMatchError: + # A repeated/generic landmark cannot corroborate or contradict a + # candidate. It abstains while any independent unique landmark + # remains available. Treating one ambiguous landmark as a veto + # makes the outcome depend on irrelevant repeated labels and + # over-halts otherwise uniquely established targets. + continue if match is None: continue estimates.append( @@ -236,12 +247,25 @@ def _landmarks_contradict( ) if not estimates: return False + if _estimates_conflict(estimates): + raise ContradictoryOcrEvidenceError( + "unique OCR landmark estimates disagree beyond tolerance" + ) return all( math.hypot(ex - point[0], ey - point[1]) > GLOBAL_LANDMARK_TOLERANCE_PX for ex, ey in estimates ) +def _estimates_conflict(estimates: list[Point]) -> bool: + """Whether retained unique landmarks disagree beyond the target tolerance.""" + return any( + math.hypot(ax - bx, ay - by) > GLOBAL_LANDMARK_TOLERANCE_PX + for index, (ax, ay) in enumerate(estimates) + for bx, by in estimates[index + 1 :] + ) + + def resolve( anchor: Anchor, screen_png: bytes, @@ -281,6 +305,11 @@ def resolve( is the screen region the evidence matched (used for healing), or None when every rung fails. ``resolution.elapsed_ms`` is the total time spent across all rungs attempted. + + Raises: + OcrResolutionRefused: When OCR target/context evidence is ambiguous or + contradictory. This is deliberately distinct from absence so the + runtime halts without retrying or downgrading to weaker evidence. """ t0 = time.monotonic() if viewport is None: @@ -364,10 +393,7 @@ def elapsed_ms() -> float: ) if match is not None: point = _scaled_click_point(anchor, tuple(match.region)) - try: - contradicted = _landmarks_contradict(anchor, point, screen_png, vision) - except AmbiguousOcrMatchError: - return None + contradicted = _landmarks_contradict(anchor, point, screen_png, vision) if not contradicted: resolution = Resolution( rung="template_global", @@ -400,16 +426,18 @@ def elapsed_ms() -> float: raise_on_ambiguity=True, ) except AmbiguousOcrMatchError: - return None + # Ambiguity is not absence. Preserve the typed refusal so the + # runtime does not retry it as a miss or downgrade to geometry, + # model grounding, healing, or coordinate actuation. + raise if match is None: continue point = (int(match.point[0]), int(match.point[1])) - try: - contradicted = _landmarks_contradict(anchor, point, screen_png, vision) - except AmbiguousOcrMatchError: - return None + contradicted = _landmarks_contradict(anchor, point, screen_png, vision) if contradicted: - return None + raise ContradictoryOcrEvidenceError( + "OCR target and unique landmark evidence disagree" + ) resolution = Resolution( rung="ocr", point=point, @@ -421,6 +449,7 @@ def elapsed_ms() -> float: # Rung 4: geometry from landmarks. estimates: list[Point] = [] confidences: list[float] = [] + ambiguous_landmark = False for landmark in anchor.landmarks: try: lm_match = vision.find_text( @@ -430,7 +459,11 @@ def elapsed_ms() -> float: raise_on_ambiguity=True, ) except AmbiguousOcrMatchError: - return None + # An ambiguous landmark contributes no coordinate. Other unique + # landmarks remain independently usable under the existing + # geometry contract, regardless of declaration order. + ambiguous_landmark = True + continue if lm_match is None: continue estimates.append( @@ -444,6 +477,14 @@ def elapsed_ms() -> float: ) confidences.append(float(lm_match.confidence)) if estimates: + if _estimates_conflict(estimates): + # Averaging inconsistent unique anchors can synthesize a point that + # matches no observed control. Preserve the disagreement as a typed + # terminal refusal; a model grounder must not override retained + # contradictory evidence. + raise ContradictoryOcrEvidenceError( + "unique OCR landmark estimates disagree beyond tolerance" + ) px = int(round(sum(p[0] for p in estimates) / len(estimates))) py = int(round(sum(p[1] for p in estimates) / len(estimates))) region = _clamp_region_of_size( @@ -458,6 +499,14 @@ def elapsed_ms() -> float: ) return resolution, region + if ambiguous_landmark: + # Every locatable landmark was ambiguous. This is a terminal typed + # refusal, not a miss: a grounder or healer must not guess beneath the + # unresolved repeated-row evidence. + raise AmbiguousOcrMatchError( + "OCR landmark evidence did not uniquely establish a target" + ) + # Rung 5: optional grounder. if grounder is not None: match = grounder.locate(screen_png, intent, anchor.ocr_text) diff --git a/openadapt_flow/vision/__init__.py b/openadapt_flow/vision/__init__.py index 6965ff48..6f544fc8 100644 --- a/openadapt_flow/vision/__init__.py +++ b/openadapt_flow/vision/__init__.py @@ -19,7 +19,9 @@ ) from openadapt_flow.vision.ocr import ( AmbiguousOcrMatchError, + ContradictoryOcrEvidenceError, OcrLine, + OcrResolutionRefused, find_text, ocr, text_present, @@ -33,8 +35,10 @@ __all__ = [ "AmbiguousOcrMatchError", + "ContradictoryOcrEvidenceError", "Match", "OcrLine", + "OcrResolutionRefused", "SettleResult", "find_structural_template", "find_template", diff --git a/openadapt_flow/vision/ocr.py b/openadapt_flow/vision/ocr.py index 8096880e..f61a97cb 100644 --- a/openadapt_flow/vision/ocr.py +++ b/openadapt_flow/vision/ocr.py @@ -23,7 +23,11 @@ _engine_lock = threading.Lock() -class AmbiguousOcrMatchError(RuntimeError): +class OcrResolutionRefused(RuntimeError): + """Base class for fail-closed OCR target-resolution refusals.""" + + +class AmbiguousOcrMatchError(OcrResolutionRefused): """Raised when target resolution sees more than one qualifying OCR line. Generic OCR consumers retain the historical best-match behavior. The @@ -33,6 +37,10 @@ class AmbiguousOcrMatchError(RuntimeError): """ +class ContradictoryOcrEvidenceError(OcrResolutionRefused): + """Raised when independently located OCR evidence disagrees on target.""" + + def _get_engine() -> Any: """Return the process-wide RapidOCR engine, creating it on first use.""" global _engine diff --git a/tests/test_ocr_resolution_safety.py b/tests/test_ocr_resolution_safety.py index 922c4988..9717f999 100644 --- a/tests/test_ocr_resolution_safety.py +++ b/tests/test_ocr_resolution_safety.py @@ -16,7 +16,12 @@ from openadapt_flow.ir import Anchor, Landmark, Region from openadapt_flow.runtime.resolver import resolve from openadapt_flow.vision.match import Match -from openadapt_flow.vision.ocr import AmbiguousOcrMatchError, OcrLine, find_text +from openadapt_flow.vision.ocr import ( + AmbiguousOcrMatchError, + ContradictoryOcrEvidenceError, + OcrLine, + find_text, +) VIEWPORT = (640, 480) ocr_module = importlib.import_module("openadapt_flow.vision.ocr") @@ -166,9 +171,9 @@ def test_local_ocr_ambiguity_halts_without_weaker_fallback() -> None: vision = _Vision({("Delete", local_region): _AMBIGUOUS}) grounder = _Grounder() - result = resolve(_anchor(), _png(), vision, grounder, viewport=VIEWPORT) + with pytest.raises(AmbiguousOcrMatchError): + resolve(_anchor(), _png(), vision, grounder, viewport=VIEWPORT) - assert result is None assert vision.text_calls == [("Delete", local_region)] assert grounder.calls == 0 @@ -184,9 +189,9 @@ def test_global_ocr_ambiguity_halts_without_weaker_fallback() -> None: ) grounder = _Grounder() - result = resolve(_anchor(), _png(), vision, grounder, viewport=VIEWPORT) + with pytest.raises(AmbiguousOcrMatchError): + resolve(_anchor(), _png(), vision, grounder, viewport=VIEWPORT) - assert result is None assert vision.text_calls == [("Delete", local_region), ("Delete", None)] assert grounder.calls == 0 @@ -210,13 +215,161 @@ def test_ambiguous_landmark_halts_instead_of_blind_geometry() -> None: vision = _Vision({("Name:", None): _AMBIGUOUS}) grounder = _Grounder() - result = resolve(anchor, _png(), vision, grounder, viewport=VIEWPORT) + with pytest.raises(AmbiguousOcrMatchError): + resolve(anchor, _png(), vision, grounder, viewport=VIEWPORT) - assert result is None assert vision.text_calls == [("Name:", None)] assert grounder.calls == 0 +@pytest.mark.parametrize("ambiguous_first", [False, True]) +def test_ambiguous_landmark_abstains_when_unique_landmark_is_sufficient( + ambiguous_first: bool, +) -> None: + """One repeated landmark cannot poison independent unique geometry.""" + ambiguous = Landmark( + relation="left_of", + ocr_text="Name:", + distance_px=80, + dx_px=80, + dy_px=0, + ) + unique = Landmark( + relation="left_of", + ocr_text="Case A17", + distance_px=80, + dx_px=80, + dy_px=0, + ) + landmarks = [ambiguous, unique] if ambiguous_first else [unique, ambiguous] + anchor = Anchor( + template="templates/icon.png", + region=(80, 90, 90, 32), + click_point=(125, 106), + landmarks=landmarks, + ) + vision = _Vision( + { + ("Name:", None): _AMBIGUOUS, + ("Case A17", None): _match((100, 106), (70, 96, 60, 20)), + } + ) + grounder = _Grounder() + + result = resolve(anchor, _png(), vision, grounder, viewport=VIEWPORT) + + assert result is not None + assert result[0].rung == "geometry" + assert result[0].point == (180, 106) + assert grounder.calls == 0 + + +def test_conflicting_unique_landmarks_refuse_without_grounder() -> None: + """Unique but inconsistent landmarks cannot be averaged into a click.""" + anchor = Anchor( + template="templates/icon.png", + region=(80, 90, 90, 32), + click_point=(125, 106), + landmarks=[ + Landmark( + relation="left_of", + ocr_text="Case A17", + distance_px=80, + dx_px=80, + dy_px=0, + ), + Landmark( + relation="above", + ocr_text="Account 42", + distance_px=60, + dx_px=0, + dy_px=60, + ), + ], + ) + vision = _Vision( + { + ("Case A17", None): _match((100, 106), (70, 96, 60, 20)), + ("Account 42", None): _match((400, 300), (360, 290, 80, 20)), + } + ) + grounder = _Grounder() + + with pytest.raises(ContradictoryOcrEvidenceError): + resolve(anchor, _png(), vision, grounder, viewport=VIEWPORT) + + assert grounder.calls == 0 + + +def test_conflicting_unique_landmarks_cannot_corroborate_ocr_target() -> None: + """One agreeing landmark cannot hide contradictory retained context.""" + local_region = (40, 50, 170, 112) + anchor = _anchor( + landmarks=[ + Landmark( + relation="left_of", + ocr_text="Case A17", + distance_px=25, + dx_px=25, + dy_px=0, + ), + Landmark( + relation="above", + ocr_text="Account 42", + distance_px=60, + dx_px=0, + dy_px=60, + ), + ] + ) + vision = _Vision( + { + ("Delete", local_region): _match((125, 106), (95, 96, 60, 20)), + ("Case A17", None): _match((100, 106), (70, 96, 60, 20)), + ("Account 42", None): _match((400, 300), (360, 290, 80, 20)), + } + ) + + with pytest.raises(ContradictoryOcrEvidenceError): + resolve(anchor, _png(), vision, viewport=VIEWPORT) + + +@pytest.mark.parametrize("ambiguous_first", [False, True]) +def test_ambiguous_landmark_abstains_from_unique_target_corroboration( + ambiguous_first: bool, +) -> None: + """Repeated context does not veto a target corroborated by unique context.""" + local_region = (40, 50, 170, 112) + ambiguous = Landmark( + relation="left_of", + ocr_text="Name:", + distance_px=80, + dx_px=80, + dy_px=0, + ) + unique = Landmark( + relation="left_of", + ocr_text="Case A17", + distance_px=25, + dx_px=25, + dy_px=0, + ) + landmarks = [ambiguous, unique] if ambiguous_first else [unique, ambiguous] + vision = _Vision( + { + ("Delete", local_region): _match((125, 106), (95, 96, 60, 20)), + ("Name:", None): _AMBIGUOUS, + ("Case A17", None): _match((100, 106), (70, 96, 60, 20)), + } + ) + + result = resolve(_anchor(landmarks=landmarks), _png(), vision, viewport=VIEWPORT) + + assert result is not None + assert result[0].rung == "ocr" + assert result[0].point == (125, 106) + + def test_labeled_far_decoy_contradicted_by_landmark_halts() -> None: """A far global label cannot outrank independent recorded-row geometry.""" local_region = (40, 50, 170, 112) @@ -235,11 +388,11 @@ def test_labeled_far_decoy_contradicted_by_landmark_halts() -> None: } ) - result = resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) + with pytest.raises(ContradictoryOcrEvidenceError): + resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) # The contradicted OCR result is a refusal. In particular, the ladder does # not turn the same landmark into an unverified geometry action. - assert result is None assert vision.text_calls == [ ("Delete", local_region), ("Delete", None), @@ -264,7 +417,7 @@ def test_local_ocr_candidate_contradicted_by_landmark_halts() -> None: } ) - result = resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) + with pytest.raises(ContradictoryOcrEvidenceError): + resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) - assert result is None assert vision.text_calls == [("Delete", local_region), ("Name:", None)] diff --git a/tests/test_replayer.py b/tests/test_replayer.py index f659fa97..79e61452 100644 --- a/tests/test_replayer.py +++ b/tests/test_replayer.py @@ -27,6 +27,7 @@ runtime_inputs_digest, ) from openadapt_flow.runtime.replayer import Replayer +from openadapt_flow.vision.ocr import AmbiguousOcrMatchError VIEWPORT = (300, 200) @@ -309,6 +310,40 @@ def test_missing_param_fails_step_and_aborts_run(bundle, run_dir): assert backend.actions == [] # nothing typed, nothing pressed +def test_ocr_ambiguity_is_an_operator_visible_safety_halt(bundle, run_dir): + """Repeated OCR targets never become a generic miss/retry or an action.""" + + class AmbiguousTargetVision(FakeVision): + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + raise_on_ambiguity=False, + ): + del screen_png, text, region, min_ratio + if raise_on_ambiguity: + raise AmbiguousOcrMatchError("two OCR target candidates qualify") + return None + + backend = FakeBackend() + report = Replayer( + backend, vision=AmbiguousTargetVision(), poll_interval_s=0.01 + ).run( + Workflow(name="wf", steps=[click_step()]), + bundle_dir=bundle, + run_dir=run_dir, + ) + + assert report.success is False + assert report.results[0].safety_halt is True + assert "OCR safety refusal" in report.results[0].error + assert "no action was admitted" in report.results[0].error + assert backend.actions == [] + + def test_param_overrides_recorded_literal_text(bundle, run_dir): """Compiled TYPE steps carry BOTH the recorded literal (step.text) and the param name; the runtime param value must win over the literal.""" From 6d0151dc4423aa223a85514ba22199f718d9e3af Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 21 Jul 2026 13:19:40 -0400 Subject: [PATCH 3/3] fix(resolver): disambiguate OCR with retained evidence --- DESIGN.md | 27 +- openadapt_flow/runtime/resolver.py | 246 +++++++++++++-- .../validation/structural_action.py | 23 +- openadapt_flow/vision/__init__.py | 6 +- openadapt_flow/vision/ocr.py | 69 ++++- tests/test_ocr_resolution_safety.py | 288 +++++++++++++++++- 6 files changed, 589 insertions(+), 70 deletions(-) diff --git a/DESIGN.md b/DESIGN.md index 388ef6e2..2339bb50 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -125,11 +125,16 @@ def ocr(screen_png: bytes, *, region: Region | None = None) -> list[OcrLine] def find_text(screen_png: bytes, text: str, *, region: Region | None = None, min_ratio: float = 0.8, raise_on_ambiguity: bool = False) -> Match | None +def find_text_candidates(screen_png: bytes, text: str, *, + region: Region | None = None, + min_ratio: float = 0.8) -> list[Match] # normalized fuzzy match (difflib ratio on lowercased/stripped text); # resolution enables the typed ambiguity signal so duplicates halt instead # of falling through, while presence/readiness callers retain best-match. - # The resolver also refuses contradictory unique landmark estimates rather - # than averaging them or delegating to a weaker model-backed rung. + # Candidate enumeration never selects by order: the resolver requires a + # unique recorded-region or landmark-relation choice for repeated labels. + # Contradictory independent choices and unresolved repeats halt rather than + # averaging coordinates or delegating to a weaker model-backed rung. # hashing.py def phash_png(png: bytes, region: Region | None = None) -> str @@ -211,15 +216,23 @@ implemented rungs are: (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, with - recorded-position locality and repeated-widget uniqueness checks + recorded-position locality; repeated-widget frames are refused when no + candidate is near the recorded position 2. `template_global` — find_template full frame; for labeled and unlabeled anchors alike, a match is rejected when locatable landmarks contradict its position, and repeated look-alikes with no expected-position candidate halt -3. `ocr` — require one qualifying `find_text(anchor.ocr_text)` candidate in the - padded local region; only a true miss searches globally. Multiple candidates - or contradictory landmarks halt instead of falling through to weaker rungs +3. `ocr` — enumerate qualifying `anchor.ocr_text` candidates in the padded + local region; only a true miss searches globally. A sole label is accepted + even when legitimate layout reflow makes old fixed landmark offsets stale. + Repeated labels require exactly one candidate established by the recorded + target region or retained landmark relations. Unresolved repeats or + independent evidence selecting different observed candidates halt instead + of falling through to weaker rungs 4. `geometry` — landmarks: locate landmark text, offset by the exact - `dx_px`/`dy_px` offsets when recorded, else by relation/distance + `dx_px`/`dy_px` offsets when recorded, else by relation/distance. Coherent + estimates may be averaged; incompatible estimates never are. A unique + recorded-region estimate or unique largest coherent cluster may resolve; + otherwise disagreement is a typed halt 5. `grounder` — optional injected `Grounder.locate(png, intent) -> Match|None` (protocol in `runtime/grounder.py`; ship a `NullGrounder`; an Anthropic implementation goes behind the `grounder` extra and is NOT used in tests) diff --git a/openadapt_flow/runtime/resolver.py b/openadapt_flow/runtime/resolver.py index baacb40b..a8aa377f 100644 --- a/openadapt_flow/runtime/resolver.py +++ b/openadapt_flow/runtime/resolver.py @@ -15,8 +15,10 @@ 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`` — unique fuzzy text match on ``anchor.ocr_text`` in - the anchor's padded local region, then globally only after a local miss. +3. ``ocr`` — uniquely established fuzzy text match on + ``anchor.ocr_text`` in the anchor's padded local region, then globally only + after a local miss. Repeated labels require independent retained locality + or landmark evidence; candidate order is never used. 4. ``geometry`` — locate landmark text and offset by relation/distance to estimate the target point. 5. ``grounder`` — optional injected model-backed grounding. @@ -248,9 +250,13 @@ def _landmarks_contradict( if not estimates: return False if _estimates_conflict(estimates): - raise ContradictoryOcrEvidenceError( - "unique OCR landmark estimates disagree beyond tolerance" - ) + # Conflicting fixed-offset context cannot safely corroborate a global + # template candidate, but it also cannot prove that a uniquely + # observed target label is wrong after legitimate layout reflow. + # Reject this template rung and let target OCR attempt an independent + # uniqueness proof. If OCR cannot do so, the geometry rung below keeps + # the same disagreement as a typed terminal refusal. + return True return all( math.hypot(ex - point[0], ey - point[1]) > GLOBAL_LANDMARK_TOLERANCE_PX for ex, ey in estimates @@ -266,6 +272,174 @@ def _estimates_conflict(estimates: list[Point]) -> bool: ) +def _point_in_region(point: Point, region: Region) -> bool: + """Whether ``point`` lies inside ``region`` (inclusive at the far edge).""" + x, y, width, height = region + return x <= point[0] <= x + width and y <= point[1] <= y + height + + +def _select_ocr_candidate( + anchor: Anchor, + candidates: list[Any], + screen_png: bytes, + vision: Any, +) -> Any: + """Select one OCR target only when independent retained evidence does so. + + A sole qualifying OCR line is already unique textual evidence. Repeated + labels require either exactly one candidate in the recorded target region + or unique support from retained landmark relations. If those independent + sources select different observed candidates, resolution terminates as a + contradiction. Candidate enumeration order and fuzzy-score arg-max are + intentionally never used. + """ + if len(candidates) == 1: + return candidates[0] + + locality = [ + candidate + for candidate in candidates + if _point_in_region( + (int(candidate.point[0]), int(candidate.point[1])), + anchor.region, + ) + ] + locality_choice = locality[0] if len(locality) == 1 else None + + supported_indices: set[int] = set() + for landmark in anchor.landmarks: + if landmark.dx_px is None or landmark.dy_px is None: + continue + try: + match = vision.find_text( + screen_png, + landmark.ocr_text, + min_ratio=OCR_MIN_RATIO, + raise_on_ambiguity=True, + ) + except AmbiguousOcrMatchError: + # A repeated context label supplies no unique relation. + continue + if match is None: + continue + estimate = ( + int(match.point[0]) + landmark.dx_px, + int(match.point[1]) + landmark.dy_px, + ) + near = [ + index + for index, candidate in enumerate(candidates) + if math.hypot( + int(candidate.point[0]) - estimate[0], + int(candidate.point[1]) - estimate[1], + ) + <= GLOBAL_LANDMARK_TOLERANCE_PX + ] + if len(near) == 1: + supported_indices.add(near[0]) + + if len(supported_indices) > 1: + raise ContradictoryOcrEvidenceError( + "independent OCR landmark relations select different target candidates" + ) + landmark_choice = ( + candidates[next(iter(supported_indices))] if supported_indices else None + ) + + if locality_choice is not None and landmark_choice is not None: + if locality_choice is not landmark_choice: + raise ContradictoryOcrEvidenceError( + "recorded target region and OCR landmark relations select " + "different candidates" + ) + return locality_choice + if locality_choice is not None: + return locality_choice + if landmark_choice is not None: + return landmark_choice + + raise AmbiguousOcrMatchError( + f"{len(candidates)} OCR target candidates remain after retained-evidence " + "disambiguation" + ) + + +def _select_geometry_estimates( + anchor: Anchor, + estimates: list[Point], + confidences: list[float], +) -> tuple[Point, float]: + """Resolve landmark estimates without averaging incompatible coordinates. + + Coherent estimates may be averaged. When layout drift leaves a stale + landmark far away, exactly one estimate inside the recorded target region + is independently supported by retained locality and can win. With three or + more estimates, a unique largest pairwise-coherent cluster can win. If that + cluster conflicts with an in-region estimate, neither silently outranks the + other: the independent disagreement remains a typed terminal refusal. + """ + selected = list(range(len(estimates))) + if _estimates_conflict(estimates): + in_region = [ + index + for index, estimate in enumerate(estimates) + if _point_in_region(estimate, anchor.region) + ] + clusters: set[frozenset[int]] = set() + for index, estimate in enumerate(estimates): + cluster = frozenset( + other + for other, candidate in enumerate(estimates) + if math.hypot( + estimate[0] - candidate[0], + estimate[1] - candidate[1], + ) + <= GLOBAL_LANDMARK_TOLERANCE_PX + ) + points = [estimates[item] for item in cluster] + if len(cluster) >= 2 and not _estimates_conflict(points): + clusters.add(cluster) + largest: list[frozenset[int]] = [] + if clusters: + largest_size = max(len(cluster) for cluster in clusters) + largest = [cluster for cluster in clusters if len(cluster) == largest_size] + + if len(largest) == 1: + cluster = largest[0] + if any(index not in cluster for index in in_region): + raise ContradictoryOcrEvidenceError( + "recorded target region and coherent OCR landmark cluster disagree" + ) + selected = sorted(cluster) + elif len(largest) > 1: + locality_clusters = [ + cluster + for cluster in largest + if in_region and all(index in cluster for index in in_region) + ] + if len(locality_clusters) == 1: + selected = sorted(locality_clusters[0]) + else: + raise ContradictoryOcrEvidenceError( + "OCR landmark estimates form tied target clusters" + ) + elif len(in_region) == 1: + selected = in_region + elif in_region and not _estimates_conflict( + [estimates[index] for index in in_region] + ): + selected = in_region + else: + raise ContradictoryOcrEvidenceError( + "unique OCR landmark estimates disagree beyond tolerance" + ) + + px = int(round(sum(estimates[index][0] for index in selected) / len(selected))) + py = int(round(sum(estimates[index][1] for index in selected) / len(selected))) + confidence = sum(confidences[index] for index in selected) / len(selected) + return (px, py), confidence + + def resolve( anchor: Anchor, screen_png: bytes, @@ -405,18 +579,43 @@ def elapsed_ms() -> float: # Rung 3: OCR text match. Search the same anchor-bounded padded region as # the local template rung first. Only after a local miss may the resolver - # search the full frame. ``vision.find_text`` refuses multiple qualifying - # lines within either scope, so repeated labels never degrade to a - # first/best-match click. + # search the full frame. On the real vision namespace, candidate enumeration + # lets repeated labels be disambiguated only by independent retained + # locality/landmark evidence. Older injected vision namespaces retain the + # strict ``raise_on_ambiguity`` contract for API compatibility. # - # Landmarks are independent positional evidence. If they contradict an - # OCR candidate, halt immediately: falling through to geometry would turn - # the same contradictory landmarks into a blind offset and could silently - # act on a control that was never established to be present. + # A sole target label remains valid under legitimate layout reflow even + # when old fixed-offset landmark geometry has gone stale. Landmarks are + # therefore used to disambiguate observed repeated target candidates, not + # to veto a uniquely observed target merely for moving independently. if anchor.ocr_text: search_region = pad_region(anchor.region, anchor.search_pad, viewport) ocr_regions: tuple[Region | None, ...] = (search_region, None) + find_candidates = getattr(vision, "find_text_candidates", None) for ocr_region in ocr_regions: + if find_candidates is not None: + candidates = find_candidates( + screen_png, + anchor.ocr_text, + region=ocr_region, + min_ratio=OCR_MIN_RATIO, + ) + if not candidates: + continue + match = _select_ocr_candidate( + anchor, + list(candidates), + screen_png, + vision, + ) + point = (int(match.point[0]), int(match.point[1])) + resolution = Resolution( + rung="ocr", + point=point, + confidence=float(match.confidence), + elapsed_ms=elapsed_ms(), + ) + return resolution, tuple(match.region) try: match = vision.find_text( screen_png, @@ -433,11 +632,6 @@ def elapsed_ms() -> float: if match is None: continue point = (int(match.point[0]), int(match.point[1])) - contradicted = _landmarks_contradict(anchor, point, screen_png, vision) - if contradicted: - raise ContradictoryOcrEvidenceError( - "OCR target and unique landmark evidence disagree" - ) resolution = Resolution( rung="ocr", point=point, @@ -477,24 +671,18 @@ def elapsed_ms() -> float: ) confidences.append(float(lm_match.confidence)) if estimates: - if _estimates_conflict(estimates): - # Averaging inconsistent unique anchors can synthesize a point that - # matches no observed control. Preserve the disagreement as a typed - # terminal refusal; a model grounder must not override retained - # contradictory evidence. - raise ContradictoryOcrEvidenceError( - "unique OCR landmark estimates disagree beyond tolerance" - ) - px = int(round(sum(p[0] for p in estimates) / len(estimates))) - py = int(round(sum(p[1] for p in estimates) / len(estimates))) + (px, py), geometry_confidence = _select_geometry_estimates( + anchor, + estimates, + confidences, + ) region = _clamp_region_of_size( (px, py), (anchor.region[2], anchor.region[3]), viewport ) resolution = Resolution( rung="geometry", point=(px, py), - confidence=(sum(confidences) / len(confidences)) - * _GEOMETRY_CONFIDENCE_SCALE, + confidence=geometry_confidence * _GEOMETRY_CONFIDENCE_SCALE, elapsed_ms=elapsed_ms(), ) return resolution, region diff --git a/openadapt_flow/validation/structural_action.py b/openadapt_flow/validation/structural_action.py index b7acd473..46d47d80 100644 --- a/openadapt_flow/validation/structural_action.py +++ b/openadapt_flow/validation/structural_action.py @@ -45,6 +45,7 @@ from openadapt_flow.backends.playwright_backend import PlaywrightBackend from openadapt_flow.ir import Anchor from openadapt_flow.runtime.resolver import resolve +from openadapt_flow.vision.ocr import OcrResolutionRefused # Grid layout (viewport is the PlaywrightBackend default 1280x800). _COL_X = 90 # baseline button column (left) @@ -179,14 +180,20 @@ def run_probe(n: int = 21, *, headless: bool = True) -> dict: 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, - ) + try: + res = resolve( + anchors[i], + drift_png, + vision_mod, + template_png=templates[i], + viewport=backend.viewport, + structural=None, + ) + except OcrResolutionRefused: + # This probe compares successful target localization counts; + # a typed ambiguity/contradiction is the visual arm's intended + # fail-closed outcome, equivalent to no visual resolution. + res = None v_ok = res is not None and _inside(res[0].point, correct_box) structural_ok += int(s_ok) diff --git a/openadapt_flow/vision/__init__.py b/openadapt_flow/vision/__init__.py index 6f544fc8..aecc60db 100644 --- a/openadapt_flow/vision/__init__.py +++ b/openadapt_flow/vision/__init__.py @@ -3,8 +3,8 @@ Public API (see DESIGN.md "Vision API"): - :class:`Match`, :func:`find_template`, :func:`find_structural_template` -- :class:`OcrLine`, :func:`ocr`, :func:`find_text`, :func:`text_present`, - :func:`upscale_png` +- :class:`OcrLine`, :func:`ocr`, :func:`find_text`, + :func:`find_text_candidates`, :func:`text_present`, :func:`upscale_png` - :func:`phash_png`, :func:`phash_distance` - :func:`pixels_changed` - :func:`wait_settled`, :func:`wait_settled_result`, :class:`SettleResult` @@ -23,6 +23,7 @@ OcrLine, OcrResolutionRefused, find_text, + find_text_candidates, ocr, text_present, upscale_png, @@ -43,6 +44,7 @@ "find_structural_template", "find_template", "find_text", + "find_text_candidates", "ocr", "phash_distance", "phash_png", diff --git a/openadapt_flow/vision/ocr.py b/openadapt_flow/vision/ocr.py index f61a97cb..07cdcda3 100644 --- a/openadapt_flow/vision/ocr.py +++ b/openadapt_flow/vision/ocr.py @@ -249,14 +249,12 @@ def find_text( A :class:`Match` centered on the best qualifying line's bounding box, or ``None`` if no line is similar enough. """ - target = normalize_text(text) - if not target: - return None - qualifying: list[tuple[float, OcrLine]] = [] - for line in ocr(screen_png, region=region): - ratio = difflib.SequenceMatcher(None, normalize_text(line.text), target).ratio() - if ratio >= min_ratio: - qualifying.append((ratio, line)) + qualifying = _qualifying_text_lines( + screen_png, + text, + region=region, + min_ratio=min_ratio, + ) if len(qualifying) > 1: if raise_on_ambiguity: raise AmbiguousOcrMatchError( @@ -267,3 +265,58 @@ def find_text( ratio, line = max(qualifying, key=lambda item: item[0]) x, y, w, h = line.region return Match(point=(x + w // 2, y + h // 2), region=line.region, confidence=ratio) + + +def find_text_candidates( + screen_png: bytes, + text: str, + *, + region: Region | None = None, + min_ratio: float = 0.8, +) -> list[Match]: + """Return every OCR line that qualifies for ``text``. + + Target resolution needs the complete candidate set so it can distinguish + repeated labels with independent retained evidence (for example the + recorded target region or landmark relations). This API deliberately + makes no selection: candidate order is not evidence, and callers must + prove uniqueness before acting. + + Generic callers should continue to use :func:`find_text`, whose default + best-match behavior remains backward compatible. + """ + matches: list[Match] = [] + for ratio, line in _qualifying_text_lines( + screen_png, + text, + region=region, + min_ratio=min_ratio, + ): + x, y, w, h = line.region + matches.append( + Match( + point=(x + w // 2, y + h // 2), + region=line.region, + confidence=ratio, + ) + ) + return matches + + +def _qualifying_text_lines( + screen_png: bytes, + text: str, + *, + region: Region | None, + min_ratio: float, +) -> list[tuple[float, OcrLine]]: + """Return qualifying ``(similarity, line)`` pairs without selecting one.""" + target = normalize_text(text) + if not target: + return [] + qualifying: list[tuple[float, OcrLine]] = [] + for line in ocr(screen_png, region=region): + ratio = difflib.SequenceMatcher(None, normalize_text(line.text), target).ratio() + if ratio >= min_ratio: + qualifying.append((ratio, line)) + return qualifying diff --git a/tests/test_ocr_resolution_safety.py b/tests/test_ocr_resolution_safety.py index 9717f999..49fa96c0 100644 --- a/tests/test_ocr_resolution_safety.py +++ b/tests/test_ocr_resolution_safety.py @@ -21,6 +21,7 @@ ContradictoryOcrEvidenceError, OcrLine, find_text, + find_text_candidates, ) VIEWPORT = (640, 480) @@ -79,6 +80,45 @@ def find_text( return result +class _CandidateVision(_Vision): + """Scripted namespace exposing the complete target-candidate API.""" + + def __init__( + self, + *, + candidates: dict[tuple[str, Region | None], list[Match]], + text_results: dict[tuple[str, Region | None], Any] | None = None, + ) -> None: + super().__init__(text_results or {}) + self.candidates = candidates + + def find_text_candidates( + self, + _screen_png: bytes, + text: str, + *, + region: Region | None = None, + min_ratio: float = 0.8, + ) -> list[Match]: + del min_ratio + self.text_calls.append((text, region)) + return list(self.candidates.get((text, region), [])) + + +class _GlobalTemplateVision(_Vision): + """Scripted global template candidate followed by OCR evidence.""" + + @staticmethod + def find_template( + *_args: Any, + search_region: Region | None = None, + **_kwargs: Any, + ) -> Match | None: + if search_region is not None: + return None + return _match((500, 400), (455, 384, 90, 32)) + + class _Grounder: def __init__(self) -> None: self.calls = 0 @@ -119,6 +159,19 @@ def test_find_text_can_signal_duplicate_qualifying_lines(monkeypatch) -> None: ) +def test_find_text_candidates_returns_all_without_choosing(monkeypatch) -> None: + """Candidate enumeration preserves all matches for evidence resolution.""" + lines = [ + OcrLine(text="Delete", region=(100, 300, 60, 20), confidence=0.98), + OcrLine(text="Delete", region=(100, 100, 60, 20), confidence=0.99), + ] + monkeypatch.setattr(ocr_module, "ocr", lambda *_args, **_kwargs: lines) + + result = find_text_candidates(b"synthetic", "Delete", min_ratio=0.9) + + assert [match.point for match in result] == [(130, 310), (130, 110)] + + def test_find_text_preserves_unique_success(monkeypatch) -> None: """A unique qualifying line still resolves exactly as before.""" lines = [ @@ -196,6 +249,70 @@ def test_global_ocr_ambiguity_halts_without_weaker_fallback() -> None: assert grounder.calls == 0 +@pytest.mark.parametrize("recorded_first", [False, True]) +def test_repeated_ocr_target_uses_unique_recorded_region( + recorded_first: bool, +) -> None: + """Recorded locality can uniquely identify a repeated target label.""" + local_region = (0, 0, 570, 480) + recorded = _match((125, 106), (95, 96, 60, 20)) + sibling = _match((400, 300), (370, 290, 60, 20)) + candidates = [recorded, sibling] if recorded_first else [sibling, recorded] + anchor = _anchor() + anchor.search_pad = 400 + vision = _CandidateVision( + candidates={("Delete", local_region): candidates}, + ) + + result = resolve(anchor, _png(), vision, viewport=VIEWPORT) + + assert result is not None + assert result[0].rung == "ocr" + assert result[0].point == (125, 106) + + +def test_repeated_ocr_target_without_unique_retained_evidence_halts() -> None: + """Repeated labels still refuse when neither locality nor context wins.""" + local_region = (40, 50, 170, 112) + vision = _CandidateVision( + candidates={ + ("Delete", local_region): [ + _match((60, 70), (30, 60, 60, 20)), + _match((190, 140), (160, 130, 60, 20)), + ] + }, + ) + grounder = _Grounder() + + with pytest.raises(AmbiguousOcrMatchError): + resolve(_anchor(), _png(), vision, grounder, viewport=VIEWPORT) + + assert grounder.calls == 0 + + +def test_repeated_target_locality_landmark_conflict_halts() -> None: + """Independent retained evidence selecting different labels is terminal.""" + local_region = (40, 50, 170, 112) + locality = _match((125, 106), (95, 96, 60, 20)) + landmark_supported = _match((190, 140), (160, 130, 60, 20)) + landmark = Landmark( + relation="left_of", + ocr_text="Case A17", + distance_px=80, + dx_px=80, + dy_px=0, + ) + vision = _CandidateVision( + candidates={("Delete", local_region): [locality, landmark_supported]}, + text_results={ + ("Case A17", None): _match((110, 140), (80, 130, 60, 20)), + }, + ) + + with pytest.raises(ContradictoryOcrEvidenceError): + resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) + + def test_ambiguous_landmark_halts_instead_of_blind_geometry() -> None: """A repeated row landmark cannot be averaged into a coordinate action.""" anchor = Anchor( @@ -301,8 +418,101 @@ def test_conflicting_unique_landmarks_refuse_without_grounder() -> None: assert grounder.calls == 0 -def test_conflicting_unique_landmarks_cannot_corroborate_ocr_target() -> None: - """One agreeing landmark cannot hide contradictory retained context.""" +@pytest.mark.parametrize("valid_first", [False, True]) +def test_geometry_uses_unique_in_region_estimate_not_stale_outlier( + valid_first: bool, +) -> None: + """Recorded locality selects one estimate; incompatible points aren't averaged.""" + valid = Landmark( + relation="left_of", + ocr_text="Consult", + distance_px=25, + dx_px=25, + dy_px=0, + ) + stale = Landmark( + relation="above", + ocr_text="Save Encounter", + distance_px=60, + dx_px=0, + dy_px=60, + ) + landmarks = [valid, stale] if valid_first else [stale, valid] + anchor = Anchor( + template="templates/field.png", + region=(80, 90, 90, 32), + click_point=(125, 106), + landmarks=landmarks, + ) + vision = _Vision( + { + ("Consult", None): _match((100, 106), (70, 96, 60, 20)), + ("Save Encounter", None): _match( + (400, 300), + (350, 290, 100, 20), + ), + } + ) + + result = resolve(anchor, _png(), vision, viewport=VIEWPORT) + + assert result is not None + assert result[0].rung == "geometry" + assert result[0].point == (125, 106) + + +@pytest.mark.parametrize("old_region_first", [False, True]) +def test_geometry_refuses_old_region_singleton_vs_current_landmark_cluster( + old_region_first: bool, +) -> None: + """A stale local singleton cannot silently beat two agreeing relations.""" + old = Landmark( + relation="left_of", + ocr_text="Old context", + distance_px=25, + dx_px=25, + dy_px=0, + ) + current_a = Landmark( + relation="left_of", + ocr_text="Current A", + distance_px=80, + dx_px=80, + dy_px=0, + ) + current_b = Landmark( + relation="above", + ocr_text="Current B", + distance_px=60, + dx_px=0, + dy_px=60, + ) + landmarks = ( + [old, current_a, current_b] if old_region_first else [current_b, current_a, old] + ) + anchor = Anchor( + template="templates/field.png", + region=(80, 90, 90, 32), + click_point=(125, 106), + landmarks=landmarks, + ) + vision = _Vision( + { + ("Old context", None): _match((100, 106), (70, 96, 60, 20)), + ("Current A", None): _match((320, 300), (290, 290, 60, 20)), + ("Current B", None): _match((402, 242), (372, 232, 60, 20)), + } + ) + grounder = _Grounder() + + with pytest.raises(ContradictoryOcrEvidenceError): + resolve(anchor, _png(), vision, grounder, viewport=VIEWPORT) + + assert grounder.calls == 0 + + +def test_unique_ocr_target_survives_stale_conflicting_landmark_geometry() -> None: + """A unique observed label is not vetoed by stale fixed-offset context.""" local_region = (40, 50, 170, 112) anchor = _anchor( landmarks=[ @@ -330,8 +540,53 @@ def test_conflicting_unique_landmarks_cannot_corroborate_ocr_target() -> None: } ) - with pytest.raises(ContradictoryOcrEvidenceError): - resolve(anchor, _png(), vision, viewport=VIEWPORT) + result = resolve(anchor, _png(), vision, viewport=VIEWPORT) + + assert result is not None + assert result[0].rung == "ocr" + assert result[0].point == (125, 106) + + +def test_conflicting_landmarks_reject_template_but_not_unique_target_ocr() -> None: + """Stale context falls through instead of vetoing unique target text.""" + local_region = (40, 50, 170, 112) + anchor = _anchor( + landmarks=[ + Landmark( + relation="left_of", + ocr_text="Case A17", + distance_px=25, + dx_px=25, + dy_px=0, + ), + Landmark( + relation="above", + ocr_text="Account 42", + distance_px=60, + dx_px=0, + dy_px=60, + ), + ] + ) + vision = _GlobalTemplateVision( + { + ("Case A17", None): _match((100, 106), (70, 96, 60, 20)), + ("Account 42", None): _match((400, 300), (360, 290, 80, 20)), + ("Delete", local_region): _match((125, 106), (95, 96, 60, 20)), + } + ) + + result = resolve( + anchor, + _png(), + vision, + template_png=b"scripted", + viewport=VIEWPORT, + ) + + assert result is not None + assert result[0].rung == "ocr" + assert result[0].point == (125, 106) @pytest.mark.parametrize("ambiguous_first", [False, True]) @@ -370,8 +625,8 @@ def test_ambiguous_landmark_abstains_from_unique_target_corroboration( assert result[0].point == (125, 106) -def test_labeled_far_decoy_contradicted_by_landmark_halts() -> None: - """A far global label cannot outrank independent recorded-row geometry.""" +def test_unique_moved_global_label_survives_stale_landmark_geometry() -> None: + """A unique moved label wins when no competing observed target exists.""" local_region = (40, 50, 170, 112) landmark = Landmark( relation="left_of", @@ -388,20 +643,19 @@ def test_labeled_far_decoy_contradicted_by_landmark_halts() -> None: } ) - with pytest.raises(ContradictoryOcrEvidenceError): - resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) + result = resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) - # The contradicted OCR result is a refusal. In particular, the ladder does - # not turn the same landmark into an unverified geometry action. + assert result is not None + assert result[0].rung == "ocr" + assert result[0].point == (545, 416) assert vision.text_calls == [ ("Delete", local_region), ("Delete", None), - ("Name:", None), ] -def test_local_ocr_candidate_contradicted_by_landmark_halts() -> None: - """Landmark contradiction is enforced on the local OCR path too.""" +def test_unique_local_label_survives_stale_landmark_geometry() -> None: + """Local unique target presence outranks an obsolete recorded offset.""" local_region = (40, 50, 170, 112) landmark = Landmark( relation="left_of", @@ -417,7 +671,9 @@ def test_local_ocr_candidate_contradicted_by_landmark_halts() -> None: } ) - with pytest.raises(ContradictoryOcrEvidenceError): - resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) + result = resolve(_anchor(landmarks=[landmark]), _png(), vision, viewport=VIEWPORT) - assert vision.text_calls == [("Delete", local_region), ("Name:", None)] + assert result is not None + assert result[0].rung == "ocr" + assert result[0].point == (190, 140) + assert vision.text_calls == [("Delete", local_region)]