fix(compiler): retain exact opaque field labels

Retain one unique row-aligned exact label for opaque select fields and halt when post-focus resolution moves.

Author: abrichr

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

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.

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,

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

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,13 +323,16 @@ 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)
     if not target:
         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))