fix(flow): the census follows the target's teach-time role on every rung

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"

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
 

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"