diff --git a/packages/flow/openadapt_flow/runtime/identity_census.py b/packages/flow/openadapt_flow/runtime/identity_census.py index c0839bc43..dc7a7d76d 100644 --- a/packages/flow/openadapt_flow/runtime/identity_census.py +++ b/packages/flow/openadapt_flow/runtime/identity_census.py @@ -138,3 +138,49 @@ def count_matching_rows( if abs(y - target_y) >= same_row_px and text.strip() and verifies(text) ) return others + 1 + + +# Structural roles (backend-neutral, as the recorder stores them) of a form +# field or a control. A target recorded with one of these is not a record row. +CONTROL_ROLES = frozenset( + { + "textbox", + "searchbox", + "combobox", + "checkbox", + "radio", + "spinbutton", + "switch", + "slider", + "button", + "tab", + "menuitem", + } +) + + +def teach_time_role(step: Any) -> Literal["row", "control", "unknown"]: + """What teach time recorded the step's target as. + + ``row`` when the recorder saw a record row (the census at record time), + ``control`` when it saw no row there, or when the target's structural + locator names a form-field or control role (:data:`CONTROL_ROLES`). + ``unknown`` otherwise, which the runtime counts like a row: a link, a + cell, a list item, or a target with no recorded role. + """ + anchor = getattr(step, "anchor", None) + if anchor is None: + return "unknown" + template = getattr(anchor, "identity_template", None) + recorded_row = getattr(template, "structured_row", None) + if recorded_row is True: + return "row" + if recorded_row is False: + return "control" + if getattr(template, "structured", None): + # A structured identity was recorded but nothing counted its rows (a + # backend without a census, a legacy bundle): it may be a row's text. + return "unknown" + structural = getattr(anchor, "structural", None) + role = (getattr(structural, "role", None) or "").strip().casefold() + return "control" if role in CONTROL_ROLES else "unknown" diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index 64d104330..389316b0d 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -15707,18 +15707,14 @@ def _apply_candidate_census( "param", ): return check + teach_role = census_mod.teach_time_role(step) if check.mode == "structured": decision = self._structured_census(step, resolution.point) else: decision = self._ocr_census(step, resolution, before_png, params, workflow) - tmpl = step.anchor.identity_template if step.anchor is not None else None - if ( - decision.verdict == "cannot_count" - and check.mode == "structured" - and tmpl is not None - and tmpl.structured_row is False - ): - # Teach time saw no record row here (a form field, a button). + if decision.verdict == "cannot_count" and teach_role == "control": + # Teach time recorded a form field or a control, and no census can + # count here. A census that can count always decides. decision = census_mod.CensusDecision("not_a_row") if decision.verdict == "not_a_row": return check.model_copy(update={"census_status": "not_a_row"}) @@ -17448,6 +17444,12 @@ def miss(reason: str) -> bool: check = self._verify_identity( step, resolution, frame_png, params, workflow, bundle_dir ) + if check.status != "verified" and check.ambiguity_halt is not None: + # The target is in view and its identity matched, but the census + # refused it. Scrolling cannot change that; the scroll loop stops + # and reports this instead of spending its budget. + self._scroll_census_refusal = (step.id, check) + return miss("census") return check.status == "verified" or miss("identity") def _act_scroll( @@ -17553,9 +17555,41 @@ def readiness_holds(frame_png: bytes) -> bool: workflow=workflow, ) + def census_refusal() -> Optional[str]: + """The halt for a target in view whose census refused it.""" + + refusal = self.__dict__.pop("_scroll_census_refusal", None) + if not misses or misses[-1] != "census" or refusal is None: + return None + target_id, check = refusal + why = { + "identical_candidates": ( + f"{check.candidate_count} shown rows carry the same identity " + "and no recorded record id tells them apart" + ), + "identity_uncounted": ( + "this surface cannot count the rows that might share it, " + "and a consequential write follows" + ), + }.get( + check.ambiguity_halt or "", + "the rows that could share its identity could not all be counted", + ) + result.safety_halt = True + return ( + f"Step '{step.id}' ({step.intent}): the target of step " + f"'{target_id}' is in view and its identity matched, but {why} " + f"({check.ambiguity_halt}); scrolling cannot change that, so " + "the scroll stops here and no input was sent for it" + ) + + self.__dict__.pop("_scroll_census_refusal", None) holds = readiness_holds(before_png) if self._governed_asset_mutation is not None: return self._governed_asset_mutation + stop = census_refusal() + if stop is not None: + return stop if holds: # Target already in view; nothing to scroll and no input sent. # Record why, only for the next-target readiness check, so the @@ -17594,6 +17628,9 @@ def readiness_holds(frame_png: bytes) -> bool: return scroll_error if readiness_holds(fresh_scroll_frame): return None + stop = census_refusal() + if stop is not None: + return stop remote_preflight_error = self._prepare_remote_scroll_input(step, result) if remote_preflight_error is not None: return remote_preflight_error @@ -17616,6 +17653,9 @@ def readiness_holds(frame_png: bytes) -> bool: holds = readiness_holds(frame) if self._governed_asset_mutation is not None: return self._governed_asset_mutation + stop = census_refusal() + if stop is not None: + return stop if holds: return None diff --git a/packages/flow/tests/test_census_teach_role.py b/packages/flow/tests/test_census_teach_role.py new file mode 100644 index 000000000..9ef238b41 --- /dev/null +++ b/packages/flow/tests/test_census_teach_role.py @@ -0,0 +1,202 @@ +"""The candidate census follows the target's teach-time role, whatever rung +resolved it. + +A form field or control recorded at teach time (OpenEMR's "Contact" section +button, resolved by its pixel template) has no look-alike rows to count, so a +verified identity stands before a save. A target recorded as a row, or with no +recorded role, is still counted and halts when it cannot be. A scroll whose +next target is in view but refused by the census stops and says so instead of +spending its budget. +""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pytest + +from openadapt_flow.ir import IdentityCheck, StructuralLocator, Workflow +from openadapt_flow.runtime import identity_template as itmpl +from openadapt_flow.runtime.identity_census import teach_time_role +from openadapt_flow.runtime.replayer import Replayer +from openadapt_flow.vision import Match +from tests.test_identity_evidence_capability import ( + _W, + _button, + _crop_only_step, + _gate, +) +from tests.test_replayer import ( + FakeBackend, + FakeVision, + OcrLine, + click_step, + context_click_step, + make_png, + scroll_step, +) + + +@pytest.fixture() +def bundle(tmp_path: Path) -> Path: + bundle_dir = tmp_path / "bundle" + (bundle_dir / "templates").mkdir(parents=True) + (bundle_dir / "templates" / "btn.png").write_bytes(make_png((50, 20))) + return bundle_dir + + +@pytest.fixture() +def run_dir(tmp_path: Path) -> Path: + return tmp_path / "run" + + +def _with_role(step, role: str | None): + assert step.anchor is not None + step.anchor.structural = ( + StructuralLocator(role=role, name="Contact") if role is not None else None + ) + return step + + +@pytest.mark.parametrize("role", ["button", "textbox", "combobox", "checkbox"]) +def test_a_pixel_resolved_form_field_before_a_save_verifies( + tmp_path: Path, role: str +) -> None: + """The OpenEMR case: a recorded control, verified by its pixel crop.""" + + step, res = _crop_only_step(tmp_path, _button()) # irreversible + _with_role(step, role) + assert teach_time_role(step) == "control" + error, check = _gate(tmp_path, step, res, _button(), pixel_verify_enabled=True) + assert error is None + assert check is not None and check.status == "verified" + assert check.census_status == "not_a_row" + + +@pytest.mark.parametrize("role", [None, "row", "listitem", "link", "gridcell"]) +def test_a_pixel_resolved_row_that_cannot_be_counted_still_halts( + tmp_path: Path, role: str | None +) -> None: + step, res = _crop_only_step(tmp_path, _button()) + _with_role(step, role) + assert teach_time_role(step) == "unknown" + error, check = _gate(tmp_path, step, res, _button(), pixel_verify_enabled=True) + assert error is not None and "identity_uncounted" in error + assert check is not None and check.ambiguity_halt == "identity_uncounted" + + +def test_a_row_recorded_at_teach_time_is_counted_even_with_a_control_role( + tmp_path: Path, +) -> None: + """A row's own button: teach time saw the row, so the role does not exempt it.""" + + step, res = _crop_only_step(tmp_path, _button()) + _with_role(step, "button") + step.anchor.identity_template = itmpl.build_identity_template( + None, structured_identity="Open", structured_row=True + ) + assert teach_time_role(step) == "row" + error, check = _gate(tmp_path, step, res, _button(), pixel_verify_enabled=True) + assert error is not None + assert check is not None and check.status != "verified" + + +def test_a_row_with_ocr_look_alikes_halts(bundle, run_dir) -> None: + """Two OCR rows read the recorded band: the step stops before the click.""" + + band = "Jane Sample Knee pain referral High" + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.99) + ] * 4 + vision.ocr_lines = [ + OcrLine(band, region=(160, 95, 240, 20)), + OcrLine(band, region=(160, 195, 240, 20)), + ] + step = context_click_step(band, risk="irreversible") + report = Replayer(FakeBackend(), vision=vision, poll_interval_s=0.01).run( + Workflow(name="wf", steps=[step]), bundle_dir=bundle, run_dir=run_dir + ) + result = report.results[0] + assert report.success is False + assert result.identity is not None + assert result.identity.ambiguity_halt == "identical_candidates" + assert result.delivery_attempted is False + + +def test_a_scroll_stops_when_the_census_refuses_a_target_in_view( + bundle, run_dir, monkeypatch +) -> None: + vision = FakeVision() + target = Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.99) + vision.template_results = [target] * 6 + backend = FakeBackend() + click = click_step() + assert click.anchor is not None + click.anchor.context_text = "Synthetic Contact Address" + click.identity_armed = True + workflow = Workflow(name="wf", steps=[scroll_step(), click]) + replayer = Replayer(backend, vision=vision, poll_interval_s=0.01) + monkeypatch.setattr( + replayer, + "_verify_identity", + lambda *args, **kwargs: IdentityCheck( + status="abstain", + mode="pixel", + ambiguity_halt="identity_uncounted", + ), + ) + report = replayer.run(workflow, bundle_dir=bundle, run_dir=run_dir) + assert report.success is False + assert len(report.results) == 1 + halted = report.results[0] + assert halted.safety_halt is True + assert "is in view" in halted.error and "identity_uncounted" in halted.error + assert "exhausted" not in halted.error + assert backend.actions == [] # in view on the first frame: no wheel input + + +def test_teach_time_role_unknown_without_any_record() -> None: + step = click_step() + assert step.anchor is not None + step.anchor.structural = None + assert teach_time_role(step) == "unknown" + assert _W > 0 and np is not None + + +@pytest.mark.parametrize("role", ["button", "menuitem", "tab", "checkbox"]) +def test_a_control_role_never_skips_an_ocr_census_that_can_count( + bundle, run_dir, role +) -> None: + """A per-row "Open" button: the role does not exempt OCR look-alikes.""" + + band = "Jane Sample Knee pain referral High" + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.99) + ] * 4 + vision.ocr_lines = [ + OcrLine(band, region=(160, 95, 240, 20)), + OcrLine(band, region=(160, 195, 240, 20)), + ] + step = context_click_step(band, risk="irreversible") + step.anchor.structural = StructuralLocator(role=role, name="Open") + backend = FakeBackend() + report = Replayer(backend, vision=vision, poll_interval_s=0.01).run( + Workflow(name="wf", steps=[step]), bundle_dir=bundle, run_dir=run_dir + ) + assert report.success is False + assert backend.actions == [] + assert report.results[0].identity.ambiguity_halt == "identical_candidates" + + +def test_a_recorded_structured_identity_with_no_teach_census_is_not_a_control() -> None: + """A backend with no census at teach time: the row text may be a row's.""" + + step = click_step() + _with_role(step, "button") + step.anchor.identity_template = itmpl.build_identity_template( + None, structured_identity="John Smith Follow-up Dr Lee" + ) + assert teach_time_role(step) == "unknown" diff --git a/packages/flow/tests/test_realism_browser.py b/packages/flow/tests/test_realism_browser.py index 8469bdae2..1f8ac623d 100644 --- a/packages/flow/tests/test_realism_browser.py +++ b/packages/flow/tests/test_realism_browser.py @@ -108,11 +108,14 @@ def _worklist(query: dict[str, list[str]]) -> str: """ +_PAGES = {"/form.html": _form, "/worklist.html": _worklist} + + class _Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: # noqa: N802 parsed = urlparse(self.path) query = parse_qs(parsed.query) - page = {"/form.html": _form, "/worklist.html": _worklist}.get(parsed.path) + page = _PAGES.get(parsed.path) if page is None: self.send_error(404) return @@ -450,3 +453,75 @@ def test_a_backend_that_cannot_count_rows_still_runs_a_plain_form( assert page.locator("#status").inner_text() == "Saved Ada" finally: page.close() + + +def _row_buttons(query: dict[str, list[str]]) -> str: + """Identical table rows, each with an Edit button whose id is positional.""" + order = query.get("order", ["ab"])[0] + charts = {"a": "1001", "b": "1002"} + rows = "".join( + f'John Smith Follow-up Dr Lee' + f'' + "" + for i, k in enumerate(order) + ) + return f"""Worklist +

