From 6445bace564eaa3f2411e3c2486e2c86b29c9cce Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 28 Jul 2026 04:35:01 -0400 Subject: [PATCH] fix(compiler): retain exact opaque field labels --- openadapt_flow/compiler/compile.py | 110 +++++++++++++++++++- openadapt_flow/ir.py | 22 ++++ openadapt_flow/runtime/replayer.py | 39 +++++++ openadapt_flow/runtime/resolver.py | 47 +++++---- openadapt_flow/vision/ocr.py | 17 ++++ tests/test_bundle_schema_v2.py | 54 ++++++++++ tests/test_compiler.py | 151 +++++++++++++++++++++++++++- tests/test_ocr_resolution_safety.py | 34 +++++++ tests/test_replayer.py | 46 ++++++++- tests/test_resolver.py | 77 ++++++++++++++ 10 files changed, 572 insertions(+), 25 deletions(-) diff --git a/openadapt_flow/compiler/compile.py b/openadapt_flow/compiler/compile.py index 7c9e3003..878b2d8d 100644 --- a/openadapt_flow/compiler/compile.py +++ b/openadapt_flow/compiler/compile.py @@ -14,6 +14,7 @@ import logging import math import re +from collections import Counter from datetime import date, datetime from pathlib import Path from typing import TYPE_CHECKING, Iterable, Literal, Optional, Sequence, cast @@ -983,6 +984,83 @@ def _field_label_from_ocr( return best[2] if best is not None else None +def _exact_left_field_label_landmark( + lines: list[OcrLine], + target_region: Region, + click: Point, + *, + exclude_texts: tuple[str, ...] = (), + reference_date: Optional[date] = None, +) -> Optional[tuple[str, Landmark]]: + """Return one unique, row-aligned label left of an opaque field. + + An open native select can repeat its current option in the closed field, + popup, and suggestion list. The option text is then not target identity. + Pixel-only recordings need an independent relation that survives the open + menu. This function mines only a label whose normalized OCR text occurs + exactly once in the recorded frame, whose box is left of and vertically + aligned with the qualified field region, and whose text is stable and + passes the configured exclusions, volatility classification, and identifier + heuristic. It records an exact normalized OCR match and the exact + label-center-to-click offset. There is no fuzzy fallback in this contract. + """ + + rx, ry, rw, rh = (int(value) for value in target_region) + if rw <= 0 or rh <= 0: + return None + recognized = [line for line in lines if normalize_text(line.text)] + counts = Counter(normalize_text(line.text) for line in recognized) + candidates: list[tuple[float, float, str, Landmark]] = [] + for line in recognized: + if line.confidence < MIN_OCR_CONFIDENCE: + continue + text = " ".join(line.text.split()) + normalized = normalize_text(text) + if counts[normalized] != 1 or len(text) > LABEL_OCR_MAX_CHARS: + continue + if _contains_excluded(text, exclude_texts): + continue + if volatility.classify_text(text, reference_date=reference_date): + continue + if _text_carries_phi(text): + continue + lx, ly, lw, lh = (int(value) for value in line.region) + if lw <= 0 or lh <= 0 or lx + lw > rx: + continue + if not ly <= click[1] <= ly + lh: + continue + gap = float(rx - (lx + lw)) + if gap > LABEL_OCR_MAX_LEFT_GAP_PX: + continue + center = (lx + lw // 2, ly + lh // 2) + dx = int(click[0] - center[0]) + dy = int(click[1] - center[1]) + if dx <= 0: + continue + candidates.append( + ( + gap, + abs(float(dy)), + text, + Landmark( + relation="left_of", + ocr_text=text, + distance_px=int(round(math.hypot(dx, dy))), + match_mode="exact", + dx_px=dx, + dy_px=dy, + ), + ) + ) + if not candidates: + return None + _gap, _vertical_delta, text, landmark = min( + candidates, + key=lambda candidate: (candidate[1], candidate[0], candidate[2].casefold()), + ) + return text, landmark + + def _identity_unarmed_reason( frame_lines: list[OcrLine], *, @@ -1944,6 +2022,35 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: key_after, demonstrated, selection_region ) ): + selected_anchor = previous_anchor.model_copy(deep=True) + selected_field_label = step.field_label + exact_label = _exact_left_field_label_landmark( + cached_lines( + int(type_event["i"]), + "before", + type_before, + ), + selected_anchor.region, + selected_anchor.click_point, + exclude_texts=exclude_texts, + reference_date=reference_date, + ) + if exact_label is not None: + selected_field_label, label_landmark = exact_label + retained_landmarks = [ + landmark + for landmark in selected_anchor.landmarks + if normalize_text(landmark.ocr_text) + != normalize_text(label_landmark.ocr_text) + ] + selected_anchor = selected_anchor.model_copy( + update={ + "landmarks": [ + label_landmark, + *retained_landmarks, + ] + } + ) merged_event = dict(type_event) for name, value in key_event.items(): if name.endswith("_after"): @@ -1958,7 +2065,8 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]: ), "selection_commit_key": commit, "selection_region": selection_region, - "anchor": previous_anchor.model_copy(deep=True), + "anchor": selected_anchor, + "field_label": selected_field_label, "identity_armed": previous.identity_armed, "identity_unarmed_reason": ( previous.identity_unarmed_reason diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index f1caf77e..f39f1c18 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -102,6 +102,14 @@ class Landmark(BaseModel): relation: Literal["left_of", "right_of", "above", "below"] ocr_text: str distance_px: int + match_mode: Literal["fuzzy", "exact"] = Field( + default="fuzzy", + description=( + "OCR comparison mode for this retained relation. Compiler-mined " + "generic context remains fuzzy; a qualified opaque-field label " + "uses exact normalized text so a near label cannot authorize input." + ), + ) dx_px: Optional[int] = Field( default=None, description="Exact x offset landmark center -> target click point", @@ -111,6 +119,20 @@ class Landmark(BaseModel): description="Exact y offset landmark center -> target click point", ) + @model_serializer(mode="wrap") + def _serialize_compatible(self, handler: Any) -> dict[str, Any]: + """Keep legacy fuzzy landmarks byte-semantically unchanged. + + The package supports Pydantic 2.5, before ``Field(exclude_if=...)``. + This v2-compatible serializer omits the additive default while keeping + an explicit exact-label contract inside new bundle digests. + """ + + data: dict[str, Any] = handler(self) + if self.match_mode == "fuzzy": + data.pop("match_mode", None) + return data + class StructuralLocator(BaseModel): """A stable structural (DOM / accessibility) locator for a step's target. diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index 5bf8a297..101f1855 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -5213,6 +5213,7 @@ def _act( # re-run target/identity before typing; the focusing click # consumed the first one-shot remote lease. if self._step_needs_consequential_revalidation(step, workflow): + focused_field_point = field_point ( refreshed, refreshed_region, @@ -5232,6 +5233,24 @@ def _act( if remote_error is not None: return remote_error if refreshed is not None: + if ( + step.action is ActionKind.SELECT_OPTION + and focused_field_point is not None + and step.anchor is not None + and not self._selection_target_continuous( + step.anchor.region, + focused_field_point, + refreshed.point, + ) + ): + self._cancel_guarded_keyboard() + result.safety_halt = True + result.failure_category = "safety_halt" + return ( + f"Step '{step.id}' ({step.intent}) re-resolved " + "to a different field after focus; refusing " + "option selection" + ) resolution = refreshed result.resolution = refreshed field_point = refreshed.point @@ -7190,6 +7209,26 @@ def _selection_region_compatible( and 0.75 <= live_h / qualified_h <= 1.25 ) + @staticmethod + def _selection_target_continuous( + qualified: Region, + focused: Point, + refreshed: Point, + ) -> bool: + """Require both select phases to name the same focused field. + + Reuse the selection contract's 25% size tolerance for point drift. + This permits small OCR/template jitter but refuses a second control. + """ + + _, _, qualified_w, qualified_h = qualified + if qualified_w <= 0 or qualified_h <= 0: + return False + return ( + abs(refreshed[0] - focused[0]) <= qualified_w * 0.25 + and abs(refreshed[1] - focused[1]) <= qualified_h * 0.25 + ) + @staticmethod def _mapped_selection_readback_region( recorded_target: Region, diff --git a/openadapt_flow/runtime/resolver.py b/openadapt_flow/runtime/resolver.py index 752f54c7..a5be6b32 100644 --- a/openadapt_flow/runtime/resolver.py +++ b/openadapt_flow/runtime/resolver.py @@ -42,7 +42,8 @@ from typing import Any, Optional from openadapt_flow.backend import StructuralResolutionRefused -from openadapt_flow.ir import Anchor, Point, Region, Resolution, Rung +from openadapt_flow.ir import Anchor, Landmark, Point, Region, Resolution, Rung +from openadapt_flow.vision.match import Match from openadapt_flow.vision.ocr import ( AmbiguousOcrMatchError, ContradictoryOcrEvidenceError, @@ -75,6 +76,7 @@ # (they fall through to the geometry rung) while true labels, which OCR # reads near-verbatim, still match at ≈ 1.0. OCR_MIN_RATIO = 0.9 +EXACT_LANDMARK_MIN_OCR_CONFIDENCE = 0.5 # The global template rung must not accept a match that contradicts the # anchor's landmarks by more than this many pixels. Repeated-widget UIs (an @@ -87,6 +89,28 @@ GLOBAL_LANDMARK_TOLERANCE_PX = 40 +def _landmark_min_ratio(landmark: Landmark) -> float: + """Use exact normalized OCR for compiler-qualified field labels.""" + + return 1.0 if landmark.match_mode == "exact" else OCR_MIN_RATIO + + +def _find_landmark_text( + vision: Any, + screen_png: bytes, + landmark: Landmark, +) -> Optional[Match]: + """Locate a landmark with the compiler's exact-label confidence floor.""" + + kwargs: dict[str, Any] = { + "min_ratio": _landmark_min_ratio(landmark), + "raise_on_ambiguity": True, + } + if landmark.match_mode == "exact": + kwargs["min_ocr_confidence"] = EXACT_LANDMARK_MIN_OCR_CONFIDENCE + return vision.find_text(screen_png, landmark.ocr_text, **kwargs) + + def is_below_ocr(rung: Rung) -> bool: """Return True if ``rung`` is weaker evidence than the ``ocr`` rung. @@ -226,12 +250,7 @@ def _landmarks_contradict( 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, - ) + match = _find_landmark_text(vision, screen_png, landmark) except AmbiguousOcrMatchError: # A repeated/generic landmark cannot corroborate or contradict a # candidate. It abstains while any independent unique landmark @@ -311,12 +330,7 @@ def _select_ocr_candidate( 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, - ) + match = _find_landmark_text(vision, screen_png, landmark) except AmbiguousOcrMatchError: # A repeated context label supplies no unique relation. continue @@ -652,12 +666,7 @@ def elapsed_ms() -> float: ambiguous_landmark = False for landmark in anchor.landmarks: try: - lm_match = vision.find_text( - screen_png, - landmark.ocr_text, - min_ratio=OCR_MIN_RATIO, - raise_on_ambiguity=True, - ) + lm_match = _find_landmark_text(vision, screen_png, landmark) except AmbiguousOcrMatchError: # An ambiguous landmark contributes no coordinate. Other unique # landmarks remain independently usable under the existing diff --git a/openadapt_flow/vision/ocr.py b/openadapt_flow/vision/ocr.py index 07cdcda3..9673ec7e 100644 --- a/openadapt_flow/vision/ocr.py +++ b/openadapt_flow/vision/ocr.py @@ -224,6 +224,7 @@ def find_text( *, region: Region | None = None, min_ratio: float = 0.8, + min_ocr_confidence: float = 0.0, raise_on_ambiguity: bool = False, ) -> Match | None: """Locate a text label on screen via OCR plus fuzzy matching. @@ -240,6 +241,9 @@ def find_text( 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. + min_ocr_confidence: Minimum OCR-engine confidence to accept. The + compatibility default keeps historical generic lookup behavior; + compiler-qualified exact labels opt into a stricter floor. 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 @@ -249,17 +253,25 @@ def find_text( A :class:`Match` centered on the best qualifying line's bounding box, or ``None`` if no line is similar enough. """ + exact_confidence_contract = ( + raise_on_ambiguity and min_ratio == 1.0 and min_ocr_confidence > 0.0 + ) qualifying = _qualifying_text_lines( screen_png, text, region=region, min_ratio=min_ratio, + min_ocr_confidence=(0.0 if exact_confidence_contract else min_ocr_confidence), ) if len(qualifying) > 1: if raise_on_ambiguity: raise AmbiguousOcrMatchError( f"{len(qualifying)} OCR lines qualify for target text" ) + if exact_confidence_contract: + qualifying = [ + item for item in qualifying if item[1].confidence >= min_ocr_confidence + ] if not qualifying: return None ratio, line = max(qualifying, key=lambda item: item[0]) @@ -273,6 +285,7 @@ def find_text_candidates( *, region: Region | None = None, min_ratio: float = 0.8, + min_ocr_confidence: float = 0.0, ) -> list[Match]: """Return every OCR line that qualifies for ``text``. @@ -291,6 +304,7 @@ def find_text_candidates( text, region=region, min_ratio=min_ratio, + min_ocr_confidence=min_ocr_confidence, ): x, y, w, h = line.region matches.append( @@ -309,6 +323,7 @@ def _qualifying_text_lines( *, region: Region | None, min_ratio: float, + min_ocr_confidence: float, ) -> list[tuple[float, OcrLine]]: """Return qualifying ``(similarity, line)`` pairs without selecting one.""" target = normalize_text(text) @@ -316,6 +331,8 @@ def _qualifying_text_lines( return [] qualifying: list[tuple[float, OcrLine]] = [] for line in ocr(screen_png, region=region): + if line.confidence < min_ocr_confidence: + continue ratio = difflib.SequenceMatcher(None, normalize_text(line.text), target).ratio() if ratio >= min_ratio: qualifying.append((ratio, line)) diff --git a/tests/test_bundle_schema_v2.py b/tests/test_bundle_schema_v2.py index 0e30c1a3..2784db7f 100644 --- a/tests/test_bundle_schema_v2.py +++ b/tests/test_bundle_schema_v2.py @@ -19,6 +19,7 @@ TEMPLATE_AAD, ActionKind, Anchor, + Landmark, LoopSpec, Predicate, PredicateKind, @@ -400,6 +401,59 @@ def test_sealed_empty_defaults_do_not_implicitly_cross_schema_versions(): assert "selection_region" in step +def test_legacy_landmark_without_match_mode_keeps_digest_and_loads_fuzzy(tmp_path): + bundle = _write_bundle_dir(tmp_path) + workflow = Workflow(name="legacy-landmark", steps=[click_step()]) + anchor = workflow.steps[0].anchor + assert anchor is not None + anchor.landmarks = [ + Landmark( + relation="left_of", + ocr_text="Field label", + distance_px=80, + dx_px=80, + dy_px=0, + ) + ] + workflow.save(bundle) + saved_digest = workflow.manifest.content_digest + raw = json.loads((bundle / "workflow.json").read_text()) + assert "match_mode" not in raw["steps"][0]["anchor"]["landmarks"][0] + + loaded = Workflow.load(bundle, verify_integrity=True) + + assert loaded.manifest.content_digest == saved_digest + assert loaded.steps[0].anchor.landmarks[0].match_mode == "fuzzy" + rendered = loaded.model_dump(mode="json", exclude={"manifest"}) + assert "match_mode" not in rendered["steps"][0]["anchor"]["landmarks"][0] + + +def test_exact_landmark_survives_sealed_digest_verification(tmp_path): + bundle = _write_bundle_dir(tmp_path) + workflow = Workflow(name="exact-landmark", steps=[click_step()]) + anchor = workflow.steps[0].anchor + assert anchor is not None + anchor.landmarks = [ + Landmark( + relation="left_of", + ocr_text="Title:", + distance_px=80, + match_mode="exact", + dx_px=80, + dy_px=0, + ) + ] + workflow.save(bundle) + saved_digest = workflow.manifest.content_digest + + loaded = Workflow.load(bundle, verify_integrity=True) + + assert loaded.manifest.content_digest == saved_digest + assert loaded.steps[0].anchor.landmarks[0].match_mode == "exact" + raw = json.loads((bundle / "workflow.json").read_text()) + assert raw["steps"][0]["anchor"]["landmarks"][0]["match_mode"] == "exact" + + def test_digest_changes_when_content_changes(tmp_path): b = _write_bundle_dir(tmp_path) wf = _good_program_workflow() diff --git a/tests/test_compiler.py b/tests/test_compiler.py index fbb8795d..4e8d219c 100644 --- a/tests/test_compiler.py +++ b/tests/test_compiler.py @@ -16,9 +16,10 @@ import pytest from openadapt_flow.compiler import compile_recording, render_workflow_py -from openadapt_flow.ir import ActionKind, PostconditionKind, Workflow +from openadapt_flow.compiler.compile import _exact_left_field_label_landmark +from openadapt_flow.ir import ActionKind, Landmark, PostconditionKind, Workflow from openadapt_flow.runtime.identity_template import token_in_template -from openadapt_flow.vision.ocr import normalize_text +from openadapt_flow.vision.ocr import OcrLine, normalize_text VIEWPORT = (1280, 800) NOTE_VALUE = "confidential follow up note" @@ -176,7 +177,7 @@ def test_workflow_metadata(self, compiled) -> None: ] def test_remote_typeahead_and_commit_compile_as_one_verified_selection( - self, tmp_path: Path + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: recording = tmp_path / "recording" (recording / "frames").mkdir(parents=True) @@ -233,6 +234,19 @@ def test_remote_typeahead_and_commit_compile_as_one_verified_selection( ) original_click = uncollapsed.steps[0] + exact_label = Landmark( + relation="left_of", + ocr_text="Title:", + distance_px=686, + match_mode="exact", + dx_px=686, + dy_px=0, + ) + monkeypatch.setattr( + "openadapt_flow.compiler.compile._exact_left_field_label_landmark", + lambda *args, **kwargs: ("Title:", exact_label), + ) + workflow = compile_recording( recording, tmp_path / "bundle", @@ -252,7 +266,17 @@ def test_remote_typeahead_and_commit_compile_as_one_verified_selection( 240, 64, ) - assert selected.anchor == original_click.anchor + assert selected.field_label == "Title:" + assert selected.anchor is not None and original_click.anchor is not None + assert selected.anchor.landmarks[0] == exact_label + serialized = json.loads((tmp_path / "bundle" / "workflow.json").read_text()) + assert serialized["steps"][0]["anchor"]["landmarks"][0]["match_mode"] == "exact" + assert ( + selected.anchor.model_copy( + update={"landmarks": original_click.anchor.landmarks} + ) + == original_click.anchor + ) assert selected.identity_armed == original_click.identity_armed assert ( selected.identity_unarmed_reason == original_click.identity_unarmed_reason @@ -354,6 +378,125 @@ def test_remote_typeahead_and_commit_compile_as_one_verified_selection( ActionKind.KEY, ] + def test_opaque_select_label_requires_unique_exact_aligned_text(self) -> None: + target_region = (300, 200, 160, 40) + click = (340, 220) + title = OcrLine(text="Title:", region=(80, 210, 20, 20), confidence=1.0) + repeated_options = [ + OcrLine( + text="Unassigned", + region=(310, 200 + index * 24, 100, 18), + confidence=1.0, + ) + for index in range(8) + ] + duplicated_context = [ + OcrLine( + text="Birth First Name", + region=(120, 205, 150, 20), + confidence=1.0, + ), + OcrLine( + text="Birth First Name", + region=(120, 305, 150, 20), + confidence=1.0, + ), + OcrLine(text="Middle N", region=(190, 210, 80, 20), confidence=1.0), + OcrLine(text="Middle N", region=(190, 310, 80, 20), confidence=1.0), + ] + + mined = _exact_left_field_label_landmark( + [title, *repeated_options, *duplicated_context], + target_region, + click, + ) + + assert mined is not None + label, landmark = mined + assert label == "Title:" + assert landmark.relation == "left_of" + assert landmark.match_mode == "exact" + assert landmark.dx_px == click[0] - (title.region[0] + title.region[2] // 2) + + padding_label = OcrLine( + text="Compact label", + region=(270, 210, 20, 20), + confidence=1.0, + ) + padded = _exact_left_field_label_landmark( + [padding_label, *repeated_options], + target_region, + click, + ) + assert padded is not None + assert padded[0] == "Compact label" + + duplicate_title = title.model_copy(update={"region": (80, 310, 20, 20)}) + boundary_alternative = OcrLine( + text="Name:", + region=(90, 230, 20, 20), + confidence=1.0, + ) + assert ( + _exact_left_field_label_landmark( + [ + title, + duplicate_title, + boundary_alternative, + *repeated_options, + *duplicated_context, + ], + target_region, + click, + ) + is None + ) + + low_confidence_duplicate = title.model_copy( + update={"region": (80, 310, 20, 20), "confidence": 0.49} + ) + assert ( + _exact_left_field_label_landmark( + [ + title, + low_confidence_duplicate, + *repeated_options, + *duplicated_context, + ], + target_region, + click, + ) + is None + ) + + far_sidebar_label = OcrLine( + text="Remote heading", + region=(0, 210, 60, 20), + confidence=1.0, + ) + assert ( + _exact_left_field_label_landmark( + [far_sidebar_label, *repeated_options], + target_region, + click, + ) + is None + ) + + adjacent_row_label = OcrLine( + text="Adjacent row", + region=(260, 180, 30, 21), + confidence=1.0, + ) + assert ( + _exact_left_field_label_landmark( + [adjacent_row_label, *repeated_options], + target_region, + click, + ) + is None + ) + def test_target_surface_cannot_contradict_recorded_surface( self, tmp_path: Path ) -> None: diff --git a/tests/test_ocr_resolution_safety.py b/tests/test_ocr_resolution_safety.py index 49fa96c0..fe0aac92 100644 --- a/tests/test_ocr_resolution_safety.py +++ b/tests/test_ocr_resolution_safety.py @@ -187,6 +187,40 @@ def test_find_text_preserves_unique_success(monkeypatch) -> None: assert result.region == (100, 100, 60, 20) +def test_find_text_exact_label_refuses_low_ocr_confidence(monkeypatch) -> None: + lines = [OcrLine(text="Title:", region=(100, 100, 60, 20), confidence=0.49)] + monkeypatch.setattr(ocr_module, "ocr", lambda *_args, **_kwargs: lines) + + assert ( + find_text( + b"synthetic", + "Title:", + min_ratio=1.0, + min_ocr_confidence=0.5, + raise_on_ambiguity=True, + ) + is None + ) + assert find_text(b"synthetic", "Title:", min_ratio=1.0) is not None + + +def test_find_text_exact_label_refuses_low_confidence_duplicate(monkeypatch) -> None: + lines = [ + OcrLine(text="Title:", region=(100, 100, 60, 20), confidence=0.9), + OcrLine(text="Title:", region=(100, 200, 60, 20), confidence=0.49), + ] + monkeypatch.setattr(ocr_module, "ocr", lambda *_args, **_kwargs: lines) + + with pytest.raises(AmbiguousOcrMatchError): + find_text( + b"synthetic", + "Title:", + min_ratio=1.0, + min_ocr_confidence=0.5, + raise_on_ambiguity=True, + ) + + 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) diff --git a/tests/test_replayer.py b/tests/test_replayer.py index 9d032e9e..e9e10dbe 100644 --- a/tests/test_replayer.py +++ b/tests/test_replayer.py @@ -111,9 +111,10 @@ def find_text( *, region=None, min_ratio=0.8, + min_ocr_confidence=0.0, raise_on_ambiguity=False, ): - del raise_on_ambiguity + del min_ocr_confidence, raise_on_ambiguity self.text_calls.append(text) result = self.text_results.get(text) if isinstance(result, list): @@ -1923,6 +1924,49 @@ def test_remote_selection_post_focus_uses_landmarks_not_duplicated_value( assert report.results[0].input_verified is True +def test_remote_selection_refuses_shifted_exact_label_after_focus(bundle, run_dir): + vision = FakeVision() + field = Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95) + vision.template_results = [field, field, None, None] + vision.text_results = { + "Field label": Match( + point=(170, 105), + region=(145, 95, 50, 20), + confidence=1.0, + ) + } + workflow = _remote_selection_workflow() + step = workflow.steps[0] + assert step.anchor is not None + step.anchor = step.anchor.model_copy( + update={ + "ocr_text": "Unassigned", + "landmarks": [ + Landmark( + relation="left_of", + ocr_text="Field label", + distance_px=40, + match_mode="exact", + dx_px=40, + dy_px=0, + ) + ], + } + ) + backend = SelectRemoteBackend(selected_value="Massachusetts") + + report = Replayer(backend, vision=vision).run( + workflow, + params={"state": "Massachusetts"}, + bundle_dir=bundle, + run_dir=run_dir, + ) + + assert report.success is False + assert backend.select_calls == [] + assert "different field after focus" in (report.results[0].error or "") + + def test_remote_selection_refuses_incompatible_live_field_geometry(bundle, run_dir): vision = FakeVision() vision.template_results = [ diff --git a/tests/test_resolver.py b/tests/test_resolver.py index 5fa7ea90..3175f9a7 100644 --- a/tests/test_resolver.py +++ b/tests/test_resolver.py @@ -19,6 +19,7 @@ png_size, resolve, ) +from openadapt_flow.vision.ocr import AmbiguousOcrMatchError VIEWPORT = (300, 200) @@ -314,6 +315,82 @@ def test_geometry_rung_single_landmark(screen): assert resolution.confidence == pytest.approx(0.7 * 0.9) +def test_exact_field_label_resolves_open_select_and_duplicate_label_refuses(screen): + anchor = Anchor( + template="templates/select.png", + region=(100, 90, 80, 30), + click_point=(140, 105), + ocr_text="Unassigned", + landmarks=[ + Landmark( + relation="left_of", + ocr_text="Title:", + distance_px=90, + match_mode="exact", + dx_px=90, + dy_px=0, + ) + ], + ) + + class OpenSelectVision(FakeVision): + def __init__(self, *, duplicate_label: bool = False): + super().__init__() + self.duplicate_label = duplicate_label + self.minimum_ratios: list[float] = [] + self.minimum_ocr_confidences: list[float] = [] + + def find_text( + self, + screen_png, + text, + *, + region=None, + min_ratio=0.8, + min_ocr_confidence=0.0, + raise_on_ambiguity=False, + ): + del screen_png, region + assert text == "Title:" # repeated option text is disabled post-focus + assert raise_on_ambiguity is True + self.minimum_ratios.append(min_ratio) + self.minimum_ocr_confidences.append(min_ocr_confidence) + if self.duplicate_label: + raise AmbiguousOcrMatchError("two exact Title labels") + return Match( + point=(50, 105), + region=(25, 95, 50, 20), + confidence=1.0, + ) + + unique = OpenSelectVision() + unique.template_results = [None, None] + resolution, _matched = resolve( + anchor, + screen, + unique, + template_png=b"closed-field-template", + viewport=VIEWPORT, + allow_target_ocr=False, + ) + assert resolution.rung == "geometry" + assert resolution.point == (140, 105) + assert unique.minimum_ratios == [1.0] + assert unique.minimum_ocr_confidences == [0.5] + + duplicate = OpenSelectVision(duplicate_label=True) + duplicate.template_results = [None, None] + with pytest.raises(AmbiguousOcrMatchError, match="did not uniquely establish"): + resolve( + anchor, + screen, + duplicate, + template_png=b"closed-field-template", + viewport=VIEWPORT, + allow_target_ocr=False, + ) + + def test_geometry_rung_averages_multiple_landmarks(screen): anchor = Anchor( template="templates/a.png",