From 58fb10cdca9e5724de86d077e0622ec58f68c721 Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 09:21:02 -0400 Subject: [PATCH 1/6] Take extra looks before stopping on a look-alike identifier On a pixel surface, an OCR identity read that abstains only because the record ID has a look-alike glyph (1/l, 0/O) now takes extra looks before the run stops: two fresh frames, the 2x re-read, the known-ID list, and, where configured, a select-and-copy of the identifier's text box. Only the list or a copy equal to a run parameter can settle the record; any look that names another record is final. Each look is recorded on the step's identity check. The copy look is input, so only the step-entry check takes it; the actuation check reuses it only on matching pixels. Co-Authored-By: Claude Opus 5.5 --- packages/flow/openadapt_flow/deployment.py | 38 ++ packages/flow/openadapt_flow/ir.py | 34 ++ .../openadapt_flow/runtime/identity_looks.py | 527 ++++++++++++++++++ .../flow/openadapt_flow/runtime/known_ids.py | 4 + .../flow/openadapt_flow/runtime/replayer.py | 299 ++++++++-- .../validation/identity_looks_dense.py | 492 ++++++++++++++++ packages/flow/tests/test_identity_looks.py | 287 ++++++++++ .../tests/test_identity_looks_replayer.py | 221 ++++++++ .../flow/tests/test_known_ids_replayer.py | 15 +- 9 files changed, 1862 insertions(+), 55 deletions(-) create mode 100644 packages/flow/openadapt_flow/runtime/identity_looks.py create mode 100644 packages/flow/openadapt_flow/validation/identity_looks_dense.py create mode 100644 packages/flow/tests/test_identity_looks.py create mode 100644 packages/flow/tests/test_identity_looks_replayer.py diff --git a/packages/flow/openadapt_flow/deployment.py b/packages/flow/openadapt_flow/deployment.py index 42c42a484..d796aea39 100644 --- a/packages/flow/openadapt_flow/deployment.py +++ b/packages/flow/openadapt_flow/deployment.py @@ -678,6 +678,24 @@ class GroundingModelConfig(BaseModel): api_key_env: str = "" +class IdentityCopyLookSection(BaseModel): + """Select-and-copy look for a look-alike identifier on a pixel surface. + + Turn on only where selecting the recorded identifier cell and copying it + changes nothing (a read-only text cell) and the session's clipboard reaches + the machine Flow runs on (RDP/Citrix clipboard redirection). See + :mod:`openadapt_flow.runtime.identity_looks`. + """ + + model_config = ConfigDict(extra="forbid") + + enabled: bool = False + #: ``drag`` across the cell at mid height, or ``double_click`` its centre. + gesture: Literal["drag", "double_click"] = "drag" + #: Keys that copy the selection in the driven session. + copy_keys: list[str] = Field(default_factory=lambda: ["Control+c"], min_length=1) + + class RuntimeSection(BaseModel): """Runtime posture: durability and model-egress opt-in.""" @@ -727,6 +745,13 @@ class RuntimeSection(BaseModel): known_ids_confirmation_path: Optional[str] = None #: Oldest export (by its export time) replay will use. known_ids_max_age_hours: float = Field(default=24.0, gt=0) + #: Before stopping on a look-alike identifier, select the identifier cell, + #: copy it and read the clipboard. Settles the record only when the copy + #: equals a run parameter in the identity band; halts when the screen + #: changes outside the cell. Off by default. Never used by governed runs. + identity_copy_look: IdentityCopyLookSection = Field( + default_factory=IdentityCopyLookSection + ) #: Font/layout drift check before each recorded click step #: (:mod:`openadapt_flow.runtime.layout_drift`). None (default) turns it on #: for remote-desktop replays (``rdp``/``citrix``) and off elsewhere; @@ -1308,8 +1333,21 @@ def build_replayer( confirmation_path=runtime_config.known_ids_confirmation_path, max_age_s=runtime_config.known_ids_max_age_hours * 3600.0, ) + identity_copy = None + read_clipboard = None + copy_cfg = runtime_config.identity_copy_look if runtime_config is not None else None + if copy_cfg is not None and copy_cfg.enabled: + from openadapt_flow.records.read import SystemClipboard + from openadapt_flow.runtime.identity_looks import IdentityCopyConfig + + identity_copy = IdentityCopyConfig( + gesture=copy_cfg.gesture, copy_keys=tuple(copy_cfg.copy_keys) + ) + read_clipboard = SystemClipboard() return Replayer( backend, + identity_copy=identity_copy, + read_clipboard=read_clipboard, layout_drift=( runtime_config.layout_drift_check if runtime_config is not None else None ), diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py index 8b080ca65..16573be3c 100644 --- a/packages/flow/openadapt_flow/ir.py +++ b/packages/flow/openadapt_flow/ir.py @@ -3163,6 +3163,26 @@ class KnownIdEvidence(BaseModel): export_sha256: Optional[str] = Field(default=None, pattern=r"^[0-9a-f]{64}$") +class IdentityLook(BaseModel): + """One extra look taken before stopping on a look-alike identifier. + + See :mod:`openadapt_flow.runtime.identity_looks`. Holds no identifier + values: only what kind of look it was, what it concluded, and the + SHA-256 of the frame it read when it read one. + """ + + model_config = ConfigDict(extra="forbid", frozen=True) + + kind: Literal["frame", "magnified", "known_ids", "clipboard"] + #: ``agrees``: consistent with the first read, settles nothing alone. + #: ``resolves``: names exactly one record, the expected one. + #: ``contradicts``: names another record; the step stops. + #: ``unavailable``: the look could not be taken or read nothing usable. + outcome: Literal["agrees", "resolves", "contradicts", "unavailable"] + reason: Optional[str] = Field(default=None, max_length=80) + frame_sha256: Optional[str] = Field(default=None, pattern=r"^[0-9a-f]{64}$") + + class IdentityCheck(BaseModel): """Outcome of the pre-click target-identity check (runtime.identity). @@ -3247,6 +3267,20 @@ class IdentityCheck(BaseModel): "confirmed, or it is older than the configured age. The step halts." ), ) + looks: list[IdentityLook] = Field( + default_factory=list, + description=( + "Extra looks taken because the first read abstained only on a " + "look-alike identifier (runtime.identity_looks), in order." + ), + ) + look_halt: Optional[Literal["identity_look_changed_screen"]] = Field( + default=None, + description=( + "Set when an extra look changed the screen outside the " + "identifier it read. The step halts." + ), + ) class HealEvent(BaseModel): diff --git a/packages/flow/openadapt_flow/runtime/identity_looks.py b/packages/flow/openadapt_flow/runtime/identity_looks.py new file mode 100644 index 000000000..2eb21206e --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/identity_looks.py @@ -0,0 +1,527 @@ +"""Extra looks before stopping on a look-alike identifier. + +On a pixel-only surface (RDP, Citrix, canvas) the OCR identity tier abstains +when the record's identifier carries a look-alike glyph (O/0, l/1/I): OCR can +read ``A1001`` and a different record's ``Al001`` as the same string. Before +the run stops on such an abstain, the replayer takes extra looks that change +nothing on the screen: + +- ``frame``: the same OCR read on :data:`FRAME_LOOKS` fresh screenshots. Each + must abstain on the same look-alike identifier classes as the first read. + A frame that names another record ends the check with a mismatch. +- ``magnified``: the replayer's existing 2x re-read of the first frame. It + must not name another record, and when it reads the identifier it must read + it in the same look-alike classes. +- ``known_ids``: the complete known-ID list (:mod:`.known_ids`) when the + deployment configured one. +- ``clipboard``: where the surface allows it, a drag across the recorded + identifier cell, a copy, and a read of the clipboard (:func:`copy_identifier`). + The clipboard holds the application's own characters, so ``0`` and ``O`` + differ. + +OCR looks never settle the question on their own: a look-alike twin can read +the same on every frame and at every magnification. Only two looks can +resolve: the known-ID list, and a clipboard read that equals an EXACT +expected identifier. An exact expected identifier comes only from a run +parameter embedded in the band (the operator's typed or supplied value). A +recorded OCR band is not exact (a recording can read ``A1001`` as ``Alo01``), +so in recorded-record mode the clipboard can contradict but never accept. + +:func:`decide` accepts only when every frame agrees, no look contradicts, and +at least one look resolves to the expected record. Any contradiction is +final. Otherwise the original abstain stands, with the looks recorded on it. +""" + +from __future__ import annotations + +import hashlib +import io +import secrets +from dataclasses import dataclass +from typing import Any, Callable, Iterable, Literal, Mapping, Optional, Sequence + +from openadapt_flow.ir import IdentityCheck, IdentityLook, Region +from openadapt_flow.runtime import identity as identity_mod +from openadapt_flow.runtime.known_ids import ( + ABSTAIN_REASON_GLYPH, + KnownIdList, + id_core, + lookalike_class, + resolve_lookalike_abstain, +) + +FRAME_LOOKS = 2 +"""Fresh screenshots read after the first one.""" + +SELECTION_PAD_PX = 6 +"""How far outside the identifier cell a selection highlight may paint.""" + +COPY_GESTURES = ("drag", "double_click") + + +def identifier_tokens(text: str) -> list[str]: + """Identifier-shaped tokens of a read, in :func:`known_ids.id_core` form.""" + out = [] + for raw in (text or "").split(): + tok = identity_mod.squash(raw) + if tok and identity_mod._is_identifier_shaped(tok): + out.append(id_core(tok)) + return out + + +def lookalike_classes(text: str) -> frozenset[str]: + """Look-alike classes of the identifier-shaped tokens in ``text``.""" + return frozenset(lookalike_class(t) for t in identifier_tokens(text)) + + +def vulnerable_classes(text: str) -> frozenset[str]: + """Classes of the identifiers OCR could have collapsed (O/0, l/1/I).""" + return frozenset( + lookalike_class(t) + for t in identifier_tokens(text) + if identity_mod._is_glyph_vulnerable_identifier(t) + ) + + +def is_glyph_abstain(check: Optional[IdentityCheck]) -> bool: + return ( + check is not None + and check.status == "abstain" + and check.abstain_reason == ABSTAIN_REASON_GLYPH + ) + + +def exact_expected_ids( + run_values: Mapping[str, Any], embedded: Iterable[str] +) -> tuple[str, ...]: + """Exact identifiers the operator gave for parameters in the band. + + Only parameters embedded in the recorded band count: a run value that is + not part of the identity band says nothing about which record this is. + """ + out: list[str] = [] + for name in embedded: + value = run_values.get(name) + if isinstance(value, str): + out.extend(identifier_tokens(value)) + return tuple(dict.fromkeys(out)) + + +@dataclass(frozen=True) +class IdentityCopyConfig: + """Turns on the select-and-copy look for a deployment. + + Only for surfaces where selecting the identifier cell and copying it + changes nothing: a read-only text cell, with the remote session's + clipboard redirected to the machine Flow runs on. Any screen change + outside the cell during the look halts the run. + """ + + gesture: str = "drag" + copy_keys: tuple[str, ...] = ("Control+c",) + + def __post_init__(self) -> None: + if self.gesture not in COPY_GESTURES: + raise ValueError( + f"identity copy gesture must be one of {COPY_GESTURES}, " + f"not {self.gesture!r}" + ) + if not self.copy_keys: + raise ValueError("identity copy needs at least one copy key") + + +@dataclass(frozen=True) +class LookBinding: + """A clipboard read taken at step entry, reusable at actuation. + + The actuation-time identity check takes no input. It reuses this read only + when the identifier cell on its frame shows the same pixels as one of the + crops seen at step entry. + """ + + step_id: str + classes: frozenset[str] + read: "ClipboardRead" + crops: tuple[bytes, ...] + + def matches( + self, step_id: str, classes: frozenset[str], crop: Optional[bytes] + ) -> bool: + return ( + step_id == self.step_id + and classes == self.classes + and any(crops_match(crop, c) for c in self.crops) + ) + + +@dataclass(frozen=True) +class ClipboardRead: + """What one select-and-copy look returned. + + Attributes: + text: The copied text, or None when nothing was copied. + changed_outside: The screen changed outside the identifier cell + while the look ran. The step halts. + pre_crop: The identifier cell's pixels just before the gesture. + post_crop: The identifier cell's pixels just after the copy. + frame_sha256: SHA-256 of the frame taken just after the copy. + reason: Why nothing was copied, when nothing was. + """ + + text: Optional[str] + changed_outside: bool = False + pre_crop: Optional[bytes] = None + post_crop: Optional[bytes] = None + frame_sha256: Optional[str] = None + reason: Optional[str] = None + + +def _look( + kind: Literal["frame", "magnified", "known_ids", "clipboard"], + outcome: Literal["agrees", "resolves", "contradicts", "unavailable"], + reason: Optional[str] = None, + frame_sha256: Optional[str] = None, +) -> IdentityLook: + return IdentityLook( + kind=kind, outcome=outcome, reason=reason, frame_sha256=frame_sha256 + ) + + +def _frame_look( + first_classes: frozenset[str], + read: IdentityCheck, + frame_sha256: Optional[str], + contradicts: Callable[[IdentityCheck], bool], +) -> IdentityLook: + if read.status == "mismatch": + reason = "names_another_record" if contradicts(read) else "read_incomplete" + return _look("frame", "contradicts", reason, frame_sha256) + if not is_glyph_abstain(read): + return _look("frame", "unavailable", f"read_{read.status}", frame_sha256) + if lookalike_classes(read.observed) != first_classes: + return _look("frame", "unavailable", "identifier_read_differs", frame_sha256) + return _look("frame", "agrees", None, frame_sha256) + + +def _magnified_look( + first_classes: frozenset[str], + read: Optional[IdentityCheck], + contradicts: Callable[[IdentityCheck], bool], +) -> IdentityLook: + if read is None: + return _look("magnified", "unavailable", "not_taken") + if read.status == "mismatch" and contradicts(read): + return _look("magnified", "contradicts", "names_another_record") + if read.status in ("unreadable", "mismatch"): + # Nothing readable, or only part of the band: says nothing either way. + return _look("magnified", "unavailable", f"read_{read.status}") + classes = lookalike_classes(read.observed) + if not classes: + return _look("magnified", "unavailable", "no_identifier_read") + if classes != first_classes: + return _look("magnified", "unavailable", "identifier_read_differs") + return _look("magnified", "agrees") + + +def _clipboard_look( + first_classes: frozenset[str], + vulnerable: frozenset[str], + copied: Optional[ClipboardRead], + exact: Sequence[str], + known_ids: Optional[KnownIdList], +) -> IdentityLook: + if copied is None: + return _look("clipboard", "unavailable", "not_configured") + sha = copied.frame_sha256 + if copied.text is None: + return _look("clipboard", "unavailable", copied.reason or "nothing_copied", sha) + tokens = identifier_tokens(copied.text) + by_class: dict[str, set[str]] = {} + for tok in tokens: + by_class.setdefault(lookalike_class(tok), set()).add(tok) + if set(by_class) != set(first_classes): + # The copy did not cover the identifier OCR read (or covered more). + return _look("clipboard", "unavailable", "copy_differs_from_read", sha) + if any(len(v) != 1 for v in by_class.values()): + return _look("clipboard", "unavailable", "copy_ambiguous", sha) + copied_ids = {k: next(iter(v)) for k, v in by_class.items()} + if known_ids is not None and known_ids.complete: + listed = {m for k in copied_ids for m in known_ids.members_of_class(k)} + if any(core not in listed for core in copied_ids.values()): + return _look("clipboard", "contradicts", "copied_id_not_listed", sha) + exact_by_class: dict[str, set[str]] = {} + for e in exact: + exact_by_class.setdefault(lookalike_class(e), set()).add(e) + for cls, core in copied_ids.items(): + expected = exact_by_class.get(cls) + if expected and core not in expected: + return _look("clipboard", "contradicts", "copied_id_differs", sha) + if vulnerable and all( + cls in exact_by_class and copied_ids[cls] in exact_by_class[cls] + for cls in vulnerable + ): + return _look("clipboard", "resolves", "copied_id_equals_run_value", sha) + return _look("clipboard", "agrees", "no_exact_expected_id", sha) + + +def decide( + first: IdentityCheck, + *, + frames: Sequence[tuple[IdentityCheck, Optional[str]]], + magnified: Optional[IdentityCheck], + contradicts: Callable[[IdentityCheck], bool], + known_ids: Optional[KnownIdList] = None, + known_ids_lift: Optional[IdentityCheck] = None, + clipboard: Optional[ClipboardRead] = None, + exact_expected: Sequence[str] = (), +) -> IdentityCheck: + """Combine the looks into one verdict. + + Args: + first: The first read: a glyph-only abstain. + frames: ``(read, frame_sha256)`` per fresh screenshot. + magnified: The 2x re-read of the first frame, if taken. + contradicts: Whether a ``mismatch`` read names another record + (``replayer.identity_read_contradicts``). + known_ids: The complete known-ID list, for the clipboard cross-check. + known_ids_lift: The list's verdict on the frame reads + (:func:`known_ids.resolve_lookalike_abstain`), or None. + clipboard: The select-and-copy read, or None when not taken. + exact_expected: Exact identifiers from run parameters in the band. + + Returns: + ``verified`` (with every look recorded) when every frame agrees, no + look contradicts and one resolves; the contradicting read (or the + first, as ``mismatch``) when any look contradicts; otherwise + ``first`` with the looks recorded. + """ + if not is_glyph_abstain(first): + return first + first_classes = lookalike_classes(first.observed) + vulnerable = vulnerable_classes(first.observed) + looks: list[IdentityLook] = [] + for read, sha in frames: + look = _frame_look(first_classes, read, sha, contradicts) + looks.append(look) + if look.outcome == "contradicts": + # A fresh frame that names another record is final. + return read.model_copy(update={"looks": looks}) + mag = _magnified_look(first_classes, magnified, contradicts) + looks.append(mag) + if mag.outcome == "contradicts": + assert magnified is not None + return magnified.model_copy(update={"looks": looks}) + if known_ids is not None or known_ids_lift is not None: + looks.append( + _look("known_ids", "resolves", "single_listed_id") + if known_ids_lift is not None + else _look("known_ids", "unavailable", "list_does_not_settle_it") + ) + if clipboard is not None: + clip = _clipboard_look( + first_classes, vulnerable, clipboard, exact_expected, known_ids + ) + looks.append(clip) + if clip.outcome == "contradicts": + return first.model_copy( + update={ + "status": "mismatch", + "abstain_reason": None, + "looks": looks, + } + ) + frames_agree = bool(frames) and all( + look.outcome == "agrees" for look in looks if look.kind == "frame" + ) + magnified_ok = mag.outcome in ("agrees", "unavailable") + resolved = any(look.outcome == "resolves" for look in looks) + if frames_agree and magnified_ok and resolved: + base = known_ids_lift if known_ids_lift is not None else first + return base.model_copy( + update={"status": "verified", "abstain_reason": None, "looks": looks} + ) + return first.model_copy(update={"looks": looks}) + + +# ----------------------------------------------------------- copy look + + +def copy_region(lines: Iterable[Any], within: Region) -> Optional[Region]: + """The text box to select: the one line inside ``within`` that holds a + look-alike identifier. + + Selecting only that line keeps the gesture off every other control in the + row. None when no line, or more than one, holds such an identifier: the + copy look is then not taken. + """ + hits = [] + for line in lines: + region = getattr(line, "region", None) + if region is None or not identity_mod.regions_intersect(region, within): + continue + if vulnerable_classes(str(getattr(line, "text", ""))): + hits.append(region) + if len(hits) != 1: + return None + x, y, w, h = (int(v) for v in hits[0]) + return (x - 2, y - 2, w + 4, h + 4) + + +def crop_png(png: bytes, region: Region) -> Optional[bytes]: + return identity_mod.crop_region(png, region) + + +def _diff_outside(pre: bytes, post: bytes, allowed: Region) -> Optional[bool]: + """Whether ``post`` differs from ``pre`` outside ``allowed``. + + None when the frames cannot be compared (a different size counts as a + change). + """ + try: + import numpy as np + from PIL import Image + + a = np.asarray(Image.open(io.BytesIO(pre)).convert("RGB"), dtype=np.int16) + b = np.asarray(Image.open(io.BytesIO(post)).convert("RGB"), dtype=np.int16) + except Exception: # noqa: BLE001 - an unreadable frame cannot be judged + return None + if a.shape != b.shape: + return True + changed = np.abs(a - b).max(axis=2) > 0 + x, y, w, h = allowed + x0 = max(0, x - SELECTION_PAD_PX) + y0 = max(0, y - SELECTION_PAD_PX) + x1 = min(changed.shape[1], x + w + SELECTION_PAD_PX) + y1 = min(changed.shape[0], y + h + SELECTION_PAD_PX) + changed[y0:y1, x0:x1] = False + return bool(changed.any()) + + +def copy_identifier( + backend: Any, + clipboard: Any, + region: Region, + *, + gesture: str = "drag", + copy_keys: Sequence[str] = ("Control+c",), + settle: Optional[Callable[[], bytes]] = None, + crop_region: Optional[Region] = None, +) -> ClipboardRead: + """Select the identifier cell, copy it, and read the clipboard. + + The clipboard's previous text is restored afterwards. The clipboard is + first set to a random marker; a clipboard still holding the marker counts + as nothing copied. The frames just before the gesture and just after the + copy must be identical outside the identifier cell (a selection highlight + may paint inside it); otherwise ``changed_outside`` is set and the caller + halts. + + Args: + backend: The run's backend. ``drag`` needs a ``drag(x, y, ex, ey)`` + method; ``double_click`` uses ``click(x, y, double=True)``. + clipboard: A :class:`~openadapt_flow.records.read.ClipboardReader`. + region: The identifier cell on the live frame. + gesture: ``drag`` (across the cell, left to right, at mid height) or + ``double_click`` (at the cell's centre). + copy_keys: Keys that copy the selection on the driven session. + settle: Returns the next settled frame; defaults to a screenshot. + crop_region: Where to cut ``pre_crop`` / ``post_crop``; defaults to + ``region``. + """ + x, y, w, h = (int(v) for v in region) + if w < 4 or h < 4: + return ClipboardRead(text=None, reason="cell_too_small") + if gesture == "drag" and not callable(getattr(backend, "drag", None)): + return ClipboardRead(text=None, reason="backend_cannot_drag") + if gesture not in COPY_GESTURES: + return ClipboardRead(text=None, reason="unknown_gesture") + capture = settle or backend.screenshot + try: + previous = clipboard.read_text() + except Exception: # noqa: BLE001 - no clipboard channel, no look + return ClipboardRead(text=None, reason="clipboard_unreadable") + marker = f"openadapt-identity-look-{secrets.token_hex(8)}" + copied: Optional[str] = None + pre = post = None + try: + clipboard.write_text(marker) + pre = capture() + mid_y = y + h // 2 + inset = max(2, min(6, w // 10)) + if gesture == "drag": + backend.drag(x + inset, mid_y, x + w - inset, mid_y) + else: + backend.click(x + w // 2, mid_y, double=True) + for key in copy_keys: + backend.press(key) + post = capture() + copied = clipboard.read_text() + except Exception as exc: # noqa: BLE001 - a failed look is no look + return ClipboardRead( + text=None, + changed_outside=bool( + pre is not None + and post is not None + and _diff_outside(pre, post, region) is not False + ), + reason=f"look_failed:{type(exc).__name__}"[:80], + ) + finally: + if previous is not None: + try: + clipboard.write_text(previous) + except Exception: # noqa: BLE001 - restoring is best effort + pass + outside = _diff_outside(pre, post, region) if pre and post else None + read = ClipboardRead( + text=None if copied in (None, marker) else copied, + changed_outside=outside is not False, + pre_crop=crop_png(pre, crop_region or region) if pre else None, + post_crop=crop_png(post, crop_region or region) if post else None, + frame_sha256=hashlib.sha256(post).hexdigest() if post else None, + reason="nothing_copied" if copied in (None, marker) else None, + ) + return read + + +def crops_match(a: Optional[bytes], b: Optional[bytes]) -> bool: + """Whether two identifier-cell crops show the same pixels. + + Tolerates encoder noise (every channel within 8 levels, mean under 1), + never a changed glyph: a different character moves whole strokes. + """ + if a is None or b is None: + return False + if a == b: + return True + try: + import numpy as np + from PIL import Image + + x = np.asarray(Image.open(io.BytesIO(a)).convert("RGB"), dtype=np.int16) + y = np.asarray(Image.open(io.BytesIO(b)).convert("RGB"), dtype=np.int16) + except Exception: # noqa: BLE001 - unreadable crops never match + return False + if x.shape != y.shape: + return False + d = np.abs(x - y) + return bool(d.max() <= 8 and d.mean() < 1.0) + + +__all__ = [ + "COPY_GESTURES", + "ClipboardRead", + "FRAME_LOOKS", + "IdentityCopyConfig", + "LookBinding", + "copy_identifier", + "copy_region", + "crops_match", + "decide", + "exact_expected_ids", + "identifier_tokens", + "is_glyph_abstain", + "lookalike_classes", + "resolve_lookalike_abstain", + "vulnerable_classes", +] diff --git a/packages/flow/openadapt_flow/runtime/known_ids.py b/packages/flow/openadapt_flow/runtime/known_ids.py index 0fa17ec9f..7d528d3ef 100644 --- a/packages/flow/openadapt_flow/runtime/known_ids.py +++ b/packages/flow/openadapt_flow/runtime/known_ids.py @@ -137,6 +137,10 @@ def members(self, token: str) -> tuple[str, ...]: """Listed identifiers OCR could read as ``token``.""" return self._classes.get(lookalike_class(token), ()) + def members_of_class(self, key: str) -> tuple[str, ...]: + """Listed identifiers in one look-alike class (a :func:`lookalike_class` key).""" + return self._classes.get(key, ()) + @property def size(self) -> int: return sum(len(v) for v in self._classes.values()) diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 68c19a824..5e3a125d6 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -97,6 +97,7 @@ FreshActuationEvent, HaltObservation, IdentityCheck, + IdentityLook, IdentitySignalEvidence, Interstitial, InterstitialActionResult, @@ -149,6 +150,7 @@ from openadapt_flow.runtime import heal as heal_mod from openadapt_flow.runtime import healing as healing_mod from openadapt_flow.runtime import identity as identity_mod +from openadapt_flow.runtime import identity_looks as looks_mod from openadapt_flow.runtime.authorization import ( GovernedRunAuthorization, RuntimeParamScalar, @@ -624,9 +626,19 @@ class Replayer: structured-text, pixel-compare, OCR, and halt with no model. known_ids: Optional complete ``runtime.known_ids.KnownIdList``. When set, an OCR identity check that abstained only because the band - rests on a look-alike identifier is lifted to verified if two OCR - reads agree and exactly one listed identifier is in that - look-alike class. None (default) keeps the abstain. + rests on a look-alike identifier is lifted to verified if the + reads of three separate screenshots agree and exactly one listed + identifier is in that look-alike class. None (default) keeps the + abstain. + identity_copy: Optional + :class:`~openadapt_flow.runtime.identity_looks.IdentityCopyConfig`. + With a ``read_clipboard`` channel, a look-alike abstain on a step + with a recorded identifier cell takes one more look: select the + cell, copy it, read the clipboard. The copy settles the record only + when it equals a run parameter embedded in the band; it stops the + run when it names another record or the screen changed outside the + cell. Never taken in governed or qualification runs. None + (default) takes no copy look. effect_verifier: Optional system-of-record ``EffectVerifier`` (``runtime.effects``) bound to the deployment's system of record (a REST/FHIR API, a document store). When set, any step that @@ -713,6 +725,8 @@ class Replayer: _render_scale_attempted: bool = False _render_scale_inner: Optional[Backend] = None _render_scale_evidence: Optional[dict[str, Any]] = None + identity_copy: Optional[looks_mod.IdentityCopyConfig] = None + _identity_input_looks: bool = False def __init__( self, @@ -755,6 +769,7 @@ def __init__( attended_draft_by: Optional[Any] = None, draft_key_dir: Optional[Path] = None, read_clipboard: Optional[Any] = None, + identity_copy: Optional[looks_mod.IdentityCopyConfig] = None, deployment_surface: Optional[str] = None, layout_drift: Optional[bool] = None, layout_drift_tolerances: Optional[Any] = None, @@ -995,6 +1010,13 @@ def __init__( # Optional complete known-ID list (runtime.known_ids): lifts a # look-alike identifier abstain only when no listed twin exists. self.known_ids = known_ids + # Optional select-and-copy look for look-alike identifiers + # (runtime.identity_looks). It is input, so only the step-entry + # identity check takes it; the actuation-time check reuses its read + # through a pixel-matched binding. + self.identity_copy = identity_copy + self._identity_input_looks = False + self._identity_look_bindings: dict[str, looks_mod.LookBinding] = {} # Optional VLM drift-oracle (state-verifier), OFF by default. When set, # it may CONFIRM a render-drift-sensitive postcondition that # deterministically false-failed (the same heal-under-drift the @@ -6947,7 +6969,7 @@ def _run_step_on_active_surface( ) last_frame = before_png pre_identity_halt = (result.safety_halt, result.failure_category) - error = self._identity_gate_error( + error = self._entry_identity_gate_error( step, resolution, before_png, @@ -7018,7 +7040,7 @@ def _run_step_on_active_surface( result.resolution = resolution # The first read's halt flags belong to that read. result.safety_halt, result.failure_category = pre_identity_halt - error = self._identity_gate_error( + error = self._entry_identity_gate_error( step, resolution, before_png, @@ -9100,6 +9122,26 @@ def _identity_band_still_changing( return False return False + def _entry_identity_gate_error( + self, + step: Step, + *args: Any, + **kwargs: Any, + ) -> Optional[str]: + """The identity gate at step entry, where an extra look may use input. + + Only this check may select and copy the identifier cell + (:mod:`openadapt_flow.runtime.identity_looks`). The actuation-time + check that follows takes no input and reuses that read only through a + pixel-matched binding. + """ + self.__dict__.setdefault("_identity_look_bindings", {}).pop(step.id, None) + self._identity_input_looks = True + try: + return self._identity_gate_error(step, *args, **kwargs) + finally: + self._identity_input_looks = False + def _identity_gate_error( self, step: Step, @@ -9203,6 +9245,13 @@ def _identity_gate_error( f"{check.quorum_required or 0} independent signals verified " "on the fresh frame — refusing to act; run aborted" ) + elif check.look_halt is not None: + error = ( + f"Step '{step.id}' ({step.intent}) took an extra look at a " + "look-alike identifier and the screen changed outside the " + f"identifier ({check.look_halt}) — refusing to act" + ) + result.safety_halt = True elif check.known_id_halt is not None: names = { "known_ids_unconfirmed": "has no admission confirmation", @@ -14110,16 +14159,20 @@ def native_attempt( ), scoped_identifier=identifier_region is not None, ) + pending_looks: list[Any] = [] if ( - self.known_ids is not None - and retry.status not in ("verified", "mismatch") - and check.status == "abstain" - and check.abstain_reason == "glyph_confusable_identifier" + retry.status not in ("verified", "mismatch") + and looks_mod.is_glyph_abstain(check) + and ( + self.known_ids is not None + or self._copy_look_possible(identifier_region) + ) ): - # The band rests on a look-alike identifier. A confirmed, - # complete known-ID list with exactly one member in that class - # rules out a twin (runtime.known_ids), provided a second read - # from a FRESH screenshot agrees with the first. + # The band rests on a look-alike identifier. Take extra looks that + # change nothing (runtime.identity_looks): fresh frames, the 2x + # re-read above, the known-ID list, and, where configured, a + # select-and-copy of the identifier cell. Only the list or an + # exact copy can settle it; any contradiction is final. matches: Optional[Callable[[str], bool]] = None if anchor.identity_template is not None and anchor.identity_template.tokens: from openadapt_flow.runtime import identity_template as itmpl @@ -14127,9 +14180,30 @@ def native_attempt( matches = functools.partial( itmpl.recorded_identifier_matches, anchor.identity_template ) - lifted = self._known_id_lift(check, native_attempt, params, matches) - if lifted is not None: - return lifted + looked = self._look_again( + step, + check, + retry, + native_attempt, + params, + workflow, + identifier_region, + before_png, + matches, + ) + if ( + looked.status != "abstain" + or looked.known_id_halt is not None + or looked.look_halt is not None + ): + return looked + pending_looks = list(looked.looks) + + def with_looks(final: IdentityCheck) -> IdentityCheck: + if not pending_looks or final.status == "verified" or final.looks: + return final + return final.model_copy(update={"looks": pending_looks}) + check = better_identity_read(check, retry) if ( check.status == "verified" @@ -14138,7 +14212,7 @@ def native_attempt( ): # A read of a different record is final: the row re-read below # must never overrule it into "verified". - return check + return with_looks(check) # Last, re-read the point's own text row alone (see # identity.row_region): the same verdict rules, on a crop that holds # no part of the neighbouring rows. @@ -14151,7 +14225,7 @@ def native_attempt( ) ) if row is None or row_png is None: - return check + return with_looks(check) f = identity_mod.ROW_REREAD_FACTOR row_check = attempt( row_png, @@ -14164,22 +14238,55 @@ def native_attempt( exclude[3] * f, ), ) - return better_identity_read(check, row_check) + return with_looks(better_identity_read(check, row_check)) - def _known_id_lift( + def _copy_look_allowed(self) -> bool: + """The copy look is input: never in governed or qualification runs.""" + return ( + self.identity_copy is not None + and self._read_clipboard is not None + and self.governed_authorization is None + and self.managed_dispatch_binding is None + and self.production_qualification_guard is None + and self.qualification_campaign_guard is None + and self.qualification_fault_driver is None + and getattr(self, "approval_gate", None) is None + ) + + def _copy_look_possible(self, identifier_region: Optional[Region]) -> bool: + return identifier_region is not None and self._copy_look_allowed() + + def _embedded_identity_params( + self, anchor: Anchor, workflow: Workflow + ) -> list[str]: + """Workflow parameters whose demonstrated value is in the identity band.""" + tmpl = anchor.identity_template + if tmpl is not None and tmpl.tokens: + return list(tmpl.param_token_indices) + if anchor.context_text: + return identity_mod.embedded_params(anchor.context_text, workflow.params) + return [] + + def _look_again( self, + step: Step, first: IdentityCheck, + magnified: IdentityCheck, read: Callable[[bytes], IdentityCheck], params: Optional[Mapping[str, RuntimeParamScalar]], + workflow: Workflow, + identifier_region: Optional[Region], + frame_png: bytes, recorded_id_matches: Optional[Callable[[str], bool]] = None, - ) -> Optional[IdentityCheck]: - """Known-ID cross-check for a look-alike abstain. - - Returns a ``verified`` check, the abstain marked with a named - ``known_id_halt`` when the list may not be used now, the fresh read - when it is a ``mismatch`` (final), or None (keep the original - verdict). The second read comes from a new screenshot, - so a transient render state on one capture cannot satisfy both. + ) -> IdentityCheck: + """Extra looks before stopping on a look-alike identifier. + + Returns the verdict of :func:`identity_looks.decide`: ``verified`` + when every fresh frame agrees and the known-ID list or an exact copy + settles the record, the contradicting read when a look names another + record, or ``first`` carrying the looks. A known-ID list that may not + be used now, or a copy look that changed the screen, returns ``first`` + marked with a named halt. """ from openadapt_flow.runtime import known_ids as known_ids_mod @@ -14188,33 +14295,123 @@ def _known_id_lift( if isinstance(self.known_ids, known_ids_mod.KnownIdAdmission) else None ) + known: Any = self.known_ids if admission is not None: halt = admission.check() if halt is not None: return first.model_copy(update={"known_id_halt": halt.reason}) - known: Any = admission.known_ids - else: - known = self.known_ids - second = read(self.backend.screenshot()) - if second.status == "mismatch": - # The fresh read names a different record; that is final. - return second - lifted = known_ids_mod.resolve_lookalike_abstain( - [first, second], - known, - run_values=runtime_params_for_gui(params or {}), - recorded_id_matches=recorded_id_matches, - ) - if lifted is None or lifted.known_id_evidence is None: - return None - update: dict[str, Any] = {"frames": 2} - if admission is not None: - update["confirmed_at"] = admission.confirmation.confirmed_at - update["export_sha256"] = admission.confirmation.sha256 - return lifted.model_copy( - update={ - "known_id_evidence": lifted.known_id_evidence.model_copy(update=update) - } + known = admission.known_ids + run_values = runtime_params_for_gui(params or {}) + + frames: list[tuple[IdentityCheck, Optional[str]]] = [] + for _ in range(looks_mod.FRAME_LOOKS): + time.sleep(self.poll_interval_s) + png = self.backend.screenshot() + frames.append((read(png), sha256_bytes(png))) + if frames[-1][0].status == "mismatch": + break # final; no further looks + + frame_reads = [first, *(r for r, _ in frames)] + known_lift: Optional[IdentityCheck] = None + if known is not None and all( + looks_mod.is_glyph_abstain(r) for r in frame_reads + ): + lifted = known_ids_mod.resolve_lookalike_abstain( + frame_reads, + known, + run_values=run_values, + recorded_id_matches=recorded_id_matches, + ) + if lifted is not None and lifted.known_id_evidence is not None: + update: dict[str, Any] = {"frames": len(frame_reads)} + if admission is not None: + update["confirmed_at"] = admission.confirmation.confirmed_at + update["export_sha256"] = admission.confirmation.sha256 + known_lift = lifted.model_copy( + update={ + "known_id_evidence": lifted.known_id_evidence.model_copy( + update=update + ) + } + ) + + copied: Optional[looks_mod.ClipboardRead] = None + classes = looks_mod.lookalike_classes(first.observed) + bindings = self.__dict__.setdefault("_identity_look_bindings", {}) + if ( + identifier_region is not None + and self._copy_look_allowed() + and all(r.status != "mismatch" for r, _ in frames) + ): + assert self.identity_copy is not None + if self._identity_input_looks: + try: + cell_lines = self.vision.ocr(frame_png, region=identifier_region) + except Exception: # noqa: BLE001 - no box, no copy look + cell_lines = [] + cell = looks_mod.copy_region(cell_lines, identifier_region) + copied = ( + looks_mod.copy_identifier( + self.backend, + self._read_clipboard, + cell, + gesture=self.identity_copy.gesture, + copy_keys=self.identity_copy.copy_keys, + crop_region=identifier_region, + ) + if cell is not None + else looks_mod.ClipboardRead( + text=None, reason="no_single_identifier_box" + ) + ) + if copied.changed_outside: + look = IdentityLook( + kind="clipboard", + outcome="unavailable", + reason="screen_changed_outside_cell", + frame_sha256=copied.frame_sha256, + ) + return first.model_copy( + update={ + "look_halt": "identity_look_changed_screen", + "looks": [look], + } + ) + crops = tuple( + c + for c in ( + looks_mod.crop_png(frame_png, identifier_region), + copied.pre_crop, + copied.post_crop, + ) + if c is not None + ) + bindings[step.id] = looks_mod.LookBinding( + step_id=step.id, classes=classes, read=copied, crops=crops + ) + else: + binding = bindings.get(step.id) + live_crop = looks_mod.crop_png(frame_png, identifier_region) + if binding is not None and binding.matches(step.id, classes, live_crop): + copied = binding.read + + known_list = None + if known is not None: + known_list = getattr(known, "known_ids", known) + return looks_mod.decide( + first, + frames=frames, + magnified=magnified, + contradicts=identity_read_contradicts, + known_ids=known_list, + known_ids_lift=known_lift, + clipboard=copied, + exact_expected=looks_mod.exact_expected_ids( + run_values, + self._embedded_identity_params(step.anchor, workflow) + if step.anchor is not None + else [], + ), ) # -- typed-input verification ------------------------------------------------- diff --git a/packages/flow/openadapt_flow/validation/identity_looks_dense.py b/packages/flow/openadapt_flow/validation/identity_looks_dense.py new file mode 100644 index 000000000..e73bd9c5e --- /dev/null +++ b/packages/flow/openadapt_flow/validation/identity_looks_dense.py @@ -0,0 +1,492 @@ +"""Extra identity looks on the dense record list: before and after. + +Drives the REAL ``Replayer._verify_identity`` (identifier-cell OCR, the 2x +re-read, the known-ID list and the extra looks of +:mod:`openadapt_flow.runtime.identity_looks`) against the dense record list of +:mod:`.dense_surface`, rendered live in a browser. The pixel path is the one +the compiler builds for a pixel recording: the identifier region is the tight +box of the row's identity text (``context_region_from_lines``) and the +recorded identity is the OCR text inside it. + +Each armed click is scored twice, at the step-entry check and, when that +check verified, again at the actuation-time check, which takes no input: + +- **correct row**: the resolver landed on the recorded record. A stop is a + false stop (an over-halt). +- **sibling row**: the resolver landed on the adjacent different record. A + verify is a wrong-record accept. Must stay 0. + +Variants: no list and no copy (the shipped default), the known-ID list, the +copy look, and both. Modes: recorded record (the band names a fixed record) +and parameter (the MRN is a run parameter, as in a worklist). The copy look +drags across the identifier's text box with the real mouse; the copied text +is the page's own selection, which is what a redirected clipboard receives. + +The same script runs on ``origin/main`` for the before numbers: variants the +code under test does not support are skipped. + +Run: + python -m openadapt_flow.validation.identity_looks_dense --out DIR +""" + +from __future__ import annotations + +import argparse +import hashlib +import inspect +import json +import subprocess +from dataclasses import dataclass, field +from datetime import date +from pathlib import Path +from typing import Any, Optional + +SCHEMA = "openadapt.identity-looks-dense.v1" +VARIANTS = ("shipped", "list", "copy", "list_copy") +MODES = ("recorded", "param") + + +def _git_commit() -> Optional[str]: + try: + here = Path(__file__).resolve().parent + out = subprocess.run( + ["git", "-C", str(here), "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=False, + ) + return out.stdout.strip() or None + except Exception: # noqa: BLE001 - provenance only + return None + + +class CachedVision: + """``openadapt_flow.vision`` with OCR memoized by (frame bytes, region). + + The fixture renders deterministically, so identical frames get identical + reads; caching changes no verdict, only the run time. + """ + + def __init__(self) -> None: + import openadapt_flow.vision as vision + + self._vision = vision + self._cache: dict[tuple[str, Any], Any] = {} + self.calls = 0 + + def __getattr__(self, name: str) -> Any: + return getattr(self._vision, name) + + def ocr(self, png: bytes, region: Any = None, **kw: Any) -> Any: + key = (hashlib.sha256(png).hexdigest(), tuple(region) if region else None) + if key not in self._cache: + self.calls += 1 + self._cache[key] = self._vision.ocr(png, region=region, **kw) + return self._cache[key] + + +@dataclass +class PageBackend: + """A live browser page seen only as pixels, with mouse drag and copy. + + Coordinates are screen pixels (CSS pixels times the device scale). The + clipboard is the page's selection at the copy key, which is what a + redirected remote clipboard would receive. + """ + + page: Any + dsf: int + viewport: tuple[int, int] + clean_png: bytes = b"" + dirty: bool = False + copied: Optional[str] = None + drags: int = 0 + # (clean frame sha, drag) -> (selection text, post frame) + _memo: dict[tuple[str, tuple[int, int, int, int]], tuple[str, bytes]] = field( + default_factory=dict + ) + _pending: Optional[tuple[str, tuple[int, int, int, int]]] = None + _post: Optional[bytes] = None + + def screenshot(self) -> bytes: + if not self.dirty: + return self.clean_png + if self._post is not None: + return self._post + return self.page.screenshot(full_page=True) + + def drag(self, x: int, y: int, end_x: int, end_y: int) -> None: + key = (hashlib.sha256(self.clean_png).hexdigest()[:16], (x, y, end_x, end_y)) + self.drags += 1 + self.dirty = True + if key in self._memo: + self._pending = key + self._post = self._memo[key][1] + return + d = self.dsf + self.page.mouse.move(x / d, y / d) + self.page.mouse.down() + self.page.mouse.move(end_x / d, end_y / d, steps=6) + self.page.mouse.up() + self._pending = key + self._post = None + + def press(self, key: str) -> None: + if key.lower() not in ("control+c", "meta+c", "controlormeta+c"): + raise AssertionError(f"unexpected key {key!r}") + if self._pending is None: + return + if self._pending in self._memo: + self.copied = self._memo[self._pending][0] + return + text = self.page.evaluate("() => String(window.getSelection())") + post = self.page.screenshot(full_page=True) + self._memo[self._pending] = (text, post) + self._post = post + self.copied = text + + def reset(self) -> None: + """Harness only: clear the selection between trials.""" + self.page.evaluate("() => window.getSelection().removeAllRanges()") + self.page.mouse.move(1, 1) + self.dirty = False + self._pending = None + self._post = None + + +class PageClipboard: + def __init__(self, backend: PageBackend) -> None: + self.backend = backend + + def read_text(self) -> Optional[str]: + return self.backend.copied + + def write_text(self, text: str) -> None: + self.backend.copied = text + + +def _open_page(pw: Any, table: Any, cond: Any, top_offset_px: int) -> tuple[Any, Any]: + from openadapt_flow.validation.dense_surface import render_table_html + + html = render_table_html( + table, + font_family=cond.font_family, + font_px=cond.font_px, + row_pad_px=cond.row_pad_px, + top_offset_px=top_offset_px, + ) + browser = pw.chromium.launch(headless=True) + page = browser.new_page( + viewport={"width": 1120, "height": 1600}, + device_scale_factor=cond.device_scale_factor, + ) + page.set_content(html, wait_until="networkidle") + full_h = page.evaluate("document.body.scrollHeight") + page.set_viewport_size({"width": 1120, "height": int(full_h) + 20}) + return browser, page + + +def _points(page: Any, n: int, dsf: int) -> dict[int, tuple[tuple[int, int], ...]]: + out: dict[int, tuple[tuple[int, int], ...]] = {} + for i in range(n): + pts = [] + for sel in (f'[data-name="{i}"]', f'[data-open="{i}"]'): + bb = page.eval_on_selector( + sel, + "el => { const r = el.getBoundingClientRect();" + " return [r.x, r.y, r.width, r.height]; }", + ) + pts.append((int((bb[0] + bb[2] / 2) * dsf), int((bb[1] + bb[3] / 2) * dsf))) + out[i] = tuple(pts) + return out + + +def _supports_copy() -> bool: + from openadapt_flow.runtime.replayer import Replayer + + return "identity_copy" in inspect.signature(Replayer.__init__).parameters + + +def _replayer(backend: Any, vision: Any, variant: str, known_ids: Any) -> Any: + from openadapt_flow.runtime.replayer import Replayer + + kwargs: dict[str, Any] = {"vision": vision, "poll_interval_s": 0.0} + if variant in ("list", "list_copy"): + kwargs["known_ids"] = known_ids + if variant in ("copy", "list_copy"): + from openadapt_flow.runtime.identity_looks import IdentityCopyConfig + + kwargs["identity_copy"] = IdentityCopyConfig() + kwargs["read_clipboard"] = PageClipboard(backend) + return Replayer(backend, **kwargs) + + +def _check_both_gates( + replayer: Any, backend: PageBackend, step: Any, res: Any, wf: Any, params: dict +) -> dict[str, Any]: + entry_png = backend.screenshot() + if hasattr(replayer, "_entry_identity_gate_error"): + replayer.__dict__.setdefault("_identity_look_bindings", {}).pop(step.id, None) + replayer._identity_input_looks = True + try: + entry = replayer._verify_identity( + step, res, entry_png, params, wf, bundle_dir=None + ) + finally: + if hasattr(replayer, "_identity_input_looks"): + replayer._identity_input_looks = False + out: dict[str, Any] = { + "entry": entry.status, + "entry_reason": entry.abstain_reason, + "looks": [ + [look.kind, look.outcome, look.reason] + for look in getattr(entry, "looks", []) + ], + "look_halt": getattr(entry, "look_halt", None), + } + final = entry.status + if entry.status == "verified" and backend.dirty: + act_png = backend.screenshot() # the selection highlight is still drawn + act = replayer._verify_identity(step, res, act_png, params, wf, bundle_dir=None) + out["actuation"] = act.status + out["actuation_looks"] = [[x.kind, x.outcome, x.reason] for x in act.looks] + final = act.status + elif entry.status == "verified": + out["actuation"] = "verified" # no input was used: same frame, same verdict + out["final"] = final + backend.reset() + return out + + +def run( + seeds: list[int], + *, + n_rows: int = 40, + progress: bool = False, +) -> dict[str, Any]: + import cv2 + import numpy as np + from playwright.sync_api import sync_playwright + + from openadapt_flow._browser_setup import ensure_chromium_installed + from openadapt_flow.compiler.compile import ( + MIN_OCR_CONFIDENCE, + _discriminative_crop_region, + ) + from openadapt_flow.ir import Anchor, Resolution, Step, Workflow + from openadapt_flow.runtime.identity import ( + band_region, + context_region_from_lines, + identifier_text_from_lines, + ) + from openadapt_flow.runtime.known_ids import KnownIdList + from openadapt_flow.validation.dense_surface import ( + CLICK_CONFIGS, + RECORD_CONDITION, + REPLAY_CONDITIONS, + build_dense_table, + render_frame, + ) + + ensure_chromium_installed() + vision = CachedVision() + variants = [v for v in VARIANTS if _supports_copy() or "copy" not in v] + today = date.today() + trials: list[dict[str, Any]] = [] + # Record frames first: render_frame runs its own Playwright session. + record_frames = { + seed: render_frame( + build_dense_table(seed, n_rows=n_rows), RECORD_CONDITION, top_offset_px=0 + ) + for seed in seeds + } + with sync_playwright() as pw: + for seed in seeds: + table = build_dense_table(seed, n_rows=n_rows) + known = KnownIdList( + ids=tuple(r.mrn for r in table.rows), + source=f"synthetic seed {seed}", + complete=True, + ) + rec = record_frames[seed] + rec_lines = vision.ocr(rec.png) + rec_bgr = cv2.imdecode(np.frombuffer(rec.png, np.uint8), cv2.IMREAD_COLOR) + idx = {id(r): i for i, r in enumerate(table.rows)} + for k, cond in enumerate(REPLAY_CONDITIONS): + browser, page = _open_page(pw, table, cond, 4 + k * 3) + try: + dsf = cond.device_scale_factor + png = page.screenshot(full_page=True) + import io + + from PIL import Image + + with Image.open(io.BytesIO(png)) as im: + viewport = im.size + backend = PageBackend(page=page, dsf=dsf, viewport=viewport) + backend.clean_png = png + live_pts = _points(page, len(table.rows), dsf) + for pair in table.pairs: + ti, si = idx[id(pair.target)], idx[id(pair.sibling)] + for config in CLICK_CONFIGS: + pi = 0 if config == "click_name" else 1 + rec_click = rec.points[ti][pi] + crop = _discriminative_crop_region(rec_bgr, rec_click) + box = context_region_from_lines( + rec_lines, + exclude_region=crop, + band=band_region(rec_click, crop[3], rec.viewport), + point=rec_click, + min_confidence=MIN_OCR_CONFIDENCE, + reference_date=today, + ) + context = ( + identifier_text_from_lines( + rec_lines, + region=box, + min_confidence=MIN_OCR_CONFIDENCE, + reference_date=today, + ) + if box + else None + ) + base = { + "seed": seed, + "collision_class": pair.collision_class, + "click_config": config, + "replay_condition": cond.name, + "armed": bool(context), + } + if not context: + trials.append(base) + continue + anchor = Anchor( + template="t.png", + region=crop, + click_point=rec_click, + context_text=context, + identifier_region=box, + ) + step = Step( + id=f"s{ti}_{config}", + intent="open record", + action="click", + anchor=anchor, + risk="irreversible", + ) + for mode in MODES: + params = ( + {"mrn": pair.target.mrn} if mode == "param" else {} + ) + wf = Workflow( + name="dense", params=dict(params), steps=[step] + ) + for variant in variants: + row = dict(base, mode=mode, variant=variant) + for side, ri in (("correct", ti), ("sibling", si)): + res = Resolution( + rung="ocr", + point=live_pts[ri][pi], + confidence=0.9, + elapsed_ms=1.0, + ) + rp = _replayer(backend, vision, variant, known) + row[side] = _check_both_gates( + rp, backend, step, res, wf, params + ) + trials.append(row) + if progress: + print( + f"seed {seed} {cond.name} {pair.collision_class} done" + ) + finally: + browser.close() + return { + "schema": SCHEMA, + "meta": { + "commit": _git_commit(), + "seeds": seeds, + "n_rows": n_rows, + "replay_conditions": [c.name for c in REPLAY_CONDITIONS], + "click_configs": list(CLICK_CONFIGS), + "variants": variants, + "modes": list(MODES), + "ocr_calls": vision.calls, + }, + "trials": trials, + } + + +def score(result: dict[str, Any]) -> dict[str, Any]: + """Counts per (mode, variant), overall and by collision class.""" + out: dict[str, Any] = {} + for t in result["trials"]: + if not t.get("armed") or "variant" not in t: + continue + key = f"{t['mode']}/{t['variant']}" + for scope in ("all", t["collision_class"]): + b = out.setdefault(key, {}).setdefault( + scope, + { + "correct_row_clicks": 0, + "false_stops": 0, + "entry_verified_actuation_stopped": 0, + "sibling_row_clicks": 0, + "wrong_record_accepts": 0, + "copy_looks": 0, + }, + ) + b["correct_row_clicks"] += 1 + b["sibling_row_clicks"] += 1 + c, s = t["correct"], t["sibling"] + b["false_stops"] += c["final"] != "verified" + b["entry_verified_actuation_stopped"] += ( + c["entry"] == "verified" and c["final"] != "verified" + ) + b["wrong_record_accepts"] += ( + s["entry"] == "verified" or s.get("final") == "verified" + ) + b["copy_looks"] += sum( + 1 for x in (c, s) for look in x["looks"] if look[0] == "clipboard" + ) + return out + + +def unarmed(result: dict[str, Any]) -> int: + return sum(1 for t in result["trials"] if not t.get("armed")) + + +def main(argv: Optional[list[str]] = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--out", required=True) + parser.add_argument("--seeds", type=int, nargs="+", default=[1, 2, 3]) + parser.add_argument("--n-rows", type=int, default=40) + parser.add_argument("--progress", action="store_true") + args = parser.parse_args(argv) + result = run(args.seeds, n_rows=args.n_rows, progress=args.progress) + scored = score(result) + out = Path(args.out) + out.parent.mkdir(parents=True, exist_ok=True) + out.write_text( + json.dumps( + { + "meta": result["meta"], + "unarmed": unarmed(result), + "scored": scored, + "trials": result["trials"], + }, + indent=1, + ) + + "\n" + ) + for key, groups in scored.items(): + a = groups["all"] + print( + f"{key}: false stops {a['false_stops']}/{a['correct_row_clicks']}, " + f"wrong-record accepts {a['wrong_record_accepts']}/{a['sibling_row_clicks']}" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/flow/tests/test_identity_looks.py b/packages/flow/tests/test_identity_looks.py new file mode 100644 index 000000000..bb4835356 --- /dev/null +++ b/packages/flow/tests/test_identity_looks.py @@ -0,0 +1,287 @@ +"""Extra looks before stopping on a look-alike identifier (runtime.identity_looks). + +Pure tests of the decision rule and the copy look. The replayer path, with a +real render and real OCR, is in ``test_identity_looks_replayer.py``. +""" + +from __future__ import annotations + +import io + +import pytest +from PIL import Image, ImageDraw + +from openadapt_flow.ir import IdentityCheck +from openadapt_flow.runtime import identity_looks as L +from openadapt_flow.runtime.known_ids import KnownIdList +from openadapt_flow.runtime.replayer import identity_read_contradicts + +BAND = "AC50061 Smith, John 01/15/1980" + + +def _abstain(observed: str = BAND) -> IdentityCheck: + return IdentityCheck( + status="abstain", + abstain_reason="glyph_confusable_identifier", + coverage=1.0, + expected=BAND, + observed=observed, + ) + + +def _frames(*reads: IdentityCheck): + return [(r, "a" * 64) for r in reads] + + +def _decide(**kw): + kw.setdefault("frames", _frames(_abstain(), _abstain())) + kw.setdefault("magnified", _abstain()) + return L.decide(_abstain(), contradicts=identity_read_contradicts, **kw) + + +def _copy(text: str) -> L.ClipboardRead: + return L.ClipboardRead(text=text, frame_sha256="b" * 64) + + +def test_ocr_looks_alone_never_accept(): + check = _decide() + assert check.status == "abstain" + assert [look.kind for look in check.looks] == ["frame", "frame", "magnified"] + assert all(look.outcome == "agrees" for look in check.looks) + + +def test_copy_equal_to_run_value_settles_the_record(): + check = _decide( + clipboard=_copy("AC50061\tSmith, John\t01/15/1980"), exact_expected=("ac50061",) + ) + assert check.status == "verified" + assert check.looks[-1].kind == "clipboard" + assert check.looks[-1].outcome == "resolves" + + +def test_copy_of_look_alike_twin_is_a_mismatch(): + check = _decide( + clipboard=_copy("AC5OO61 Smith, John 01/15/1980"), exact_expected=("ac50061",) + ) + assert check.status == "mismatch" + assert check.looks[-1].outcome == "contradicts" + + +def test_copy_without_an_exact_expected_id_cannot_accept(): + # Recorded-record mode: the recorded band is OCR text, not exact. + check = _decide(clipboard=_copy("AC50061 Smith, John 01/15/1980")) + assert check.status == "abstain" + assert check.looks[-1].outcome == "agrees" + + +def test_copy_of_an_unlisted_id_contradicts_the_list(): + ids = KnownIdList(ids=("AC50061", "ZZ1000"), source="t", complete=True) + check = _decide( + clipboard=_copy("AC5OO61"), + known_ids=ids, + known_ids_lift=_abstain().model_copy(update={"status": "verified"}), + ) + assert check.status == "mismatch" + + +def test_known_id_list_resolves_when_frames_agree(): + ids = KnownIdList(ids=("AC50061", "ZZ1000"), source="t", complete=True) + lift = _abstain().model_copy(update={"status": "verified"}) + check = _decide(known_ids=ids, known_ids_lift=lift) + assert check.status == "verified" + assert check.looks[-1].kind == "known_ids" + + +def test_a_frame_naming_another_record_is_final(): + other = IdentityCheck( + status="mismatch", expected=BAND, observed="ZZ9999 Doe, Jane 02/02/1990" + ) + check = _decide( + frames=_frames(other, _abstain()), + clipboard=_copy("AC50061"), + exact_expected=("ac50061",), + ) + assert check.status == "mismatch" + assert check.looks[0].outcome == "contradicts" + + +def test_frames_that_read_another_identifier_block_acceptance(): + drifted = _abstain("AC5OO6l Smith, John 01/15/1980") # same class, still ok + moved = _abstain("AC50071 Smith, John 01/15/1980") # different class + assert ( + _decide( + frames=_frames(drifted, _abstain()), + clipboard=_copy("AC50061"), + exact_expected=("ac50061",), + ).status + == "verified" + ) + check = _decide( + frames=_frames(moved, _abstain()), + clipboard=_copy("AC50061"), + exact_expected=("ac50061",), + ) + assert check.status == "abstain" + assert check.looks[0].outcome == "unavailable" + + +def test_no_fresh_frames_means_no_acceptance(): + check = _decide(frames=[], clipboard=_copy("AC50061"), exact_expected=("ac50061",)) + assert check.status == "abstain" + + +def test_magnified_read_naming_another_record_is_final(): + other = IdentityCheck( + status="mismatch", expected=BAND, observed="ZZ9999 Doe, Jane 02/02/1990" + ) + check = _decide( + magnified=other, clipboard=_copy("AC50061"), exact_expected=("ac50061",) + ) + assert check.status == "mismatch" + + +def test_copy_that_covers_a_different_cell_settles_nothing(): + check = _decide(clipboard=_copy("ZZ1000 Filler, Pat"), exact_expected=("ac50061",)) + assert check.status == "abstain" + assert check.looks[-1].reason == "copy_differs_from_read" + + +def test_only_glyph_abstains_take_looks(): + short = IdentityCheck(status="abstain", abstain_reason="identity_evidence_short") + assert ( + L.decide( + short, frames=[], magnified=None, contradicts=identity_read_contradicts + ) + is short + ) + + +def test_exact_expected_ids_come_only_from_band_params(): + run = {"mrn": "AC50061", "note": "AC99999"} + assert L.exact_expected_ids(run, ["mrn"]) == ("ac50061",) + assert L.exact_expected_ids(run, []) == () + + +def test_copy_config_rejects_unknown_gesture(): + with pytest.raises(ValueError): + L.IdentityCopyConfig(gesture="triple_click") + + +# ------------------------------------------------------------ copy look + + +def _png(highlight=None, extra=None) -> bytes: + img = Image.new("RGB", (200, 100), "white") + draw = ImageDraw.Draw(img) + draw.text((20, 40), "AC50061", fill="black") + if highlight: + draw.rectangle(highlight, fill=(51, 144, 255)) + if extra: + draw.rectangle(extra, fill="red") + out = io.BytesIO() + img.save(out, format="PNG") + return out.getvalue() + + +class _Clipboard: + def __init__(self, text="user's own text"): + self.text = text + self.writes = [] + + def read_text(self): + return self.text + + def write_text(self, text): + self.writes.append(text) + self.text = text + + +class _Backend: + def __init__(self, clipboard, copied="AC50061", after=None): + self.clipboard = clipboard + self.copied = copied + self.after = after + self.frame = _png() + self.drags = [] + self.keys = [] + + def screenshot(self): + return self.frame + + def drag(self, x, y, ex, ey): + self.drags.append((x, y, ex, ey)) + self.frame = self.after if self.after is not None else self.frame + + def click(self, x, y, *, double=False): + raise AssertionError("drag gesture must not click") + + def press(self, key): + self.keys.append(key) + if self.copied is not None: + self.clipboard.text = self.copied + + +REGION = (15, 35, 80, 20) + + +def test_copy_reads_the_cell_and_restores_the_clipboard(): + clip = _Clipboard() + backend = _Backend(clip, after=_png(highlight=(16, 36, 90, 52))) + read = L.copy_identifier(backend, clip, REGION) + assert read.text == "AC50061" + assert not read.changed_outside + assert backend.drags and backend.drags[0][1] == 45 # mid height + assert backend.keys == ["Control+c"] + assert clip.text == "user's own text" # restored + + +def test_marker_left_in_clipboard_counts_as_nothing_copied(): + clip = _Clipboard() + read = L.copy_identifier(_Backend(clip, copied=None), clip, REGION) + assert read.text is None and read.reason == "nothing_copied" + + +def test_screen_change_outside_the_cell_is_flagged(): + clip = _Clipboard() + backend = _Backend(clip, after=_png(extra=(150, 80, 190, 95))) + read = L.copy_identifier(backend, clip, REGION) + assert read.changed_outside + + +def test_backend_without_drag_takes_no_look(): + class NoDrag: + def screenshot(self): + return _png() + + read = L.copy_identifier(NoDrag(), _Clipboard(), REGION) + assert read.text is None and read.reason == "backend_cannot_drag" + + +def test_crops_match_tolerates_noise_not_a_changed_glyph(): + a = L.crop_png(_png(), REGION) + img = Image.open(io.BytesIO(a)).convert("RGB") + px = img.load() + px[0, 0] = tuple(max(0, c - 3) for c in px[0, 0]) + noisy = io.BytesIO() + img.save(noisy, format="PNG") + assert L.crops_match(a, noisy.getvalue()) + other = Image.new("RGB", (200, 100), "white") + ImageDraw.Draw(other).text((20, 40), "AC5OO61", fill="black") + buf = io.BytesIO() + other.save(buf, format="PNG") + assert not L.crops_match(a, L.crop_png(buf.getvalue(), REGION)) + + +def test_copy_region_is_the_single_line_holding_a_lookalike_id(): + from types import SimpleNamespace as NS + + within = (0, 0, 600, 40) + mrn = NS(text="AC50061", region=(10, 5, 90, 20)) + name = NS(text="Smith, John", region=(200, 5, 120, 20)) + dob = NS(text="01/15/1980", region=(400, 5, 110, 20)) + assert L.copy_region([mrn, name, dob], within) == (8, 3, 94, 24) + other = NS(text="ZZ10O1", region=(520, 5, 60, 20)) + assert L.copy_region([mrn, other], within) is None # two candidates + assert L.copy_region([name, dob], within) is None # none + outside = NS(text="AC50061", region=(10, 100, 90, 20)) + assert L.copy_region([outside], within) is None diff --git a/packages/flow/tests/test_identity_looks_replayer.py b/packages/flow/tests/test_identity_looks_replayer.py new file mode 100644 index 000000000..aadf7526e --- /dev/null +++ b/packages/flow/tests/test_identity_looks_replayer.py @@ -0,0 +1,221 @@ +"""Extra looks on the real replayer path, real render and real OCR. + +A record row whose identifier carries look-alike glyphs (``AC50061``). OCR +abstains on it. The select-and-copy look settles the record only when the +copied identifier equals the run's parameter; the look-alike twin +``AC5OO61`` stops the run. The actuation-time check reuses the step-entry +copy only when the identifier cell shows the same pixels. +""" + +from __future__ import annotations + +import io +from datetime import date + +import pytest + +pytest.importorskip("playwright") +pytest.importorskip("rapidocr_onnxruntime") + +from PIL import Image, ImageDraw # noqa: E402 + +import openadapt_flow.vision as vision # noqa: E402 +from openadapt_flow.ir import ( # noqa: E402 + Anchor, + IdentityCheck, + Resolution, + Step, + StepResult, + Workflow, +) +from openadapt_flow.runtime import identity as I # noqa: E402 +from openadapt_flow.runtime.identity_looks import IdentityCopyConfig # noqa: E402 +from openadapt_flow.runtime.replayer import Replayer # noqa: E402 +from tests.test_identity_homonym_8th import _ROW_K, _render # noqa: E402 + +_FRAMES: dict[str, object] = {} + + +def _frame(mrn: str): + if mrn not in _FRAMES: + _FRAMES[mrn] = _render(mrn) + return _FRAMES[mrn] + + +def _identifier_region(frame) -> tuple[int, int, int, int]: + """The MRN + name + DOB cells of the row under test.""" + _, _, _, row = frame.points[_ROW_K] + lines = sorted(vision.ocr(frame.png, region=row), key=lambda ln: ln.region[0]) + mrn = next(ln for ln in lines if "AC5" in ln.text.upper()) + dob = next(ln for ln in lines if "1980" in ln.text) + x0 = mrn.region[0] - 4 + x1 = dob.region[0] + dob.region[2] + 4 + return (x0, row[1], x1 - x0, row[3]) + + +class _Clipboard: + def __init__(self) -> None: + self.text = "operator's clipboard" + + def read_text(self): + return self.text + + def write_text(self, text): + self.text = text + + +def _highlight(png: bytes, region) -> bytes: + img = Image.open(io.BytesIO(png)).convert("RGB") + x, y, w, h = region + draw = ImageDraw.Draw(img) + draw.rectangle((x + 2, y + 2, x + w - 3, y + 4), fill=(51, 144, 255)) + out = io.BytesIO() + img.save(out, format="PNG") + return out.getvalue() + + +class _CopyBackend: + """Pixel-only session whose identifier cell can be selected and copied.""" + + def __init__(self, frame, mrn: str, clipboard: _Clipboard, *, spill=False): + self.viewport = frame.viewport + self.frame = frame.png + self.region = _identifier_region(frame) + self.copy_text = mrn # the drag spans the identifier's own text box + self.clipboard = clipboard + self.spill = spill + self.drags: list[tuple[int, int, int, int]] = [] + self.selected = False + + def screenshot(self) -> bytes: + return self.frame + + def drag(self, x, y, ex, ey): + self.drags.append((x, y, ex, ey)) + rx, ry, rw, rh = self.region + self.selected = ry <= y <= ry + rh + if self.selected: # a selection paints between the drag's ends + self.frame = _highlight(self.frame, (x, y - 8, ex - x, 16)) + if self.spill: # the gesture opened something elsewhere on screen + self.frame = _highlight(self.frame, (5, 5, 60, 20)) + + def press(self, key): + assert key == "Control+c" + if self.selected: + self.clipboard.text = self.copy_text + + +def _setup(live_mrn: str, *, param_mode: bool, spill=False, copy=True): + rec = _frame("AC50061") + live = _frame(live_mrn) + _, open_pt, _, _ = rec.points[_ROW_K] + _, live_pt, _, _ = live.points[_ROW_K] + region = _identifier_region(rec) + context = I.identifier_text_from_lines( + vision.ocr(rec.png), region=region, reference_date=date.today() + ) + assert context and "1980" in context + anchor = Anchor( + template="t.png", + region=(open_pt[0] - 30, open_pt[1] - 12, 60, 24), + click_point=(open_pt[0], open_pt[1]), + context_text=context, + identifier_region=region, + ) + step = Step( + id="open_record", + intent="open record", + action="click", + anchor=anchor, + risk="irreversible", + ) + params = {"mrn": "AC50061"} if param_mode else {} + wf = Workflow(name="looks", params=dict(params), steps=[step]) + res = Resolution(rung="ocr", point=live_pt, confidence=0.9, elapsed_ms=1.0) + clip = _Clipboard() + backend = _CopyBackend(live, live_mrn, clip, spill=spill) + replayer = Replayer( + backend, + vision=vision, + read_clipboard=clip if copy else None, + identity_copy=IdentityCopyConfig() if copy else None, + ) + return replayer, backend, step, res, wf, params, live + + +def _entry(replayer, step, res, wf, params, png) -> IdentityCheck: + replayer._identity_input_looks = True + try: + return replayer._verify_identity(step, res, png, params, wf, bundle_dir=None) + finally: + replayer._identity_input_looks = False + + +def test_without_copy_look_the_correct_row_still_abstains(): + r, b, step, res, wf, params, live = _setup("AC50061", param_mode=True, copy=False) + check = _entry(r, step, res, wf, params, live.png) + assert check.status == "abstain" + assert not b.drags + + +def test_copy_equal_to_the_run_value_verifies_the_correct_row(): + r, b, step, res, wf, params, live = _setup("AC50061", param_mode=True) + check = _entry(r, step, res, wf, params, live.png) + assert check.status == "verified", check + kinds = [(look.kind, look.outcome) for look in check.looks] + assert ("clipboard", "resolves") in kinds + assert [k for k, _ in kinds].count("frame") == 2 + assert len(b.drags) == 1 + assert b.clipboard.text == "operator's clipboard" + + +def test_copy_of_the_look_alike_twin_stops_the_run(): + r, b, step, res, wf, params, live = _setup("AC5OO61", param_mode=True) + check = _entry(r, step, res, wf, params, live.png) + assert check.status == "mismatch" + assert check.looks[-1].kind == "clipboard" + assert check.looks[-1].outcome == "contradicts" + + +def test_recorded_record_mode_copy_cannot_accept(): + r, b, step, res, wf, params, live = _setup("AC50061", param_mode=False) + check = _entry(r, step, res, wf, params, live.png) + assert check.status == "abstain" + assert check.looks[-1].kind == "clipboard" + assert check.looks[-1].outcome == "agrees" + + +def test_screen_change_outside_the_cell_halts(tmp_path): + r, b, step, res, wf, params, live = _setup("AC50061", param_mode=True, spill=True) + result = StepResult(step_id=step.id, intent=step.intent, ok=False) + error = r._entry_identity_gate_error( + step, res, live.png, params, wf, tmp_path, result + ) + assert result.identity is not None + assert result.identity.look_halt == "identity_look_changed_screen" + assert error is not None and "identity_look_changed_screen" in error + assert result.safety_halt is True + + +def test_governed_runs_take_no_copy_look(): + r, b, step, res, wf, params, live = _setup("AC50061", param_mode=True) + r.governed_authorization = object() # type: ignore[assignment] + check = _entry(r, step, res, wf, params, live.png) + assert check.status == "abstain" + assert not b.drags + + +def test_actuation_check_reuses_the_copy_only_on_matching_pixels(): + r, b, step, res, wf, params, live = _setup("AC50061", param_mode=True) + assert _entry(r, step, res, wf, params, live.png).status == "verified" + # Actuation-time check: no input, the selection highlight is still drawn. + again = r._verify_identity(step, res, b.frame, params, wf, bundle_dir=None) + assert len(b.drags) == 1 + assert again.status == "verified" + assert ("clipboard", "resolves") in [(x.kind, x.outcome) for x in again.looks] + # The twin now sits where the record was: the binding must not carry over. + twin = _frame("AC5OO61") + b.frame = twin.png + swapped = r._verify_identity(step, res, twin.png, params, wf, bundle_dir=None) + assert swapped.status != "verified" + assert len(b.drags) == 1 diff --git a/packages/flow/tests/test_known_ids_replayer.py b/packages/flow/tests/test_known_ids_replayer.py index c204fdeab..6564ef347 100644 --- a/packages/flow/tests/test_known_ids_replayer.py +++ b/packages/flow/tests/test_known_ids_replayer.py @@ -86,14 +86,21 @@ def test_correct_row_abstains_without_list(): assert _check("AC50061", "AC50061", None).status == "abstain" -def test_correct_row_verifies_with_complete_list_using_a_fresh_capture(): +def test_correct_row_verifies_with_complete_list_using_fresh_captures(): out: dict = {} check = _check("AC50061", "AC50061", _ids(), out=out) assert check.status == "verified" assert check.known_id_evidence is not None - assert check.known_id_evidence.reads == 2 - assert check.known_id_evidence.frames == 2 - assert out["captures"] == 1 # the second read took its own screenshot + assert check.known_id_evidence.reads == 3 + assert check.known_id_evidence.frames == 3 + assert out["captures"] == 2 # each later read took its own screenshot + assert [look.kind for look in check.looks] == [ + "frame", + "frame", + "magnified", + "known_ids", + ] + assert check.looks[-1].outcome == "resolves" def test_fresh_capture_showing_another_record_blocks_the_lift(): From ebe48b92324b8c6f1ca72bc39d1a28018e45cf21 Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 09:27:46 -0400 Subject: [PATCH 2/6] Record each dense-list condition at its replay's device scale Co-Authored-By: Claude Opus 5.5 --- .../validation/identity_looks_dense.py | 23 ++++++++++++++----- 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/packages/flow/openadapt_flow/validation/identity_looks_dense.py b/packages/flow/openadapt_flow/validation/identity_looks_dense.py index e73bd9c5e..e901ddfb6 100644 --- a/packages/flow/openadapt_flow/validation/identity_looks_dense.py +++ b/packages/flow/openadapt_flow/validation/identity_looks_dense.py @@ -32,6 +32,7 @@ from __future__ import annotations import argparse +import dataclasses import hashlib import inspect import json @@ -293,12 +294,19 @@ def run( variants = [v for v in VARIANTS if _supports_copy() or "copy" not in v] today = date.today() trials: list[dict[str, Any]] = [] - # Record frames first: render_frame runs its own Playwright session. + # Record frames first: render_frame runs its own Playwright session. Each + # replay condition is recorded at its own device scale: the identifier + # box is a pixel region, and a display-scale change is handled by a + # separate mechanism (runtime.render_scale), not by this check. + dsfs = sorted({c.device_scale_factor for c in REPLAY_CONDITIONS}) record_frames = { - seed: render_frame( - build_dense_table(seed, n_rows=n_rows), RECORD_CONDITION, top_offset_px=0 + (seed, d): render_frame( + build_dense_table(seed, n_rows=n_rows), + dataclasses.replace(RECORD_CONDITION, device_scale_factor=d), + top_offset_px=0, ) for seed in seeds + for d in dsfs } with sync_playwright() as pw: for seed in seeds: @@ -308,11 +316,13 @@ def run( source=f"synthetic seed {seed}", complete=True, ) - rec = record_frames[seed] - rec_lines = vision.ocr(rec.png) - rec_bgr = cv2.imdecode(np.frombuffer(rec.png, np.uint8), cv2.IMREAD_COLOR) idx = {id(r): i for i, r in enumerate(table.rows)} for k, cond in enumerate(REPLAY_CONDITIONS): + rec = record_frames[(seed, cond.device_scale_factor)] + rec_lines = vision.ocr(rec.png) + rec_bgr = cv2.imdecode( + np.frombuffer(rec.png, np.uint8), cv2.IMREAD_COLOR + ) browser, page = _open_page(pw, table, cond, 4 + k * 3) try: dsf = cond.device_scale_factor @@ -408,6 +418,7 @@ def run( "seeds": seeds, "n_rows": n_rows, "replay_conditions": [c.name for c in REPLAY_CONDITIONS], + "record_condition": f"{RECORD_CONDITION.name} at each replay's device scale", "click_configs": list(CLICK_CONFIGS), "variants": variants, "modes": list(MODES), From 52f6dffe6aeaae20856c2b212480224b3dc483b6 Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 09:27:59 -0400 Subject: [PATCH 3/6] Report the commit of the package under test in the dense-list harness Co-Authored-By: Claude Opus 5.5 --- .../flow/openadapt_flow/validation/identity_looks_dense.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/flow/openadapt_flow/validation/identity_looks_dense.py b/packages/flow/openadapt_flow/validation/identity_looks_dense.py index e901ddfb6..59193e772 100644 --- a/packages/flow/openadapt_flow/validation/identity_looks_dense.py +++ b/packages/flow/openadapt_flow/validation/identity_looks_dense.py @@ -48,8 +48,12 @@ def _git_commit() -> Optional[str]: + """Commit of the openadapt_flow under test (the before run imports + another checkout's package with this script).""" try: - here = Path(__file__).resolve().parent + import openadapt_flow + + here = Path(openadapt_flow.__file__).resolve().parent out = subprocess.run( ["git", "-C", str(here), "rev-parse", "HEAD"], capture_output=True, From a80564cbd2a2785c8567a30dacd5e263e35956b1 Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 09:30:39 -0400 Subject: [PATCH 4/6] Describe the extra identity looks and their limits in LIMITS.md Co-Authored-By: Claude Opus 5.5 --- packages/flow/docs/LIMITS.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/packages/flow/docs/LIMITS.md b/packages/flow/docs/LIMITS.md index e1a48764f..df8274cf4 100644 --- a/packages/flow/docs/LIMITS.md +++ b/packages/flow/docs/LIMITS.md @@ -163,6 +163,30 @@ halts with a named reason when the confirmation is missing before and after numbers are in [`benchmark/known_ids/KNOWN_IDS.md`](../benchmark/known_ids/KNOWN_IDS.md). +Before a step stops on a look-alike identifier, the replayer takes extra looks +that change nothing +([`runtime/identity_looks.py`](../openadapt_flow/runtime/identity_looks.py)): +the same read on two fresh screenshots, the 2x re-read, the known-ID list when +one is configured, and, where the surface allows it, a select-and-copy of the +identifier's text box (`runtime.identity_copy_look`). OCR looks can't settle +the record on their own, because a look-alike twin can read the same on every +frame and at every size. Only two looks can: the known-ID list, and a copied +identifier that equals a run parameter in the identity band. A look that names +another record stops the run. Each look is recorded on the step's identity +check. + +The copy look has limits. It needs a surface where selecting the identifier +and copying it changes nothing, and a clipboard that reaches the machine Flow +runs on (RDP or Citrix clipboard redirection). It drags across only the one +text box that holds the identifier, and the run stops if the screen changes +anywhere else during the look. A canvas with no selectable text has no copy +look. A recording's own identifier is OCR text and can be a misread (the RDP +kiosk recorded `A1001` as `Alo01`), so in a workflow that always opens the +recorded record, the copy can stop a run but never confirm one. It confirms +only a record chosen by a run parameter. Governed and qualification runs never +take it. The check just before the click takes no input; it reuses the copy +only when the identifier's pixels are unchanged since the look. + Permissive `replay` may proceed on an unreadable reversible target and reports that condition. Governed `run` requires an affirmative live verdict for every identity-required admitted step; mismatch, unreadable, or abstain halts before From a708038895d806ef54aafa2d5fad43af4d60c48a Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 10:07:54 -0400 Subject: [PATCH 5/6] Fix four false stops in the extra identity looks - The copy drag now runs to the edges of the identifier's text box; it stopped inside the last glyph and copied one character short. - A split identifier (103 045 against 103045) is matched as one string, against the read and the known-ID list. - The check just before the click reuses the step-entry verdict when the identifier shows the same pixels, instead of re-reading the selected text. - A run-parameter mismatch that is only a look-alike misread (C0X1234 read as COX1234) gets the same looks. The mismatch stands unless the list or an exact copy of the run value settles the record. Co-Authored-By: Claude Opus 5.5 --- .../openadapt_flow/runtime/identity_looks.py | 72 ++++--- .../flow/openadapt_flow/runtime/replayer.py | 184 +++++++++++++++++- packages/flow/tests/test_identity_looks.py | 78 +++++++- .../tests/test_identity_looks_replayer.py | 2 +- 4 files changed, 307 insertions(+), 29 deletions(-) diff --git a/packages/flow/openadapt_flow/runtime/identity_looks.py b/packages/flow/openadapt_flow/runtime/identity_looks.py index 2eb21206e..5e28d6ed2 100644 --- a/packages/flow/openadapt_flow/runtime/identity_looks.py +++ b/packages/flow/openadapt_flow/runtime/identity_looks.py @@ -134,24 +134,25 @@ def __post_init__(self) -> None: class LookBinding: """A clipboard read taken at step entry, reusable at actuation. - The actuation-time identity check takes no input. It reuses this read only - when the identifier cell on its frame shows the same pixels as one of the - crops seen at step entry. + The actuation-time identity check takes no input. It reuses this read, or + the verdict it settled, only when the identifier cell on its frame shows + the same pixels as one of the crops seen at step entry. """ step_id: str classes: frozenset[str] read: "ClipboardRead" crops: tuple[bytes, ...] + #: The step-entry verdict when the copy settled the record. + verdict: Optional[IdentityCheck] = None + + def same_pixels(self, step_id: str, crop: Optional[bytes]) -> bool: + return step_id == self.step_id and any(crops_match(crop, c) for c in self.crops) def matches( self, step_id: str, classes: frozenset[str], crop: Optional[bytes] ) -> bool: - return ( - step_id == self.step_id - and classes == self.classes - and any(crops_match(crop, c) for c in self.crops) - ) + return classes == self.classes and self.same_pixels(step_id, crop) @dataclass(frozen=True) @@ -223,34 +224,61 @@ def _magnified_look( return _look("magnified", "agrees") +def _listed(known_ids: KnownIdList, core: str) -> bool: + return core in known_ids.members_of_class(lookalike_class(core)) + + def _clipboard_look( - first_classes: frozenset[str], - vulnerable: frozenset[str], + first_tokens: Sequence[str], copied: Optional[ClipboardRead], exact: Sequence[str], known_ids: Optional[KnownIdList], ) -> IdentityLook: + """Judge a select-and-copy read against the first OCR read. + + The copy and the read must name the same identifiers up to look-alike + glyphs: token by token, or, when one side splits an identifier the other + keeps whole (``103 045`` against ``103045``), as the joined string. + """ if copied is None: return _look("clipboard", "unavailable", "not_configured") sha = copied.frame_sha256 if copied.text is None: return _look("clipboard", "unavailable", copied.reason or "nothing_copied", sha) tokens = identifier_tokens(copied.text) + if not tokens: + return _look("clipboard", "unavailable", "no_identifier_copied", sha) + vulnerable_read = any( + identity_mod._is_glyph_vulnerable_identifier(t) for t in first_tokens + ) by_class: dict[str, set[str]] = {} for tok in tokens: by_class.setdefault(lookalike_class(tok), set()).add(tok) - if set(by_class) != set(first_classes): + read_classes = {lookalike_class(t) for t in first_tokens} + exact_set = set(exact) + if set(by_class) == read_classes: + if any(len(v) != 1 for v in by_class.values()): + return _look("clipboard", "unavailable", "copy_ambiguous", sha) + copied_ids = {k: next(iter(v)) for k, v in by_class.items()} + vulnerable = { + lookalike_class(t) + for t in first_tokens + if identity_mod._is_glyph_vulnerable_identifier(t) + } + elif lookalike_class("".join(tokens)) == lookalike_class("".join(first_tokens)): + joined = "".join(tokens) + copied_ids = {lookalike_class(joined): joined} + vulnerable = set(copied_ids) if vulnerable_read else set() + exact_set |= {"".join(exact)} if exact else set() + else: # The copy did not cover the identifier OCR read (or covered more). return _look("clipboard", "unavailable", "copy_differs_from_read", sha) - if any(len(v) != 1 for v in by_class.values()): - return _look("clipboard", "unavailable", "copy_ambiguous", sha) - copied_ids = {k: next(iter(v)) for k, v in by_class.items()} if known_ids is not None and known_ids.complete: - listed = {m for k in copied_ids for m in known_ids.members_of_class(k)} - if any(core not in listed for core in copied_ids.values()): + all_listed = all(_listed(known_ids, c) for c in copied_ids.values()) + if not all_listed and not _listed(known_ids, "".join(tokens)): return _look("clipboard", "contradicts", "copied_id_not_listed", sha) exact_by_class: dict[str, set[str]] = {} - for e in exact: + for e in exact_set: exact_by_class.setdefault(lookalike_class(e), set()).add(e) for cls, core in copied_ids.items(): expected = exact_by_class.get(cls) @@ -298,7 +326,6 @@ def decide( if not is_glyph_abstain(first): return first first_classes = lookalike_classes(first.observed) - vulnerable = vulnerable_classes(first.observed) looks: list[IdentityLook] = [] for read, sha in frames: look = _frame_look(first_classes, read, sha, contradicts) @@ -319,7 +346,7 @@ def decide( ) if clipboard is not None: clip = _clipboard_look( - first_classes, vulnerable, clipboard, exact_expected, known_ids + identifier_tokens(first.observed), clipboard, exact_expected, known_ids ) looks.append(clip) if clip.outcome == "contradicts": @@ -364,7 +391,7 @@ def copy_region(lines: Iterable[Any], within: Region) -> Optional[Region]: if len(hits) != 1: return None x, y, w, h = (int(v) for v in hits[0]) - return (x - 2, y - 2, w + 4, h + 4) + return (x - 4, y - 2, w + 8, h + 4) def crop_png(png: bytes, region: Region) -> Optional[bytes]: @@ -447,9 +474,10 @@ def copy_identifier( clipboard.write_text(marker) pre = capture() mid_y = y + h // 2 - inset = max(2, min(6, w // 10)) if gesture == "drag": - backend.drag(x + inset, mid_y, x + w - inset, mid_y) + # From just before the first glyph to just past the last one: a + # drag that ends inside the last glyph leaves it unselected. + backend.drag(x + 1, mid_y, x + w - 1, mid_y) else: backend.click(x + w // 2, mid_y, double=True) for key in copy_keys: diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 5e3a125d6..a49614da2 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -35,6 +35,7 @@ from __future__ import annotations +import dataclasses import functools import hashlib import hmac @@ -185,6 +186,9 @@ from openadapt_flow.runtime.effects.adapter import ( verifier_identity as effect_verifier_identity, ) +from openadapt_flow.runtime.known_ids import ( + lookalike_class as known_ids_lookalike_class, +) from openadapt_flow.runtime.program_predicates import ( evaluate_program_predicate, exact_png_size, @@ -14038,6 +14042,26 @@ def _verify_identity_ocr( rw, rh, ) + if identifier_region is not None and not self._identity_input_looks: + # The check just before the click takes no input. When the step-entry + # check settled the record with a copy and the identifier shows the + # same pixels now, the same record is on screen: reuse that verdict. + binding = self.__dict__.get("_identity_look_bindings", {}).get(step.id) + if ( + binding is not None + and binding.verdict is not None + and binding.same_pixels( + step.id, looks_mod.crop_png(before_png, identifier_region) + ) + ): + reused = IdentityLook( + kind="clipboard", + outcome="resolves", + reason="same_pixels_as_step_entry_copy", + ) + return binding.verdict.model_copy( + update={"looks": [*binding.verdict.looks, reused]} + ) band = identity_mod.band_region( resolution.point, anchor.region[3], self.backend.viewport ) @@ -14137,10 +14161,23 @@ def native_attempt( ) check = native_attempt(before_png, first_row_lines) - if check.status == "verified" or identity_read_contradicts(check): - # A read that names a different record is final: a later read of - # the same screen must never overrule it into "verified". + if check.status == "verified": return check + if identity_read_contradicts(check): + # A read that names a different record is final: a later read of + # the same screen must never overrule it into "verified". The one + # exception is a run parameter OCR read with look-alike glyphs + # (C0X1234 read as COX1234): extra looks may confirm the exact + # value; otherwise this mismatch stands. + return self._lookalike_param_looks( + step, + check, + native_attempt, + params, + workflow, + identifier_region, + before_png, + ) upscaled = identity_mod.upscale_crop(before_png, live_region) if upscaled is None: return check @@ -14240,6 +14277,131 @@ def with_looks(final: IdentityCheck) -> IdentityCheck: ) return with_looks(better_identity_read(check, row_check)) + def _band_check( + self, + anchor: Anchor, + observed: str, + run_values: Mapping[str, str], + workflow: Workflow, + ) -> IdentityCheck: + """The OCR-tier verdict for ``observed`` with the given run values.""" + tmpl = anchor.identity_template + if tmpl is not None and tmpl.tokens: + from openadapt_flow.runtime import identity_template as itmpl + + return itmpl.verify_template_identity( + tmpl, observed, params=dict(run_values), param_examples=workflow.params + ) + return identity_mod.verify_target_identity( + anchor.context_text or "", + observed, + params=dict(run_values), + param_examples=workflow.params, + ) + + def _lookalike_param_read( + self, + check: IdentityCheck, + anchor: Anchor, + params: Mapping[str, RuntimeParamScalar], + workflow: Workflow, + ) -> Optional[IdentityCheck]: + """A run-parameter mismatch that is only a look-alike misread, as the + look-alike abstain it would be, or None. + + The parameter's identifier is read with look-alike glyphs (the run + value ``C0X1234`` read as ``COX1234``). The band is checked again with + the value as OCR read it; when every other identity budget then passes + and only the look-alike identifier remains, OCR cannot say which of the + two it is. Used only to decide whether extra looks may run; the caller + keeps the mismatch unless a look settles the record. + """ + if check.status != "mismatch" or check.mode != "param" or not check.param: + return None + run_values = runtime_params_for_gui(params) + run = run_values.get(check.param) + if not isinstance(run, str): + return None + run_tokens = looks_mod.identifier_tokens(run) + if not any(identity_mod._is_glyph_vulnerable_identifier(t) for t in run_tokens): + return None + seen = looks_mod.identifier_tokens(check.observed) + as_read: list[str] = [] + for token in run_tokens: + same = { + o + for o in seen + if known_ids_lookalike_class(o) == known_ids_lookalike_class(token) + } + if len(same) != 1: + return None + as_read.append(same.pop()) + if as_read == run_tokens: + return None + again = self._band_check( + anchor, + check.observed, + {**run_values, check.param: " ".join(as_read)}, + workflow, + ) + if not looks_mod.is_glyph_abstain(again): + return None + return again + + def _lookalike_param_looks( + self, + step: Step, + check: IdentityCheck, + native_attempt: Callable[[bytes], IdentityCheck], + params: Mapping[str, RuntimeParamScalar], + workflow: Workflow, + identifier_region: Optional[Region], + frame_png: bytes, + ) -> IdentityCheck: + """Extra looks for a look-alike run-parameter mismatch. + + Returns the looks' ``verified`` verdict, a named halt, or ``check`` + unchanged (with the looks recorded). Never turns a mismatch into an + abstain. + """ + anchor = step.anchor + assert anchor is not None + first = self._lookalike_param_read(check, anchor, params, workflow) + if first is None or not ( + self.known_ids is not None or self._copy_look_possible(identifier_region) + ): + return check + + def read(png: bytes) -> IdentityCheck: + got = native_attempt(png) + return self._lookalike_param_read(got, anchor, params, workflow) or got + + matches: Optional[Callable[[str], bool]] = None + if anchor.identity_template is not None and anchor.identity_template.tokens: + from openadapt_flow.runtime import identity_template as itmpl + + matches = functools.partial( + itmpl.recorded_identifier_matches, anchor.identity_template + ) + looked = self._look_again( + step, + first, + None, + read, + params, + workflow, + identifier_region, + frame_png, + matches, + ) + if ( + looked.status == "verified" + or looked.known_id_halt is not None + or looked.look_halt is not None + ): + return looked + return check.model_copy(update={"looks": list(looked.looks)}) + def _copy_look_allowed(self) -> bool: """The copy look is input: never in governed or qualification runs.""" return ( @@ -14271,7 +14433,7 @@ def _look_again( self, step: Step, first: IdentityCheck, - magnified: IdentityCheck, + magnified: Optional[IdentityCheck], read: Callable[[bytes], IdentityCheck], params: Optional[Mapping[str, RuntimeParamScalar]], workflow: Workflow, @@ -14398,7 +14560,7 @@ def _look_again( known_list = None if known is not None: known_list = getattr(known, "known_ids", known) - return looks_mod.decide( + decided = looks_mod.decide( first, frames=frames, magnified=magnified, @@ -14413,6 +14575,18 @@ def _look_again( else [], ), ) + binding = bindings.get(step.id) + if ( + self._identity_input_looks + and binding is not None + and decided.status == "verified" + and any( + look.kind == "clipboard" and look.outcome == "resolves" + for look in decided.looks + ) + ): + bindings[step.id] = dataclasses.replace(binding, verdict=decided) + return decided # -- typed-input verification ------------------------------------------------- diff --git a/packages/flow/tests/test_identity_looks.py b/packages/flow/tests/test_identity_looks.py index bb4835356..03e47211e 100644 --- a/packages/flow/tests/test_identity_looks.py +++ b/packages/flow/tests/test_identity_looks.py @@ -279,9 +279,85 @@ def test_copy_region_is_the_single_line_holding_a_lookalike_id(): mrn = NS(text="AC50061", region=(10, 5, 90, 20)) name = NS(text="Smith, John", region=(200, 5, 120, 20)) dob = NS(text="01/15/1980", region=(400, 5, 110, 20)) - assert L.copy_region([mrn, name, dob], within) == (8, 3, 94, 24) + assert L.copy_region([mrn, name, dob], within) == (6, 3, 98, 24) other = NS(text="ZZ10O1", region=(520, 5, 60, 20)) assert L.copy_region([mrn, other], within) is None # two candidates assert L.copy_region([name, dob], within) is None # none outside = NS(text="AC50061", region=(10, 100, 90, 20)) assert L.copy_region([outside], within) is None + + +def test_a_split_identifier_is_matched_as_one_string(): + first = _abstain("103 O45 Smith, John 01/15/1980") + kw = dict( + frames=_frames(first, first), + magnified=first, + contradicts=identity_read_contradicts, + exact_expected=("103", "045"), + ) + assert L.decide(first, clipboard=_copy("103045"), **kw).status == "verified" + assert L.decide(first, clipboard=_copy("1O3045"), **kw).status == "mismatch" + ids = KnownIdList(ids=("103 045", "ZZ1000"), source="t", complete=True) + listed = L.decide(first, clipboard=_copy("103 045"), known_ids=ids, **kw) + assert listed.status == "verified" + + +def test_lookalike_param_mismatch_takes_looks_but_never_softens(monkeypatch): + from openadapt_flow.ir import Anchor, Step, Workflow + from openadapt_flow.runtime import identity as I + from openadapt_flow.runtime.replayer import Replayer + + band = "AC50061 Smith, John 01/15/1980" + anchor = Anchor( + template="t.png", + region=(0, 0, 10, 10), + click_point=(5, 5), + context_text=band, + identifier_region=(0, 0, 400, 20), + ) + step = Step(id="s", intent="open", action="click", anchor=anchor) + wf = Workflow(name="p", params={"mrn": "AC50061"}, steps=[step]) + params = {"mrn": "AC50061"} + + def read(observed): + return I.verify_target_identity( + band, observed, params=params, param_examples=wf.params + ) + + misread = read("AC5OO61 Smith, John 01/15/1980") + assert misread.status == "mismatch" and misread.mode == "param" + + class _Clip: + def read_text(self): + return None + + def write_text(self, text): + pass + + r = Replayer( + object(), # type: ignore[arg-type] + vision=object(), + read_clipboard=_Clip(), + identity_copy=L.IdentityCopyConfig(), + ) + normal = r._lookalike_param_read(misread, anchor, params, wf) + assert L.is_glyph_abstain(normal) + other = read("AC5OO61 Jones, Mary 02/02/1990") + assert r._lookalike_param_read(other, anchor, params, wf) is None + + def looked(status): + return lambda *a, **k: misread.model_copy( + update={"status": status, "abstain_reason": None} + ) + + native = lambda png: misread # noqa: E731 + monkeypatch.setattr(r, "_look_again", looked("verified")) + got = r._lookalike_param_looks( + step, misread, native, params, wf, (0, 0, 400, 20), b"" + ) + assert got.status == "verified" + monkeypatch.setattr(r, "_look_again", looked("abstain")) + got = r._lookalike_param_looks( + step, misread, native, params, wf, (0, 0, 400, 20), b"" + ) + assert got.status == "mismatch" # the original mismatch stands diff --git a/packages/flow/tests/test_identity_looks_replayer.py b/packages/flow/tests/test_identity_looks_replayer.py index aadf7526e..27d23ccd5 100644 --- a/packages/flow/tests/test_identity_looks_replayer.py +++ b/packages/flow/tests/test_identity_looks_replayer.py @@ -212,7 +212,7 @@ def test_actuation_check_reuses_the_copy_only_on_matching_pixels(): again = r._verify_identity(step, res, b.frame, params, wf, bundle_dir=None) assert len(b.drags) == 1 assert again.status == "verified" - assert ("clipboard", "resolves") in [(x.kind, x.outcome) for x in again.looks] + assert again.looks[-1].reason == "same_pixels_as_step_entry_copy" # The twin now sits where the record was: the binding must not carry over. twin = _frame("AC5OO61") b.frame = twin.png From b97f6715ac3daa2c22972b22a212b1bd62fcaa8b Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 15:02:02 -0400 Subject: [PATCH 6/6] Fix CI: harness types, a docs path, and an OCR-dependent test - identity_looks_dense: pass a typed click point and ActionKind (mypy). - LIMITS.md: name identity_looks.py by its package path, which the docs consistency check resolves. - The look-alike twin test asserts the stop; it accepts a read that is already final, as Linux OCR builds read it. Co-Authored-By: Claude Opus 5.5 --- packages/flow/docs/LIMITS.md | 2 +- .../openadapt_flow/validation/identity_looks_dense.py | 7 ++++--- packages/flow/tests/test_identity_looks_replayer.py | 8 ++++++-- 3 files changed, 11 insertions(+), 6 deletions(-) diff --git a/packages/flow/docs/LIMITS.md b/packages/flow/docs/LIMITS.md index c69bd8b65..01ca44f31 100644 --- a/packages/flow/docs/LIMITS.md +++ b/packages/flow/docs/LIMITS.md @@ -190,7 +190,7 @@ before and after numbers are in Before a step stops on a look-alike identifier, the replayer takes extra looks that change nothing -([`runtime/identity_looks.py`](../openadapt_flow/runtime/identity_looks.py)): +([`openadapt_flow/runtime/identity_looks.py`](../openadapt_flow/runtime/identity_looks.py)): the same read on two fresh screenshots, the 2x re-read, the known-ID list when one is configured, and, where the surface allows it, a select-and-copy of the identifier's text box (`runtime.identity_copy_look`). OCR looks can't settle diff --git a/packages/flow/openadapt_flow/validation/identity_looks_dense.py b/packages/flow/openadapt_flow/validation/identity_looks_dense.py index 59193e772..259d02e9b 100644 --- a/packages/flow/openadapt_flow/validation/identity_looks_dense.py +++ b/packages/flow/openadapt_flow/validation/identity_looks_dense.py @@ -278,7 +278,7 @@ def run( MIN_OCR_CONFIDENCE, _discriminative_crop_region, ) - from openadapt_flow.ir import Anchor, Resolution, Step, Workflow + from openadapt_flow.ir import ActionKind, Anchor, Resolution, Step, Workflow from openadapt_flow.runtime.identity import ( band_region, context_region_from_lines, @@ -344,7 +344,8 @@ def run( ti, si = idx[id(pair.target)], idx[id(pair.sibling)] for config in CLICK_CONFIGS: pi = 0 if config == "click_name" else 1 - rec_click = rec.points[ti][pi] + name_pt, open_pt = rec.points[ti][0], rec.points[ti][1] + rec_click = name_pt if pi == 0 else open_pt crop = _discriminative_crop_region(rec_bgr, rec_click) box = context_region_from_lines( rec_lines, @@ -384,7 +385,7 @@ def run( step = Step( id=f"s{ti}_{config}", intent="open record", - action="click", + action=ActionKind.CLICK, anchor=anchor, risk="irreversible", ) diff --git a/packages/flow/tests/test_identity_looks_replayer.py b/packages/flow/tests/test_identity_looks_replayer.py index 27d23ccd5..8f4f3f2d9 100644 --- a/packages/flow/tests/test_identity_looks_replayer.py +++ b/packages/flow/tests/test_identity_looks_replayer.py @@ -172,9 +172,13 @@ def test_copy_equal_to_the_run_value_verifies_the_correct_row(): def test_copy_of_the_look_alike_twin_stops_the_run(): r, b, step, res, wf, params, live = _setup("AC5OO61", param_mode=True) check = _entry(r, step, res, wf, params, live.png) + # It stops either way. Where OCR reads the twin as a look-alike of the + # run value, the copy is what names it; where OCR already reads another + # record (some OCR builds), the read is final and takes no looks. assert check.status == "mismatch" - assert check.looks[-1].kind == "clipboard" - assert check.looks[-1].outcome == "contradicts" + if check.looks: + assert check.looks[-1].kind == "clipboard" + assert check.looks[-1].outcome == "contradicts" def test_recorded_record_mode_copy_cannot_accept():