Worklist

{rows}
+
No chart open
+ +
Nothing signed
+""" + + +_PAGES["/rows.html"] = _row_buttons + + +def test_a_row_button_taught_without_a_census_halts_after_a_swap( + app_url, browser, tmp_path +) -> None: + """Taught and replayed on a surface that cannot count rows: the Edit + button's role must not exempt the row it sits in.""" + + url = app_url + "/rows.html" + page = _page(browser, url) + try: + recorder = Recorder( + _CannotCountBackend(page), + tmp_path / "recording", + app_url=url, + settle_interval_s=0.05, + settle_timeout_s=2.0, + ) + recorder.click(*_center(page, "#edit-0")) + recorder.click(*_center(page, "#sign")) + assert page.locator("#signed").inner_text() == "Signed chart 1001" + recording = recorder.finish() + finally: + page.close() + compile_recording(recording, tmp_path / "bundle", name="rows") + report, page = _replay( + browser, + url + "?order=ba", + tmp_path / "bundle", + tmp_path / "run", + backend=_CannotCountBackend, + ) + try: + assert page.locator("#signed").inner_text() != "Signed chart 1002", _summary( + report + ) + assert not report.success, _summary(report) + assert report.results[0].identity.ambiguity_halt == "identity_uncounted" + finally: + page.close()