From 5476285a233d2dfe73845aa99dafde6cd6e3195b Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 12:24:12 -0400
Subject: [PATCH 1/4] fix(flow): halt when another shown row has the same
 identity

Replay could open and sign the wrong patient's chart and report success.
Two worklist rows read "John Smith Follow-up Dr Lee" (data-id 1001 and
1002). After they swapped order, the template rung landed on 1002 and the
structured identity check verified it, because the text matched exactly.
Nothing checked whether the identity picked one record.

After the identity ladder verifies, the runtime now counts the shown rows
with the same identity (a DOM census on the browser backend, an OCR pass
over the frame on pixel-only screens):

- one row: the verdict stands;
- several rows and the bundle recorded the row's stable record id (an
  app-owned data-* id or the row link's path, captured at teach time and
  stored as a salted hash): act only on the row with that id, moving a
  visual resolution onto it when needed;
- several rows and no recorded id: the step halts before it acts,
  whatever its risk (IdentityCheck.ambiguity_halt);
- a recorded id that no shown row carries: identity mismatch.

The realism audit's real-browser test passes and loses its xfail. New
browser cases cover the recorded order, the swap with ids, two and three
identical rows without ids, and a unique row after a swap.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/flow/openadapt_flow/backend.py       |  26 ++
 .../backends/playwright_backend.py            | 167 ++++++++
 .../flow/openadapt_flow/compiler/compile.py   |   9 +
 packages/flow/openadapt_flow/ir.py            |  56 +++
 packages/flow/openadapt_flow/recorder.py      |  22 +
 .../openadapt_flow/runtime/identity_census.py | 127 ++++++
 .../runtime/identity_template.py              |  24 ++
 .../flow/openadapt_flow/runtime/replayer.py   | 230 +++++++++-
 .../flow/openadapt_flow/sanitized_artifact.py |   1 +
 packages/flow/tests/test_identity_census.py   | 144 +++++++
 packages/flow/tests/test_realism_browser.py   | 402 ++++++++++++++++++
 11 files changed, 1207 insertions(+), 1 deletion(-)
 create mode 100644 packages/flow/openadapt_flow/runtime/identity_census.py
 create mode 100644 packages/flow/tests/test_identity_census.py
 create mode 100644 packages/flow/tests/test_realism_browser.py

diff --git a/packages/flow/openadapt_flow/backend.py b/packages/flow/openadapt_flow/backend.py
index 4e591623b..d782cf9eb 100644
--- a/packages/flow/openadapt_flow/backend.py
+++ b/packages/flow/openadapt_flow/backend.py
@@ -218,6 +218,32 @@ def structured_text_at(self, x: int, y: int) -> Optional[str]:
         ...
 
 
+@runtime_checkable
+class IdentityCandidatesBackend(Protocol):
+    """Optional census of rows that share the target's record identity.
+
+    ``structured_text_at`` says what the row under a point is. It cannot say
+    whether that text picks one record: two worklist rows that both read
+    "John Smith Follow-up Dr Lee" are each an exact match for the recorded
+    identity, and after the rows swap order the resolver lands on the other
+    patient. ``identity_candidates_at`` counts every shown row whose identity
+    equals the identity under the point and reads each row's stable record id
+    (app-owned ``data-*`` ids, the row link's path). The runtime acts only
+    when the target is the one shown row with that identity, or when the
+    recorded record id picks exactly one of them.
+
+    Returns a dict ``{"record_key": str | None, "incomplete": bool,
+    "candidates": [{"self": bool, "record_key": str | None, "dx": float,
+    "dy": float}, ...], "top_level": bool}``, where ``dx``/``dy`` are each
+    row's offset from the row under the point in click coordinates, or None
+    when the point has no row identity or the census failed (never raises).
+    """
+
+    def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]:
+        """Census of the shown rows with the same identity as (x, y)."""
+        ...
+
+
 @runtime_checkable
 class FieldLabelBackend(Protocol):
     """Optional RECORD-TIME field-label capability a backend MAY expose.
diff --git a/packages/flow/openadapt_flow/backends/playwright_backend.py b/packages/flow/openadapt_flow/backends/playwright_backend.py
index 8e95edacb..8bd3bb25e 100644
--- a/packages/flow/openadapt_flow/backends/playwright_backend.py
+++ b/packages/flow/openadapt_flow/backends/playwright_backend.py
@@ -700,6 +700,138 @@ class _FramePoint:
 }"""
 )
 
+# The stable record id a row carries in the DOM: app-owned ``data-*`` id
+# attributes on the target, its ancestors up to the row, and the row, plus the
+# path and id-like query parameters of the row's link. Index-like attribute
+# names, and values that are only the row's position, are left out: a key that
+# moves with the row would point at whichever record now sits there. Null when
+# the row exposes no such id.
+_RECORD_KEY_JS = r"""(row, node) => {
+    const ID_NAME = /^data-(?:[a-z0-9_-]*-)?(?:id|key|uid|uuid|guid|mrn|pid|record|recordid)$/i;
+    const POSITIONAL = /(?:index|idx|position|pos|order|rank|seq)/i;
+    const parent = row.parentElement;
+    const index = parent ? Array.prototype.indexOf.call(parent.children, row) : -1;
+    const positional = (value) => {
+        const m = /^\D*(\d+)\D*$/.exec(value);
+        if (!m) return false;
+        const n = Number(m[1]);
+        return n === index || n === index + 1;
+    };
+    const parts = [];
+    let cursor = node;
+    for (let depth = 0; cursor && depth < 12; depth += 1) {
+        for (const attr of Array.from(cursor.attributes || [])) {
+            const name = attr.name.toLowerCase();
+            if (name.startsWith('data-openadapt-actuation-')) continue;
+            if (name !== 'data-openadapt-record' &&
+                    (!ID_NAME.test(name) || POSITIONAL.test(name))) continue;
+            const value = (attr.value || '').trim();
+            if (!value || value.length > 256 || positional(value)) continue;
+            parts.push(name + '=' + value);
+        }
+        if (cursor === row) break;
+        cursor = cursor.parentElement;
+    }
+    let link = node.closest('a[href]');
+    if (!link || !row.contains(link)) {
+        const links = row.querySelectorAll('a[href]');
+        link = row.matches('a[href]') ? row : (links.length === 1 ? links[0] : null);
+    }
+    if (link) {
+        try {
+            const url = new URL(link.getAttribute('href'), location.href);
+            if (url.protocol === 'http:' || url.protocol === 'https:') {
+                const ids = [];
+                for (const [k, v] of url.searchParams) {
+                    if (/(?:^|[_-])(?:id|key|uid|uuid|guid|mrn|pid)$/i.test(k) ||
+                            /^(?:id|pid|mrn)$/i.test(k)) ids.push(k + '=' + v);
+                }
+                ids.sort();
+                const here = url.pathname === location.pathname && !ids.length;
+                const value = url.pathname + (ids.length ? '?' + ids.join('&') : '') + url.hash;
+                if (!(here && !url.hash) && value.length <= 512 && !positional(value)) {
+                    parts.push('href=' + value);
+                }
+            }
+        } catch (e) {}
+    }
+    if (!parts.length) return null;
+    return Array.from(new Set(parts)).sort().join('\n');
+}"""
+
+# Every shown row whose record identity equals the identity of the row under
+# the point, with each row's stable record id and position. The identity is
+# computed exactly as ``structured_text_at`` does, for the cell at the same
+# place in each row. Null when the point has no row identity (a form field
+# group, a standalone control): there is no row to compare.
+_IDENTITY_CANDIDATES_AT_JS = (
+    r"""([px, py, cap]) => {
+    const el = document.elementFromPoint(px, py);
+    if (!el) return null;
+    const describe = """
+    + _DESCRIBE_TARGET_JS
+    + r""";
+    const recordKey = """
+    + _RECORD_KEY_JS
+    + r""";
+    const target = describe(el);
+    if (!target.rowIdentity || !target.row) return null;
+    const row = target.row;
+    const own = el.closest('td, th, [role="cell"], [role="gridcell"]') || el;
+    const path = [];
+    let cursor = own;
+    while (cursor && cursor !== row) {
+        const parent = cursor.parentElement;
+        if (!parent) break;
+        path.unshift(Array.prototype.indexOf.call(parent.children, cursor));
+        cursor = parent;
+    }
+    const ownInRow = cursor === row;
+    const shown = (node) => {
+        if (!node.isConnected) return false;
+        if (typeof node.checkVisibility === 'function') {
+            return node.checkVisibility({visibilityProperty: true, opacityProperty: true});
+        }
+        if (!node.getClientRects().length) return false;
+        const style = window.getComputedStyle(node);
+        return style.visibility !== 'hidden' && style.display !== 'none';
+    };
+    const rows = document.querySelectorAll('tr, [role="row"], li, [role="listitem"]');
+    const rowRect = row.getBoundingClientRect();
+    const result = {
+        record_key: recordKey(row, el),
+        incomplete: rows.length > cap,
+        candidates: [],
+    };
+    if (result.incomplete) return result;
+    for (const candidate of rows) {
+        if (candidate !== row && !shown(candidate)) continue;
+        let node = candidate;
+        if (ownInRow) {
+            for (const index of path) node = node && node.children[index];
+        }
+        if (!node) continue;
+        const observed = candidate === row ? target : describe(node);
+        if (observed.row !== candidate || observed.rowIdentity !== target.rowIdentity) {
+            continue;
+        }
+        const rect = candidate.getBoundingClientRect();
+        result.candidates.push({
+            self: candidate === row,
+            record_key: candidate === row ? result.record_key : recordKey(candidate, node),
+            dx: rect.x - rowRect.x,
+            dy: rect.y - rowRect.y,
+        });
+    }
+    return result;
+}"""
+)
+
+# Rows scanned by the candidate census before it gives up. A document with
+# more rows than this reports ``incomplete`` and the runtime cannot prove the
+# target unique.
+IDENTITY_CANDIDATE_ROW_CAP = 5000
+
 _TEXT_VALUE_AT_JS = r"""([px, py]) => {
     const hit = document.elementFromPoint(px, py);
     if (!hit) return null;
@@ -1640,6 +1772,41 @@ def structured_text_at(self, x: int, y: int) -> Optional[str]:
             return None
         return result or None
 
+    def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]:
+        """Census of the rows that show the same record identity as (x, y).
+
+        ``structured_text_at`` says what the row under the point is; this says
+        whether any OTHER shown row says the same thing. Two worklist rows
+        reading "John Smith Follow-up Dr Lee" are both a verified match for
+        the recorded text, so the text alone cannot pick the recorded record
+        (see :class:`openadapt_flow.backend.IdentityCandidatesBackend`).
+
+        Returns ``{"record_key", "incomplete", "candidates"}``: the stable
+        record id of the row under the point (app-owned ``data-*`` ids and the
+        row link's path, or None), whether the document had too many rows to
+        scan, and one entry per shown row with the same identity (the row
+        under the point included, ``self`` true) carrying its record id and
+        its offset ``(dx, dy)`` from the row under the point, in viewport CSS
+        pixels. None when the point has no row identity or on any failure.
+        """
+        try:
+            point = self._frame_point(int(x), int(y))
+            if point is None:
+                return None
+            result = point.scope.evaluate(
+                _IDENTITY_CANDIDATES_AT_JS,
+                [point.x, point.y, IDENTITY_CANDIDATE_ROW_CAP],
+            )
+        except Exception:
+            return None
+        if not isinstance(result, dict):
+            return None
+        # Offsets are in the hit frame's own pixels. Outside the top-level
+        # document they cannot be mapped back with certainty, so a retarget
+        # is not offered there; the count still holds.
+        result["top_level"] = not point.frame_path
+        return result
+
     def text_value_at(self, x: int, y: int) -> Optional[str]:
         """Return the exact value of the editable control under ``(x, y)``.
 
diff --git a/packages/flow/openadapt_flow/compiler/compile.py b/packages/flow/openadapt_flow/compiler/compile.py
index 0433bf997..d3bdedf68 100644
--- a/packages/flow/openadapt_flow/compiler/compile.py
+++ b/packages/flow/openadapt_flow/compiler/compile.py
@@ -2274,6 +2274,12 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]:
             # OCR context band; replay prefers it (no OCR glyph ambiguity) and
             # falls back to the band on pixel-only substrates.
             structured_identity = event.get("structured_identity") or None
+            # The row's stable record id (a DOM data-* id, the row link's
+            # path), captured with the structured identity. Hashed into the
+            # identity template so replay acts only on the row with this id.
+            record_key = (
+                event.get("record_key") or None if structured_identity else None
+            )
             # Structural locator (DOM selector / role+name, or UIA identifiers)
             # of the clicked element, captured by the recorder when the
             # recording backend exposed it (StructuralActionBackend). Drives the
@@ -2381,6 +2387,7 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]:
                 ocr_text=anchor_ocr_text,
                 context_text=context_text,
                 structured_identity=structured_identity,
+                record_key=record_key,
                 structural=structural,
                 landmarks=anchor_landmarks,
                 identifier_crop=identifier_crop_rel,
@@ -3590,9 +3597,11 @@ def _phi_free_identity(workflow: Workflow) -> None:
             structured_identity=anchor.structured_identity,
             param_examples=workflow.params,
             salt_hex=salt,
+            record_key=anchor.record_key,
         )
         if template is None:
             continue
         anchor.identity_template = template
         anchor.context_text = None
         anchor.structured_identity = None
+        anchor.record_key = None
diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py
index d79a7734d..ac9693e55 100644
--- a/packages/flow/openadapt_flow/ir.py
+++ b/packages/flow/openadapt_flow/ir.py
@@ -323,6 +323,14 @@ class IdentityTemplate(BaseModel):
     )
     param_token_indices: dict[str, list[int]] = Field(default_factory=dict)
     rests_on_confusable_identifier: bool = False
+    record_key: Optional[str] = Field(
+        default=None,
+        description=(
+            "Salted hash of the stable record id the target's row carried at "
+            "teach time (app-owned DOM data-* ids, the row link's path). When "
+            "set, replay acts only on the shown row with the same id."
+        ),
+    )
 
     @model_serializer(mode="wrap")
     def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
@@ -341,6 +349,8 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
             data.pop("signal_hashes", None)
         if not self.context_params:
             data.pop("context_params", None)
+        if self.record_key is None:
+            data.pop("record_key", None)
         return data
 
 
@@ -400,6 +410,14 @@ class Anchor(BaseModel):
             " ladder then falls back to the OCR context_text tier."
         ),
     )
+    record_key: Optional[str] = Field(
+        default=None,
+        description=(
+            "Stable record id of the target's row at record time (plaintext,"
+            " compile-time only). The PHI-free pass hashes it into"
+            " ``identity_template.record_key`` and clears this field."
+        ),
+    )
     identity_template: Optional[IdentityTemplate] = Field(
         default=None,
         description=(
@@ -451,6 +469,14 @@ class Anchor(BaseModel):
         description="Pixels of padding around `region` for the local search",
     )
 
+    @model_serializer(mode="wrap")
+    def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
+        """Leave an unset ``record_key`` out, so older bundle bytes stay equal."""
+        data: dict[str, Any] = handler(self)
+        if self.record_key is None:
+            data.pop("record_key", None)
+        return data
+
 
 class PostconditionKind(str, Enum):
     TEXT_PRESENT = "text_present"
@@ -3693,6 +3719,36 @@ class IdentityCheck(BaseModel):
             "identifier it read. The step halts."
         ),
     )
+    candidate_count: Optional[int] = Field(
+        default=None,
+        ge=0,
+        description=(
+            "How many shown rows carry the same record identity as the "
+            "resolved target, the target included (runtime.identity_census). "
+            "None when the backend cannot count them."
+        ),
+    )
+    ambiguity_halt: Optional[
+        Literal["identical_candidates", "candidates_uncounted"]
+    ] = Field(
+        default=None,
+        description=(
+            "Set when the identity matched but does not pick one record: "
+            "another shown row has the same identity and no stable record id "
+            "tells them apart, or there were too many rows to count. The "
+            "step halts whatever its risk."
+        ),
+    )
+
+    @model_serializer(mode="wrap")
+    def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
+        """Leave unset census fields out, so older report bytes stay equal."""
+        data: dict[str, Any] = handler(self)
+        if self.candidate_count is None:
+            data.pop("candidate_count", None)
+        if self.ambiguity_halt is None:
+            data.pop("ambiguity_halt", None)
+        return data
 
 
 class HealEvent(BaseModel):
diff --git a/packages/flow/openadapt_flow/recorder.py b/packages/flow/openadapt_flow/recorder.py
index f27a6b6a9..6b396c3cf 100644
--- a/packages/flow/openadapt_flow/recorder.py
+++ b/packages/flow/openadapt_flow/recorder.py
@@ -687,6 +687,9 @@ def _record(self, event: dict[str, Any], act: Callable[[], None]) -> None:
             structured = self._structured_identity_at(int(event["x"]), int(event["y"]))
             if structured:
                 event = {**event, "structured_identity": structured}
+                record_key = self._record_key_at(int(event["x"]), int(event["y"]))
+                if record_key:
+                    event = {**event, "record_key": record_key}
             # The part of the form the target sits in (a legend, an accordion
             # or tab header, a heading), for the program's named sections.
             # Display only: the compiler never puts it on a step.
@@ -1018,6 +1021,25 @@ def _structured_identity_at(self, x: int, y: int) -> Optional[str]:
         except Exception:
             return None
 
+    def _record_key_at(self, x: int, y: int) -> Optional[str]:
+        """Stable record id of the row under (x, y), if the backend exposes it.
+
+        Backends MAY expose ``identity_candidates_at``
+        (``openadapt_flow.backend.IdentityCandidatesBackend``). The row's
+        app-owned id (a DOM ``data-*`` id, the row link's path) lets replay
+        tell apart rows whose visible text is identical. None when the
+        backend cannot read one or the row has none.
+        """
+        getter = getattr(self._backend, "identity_candidates_at", None)
+        if getter is None:
+            return None
+        try:
+            census = getter(int(x), int(y))
+        except Exception:
+            return None
+        key = census.get("record_key") if isinstance(census, dict) else None
+        return key if isinstance(key, str) and key else None
+
     def _structural_state(self) -> dict[str, Any]:
         """Structural observations the backend can provide right now.
 
diff --git a/packages/flow/openadapt_flow/runtime/identity_census.py b/packages/flow/openadapt_flow/runtime/identity_census.py
new file mode 100644
index 000000000..e942b0d99
--- /dev/null
+++ b/packages/flow/openadapt_flow/runtime/identity_census.py
@@ -0,0 +1,127 @@
+"""Does the verified identity pick one record? Count the look-alikes.
+
+The identity ladder (:mod:`openadapt_flow.runtime.identity`) answers "does the
+row at the resolved point read like the recorded row?". That is not the same
+question as "is this the recorded record?". Two worklist rows that both read
+"John Smith Follow-up Dr Lee" each match the recorded text exactly. When they
+swap order, every resolution rung (structural, template, text) can land on the
+other row, and the ladder verifies it, because the text cannot tell the two
+apart.
+
+So after the ladder verifies, the runtime counts the shown rows that carry the
+same identity (``IdentityCandidatesBackend.identity_candidates_at`` for DOM
+rows, an OCR pass over the frame for pixel-only screens) and decides:
+
+* one such row: the identity picks one record; the verdict stands;
+* several rows, and the bundle recorded the row's stable record id (an
+  app-owned ``data-*`` id, the row link's path): act only on the one row with
+  that id. When the resolver landed on another row, move to the right one
+  (``retarget``) or, when it cannot move, refuse (``wrong_record``);
+* several rows and no recorded id, or more than one row with the recorded id:
+  ``ambiguous``. The step halts before it acts, whatever its risk;
+* the rows could not all be counted: ``uncounted``, which also halts.
+
+A recorded id is required even when the row is unique: a single row with
+another id is a different record that happens to read the same
+(``wrong_record``).
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Literal, Mapping, NamedTuple, Optional
+
+CensusVerdict = Literal[
+    "unique", "retarget", "ambiguous", "uncounted", "wrong_record", "unavailable"
+]
+
+
+class CensusDecision(NamedTuple):
+    """What the candidate census says about the resolved target.
+
+    Attributes:
+        verdict: See the module docstring.
+        candidate_count: Shown rows with the target's identity, the target
+            included; None when they could not be counted.
+        offset: For ``retarget``, the ``(dx, dy)`` from the resolved row to
+            the row with the recorded record id, in click coordinates.
+    """
+
+    verdict: CensusVerdict
+    candidate_count: Optional[int] = None
+    offset: Optional[tuple[float, float]] = None
+
+
+def decide(
+    census: Optional[Mapping[str, Any]],
+    *,
+    has_recorded_key: bool,
+    key_matches: Callable[[Optional[str]], bool],
+) -> CensusDecision:
+    """Turn a backend census into a decision.
+
+    Args:
+        census: ``identity_candidates_at``'s result, or None when the backend
+            has no census or the point has no row identity.
+        has_recorded_key: Whether the bundle recorded the row's record id
+            (and the identity is not bound to a run parameter).
+        key_matches: Whether a live record id equals the recorded one.
+    """
+    if not isinstance(census, Mapping):
+        return CensusDecision("unavailable")
+    if census.get("incomplete"):
+        return CensusDecision("uncounted")
+    raw = census.get("candidates")
+    if not isinstance(raw, list):
+        return CensusDecision("uncounted")
+    candidates = [c for c in raw if isinstance(c, Mapping)]
+    if len(candidates) != len(raw) or sum(bool(c.get("self")) for c in raw) != 1:
+        # The row under the point must appear exactly once in its own census.
+        return CensusDecision("uncounted")
+    count = len(candidates)
+    if has_recorded_key:
+        matching = [c for c in candidates if key_matches(c.get("record_key"))]
+        if len(matching) > 1:
+            return CensusDecision("ambiguous", count)
+        if not matching:
+            return CensusDecision("wrong_record", count)
+        chosen = matching[0]
+        if chosen.get("self"):
+            return CensusDecision("unique", count)
+        if not census.get("top_level", True):
+            return CensusDecision("wrong_record", count)
+        try:
+            offset = (float(chosen["dx"]), float(chosen["dy"]))
+        except (KeyError, TypeError, ValueError):
+            return CensusDecision("wrong_record", count)
+        return CensusDecision("retarget", count, offset)
+    if count == 1:
+        return CensusDecision("unique", count)
+    return CensusDecision("ambiguous", count)
+
+
+def count_matching_rows(
+    rows: list[tuple[int, str]],
+    *,
+    target_y: int,
+    verifies: Callable[[str], bool],
+    same_row_px: int,
+) -> int:
+    """Count OCR rows whose text verifies against the recorded identity.
+
+    Pixel-only screens have no DOM to ask. Each OCR text row of the frame,
+    other than the resolved target's own row (within ``same_row_px`` of
+    ``target_y``), is checked with the same verifier the ladder used; the
+    result counts the target too.
+
+    Args:
+        rows: ``(center_y, text)`` per OCR text row of the frame.
+        target_y: The resolved point's y.
+        verifies: Whether a row's text verifies as the recorded identity.
+        same_row_px: Rows closer than this to ``target_y`` are the target's.
+    """
+    others = sum(
+        1
+        for y, text in rows
+        if abs(y - target_y) >= same_row_px and text.strip() and verifies(text)
+    )
+    return others + 1
diff --git a/packages/flow/openadapt_flow/runtime/identity_template.py b/packages/flow/openadapt_flow/runtime/identity_template.py
index 8bbd52caa..b733f1da0 100644
--- a/packages/flow/openadapt_flow/runtime/identity_template.py
+++ b/packages/flow/openadapt_flow/runtime/identity_template.py
@@ -217,6 +217,7 @@ def build_identity_template(
     structured_identity: Optional[str] = None,
     param_examples: Optional[dict[str, str]] = None,
     salt_hex: Optional[str] = None,
+    record_key: Optional[str] = None,
 ) -> Optional[IdentityTemplate]:
     """Build a PHI-free identity template from the recorded plaintext band.
 
@@ -228,6 +229,11 @@ def build_identity_template(
         param_examples: ``workflow.params`` (param name -> demo value), used to
             mark which band tokens were a parameter value (param-mode re-anchor).
         salt_hex: Reuse an existing per-bundle salt; a fresh one otherwise.
+        record_key: The stable record id of the target's row at record time
+            (``IdentityCandidatesBackend``). Hashed in only when the
+            structured identity embeds no parameter: a parameterized row is a
+            different record on every run, so the demonstrated id is not the
+            run's.
 
     Returns:
         An :class:`IdentityTemplate`, or None when there is neither a usable
@@ -252,6 +258,8 @@ def build_identity_template(
             param_examples=param_examples,
             names=structured_params,
         )
+        if record_key and not structured_params:
+            tmpl.record_key = record_key_hash(salt, record_key)
 
     # Mirror the compiler's context_from_lines gate: a band shorter than
     # MIN_CONTEXT_CHARS is too generic to discriminate and is never stored (any
@@ -358,6 +366,22 @@ def build_identity_template(
 # ---------------------------------------------------------------------------
 
 
+def record_key_hash(salt: bytes, record_key: str) -> str:
+    """Full-width salted hash of a row's stable record id."""
+
+    return _signal_hash(salt, "record_key|" + record_key)
+
+
+def record_key_matches(tmpl: IdentityTemplate, live_key: Optional[str]) -> bool:
+    """Whether a live row's record id is the one the template recorded."""
+
+    if tmpl.record_key is None or not live_key:
+        return False
+    return hmac.compare_digest(
+        tmpl.record_key, record_key_hash(_salt_bytes(tmpl.salt), live_key)
+    )
+
+
 def _keyfn(salt: bytes):
     def canon(squashed: str) -> str:
         return _hash(salt, _id.ocr_canonical(squashed))
diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py
index 38756fb7c..71af017e4 100644
--- a/packages/flow/openadapt_flow/runtime/replayer.py
+++ b/packages/flow/openadapt_flow/runtime/replayer.py
@@ -159,6 +159,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_census as census_mod
 from openadapt_flow.runtime import identity_looks as looks_mod
 from openadapt_flow.runtime.authorization import (
     GovernedRunAuthorization,
@@ -10663,6 +10664,20 @@ def _identity_gate_error(
                 "target, the screen answered differently from the recording "
                 f"({reason}) — refusing to click; run aborted"
             )
+        elif check.ambiguity_halt is not None:
+            if check.ambiguity_halt == "identical_candidates":
+                why = (
+                    f"{check.candidate_count} shown rows carry the same "
+                    "identity and no recorded record id tells them apart"
+                )
+            else:
+                why = "the rows that could share its identity could not all be counted"
+            error = (
+                f"Step '{step.id}' ({step.intent}) matched the recorded "
+                f"identity, but {why} (rung '{resolution.rung}') — acting "
+                "could open another record; refusing to act"
+            )
+            result.safety_halt = True
         elif check.mode == "hover":
             # Hover evidence only adds a stop. A step that had no identity
             # check before keeps going when the response is absent or cannot
@@ -11712,6 +11727,9 @@ def _resolve_step(
                 ),
             )
         resolution, matched_region = resolved
+        resolution, matched_region = self._retarget_by_record_key(
+            step, resolution, matched_region
+        )
 
         if step.risk == "irreversible" and is_below_ocr(resolution.rung):
             return (
@@ -15480,9 +15498,219 @@ def ocr_tier() -> Optional[IdentityCheck]:
                 step, resolution, before_png, params, workflow
             )
 
-        return identity_mod.run_identity_ladder(
+        check = identity_mod.run_identity_ladder(
             [structured_tier, pixel_tier, vlm_tier, ocr_tier]
         )
+        return self._apply_candidate_census(
+            step, resolution, check, before_png, params, workflow
+        )
+
+    def _structured_census(self, step: Step, point: Point) -> census_mod.CensusDecision:
+        """The DOM/a11y candidate census at ``point`` (see identity_census)."""
+
+        anchor = step.anchor
+        assert anchor is not None
+        getter = getattr(self.backend, "identity_candidates_at", None)
+        if getter is None:
+            return census_mod.CensusDecision("unavailable")
+        try:
+            census = getter(int(point[0]), int(point[1]))
+        except Exception:
+            census = None
+        tmpl = anchor.identity_template
+        recorded = tmpl if tmpl is not None and tmpl.record_key else None
+        from openadapt_flow.runtime import identity_template as itmpl
+
+        return census_mod.decide(
+            census,
+            has_recorded_key=recorded is not None,
+            key_matches=lambda key: (
+                recorded is not None and itmpl.record_key_matches(recorded, key)
+            ),
+        )
+
+    def _ocr_census(
+        self,
+        step: Step,
+        resolution: Resolution,
+        before_png: bytes,
+        params: Mapping[str, RuntimeParamScalar],
+        workflow: Workflow,
+    ) -> census_mod.CensusDecision:
+        """Count the frame's OCR rows that verify as the recorded identity.
+
+        The pixel-only counterpart of :meth:`_structured_census`: every text
+        row of the frame is read the way the OCR tier reads the target's row
+        (the same band height, the target's own crop excluded at the same
+        offset, volatile lines dropped) and checked against the recorded band.
+        """
+        anchor = step.anchor
+        assert anchor is not None
+        tmpl = anchor.identity_template
+        if not (anchor.context_text or (tmpl is not None and tmpl.tokens)):
+            return census_mod.CensusDecision("unavailable")
+        today = date.today()
+        try:
+            lines = [
+                line
+                for line in self.vision.ocr(before_png, region=None)
+                if line.text.strip()
+                and not identity_mod.is_volatile_line(line.text, reference_date=today)
+            ]
+        except Exception:
+            return census_mod.CensusDecision("uncounted")
+        viewport = self.backend.viewport
+        px, py = int(resolution.point[0]), int(resolution.point[1])
+        ex, ey = (
+            anchor.region[0] - anchor.click_point[0],
+            anchor.region[1] - (anchor.click_point[1]),
+        )
+        run_params = runtime_params_for_gui(params)
+
+        def row_text(y: int) -> str:
+            band = identity_mod.band_region((px, y), anchor.region[3], viewport)
+            exclude = (px + ex, y + ey, anchor.region[2], anchor.region[3])
+            kept = [
+                line
+                for line in lines
+                if identity_mod.regions_intersect(line.region, band)
+                and not identity_mod.regions_intersect(line.region, exclude)
+            ]
+            return " ".join(
+                line.text.strip() for line in identity_mod.lines_near_point(kept, y)
+            )
+
+        def verifies(text: str) -> bool:
+            if tmpl is not None and tmpl.tokens:
+                from openadapt_flow.runtime import identity_template as itmpl
+
+                verdict = itmpl.verify_template_identity(
+                    tmpl, text, params=run_params, param_examples=workflow.params
+                )
+            else:
+                assert anchor.context_text is not None
+                verdict = identity_mod.verify_target_identity(
+                    anchor.context_text,
+                    text,
+                    params=run_params,
+                    param_examples=workflow.params,
+                )
+            return verdict.status == "verified"
+
+        target_lines = identity_mod.lines_near_point(lines, py)
+        heights = sorted(line.region[3] for line in target_lines) or [16]
+        same_row_px = max(4, heights[len(heights) // 2])
+        centers: list[int] = []
+        for line in sorted(lines, key=lambda item: item.region[1]):
+            center = line.region[1] + line.region[3] // 2
+            if not centers or center - centers[-1] > 3:
+                centers.append(center)
+        rows = [(y, row_text(y)) for y in centers]
+        count = census_mod.count_matching_rows(
+            rows, target_y=py, verifies=verifies, same_row_px=same_row_px
+        )
+        if count > 1:
+            return census_mod.CensusDecision("ambiguous", count)
+        return census_mod.CensusDecision("unique", count)
+
+    def _apply_candidate_census(
+        self,
+        step: Step,
+        resolution: Resolution,
+        check: IdentityCheck,
+        before_png: bytes,
+        params: Mapping[str, RuntimeParamScalar],
+        workflow: Workflow,
+    ) -> IdentityCheck:
+        """Keep ``verified`` only when the identity picks one shown record.
+
+        A verified ladder verdict says the target reads like the recorded
+        row. When another shown row reads the same, the verdict cannot tell
+        them apart, whichever rung resolved the point. The census
+        (:mod:`openadapt_flow.runtime.identity_census`) then decides: the
+        recorded record id picks the row, or the step halts.
+        """
+        if check.status != "verified" or check.mode not in (
+            "structured",
+            "pixel",
+            "vlm",
+            "context",
+            "param",
+        ):
+            return check
+        decision = self._structured_census(step, resolution.point)
+        if decision.verdict == "unavailable" and check.mode != "structured":
+            decision = self._ocr_census(step, resolution, before_png, params, workflow)
+        if decision.verdict == "unavailable":
+            return check
+        if decision.verdict == "unique":
+            return check.model_copy(
+                update={"candidate_count": decision.candidate_count}
+            )
+        if decision.verdict in ("wrong_record", "retarget"):
+            return check.model_copy(
+                update={
+                    "status": "mismatch",
+                    "coverage": 0.0,
+                    "expected": "<recorded record id>",
+                    "candidate_count": decision.candidate_count,
+                }
+            )
+        return check.model_copy(
+            update={
+                "status": "abstain",
+                "candidate_count": decision.candidate_count,
+                "ambiguity_halt": (
+                    "candidates_uncounted"
+                    if decision.verdict == "uncounted"
+                    else "identical_candidates"
+                ),
+            }
+        )
+
+    def _retarget_by_record_key(
+        self,
+        step: Step,
+        resolution: Resolution,
+        matched_region: Region,
+    ) -> tuple[Resolution, Region]:
+        """Move a visual resolution onto the row with the recorded record id.
+
+        When rows read identically, a template or text rung lands on whichever
+        row now sits where the recorded one was. The recorded record id names
+        the right row; move the point (and matched region) by that row's
+        offset. A structural resolution names one exact element and is never
+        moved: if it is the wrong record, the identity check refuses it.
+        """
+        anchor = step.anchor
+        if (
+            anchor is None
+            or resolution.structural_handle is not None
+            or anchor.identity_template is None
+            or not anchor.identity_template.record_key
+        ):
+            return resolution, matched_region
+        decision = self._structured_census(step, resolution.point)
+        if decision.verdict != "retarget" or decision.offset is None:
+            return resolution, matched_region
+        dx, dy = (int(round(v)) for v in decision.offset)
+        x, y = resolution.point[0] + dx, resolution.point[1] + dy
+        width, height = self.backend.viewport
+        if not (0 <= x < width and 0 <= y < height):
+            return resolution, matched_region
+        moved = resolution.model_copy(
+            update={
+                "point": (x, y),
+                # The retained visual evidence reproduces the old point.
+                "visual_evidence": None,
+                "resolution_evidence_regions": tuple(
+                    (rx + dx, ry + dy, rw, rh)
+                    for rx, ry, rw, rh in resolution.resolution_evidence_regions
+                ),
+            }
+        )
+        rx, ry, rw, rh = matched_region
+        return moved, (rx + dx, ry + dy, rw, rh)
 
     def _identifier_displacement(
         self,
diff --git a/packages/flow/openadapt_flow/sanitized_artifact.py b/packages/flow/openadapt_flow/sanitized_artifact.py
index b76b806a5..19e2399b9 100644
--- a/packages/flow/openadapt_flow/sanitized_artifact.py
+++ b/packages/flow/openadapt_flow/sanitized_artifact.py
@@ -244,6 +244,7 @@ def _apply_text_redactions(
         "pages_before",
         "param",
         "params",
+        "record_key",
         "role",
         "secret",
         "secret_params",
diff --git a/packages/flow/tests/test_identity_census.py b/packages/flow/tests/test_identity_census.py
new file mode 100644
index 000000000..04eb7fcaa
--- /dev/null
+++ b/packages/flow/tests/test_identity_census.py
@@ -0,0 +1,144 @@
+"""The identity candidate census: does a verified identity pick one record?
+
+Real-browser coverage is in ``tests/test_realism_browser.py`` (the worklist
+cases). These tests pin the decision rules and the teach-time record id.
+"""
+
+from __future__ import annotations
+
+from openadapt_flow.ir import IdentityCheck, IdentityTemplate
+from openadapt_flow.runtime import identity_template as itmpl
+from openadapt_flow.runtime.identity_census import (
+    CensusDecision,
+    count_matching_rows,
+    decide,
+)
+
+ROW = "John Smith Follow-up Dr Lee"
+
+
+def _census(*rows: tuple[bool, str | None, float], top_level: bool = True) -> dict:
+    """Rows as ``(is_target, record_key, dy)``."""
+
+    target_key = next(key for is_target, key, _ in rows if is_target)
+    return {
+        "record_key": target_key,
+        "incomplete": False,
+        "top_level": top_level,
+        "candidates": [
+            {"self": is_target, "record_key": key, "dx": 0.0, "dy": dy}
+            for is_target, key, dy in rows
+        ],
+    }
+
+
+def _template(record_key: str | None) -> IdentityTemplate:
+    tmpl = itmpl.build_identity_template(
+        None, structured_identity=ROW, record_key=record_key
+    )
+    assert tmpl is not None
+    return tmpl
+
+
+def _decide(census: dict, record_key: str | None) -> CensusDecision:
+    tmpl = _template(record_key)
+    return decide(
+        census,
+        has_recorded_key=tmpl.record_key is not None,
+        key_matches=lambda key: itmpl.record_key_matches(tmpl, key),
+    )
+
+
+def test_two_identical_rows_without_ids_are_ambiguous() -> None:
+    census = _census((True, None, 0.0), (False, None, 49.0))
+    assert _decide(census, None) == CensusDecision("ambiguous", 2)
+
+
+def test_three_identical_rows_without_ids_are_ambiguous() -> None:
+    census = _census((False, None, -49.0), (True, None, 0.0), (False, None, 49.0))
+    assert _decide(census, None) == CensusDecision("ambiguous", 3)
+
+
+def test_distinct_ids_on_identical_text_pick_the_recorded_row() -> None:
+    """The resolver landed on 1002; the recorded 1001 is one row below."""
+
+    census = _census((True, "data-id=1002", 0.0), (False, "data-id=1001", 49.0))
+    assert _decide(census, "data-id=1001") == CensusDecision("retarget", 2, (0.0, 49.0))
+
+
+def test_the_recorded_id_under_the_point_verifies_among_homonyms() -> None:
+    census = _census((True, "data-id=1001", 0.0), (False, "data-id=1002", 49.0))
+    assert _decide(census, "data-id=1001") == CensusDecision("unique", 2)
+
+
+def test_live_ids_without_a_recorded_id_do_not_pick_a_row() -> None:
+    """Distinct live ids say the rows differ, not which one was recorded."""
+
+    census = _census((True, "data-id=1002", 0.0), (False, "data-id=1001", 49.0))
+    assert _decide(census, None) == CensusDecision("ambiguous", 2)
+
+
+def test_a_unique_row_verifies() -> None:
+    assert _decide(_census((True, None, 0.0)), None) == CensusDecision("unique", 1)
+    keyed = _census((True, "data-id=1003", 0.0))
+    assert _decide(keyed, "data-id=1003") == CensusDecision("unique", 1)
+
+
+def test_a_unique_row_with_another_record_id_is_the_wrong_record() -> None:
+    census = _census((True, "data-id=2002", 0.0))
+    assert _decide(census, "data-id=1001") == CensusDecision("wrong_record", 1)
+
+
+def test_the_recorded_id_on_two_rows_is_ambiguous() -> None:
+    census = _census((True, "data-id=1001", 0.0), (False, "data-id=1001", 49.0))
+    assert _decide(census, "data-id=1001") == CensusDecision("ambiguous", 2)
+
+
+def test_no_retarget_inside_a_frame() -> None:
+    census = _census(
+        (True, "data-id=1002", 0.0), (False, "data-id=1001", 49.0), top_level=False
+    )
+    assert _decide(census, "data-id=1001").verdict == "wrong_record"
+
+
+def test_incomplete_or_malformed_census_is_uncounted() -> None:
+    assert _decide({"incomplete": True}, None).verdict == "uncounted"
+    assert _decide({"candidates": "x"}, None).verdict == "uncounted"
+    # The target must appear exactly once in its own census.
+    assert _decide({"candidates": []}, None).verdict == "uncounted"
+    assert _decide(None, None).verdict == "unavailable"
+
+
+def test_record_id_is_stored_hashed_and_only_for_unparameterized_rows() -> None:
+    tmpl = _template("data-id=1001")
+    assert tmpl.record_key is not None and "1001" not in tmpl.record_key
+    assert itmpl.record_key_matches(tmpl, "data-id=1001")
+    assert not itmpl.record_key_matches(tmpl, "data-id=1002")
+    assert not itmpl.record_key_matches(tmpl, None)
+    # A row picked by a run parameter is a different record on every run.
+    param = itmpl.build_identity_template(
+        None,
+        structured_identity=ROW,
+        param_examples={"patient": "John Smith"},
+        record_key="data-id=1001",
+    )
+    assert param is not None and param.structured_params == ["patient"]
+    assert param.record_key is None
+
+
+def test_unset_record_id_and_census_fields_leave_serialized_bytes_unchanged() -> None:
+    assert "record_key" not in _template(None).model_dump()
+    dumped = IdentityCheck(status="verified").model_dump()
+    assert "candidate_count" not in dumped and "ambiguity_halt" not in dumped
+
+
+def test_ocr_rows_count_every_verifying_row_but_the_target() -> None:
+    rows = [(100, ROW), (149, ROW), (198, "Maria Garcia Intake Dr Lee")]
+    count = count_matching_rows(
+        rows, target_y=100, verifies=lambda text: text == ROW, same_row_px=16
+    )
+    assert count == 2
+    unique = count_matching_rows(
+        rows[2:], target_y=198, verifies=lambda text: text == ROW, same_row_px=16
+    )
+    assert unique == 1
diff --git a/packages/flow/tests/test_realism_browser.py b/packages/flow/tests/test_realism_browser.py
new file mode 100644
index 000000000..4c311c96f
--- /dev/null
+++ b/packages/flow/tests/test_realism_browser.py
@@ -0,0 +1,402 @@
+"""Real-browser cases the fakes do not model (docs/audits/REALISM_AUDIT.md).
+
+Each test records a short task on a local web app in headless Chromium,
+compiles it, then replays it after the page changes in a way real apps do:
+
+* the field the task types into already holds text (form restore, autofill,
+  a search box that keeps its last query);
+* the app raises a ``confirm()`` the recording never saw;
+* two worklist rows show the same visible text and swap order;
+* the page retitles on every navigation.
+
+Every replay must end in one of two ways: the run halts, or it did exactly
+what was recorded. A run that reports success after doing something else is
+the failure these tests exist to catch.
+"""
+
+from __future__ import annotations
+
+import threading
+from collections.abc import Iterator
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+
+pytest.importorskip("playwright.sync_api")
+
+from openadapt_flow.backends.playwright_backend import PlaywrightBackend  # noqa: E402
+from openadapt_flow.compiler import compile_recording  # noqa: E402
+from openadapt_flow.ir import Workflow  # noqa: E402
+from openadapt_flow.recorder import Recorder  # noqa: E402
+from openadapt_flow.runtime.replayer import Replayer  # noqa: E402
+
+pytestmark = pytest.mark.timeout(600)
+
+VIEWPORT = {"width": 900, "height": 600}
+
+_STYLE = """
+body { font-family: sans-serif; margin: 24px; font-size: 18px; }
+input, button { font-size: 18px; padding: 6px; margin: 4px; }
+li { padding: 10px; border: 1px solid #888; margin: 6px; list-style: none;
+     cursor: pointer; width: 420px; }
+#status { margin-top: 16px; font-weight: bold; }
+"""
+
+
+def _form(query: dict[str, list[str]]) -> str:
+    prefill = query.get("prefill", [""])[0]
+    ask = "confirm" in query
+    return f"""<!doctype html><html><head><title>New note</title>
+<style>{_STYLE}</style></head><body>
+<h1>New note</h1>
+<p><label for="name">Patient</label> <input id="name" autocomplete="off"></p>
+<p style="margin-top:120px">Clinic intake form, draft copy
+<button id="save">Save note</button></p>
+<div id="status">Not saved</div>
+<script>
+const prefill = {prefill!r};
+// Autofill (or an app restoring the last value) writes the field when it
+// gains focus, after the click that the recording saw on an empty field.
+const field = document.getElementById('name');
+field.addEventListener('focus', () => {{
+  if (prefill && !field.dataset.filled) {{ field.dataset.filled = '1'; field.value = prefill; }}
+}});
+document.getElementById('save').onclick = () => {{
+  if ({str(ask).lower()} && !confirm('Discard the draft on the other tab?')) return;
+  const v = document.getElementById('name').value;
+  document.getElementById('status').textContent = 'Saved ' + v;
+  document.title = 'Saved ' + v;
+}};
+</script></body></html>"""
+
+
+def _worklist(query: dict[str, list[str]]) -> str:
+    order = query.get("order", ["ab"])[0]
+    # ``noid`` keeps each row's id in an attribute that is not a record id,
+    # so only the visible text can tell the rows apart.
+    attr = "data-chart" if "noid" in query else "data-id"
+    rows = {
+        "a": ("1001", "John Smith Follow-up Dr Lee"),
+        "b": ("1002", "John Smith Follow-up Dr Lee"),
+        "c": ("1003", "Maria Garcia Intake Dr Lee"),
+        "d": ("1004", "John Smith Follow-up Dr Lee"),
+    }
+    items = "".join(
+        f'<li {attr}="{rows[k][0]}" onclick="openRow(this)">{rows[k][1]}</li>'
+        for k in order + "c"
+    )
+    return f"""<!doctype html><html><head><title>Worklist</title>
+<style>{_STYLE}</style></head><body>
+<h1>Worklist</h1><ul>{items}</ul><div id="status">No chart open</div>
+<div id="chart" style="display:none; margin-top:24px">
+  <p>Chart header: John Smith, follow-up visit</p>
+  <button id="sign">Sign visit note</button>
+</div>
+<div id="signed">Nothing signed</div>
+<script>
+let open = null;
+function openRow(li) {{
+  open = li.dataset.id || li.dataset.chart;
+  document.getElementById('status').textContent = 'Opened chart ' + open;
+  document.getElementById('chart').style.display = 'block';
+}}
+document.getElementById('sign').onclick = () => {{
+  document.getElementById('signed').textContent = 'Signed chart ' + open;
+}};
+</script></body></html>"""
+
+
+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)
+        if page is None:
+            self.send_error(404)
+            return
+        body = page(query).encode()
+        self.send_response(200)
+        self.send_header("Content-Type", "text/html; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *_args: object) -> None:
+        pass
+
+
+@pytest.fixture(scope="module")
+def app_url() -> Iterator[str]:
+    server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    yield f"http://127.0.0.1:{server.server_address[1]}"
+    server.shutdown()
+
+
+@pytest.fixture(scope="module")
+def browser() -> Iterator[object]:
+    from openadapt_flow._browser_setup import ensure_chromium_installed
+
+    ensure_chromium_installed()
+    from playwright.sync_api import sync_playwright
+
+    with sync_playwright() as pw:
+        chromium = pw.chromium.launch(headless=True)
+        yield chromium
+        chromium.close()
+
+
+def _page(browser: object, url: str):
+    page = browser.new_page(viewport=VIEWPORT, device_scale_factor=1)  # type: ignore[attr-defined]
+    page.goto(url)
+    page.wait_for_load_state("load")
+    page.wait_for_timeout(150)
+    return page
+
+
+def _center(page, selector: str) -> tuple[int, int]:
+    box = page.locator(selector).first.bounding_box()
+    assert box is not None, selector
+    return int(box["x"] + box["width"] / 2), int(box["y"] + box["height"] / 2)
+
+
+def _recorder(page, root: Path, url: str) -> Recorder:
+    return Recorder(
+        PlaywrightBackend(page),
+        root / "recording",
+        app_url=url,
+        settle_interval_s=0.05,
+        settle_timeout_s=2.0,
+    )
+
+
+def _replay(browser, url, bundle_dir: Path, run_dir: Path, params=None):
+    page = _page(browser, url)
+    report = Replayer(PlaywrightBackend(page), poll_interval_s=0.05).run(
+        Workflow.load(bundle_dir),
+        params=params or {},
+        bundle_dir=bundle_dir,
+        run_dir=run_dir,
+    )
+    return report, page
+
+
+def _summary(report) -> str:
+    return "\n".join(f"{r.step_id} ok={r.ok} err={r.error}" for r in report.results)
+
+
+@pytest.fixture(scope="module")
+def note_bundle(app_url, browser, tmp_path_factory) -> Path:
+    root = tmp_path_factory.mktemp("note")
+    url = app_url + "/form.html"
+    page = _page(browser, url)
+    try:
+        recorder = _recorder(page, root, url)
+        recorder.click(*_center(page, "#name"))
+        recorder.type_text("Jane Doe", param="patient")
+        recorder.click(*_center(page, "#save"))
+        assert page.locator("#status").inner_text() == "Saved Jane Doe"
+        recording = recorder.finish()
+    finally:
+        page.close()
+    compile_recording(recording, root / "bundle", name="note")
+    return root / "bundle"
+
+
+def test_note_replays_on_the_page_it_was_recorded_on(
+    note_bundle, app_url, browser, tmp_path
+) -> None:
+    """Control: an unchanged page with a new parameter saves the new value."""
+
+    report, page = _replay(
+        browser,
+        app_url + "/form.html",
+        note_bundle,
+        tmp_path / "run",
+        params={"patient": "Ada"},
+    )
+    try:
+        assert report.success, _summary(report)
+        assert page.locator("#status").inner_text() == "Saved Ada"
+    finally:
+        page.close()
+
+
+def test_a_field_autofilled_on_focus_never_saves_a_concatenation(
+    note_bundle, app_url, browser, tmp_path
+) -> None:
+    """The field fills itself when focused (autofill, a restored draft).
+
+    Typing appends. A run that reports success with "John SmithAda" saved
+    has written a wrong value silently. Today the click's region check halts
+    the run before typing; this pins that.
+    """
+
+    report, page = _replay(
+        browser,
+        app_url + "/form.html?prefill=John%20Smith",
+        note_bundle,
+        tmp_path / "run",
+        params={"patient": "Ada"},
+    )
+    try:
+        status = page.locator("#status").inner_text()
+        if report.success:
+            assert status == "Saved Ada", _summary(report)
+        else:
+            assert status in {"Not saved", "Saved Ada"}, status
+    finally:
+        page.close()
+
+
+def test_an_unrecorded_confirm_dialog_never_ends_in_success(
+    note_bundle, app_url, browser, tmp_path
+) -> None:
+    """Save raises a ``confirm()`` the recording never saw.
+
+    The browser dismisses it, so nothing is saved. The run must not report
+    success on a save that did not happen.
+    """
+
+    report, page = _replay(
+        browser,
+        app_url + "/form.html?confirm=1",
+        note_bundle,
+        tmp_path / "run",
+        params={"patient": "Ada"},
+    )
+    try:
+        status = page.locator("#status").inner_text()
+        assert not (report.success and status == "Not saved"), _summary(report)
+    finally:
+        page.close()
+
+
+def _record_worklist(app_url, browser, root: Path, query: str, chart: str) -> Path:
+    url = app_url + "/worklist.html" + query
+    page = _page(browser, url)
+    try:
+        recorder = _recorder(page, root, url)
+        ids = page.locator("li").evaluate_all(
+            "els => els.map(e => e.dataset.id || e.dataset.chart)"
+        )
+        recorder.click(*_center(page, f"li >> nth={ids.index(chart)}"))
+        assert page.locator("#status").inner_text() == f"Opened chart {chart}"
+        recorder.click(*_center(page, "#sign"))
+        assert page.locator("#signed").inner_text() == f"Signed chart {chart}"
+        recording = recorder.finish()
+    finally:
+        page.close()
+    compile_recording(recording, root / "bundle", name="worklist")
+    return root / "bundle"
+
+
+@pytest.fixture(scope="module")
+def worklist_bundle(app_url, browser, tmp_path_factory) -> Path:
+    return _record_worklist(
+        app_url, browser, tmp_path_factory.mktemp("worklist"), "", "1001"
+    )
+
+
+@pytest.fixture(scope="module")
+def unkeyed_worklist_bundle(app_url, browser, tmp_path_factory) -> Path:
+    return _record_worklist(
+        app_url, browser, tmp_path_factory.mktemp("unkeyed"), "?noid=1", "1001"
+    )
+
+
+def test_identical_rows_that_swap_order_never_sign_the_other_chart(
+    worklist_bundle, app_url, browser, tmp_path
+) -> None:
+    """Two rows show the same text; at replay time they have swapped.
+
+    The task opens chart 1001 and signs its note. Signing chart 1002 and
+    reporting success acts on the wrong record. The run must halt (the
+    target is ambiguous) or sign 1001.
+    """
+
+    report, page = _replay(
+        browser,
+        app_url + "/worklist.html?order=ba",
+        worklist_bundle,
+        tmp_path / "run",
+    )
+    try:
+        status = page.locator("#signed").inner_text()
+        assert not (report.success and status == "Signed chart 1002"), _summary(report)
+    finally:
+        page.close()
+
+
+def test_identical_rows_in_recorded_order_still_sign_the_recorded_chart(
+    worklist_bundle, app_url, browser, tmp_path
+) -> None:
+    """Control: homonym rows in the recorded order. No false stop."""
+
+    report, page = _replay(
+        browser, app_url + "/worklist.html", worklist_bundle, tmp_path / "run"
+    )
+    try:
+        assert report.success, _summary(report)
+        assert page.locator("#signed").inner_text() == "Signed chart 1001"
+    finally:
+        page.close()
+
+
+def test_swapped_identical_rows_with_record_ids_sign_the_recorded_chart(
+    worklist_bundle, app_url, browser, tmp_path
+) -> None:
+    """The rows carry ``data-id``. Replay picks the row by id, not position."""
+
+    report, page = _replay(
+        browser, app_url + "/worklist.html?order=ba", worklist_bundle, tmp_path / "run"
+    )
+    try:
+        assert report.success, _summary(report)
+        assert page.locator("#signed").inner_text() == "Signed chart 1001"
+        assert report.results[0].identity.candidate_count == 2
+    finally:
+        page.close()
+
+
+@pytest.mark.parametrize("order", ["ab", "ba", "bad"])
+def test_identical_rows_without_record_ids_halt_before_opening_a_chart(
+    unkeyed_worklist_bundle, app_url, browser, tmp_path, order
+) -> None:
+    """No record id tells the rows apart: the run stops before the first click."""
+
+    report, page = _replay(
+        browser,
+        app_url + f"/worklist.html?noid=1&order={order}",
+        unkeyed_worklist_bundle,
+        tmp_path / "run",
+    )
+    try:
+        assert not report.success, _summary(report)
+        first = report.results[0]
+        assert not first.ok and first.safety_halt, _summary(report)
+        assert first.identity.ambiguity_halt == "identical_candidates"
+        assert first.identity.candidate_count == len(order)
+        assert page.locator("#status").inner_text() == "No chart open"
+        assert page.locator("#signed").inner_text() == "Nothing signed"
+    finally:
+        page.close()
+
+
+def test_a_unique_row_among_homonyms_replays_after_a_swap(
+    app_url, browser, tmp_path
+) -> None:
+    """The recorded row's text is unique on the page: no false stop."""
+
+    bundle = _record_worklist(app_url, browser, tmp_path / "rec", "?noid=1", "1003")
+    report, page = _replay(
+        browser, app_url + "/worklist.html?noid=1&order=ba", bundle, tmp_path / "run"
+    )
+    try:
+        assert report.success, _summary(report)
+        assert page.locator("#signed").inner_text() == "Signed chart 1003"
+        assert report.results[0].identity.candidate_count == 1
+    finally:
+        page.close()

From d5d4ee96aeced6b24f17eb39f50afb81b2a1bfea Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 12:39:12 -0400
Subject: [PATCH 2/4] docs: list the identity candidate census in CAPABILITIES

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CAPABILITIES.md | 1 +
 llms-full.txt   | 1 +
 2 files changed, 2 insertions(+)

diff --git a/CAPABILITIES.md b/CAPABILITIES.md
index 6b511d68b..f25145932 100644
--- a/CAPABILITIES.md
+++ b/CAPABILITIES.md
@@ -186,6 +186,7 @@ improvising or retrying when something is off.
 | Chain programs across surfaces, passing confirmed values | Released | Synthetic | `openadapt-flow compose`, `process`; [PROCESS_CONTRACT.md](packages/flow/docs/PROCESS_CONTRACT.md) |
 | One program that switches surfaces, tabs and popups mid-run | On main ([#1197](https://github.com/OpenAdaptAI/OpenAdapt/pull/1197)) | Real Chromium; Windows switch on a mock | [#1197](https://github.com/OpenAdaptAI/OpenAdapt/pull/1197) |
 | Halt when a workflow qualified with page structure runs on a pixels-only surface | Released | Synthetic | `openadapt-flow qualify` |
+| Stop before acting when another shown row has the same identity; when the rows carry an app-owned record id (a `data-*` id, the row link), act on the row with the recorded id | On main ([#1504](https://github.com/OpenAdaptAI/OpenAdapt/pull/1504)) | Real browser (headless Chromium worklist with identical rows) | `runtime/identity_census.py`; `PlaywrightBackend.identity_candidates_at` |
 | Lift look-alike identity stops with a complete known-ID list | Released | Synthetic (rendered record list) | `openadapt-flow known-ids` |
 | Take extra looks before stopping on a look-alike record ID: fresh frames, the known-ID list, and a select-and-copy of the ID where the surface allows it | On main ([#1333](https://github.com/OpenAdaptAI/OpenAdapt/pull/1333)) | Synthetic (dense record list in a browser) | `runtime.identity_copy_look` in deployment.yaml; [IDENTITY_LOOKS.md](packages/flow/benchmark/identity_looks/IDENTITY_LOOKS.md) |
 | Check an icon-only click target's hover response (highlight, tooltip) before clicking on a pixel surface, and re-check the target and identity regions on a fresh frame right before every pixel click | On main ([#1334](https://github.com/OpenAdaptAI/OpenAdapt/pull/1334)) | Synthetic (in-process kiosk) | `Recorder(capture_hover=True)`; [hover_identity](packages/flow/benchmark/hover_identity/README.md) |
diff --git a/llms-full.txt b/llms-full.txt
index 02ba1a00d..8f2399dd9 100644
--- a/llms-full.txt
+++ b/llms-full.txt
@@ -583,6 +583,7 @@ improvising or retrying when something is off.
 | Chain programs across surfaces, passing confirmed values | Released | Synthetic | `openadapt-flow compose`, `process`; [PROCESS_CONTRACT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/PROCESS_CONTRACT.md) |
 | One program that switches surfaces, tabs and popups mid-run | On main ([#1197](https://github.com/OpenAdaptAI/OpenAdapt/pull/1197)) | Real Chromium; Windows switch on a mock | [#1197](https://github.com/OpenAdaptAI/OpenAdapt/pull/1197) |
 | Halt when a workflow qualified with page structure runs on a pixels-only surface | Released | Synthetic | `openadapt-flow qualify` |
+| Stop before acting when another shown row has the same identity; when the rows carry an app-owned record id (a `data-*` id, the row link), act on the row with the recorded id | On main ([#1504](https://github.com/OpenAdaptAI/OpenAdapt/pull/1504)) | Real browser (headless Chromium worklist with identical rows) | `runtime/identity_census.py`; `PlaywrightBackend.identity_candidates_at` |
 | Lift look-alike identity stops with a complete known-ID list | Released | Synthetic (rendered record list) | `openadapt-flow known-ids` |
 | Take extra looks before stopping on a look-alike record ID: fresh frames, the known-ID list, and a select-and-copy of the ID where the surface allows it | On main ([#1333](https://github.com/OpenAdaptAI/OpenAdapt/pull/1333)) | Synthetic (dense record list in a browser) | `runtime.identity_copy_look` in deployment.yaml; [IDENTITY_LOOKS.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/benchmark/identity_looks/IDENTITY_LOOKS.md) |
 | Check an icon-only click target's hover response (highlight, tooltip) before clicking on a pixel surface, and re-check the target and identity regions on a fresh frame right before every pixel click | On main ([#1334](https://github.com/OpenAdaptAI/OpenAdapt/pull/1334)) | Synthetic (in-process kiosk) | `Recorder(capture_hover=True)`; [hover_identity](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/benchmark/hover_identity/README.md) |

From 9042f1603a5eea9c4637105320d7e0eaabad541e Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 14:39:20 -0400
Subject: [PATCH 3/4] fix(flow): halt before a write when the rows cannot be
 counted

The census treated "this backend cannot count rows" like "the target is
not a row" and let the run continue. On native, RDP and Citrix surfaces
the homonym-row fail-open stayed open.

The census now tells the two apart:

- not_a_row: the target has no row identity (the browser reports it, or
  teach time recorded the identity as a form field or control). The
  verdict stands.
- cannot_count: the target reads as a row (a structured row identity, or
  a pixel match with no OCR band to read), but the surface has no census.
  A step that leads into a consequential write halts with
  ambiguity_halt="identity_uncounted" under every profile. Elsewhere it
  runs and records census_status="cannot_count".

"Sign" now counts as a write verb ("Sign visit note", "Co-sign",
"e-sign"), never as "sign in", "sign on" or "sign out". Before this, the
compiled Sign step in the audit repro was not consequential.

Qualification compares retained identity evidence with the ladder verdict
only; the census reads the whole screen, which that evidence doesn't hold.

CI shards 1/4 and 4/4 failed on the first push for real reasons: the OCR
census's extra screen read used up scripted fake OCR results, and the
exact-equality check in qualification saw the new candidate_count. Test
scripts now include the census read. Fakes that model one record row
expose a one-row census.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CAPABILITIES.md                               |  2 +-
 llms-full.txt                                 |  2 +-
 packages/flow/openadapt_flow/backend.py       |  7 ++-
 .../backends/playwright_backend.py            |  9 ++--
 .../flow/openadapt_flow/compiler/compile.py   |  6 +++
 packages/flow/openadapt_flow/ir.py            | 40 ++++++++++++--
 .../qualification_identity_evidence.py        | 24 ++++++++-
 packages/flow/openadapt_flow/recorder.py      | 28 ++++++----
 packages/flow/openadapt_flow/risk.py          | 10 ++--
 .../openadapt_flow/runtime/identity_census.py | 23 ++++++--
 .../runtime/identity_template.py              |  4 ++
 .../flow/openadapt_flow/runtime/replayer.py   | 54 ++++++++++++++++---
 .../flow/openadapt_flow/sanitized_artifact.py |  1 +
 .../openadapt_flow/validation/governed_run.py |  8 +++
 .../validation/identity_ladder.py             | 14 +++++
 packages/flow/tests/test_governed_healing.py  |  8 +++
 packages/flow/tests/test_identity_census.py   |  7 ++-
 .../test_identity_evidence_capability.py      | 18 +++++++
 packages/flow/tests/test_loop_authoring.py    |  8 ++-
 .../test_qualification_fault_contract.py      |  8 +++
 packages/flow/tests/test_realism_browser.py   | 54 ++++++++++++++++++-
 packages/flow/tests/test_replayer.py          | 10 ++++
 22 files changed, 300 insertions(+), 45 deletions(-)

diff --git a/CAPABILITIES.md b/CAPABILITIES.md
index f25145932..f650c490b 100644
--- a/CAPABILITIES.md
+++ b/CAPABILITIES.md
@@ -186,7 +186,7 @@ improvising or retrying when something is off.
 | Chain programs across surfaces, passing confirmed values | Released | Synthetic | `openadapt-flow compose`, `process`; [PROCESS_CONTRACT.md](packages/flow/docs/PROCESS_CONTRACT.md) |
 | One program that switches surfaces, tabs and popups mid-run | On main ([#1197](https://github.com/OpenAdaptAI/OpenAdapt/pull/1197)) | Real Chromium; Windows switch on a mock | [#1197](https://github.com/OpenAdaptAI/OpenAdapt/pull/1197) |
 | Halt when a workflow qualified with page structure runs on a pixels-only surface | Released | Synthetic | `openadapt-flow qualify` |
-| Stop before acting when another shown row has the same identity; when the rows carry an app-owned record id (a `data-*` id, the row link), act on the row with the recorded id | On main ([#1504](https://github.com/OpenAdaptAI/OpenAdapt/pull/1504)) | Real browser (headless Chromium worklist with identical rows) | `runtime/identity_census.py`; `PlaywrightBackend.identity_candidates_at` |
+| Stop before acting when another shown row has the same identity; when the rows carry an app-owned record id (a `data-*` id, the row link), act on the row with the recorded id; on a surface that cannot count the rows, stop before any step that leads into a write | On main ([#1504](https://github.com/OpenAdaptAI/OpenAdapt/pull/1504)) | Real browser (headless Chromium worklist with identical rows) | `runtime/identity_census.py`; `PlaywrightBackend.identity_candidates_at` |
 | Lift look-alike identity stops with a complete known-ID list | Released | Synthetic (rendered record list) | `openadapt-flow known-ids` |
 | Take extra looks before stopping on a look-alike record ID: fresh frames, the known-ID list, and a select-and-copy of the ID where the surface allows it | On main ([#1333](https://github.com/OpenAdaptAI/OpenAdapt/pull/1333)) | Synthetic (dense record list in a browser) | `runtime.identity_copy_look` in deployment.yaml; [IDENTITY_LOOKS.md](packages/flow/benchmark/identity_looks/IDENTITY_LOOKS.md) |
 | Check an icon-only click target's hover response (highlight, tooltip) before clicking on a pixel surface, and re-check the target and identity regions on a fresh frame right before every pixel click | On main ([#1334](https://github.com/OpenAdaptAI/OpenAdapt/pull/1334)) | Synthetic (in-process kiosk) | `Recorder(capture_hover=True)`; [hover_identity](packages/flow/benchmark/hover_identity/README.md) |
diff --git a/llms-full.txt b/llms-full.txt
index 8f2399dd9..83e688986 100644
--- a/llms-full.txt
+++ b/llms-full.txt
@@ -583,7 +583,7 @@ improvising or retrying when something is off.
 | Chain programs across surfaces, passing confirmed values | Released | Synthetic | `openadapt-flow compose`, `process`; [PROCESS_CONTRACT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/PROCESS_CONTRACT.md) |
 | One program that switches surfaces, tabs and popups mid-run | On main ([#1197](https://github.com/OpenAdaptAI/OpenAdapt/pull/1197)) | Real Chromium; Windows switch on a mock | [#1197](https://github.com/OpenAdaptAI/OpenAdapt/pull/1197) |
 | Halt when a workflow qualified with page structure runs on a pixels-only surface | Released | Synthetic | `openadapt-flow qualify` |
-| Stop before acting when another shown row has the same identity; when the rows carry an app-owned record id (a `data-*` id, the row link), act on the row with the recorded id | On main ([#1504](https://github.com/OpenAdaptAI/OpenAdapt/pull/1504)) | Real browser (headless Chromium worklist with identical rows) | `runtime/identity_census.py`; `PlaywrightBackend.identity_candidates_at` |
+| Stop before acting when another shown row has the same identity; when the rows carry an app-owned record id (a `data-*` id, the row link), act on the row with the recorded id; on a surface that cannot count the rows, stop before any step that leads into a write | On main ([#1504](https://github.com/OpenAdaptAI/OpenAdapt/pull/1504)) | Real browser (headless Chromium worklist with identical rows) | `runtime/identity_census.py`; `PlaywrightBackend.identity_candidates_at` |
 | Lift look-alike identity stops with a complete known-ID list | Released | Synthetic (rendered record list) | `openadapt-flow known-ids` |
 | Take extra looks before stopping on a look-alike record ID: fresh frames, the known-ID list, and a select-and-copy of the ID where the surface allows it | On main ([#1333](https://github.com/OpenAdaptAI/OpenAdapt/pull/1333)) | Synthetic (dense record list in a browser) | `runtime.identity_copy_look` in deployment.yaml; [IDENTITY_LOOKS.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/benchmark/identity_looks/IDENTITY_LOOKS.md) |
 | Check an icon-only click target's hover response (highlight, tooltip) before clicking on a pixel surface, and re-check the target and identity regions on a fresh frame right before every pixel click | On main ([#1334](https://github.com/OpenAdaptAI/OpenAdapt/pull/1334)) | Synthetic (in-process kiosk) | `Recorder(capture_hover=True)`; [hover_identity](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/benchmark/hover_identity/README.md) |
diff --git a/packages/flow/openadapt_flow/backend.py b/packages/flow/openadapt_flow/backend.py
index d782cf9eb..b7a5be849 100644
--- a/packages/flow/openadapt_flow/backend.py
+++ b/packages/flow/openadapt_flow/backend.py
@@ -235,8 +235,11 @@ class IdentityCandidatesBackend(Protocol):
     Returns a dict ``{"record_key": str | None, "incomplete": bool,
     "candidates": [{"self": bool, "record_key": str | None, "dx": float,
     "dy": float}, ...], "top_level": bool}``, where ``dx``/``dy`` are each
-    row's offset from the row under the point in click coordinates, or None
-    when the point has no row identity or the census failed (never raises).
+    row's offset from the row under the point in click coordinates;
+    ``{"row": False}`` when the point has no row identity (a standalone
+    control); or None when the census failed (never raises). A backend without
+    this method cannot count rows, and the runtime halts a step that leads
+    into a consequential write rather than assume the row is unique.
     """
 
     def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]:
diff --git a/packages/flow/openadapt_flow/backends/playwright_backend.py b/packages/flow/openadapt_flow/backends/playwright_backend.py
index 8bd3bb25e..e0864d5d4 100644
--- a/packages/flow/openadapt_flow/backends/playwright_backend.py
+++ b/packages/flow/openadapt_flow/backends/playwright_backend.py
@@ -762,8 +762,8 @@ class _FramePoint:
 # Every shown row whose record identity equals the identity of the row under
 # the point, with each row's stable record id and position. The identity is
 # computed exactly as ``structured_text_at`` does, for the cell at the same
-# place in each row. Null when the point has no row identity (a form field
-# group, a standalone control): there is no row to compare.
+# place in each row. ``{row: false}`` when the point has no row identity (a
+# form field group, a standalone control): there is no row to compare.
 _IDENTITY_CANDIDATES_AT_JS = (
     r"""([px, py, cap]) => {
     const el = document.elementFromPoint(px, py);
@@ -775,7 +775,7 @@ class _FramePoint:
     + _RECORD_KEY_JS
     + r""";
     const target = describe(el);
-    if (!target.rowIdentity || !target.row) return null;
+    if (!target.rowIdentity || !target.row) return {row: false};
     const row = target.row;
     const own = el.closest('td, th, [role="cell"], [role="gridcell"]') || el;
     const path = [];
@@ -1787,7 +1787,8 @@ def identity_candidates_at(self, x: int, y: int) -> Optional[dict[str, Any]]:
         scan, and one entry per shown row with the same identity (the row
         under the point included, ``self`` true) carrying its record id and
         its offset ``(dx, dy)`` from the row under the point, in viewport CSS
-        pixels. None when the point has no row identity or on any failure.
+        pixels. ``{"row": False}`` when the point has no row identity (a
+        standalone control, a form field); None when the census failed.
         """
         try:
             point = self._frame_point(int(x), int(y))
diff --git a/packages/flow/openadapt_flow/compiler/compile.py b/packages/flow/openadapt_flow/compiler/compile.py
index d3bdedf68..2b8713598 100644
--- a/packages/flow/openadapt_flow/compiler/compile.py
+++ b/packages/flow/openadapt_flow/compiler/compile.py
@@ -2280,6 +2280,9 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]:
             record_key = (
                 event.get("record_key") or None if structured_identity else None
             )
+            structured_row = event.get("structured_row")
+            if not structured_identity or not isinstance(structured_row, bool):
+                structured_row = None
             # Structural locator (DOM selector / role+name, or UIA identifiers)
             # of the clicked element, captured by the recorder when the
             # recording backend exposed it (StructuralActionBackend). Drives the
@@ -2388,6 +2391,7 @@ def cached_lines(i: int, suffix: str, png: bytes) -> list[OcrLine]:
                 context_text=context_text,
                 structured_identity=structured_identity,
                 record_key=record_key,
+                structured_row=structured_row,
                 structural=structural,
                 landmarks=anchor_landmarks,
                 identifier_crop=identifier_crop_rel,
@@ -3598,6 +3602,7 @@ def _phi_free_identity(workflow: Workflow) -> None:
             param_examples=workflow.params,
             salt_hex=salt,
             record_key=anchor.record_key,
+            structured_row=anchor.structured_row,
         )
         if template is None:
             continue
@@ -3605,3 +3610,4 @@ def _phi_free_identity(workflow: Workflow) -> None:
         anchor.context_text = None
         anchor.structured_identity = None
         anchor.record_key = None
+        anchor.structured_row = None
diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py
index ac9693e55..68037b5f4 100644
--- a/packages/flow/openadapt_flow/ir.py
+++ b/packages/flow/openadapt_flow/ir.py
@@ -331,6 +331,15 @@ class IdentityTemplate(BaseModel):
             "set, replay acts only on the shown row with the same id."
         ),
     )
+    structured_row: Optional[bool] = Field(
+        default=None,
+        description=(
+            "Whether the structured identity came from a record row at teach "
+            "time. False lets a replay surface that cannot count look-alike "
+            "rows still act on the target (a form field, a button); None "
+            "(unknown) is treated as a row."
+        ),
+    )
 
     @model_serializer(mode="wrap")
     def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
@@ -351,6 +360,8 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
             data.pop("context_params", None)
         if self.record_key is None:
             data.pop("record_key", None)
+        if self.structured_row is None:
+            data.pop("structured_row", None)
         return data
 
 
@@ -418,6 +429,13 @@ class Anchor(BaseModel):
             " ``identity_template.record_key`` and clears this field."
         ),
     )
+    structured_row: Optional[bool] = Field(
+        default=None,
+        description=(
+            "Whether ``structured_identity`` came from a record row (compile-"
+            "time only; moved to ``identity_template.structured_row``)."
+        ),
+    )
     identity_template: Optional[IdentityTemplate] = Field(
         default=None,
         description=(
@@ -471,10 +489,12 @@ class Anchor(BaseModel):
 
     @model_serializer(mode="wrap")
     def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
-        """Leave an unset ``record_key`` out, so older bundle bytes stay equal."""
+        """Leave unset census metadata out, so older bundle bytes stay equal."""
         data: dict[str, Any] = handler(self)
         if self.record_key is None:
             data.pop("record_key", None)
+        if self.structured_row is None:
+            data.pop("structured_row", None)
         return data
 
 
@@ -3729,14 +3749,24 @@ class IdentityCheck(BaseModel):
         ),
     )
     ambiguity_halt: Optional[
-        Literal["identical_candidates", "candidates_uncounted"]
+        Literal["identical_candidates", "candidates_uncounted", "identity_uncounted"]
     ] = Field(
         default=None,
         description=(
             "Set when the identity matched but does not pick one record: "
             "another shown row has the same identity and no stable record id "
-            "tells them apart, or there were too many rows to count. The "
-            "step halts whatever its risk."
+            "tells them apart (identical_candidates), there were too many rows "
+            "to count (candidates_uncounted), or this surface cannot count the "
+            "rows at all and the step leads into a consequential write "
+            "(identity_uncounted). The step halts whatever its risk."
+        ),
+    )
+    census_status: Optional[Literal["not_a_row", "cannot_count"]] = Field(
+        default=None,
+        description=(
+            "Set when the identity verified without a candidate count: the "
+            "target is not a row (not_a_row), or the surface cannot count rows "
+            "and nothing consequential follows the step (cannot_count)."
         ),
     )
 
@@ -3748,6 +3778,8 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
             data.pop("candidate_count", None)
         if self.ambiguity_halt is None:
             data.pop("ambiguity_halt", None)
+        if self.census_status is None:
+            data.pop("census_status", None)
         return data
 
 
diff --git a/packages/flow/openadapt_flow/qualification_identity_evidence.py b/packages/flow/openadapt_flow/qualification_identity_evidence.py
index 442e1deb0..5927f0cd2 100644
--- a/packages/flow/openadapt_flow/qualification_identity_evidence.py
+++ b/packages/flow/openadapt_flow/qualification_identity_evidence.py
@@ -105,6 +105,20 @@ def _read_pixel_crop(
     return payload
 
 
+def _ladder_part(check: IdentityCheck) -> IdentityCheck:
+    """The check without the candidate census the runtime adds after the ladder.
+
+    The census (runtime.identity_census) reads the whole live screen, which the
+    retained evidence does not hold, so only the ladder verdict is reproduced.
+    A census that changed the verdict (a halt or a mismatch) still fails the
+    status comparison.
+    """
+
+    return check.model_copy(
+        update={"candidate_count": None, "ambiguity_halt": None, "census_status": None}
+    )
+
+
 def _canonical_ladder_error(
     check: IdentityCheck,
     *,
@@ -166,7 +180,9 @@ def _canonical_ladder_error(
         if pixel_result is None:
             return "pixel identity crop evidence does not reproduce a verdict"
         pixel_result.pixel_evidence = evidence
-        if pixel_result.status != expected_status or pixel_result != check:
+        if pixel_result.status != expected_status or pixel_result != _ladder_part(
+            check
+        ):
             return "pixel identity crop evidence does not reproduce the runtime verdict"
         return None
 
@@ -224,7 +240,11 @@ def _canonical_ladder_error(
     except (TypeError, ValueError):
         return "canonical ladder evidence could not be reproduced"
 
-    if expected is None or expected.status != expected_status or expected != check:
+    if (
+        expected is None
+        or expected.status != expected_status
+        or expected != _ladder_part(check)
+    ):
         return "canonical ladder evidence does not reproduce the runtime verdict"
     return None
 
diff --git a/packages/flow/openadapt_flow/recorder.py b/packages/flow/openadapt_flow/recorder.py
index 6b396c3cf..9cf74995c 100644
--- a/packages/flow/openadapt_flow/recorder.py
+++ b/packages/flow/openadapt_flow/recorder.py
@@ -687,9 +687,15 @@ def _record(self, event: dict[str, Any], act: Callable[[], None]) -> None:
             structured = self._structured_identity_at(int(event["x"]), int(event["y"]))
             if structured:
                 event = {**event, "structured_identity": structured}
-                record_key = self._record_key_at(int(event["x"]), int(event["y"]))
-                if record_key:
-                    event = {**event, "record_key": record_key}
+                census = self._identity_census_at(int(event["x"]), int(event["y"]))
+                if census is not None:
+                    # Whether the identity came from a record row. A replay
+                    # surface that cannot count look-alike rows still runs a
+                    # target that is not a row (a form field, a button).
+                    event = {**event, "structured_row": census.get("row") is not False}
+                    key = census.get("record_key")
+                    if isinstance(key, str) and key:
+                        event = {**event, "record_key": key}
             # The part of the form the target sits in (a legend, an accordion
             # or tab header, a heading), for the program's named sections.
             # Display only: the compiler never puts it on a step.
@@ -1021,14 +1027,15 @@ def _structured_identity_at(self, x: int, y: int) -> Optional[str]:
         except Exception:
             return None
 
-    def _record_key_at(self, x: int, y: int) -> Optional[str]:
-        """Stable record id of the row under (x, y), if the backend exposes it.
+    def _identity_census_at(self, x: int, y: int) -> Optional[dict[str, Any]]:
+        """The row census under (x, y), if the backend exposes one.
 
         Backends MAY expose ``identity_candidates_at``
-        (``openadapt_flow.backend.IdentityCandidatesBackend``). The row's
-        app-owned id (a DOM ``data-*`` id, the row link's path) lets replay
-        tell apart rows whose visible text is identical. None when the
-        backend cannot read one or the row has none.
+        (``openadapt_flow.backend.IdentityCandidatesBackend``). It says
+        whether the target is a record row and reads the row's app-owned id
+        (a DOM ``data-*`` id, the row link's path), which lets replay tell
+        apart rows whose visible text is identical. None when the backend has
+        no census or it failed.
         """
         getter = getattr(self._backend, "identity_candidates_at", None)
         if getter is None:
@@ -1037,8 +1044,7 @@ def _record_key_at(self, x: int, y: int) -> Optional[str]:
             census = getter(int(x), int(y))
         except Exception:
             return None
-        key = census.get("record_key") if isinstance(census, dict) else None
-        return key if isinstance(key, str) and key else None
+        return census if isinstance(census, dict) else None
 
     def _structural_state(self) -> dict[str, Any]:
         """Structural observations the backend can provide right now.
diff --git a/packages/flow/openadapt_flow/risk.py b/packages/flow/openadapt_flow/risk.py
index f3b72f4be..8a61623f1 100644
--- a/packages/flow/openadapt_flow/risk.py
+++ b/packages/flow/openadapt_flow/risk.py
@@ -299,10 +299,13 @@ def string_tuple(field: str) -> tuple[str, ...]:
 # "Add note" but NOT "address", and `post` (were it present) would not match
 # "postal". Ordered roughly by how common they are on real controls.
 #
+# "sign" counts in its writing sense ("Sign visit note", "Co-sign", "Signed"),
+# the clinical attestation that locks a note, but never as "sign in", "sign
+# on" or "sign out", which are session chrome.
+#
 # Deliberately EXCLUDED to avoid noisy false positives on navigation chrome:
-# bare "sign" (collides with "sign in"; the writing sense is covered by
-# save/submit/confirm and the explicit "sign up" below), bare "post"/"order"/
-# "book"/"complete" (collide with "Posts"/"Orders"/"Bookings"/"Completed" tabs).
+# bare "post"/"order"/"book"/"complete" (collide with "Posts"/"Orders"/
+# "Bookings"/"Completed" tabs).
 _WRITE_STEMS: tuple[str, ...] = (
     r"sav(?:e|es|ing)",
     r"submit(?:s|ted|ting)?",
@@ -319,6 +322,7 @@ def string_tuple(field: str) -> tuple[str, ...]:
     r"pay(?:s|ing)?",
     r"sign[\s\-_]?up",
     r"signup",
+    r"(?:co[\-_]?|e[\-_]?)?sign(?:s|ed|ing)?(?![\s\-_]*(?:in|on|out)\b)",
     r"regist(?:er|ers|ering|ration)",
     r"enroll(?:s|ed|ing|ment)?",
     r"add(?:s|ed|ing)?",
diff --git a/packages/flow/openadapt_flow/runtime/identity_census.py b/packages/flow/openadapt_flow/runtime/identity_census.py
index e942b0d99..c0839bc43 100644
--- a/packages/flow/openadapt_flow/runtime/identity_census.py
+++ b/packages/flow/openadapt_flow/runtime/identity_census.py
@@ -19,7 +19,12 @@
   (``retarget``) or, when it cannot move, refuse (``wrong_record``);
 * several rows and no recorded id, or more than one row with the recorded id:
   ``ambiguous``. The step halts before it acts, whatever its risk;
-* the rows could not all be counted: ``uncounted``, which also halts.
+* the rows could not all be counted: ``uncounted``, which also halts;
+* the target is not a row (a standalone control, a form field):
+  ``not_a_row``; the verdict stands;
+* the surface cannot count rows at all (a backend without a census, a pixel
+  identity with no OCR band to read): ``cannot_count``. The runtime halts a
+  step that leads into a consequential write, and records it elsewhere.
 
 A recorded id is required even when the row is unique: a single row with
 another id is a different record that happens to read the same
@@ -31,7 +36,13 @@
 from typing import Any, Callable, Literal, Mapping, NamedTuple, Optional
 
 CensusVerdict = Literal[
-    "unique", "retarget", "ambiguous", "uncounted", "wrong_record", "unavailable"
+    "unique",
+    "retarget",
+    "ambiguous",
+    "uncounted",
+    "wrong_record",
+    "not_a_row",
+    "cannot_count",
 ]
 
 
@@ -60,14 +71,16 @@ def decide(
     """Turn a backend census into a decision.
 
     Args:
-        census: ``identity_candidates_at``'s result, or None when the backend
-            has no census or the point has no row identity.
+        census: ``identity_candidates_at``'s result: ``{"row": False}`` when
+            the point has no row identity, None when the census failed.
         has_recorded_key: Whether the bundle recorded the row's record id
             (and the identity is not bound to a run parameter).
         key_matches: Whether a live record id equals the recorded one.
     """
     if not isinstance(census, Mapping):
-        return CensusDecision("unavailable")
+        return CensusDecision("cannot_count")
+    if census.get("row") is False:
+        return CensusDecision("not_a_row")
     if census.get("incomplete"):
         return CensusDecision("uncounted")
     raw = census.get("candidates")
diff --git a/packages/flow/openadapt_flow/runtime/identity_template.py b/packages/flow/openadapt_flow/runtime/identity_template.py
index b733f1da0..acc0ada0d 100644
--- a/packages/flow/openadapt_flow/runtime/identity_template.py
+++ b/packages/flow/openadapt_flow/runtime/identity_template.py
@@ -218,6 +218,7 @@ def build_identity_template(
     param_examples: Optional[dict[str, str]] = None,
     salt_hex: Optional[str] = None,
     record_key: Optional[str] = None,
+    structured_row: Optional[bool] = None,
 ) -> Optional[IdentityTemplate]:
     """Build a PHI-free identity template from the recorded plaintext band.
 
@@ -234,6 +235,8 @@ def build_identity_template(
             structured identity embeds no parameter: a parameterized row is a
             different record on every run, so the demonstrated id is not the
             run's.
+        structured_row: Whether the structured identity came from a record
+            row at teach time (None: unknown).
 
     Returns:
         An :class:`IdentityTemplate`, or None when there is neither a usable
@@ -260,6 +263,7 @@ def build_identity_template(
         )
         if record_key and not structured_params:
             tmpl.record_key = record_key_hash(salt, record_key)
+        tmpl.structured_row = structured_row
 
     # Mirror the compiler's context_from_lines gate: a band shorter than
     # MIN_CONTEXT_CHARS is too generic to discriminate and is never stored (any
diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py
index 71af017e4..ac227ace6 100644
--- a/packages/flow/openadapt_flow/runtime/replayer.py
+++ b/packages/flow/openadapt_flow/runtime/replayer.py
@@ -10670,6 +10670,11 @@ def _identity_gate_error(
                     f"{check.candidate_count} shown rows carry the same "
                     "identity and no recorded record id tells them apart"
                 )
+            elif check.ambiguity_halt == "identity_uncounted":
+                why = (
+                    "this surface cannot count the rows that might share it, "
+                    "and a consequential write follows (identity_uncounted)"
+                )
             else:
                 why = "the rows that could share its identity could not all be counted"
             error = (
@@ -15512,7 +15517,7 @@ def _structured_census(self, step: Step, point: Point) -> census_mod.CensusDecis
         assert anchor is not None
         getter = getattr(self.backend, "identity_candidates_at", None)
         if getter is None:
-            return census_mod.CensusDecision("unavailable")
+            return census_mod.CensusDecision("cannot_count")
         try:
             census = getter(int(point[0]), int(point[1]))
         except Exception:
@@ -15548,7 +15553,7 @@ def _ocr_census(
         assert anchor is not None
         tmpl = anchor.identity_template
         if not (anchor.context_text or (tmpl is not None and tmpl.tokens)):
-            return census_mod.CensusDecision("unavailable")
+            return census_mod.CensusDecision("cannot_count")
         today = date.today()
         try:
             lines = [
@@ -15558,7 +15563,7 @@ def _ocr_census(
                 and not identity_mod.is_volatile_line(line.text, reference_date=today)
             ]
         except Exception:
-            return census_mod.CensusDecision("uncounted")
+            return census_mod.CensusDecision("cannot_count")
         viewport = self.backend.viewport
         px, py = int(resolution.point[0]), int(resolution.point[1])
         ex, ey = (
@@ -15638,11 +15643,27 @@ def _apply_candidate_census(
             "param",
         ):
             return check
-        decision = self._structured_census(step, resolution.point)
-        if decision.verdict == "unavailable" and check.mode != "structured":
+        if check.mode == "structured":
+            decision = self._structured_census(step, resolution.point)
+        else:
             decision = self._ocr_census(step, resolution, before_png, params, workflow)
-        if decision.verdict == "unavailable":
-            return check
+        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).
+            decision = census_mod.CensusDecision("not_a_row")
+        if decision.verdict == "not_a_row":
+            return check.model_copy(update={"census_status": "not_a_row"})
+        if decision.verdict == "cannot_count":
+            if not self._leads_into_consequential(step, workflow):
+                return check.model_copy(update={"census_status": "cannot_count"})
+            return check.model_copy(
+                update={"status": "abstain", "ambiguity_halt": "identity_uncounted"}
+            )
         if decision.verdict == "unique":
             return check.model_copy(
                 update={"candidate_count": decision.candidate_count}
@@ -15668,6 +15689,25 @@ def _apply_candidate_census(
             }
         )
 
+    def _leads_into_consequential(
+        self, step: Step, workflow: Optional[Workflow]
+    ) -> bool:
+        """Whether this step or any step after it is a consequential write.
+
+        A step that picks a record (opening a chart) is not itself a write, but
+        the write that follows lands on whatever it opened. Program graphs and
+        steps not found in the linear list count every step of the workflow.
+        """
+
+        if workflow is None:
+            return is_consequential(step, None)
+        steps = list(workflow.steps)
+        index = next((i for i, item in enumerate(steps) if item.id == step.id), None)
+        following = steps[index:] if index is not None else steps
+        return self._step_is_consequential(step, workflow) or any(
+            self._step_is_consequential(item, workflow) for item in following
+        )
+
     def _retarget_by_record_key(
         self,
         step: Step,
diff --git a/packages/flow/openadapt_flow/sanitized_artifact.py b/packages/flow/openadapt_flow/sanitized_artifact.py
index 19e2399b9..5e65c5795 100644
--- a/packages/flow/openadapt_flow/sanitized_artifact.py
+++ b/packages/flow/openadapt_flow/sanitized_artifact.py
@@ -254,6 +254,7 @@ def _apply_text_redactions(
         "source",
         "structural",
         "structured_identity",
+        "structured_row",
         "t",
         "task_description",
         "text",
diff --git a/packages/flow/openadapt_flow/validation/governed_run.py b/packages/flow/openadapt_flow/validation/governed_run.py
index 64f712d9c..77452b365 100644
--- a/packages/flow/openadapt_flow/validation/governed_run.py
+++ b/packages/flow/openadapt_flow/validation/governed_run.py
@@ -114,6 +114,14 @@ def scroll(self, dx: int, dy: int) -> None:
     def structured_text_at(self, x: int, y: int) -> str | None:
         return self.structured_identity
 
+    def identity_candidates_at(self, x: int, y: int) -> dict[str, object]:
+        # One record row on this screen (IdentityCandidatesBackend).
+        return {
+            "record_key": None,
+            "incomplete": False,
+            "candidates": [{"self": True, "record_key": None, "dx": 0, "dy": 0}],
+        }
+
 
 class _Vision:
     @staticmethod
diff --git a/packages/flow/openadapt_flow/validation/identity_ladder.py b/packages/flow/openadapt_flow/validation/identity_ladder.py
index a8f69649a..5208e66c3 100644
--- a/packages/flow/openadapt_flow/validation/identity_ladder.py
+++ b/packages/flow/openadapt_flow/validation/identity_ladder.py
@@ -279,12 +279,26 @@ def structured_text_at(self, x, y):
             raise AttributeError("pixel-only substrate has no structured text")
         return self._structured_at
 
+    # The DOM row census (IdentityCandidatesBackend). Each case screen shows
+    # one record row: the homonym under test differs from the recorded row in
+    # its identifier glyphs, so no other shown row reads the same.
+    def identity_candidates_at(self, x, y):
+        if self._structured_at is None:
+            return None
+        return {
+            "record_key": None,
+            "incomplete": False,
+            "top_level": True,
+            "candidates": [{"self": True, "record_key": None, "dx": 0, "dy": 0}],
+        }
+
 
 def _make_backend(viewport, live_png, structured_live):
     b = _Backend(viewport, live_png)
     if structured_live is None:
         # pixel-only: remove structured_text_at so the tier is UNAVAILABLE.
         b.structured_text_at = None  # type: ignore[assignment]
+        b.identity_candidates_at = None  # type: ignore[assignment]
     else:
         b._structured_at = structured_live
     return b
diff --git a/packages/flow/tests/test_governed_healing.py b/packages/flow/tests/test_governed_healing.py
index 4d0d9dca7..7ab48c389 100644
--- a/packages/flow/tests/test_governed_healing.py
+++ b/packages/flow/tests/test_governed_healing.py
@@ -143,6 +143,14 @@ def viewport(self):
     def structured_text_at(self, x, y):
         return self._structured
 
+    def identity_candidates_at(self, x, y):
+        # One record row on this screen (IdentityCandidatesBackend).
+        return {
+            "record_key": None,
+            "incomplete": False,
+            "candidates": [{"self": True, "record_key": None, "dx": 0, "dy": 0}],
+        }
+
     def screenshot(self):
         return self._frame
 
diff --git a/packages/flow/tests/test_identity_census.py b/packages/flow/tests/test_identity_census.py
index 04eb7fcaa..78a127747 100644
--- a/packages/flow/tests/test_identity_census.py
+++ b/packages/flow/tests/test_identity_census.py
@@ -106,7 +106,12 @@ def test_incomplete_or_malformed_census_is_uncounted() -> None:
     assert _decide({"candidates": "x"}, None).verdict == "uncounted"
     # The target must appear exactly once in its own census.
     assert _decide({"candidates": []}, None).verdict == "uncounted"
-    assert _decide(None, None).verdict == "unavailable"
+    assert _decide(None, None).verdict == "cannot_count"
+
+
+def test_a_target_that_is_not_a_row_is_not_counted() -> None:
+    assert _decide({"row": False}, None) == CensusDecision("not_a_row")
+    assert _decide({"row": False}, "data-id=1001") == CensusDecision("not_a_row")
 
 
 def test_record_id_is_stored_hashed_and_only_for_unparameterized_rows() -> None:
diff --git a/packages/flow/tests/test_identity_evidence_capability.py b/packages/flow/tests/test_identity_evidence_capability.py
index 1ce373b76..75d01b6f6 100644
--- a/packages/flow/tests/test_identity_evidence_capability.py
+++ b/packages/flow/tests/test_identity_evidence_capability.py
@@ -109,10 +109,28 @@ def test_no_surrounding_text_valid_crop_verifies_where_crops_are_verifiable(
     tmp_path: Path,
 ) -> None:
     step, res = _crop_only_step(tmp_path, _button())
+    # A read-only step: nothing consequential follows it.
+    step.risk = "reversible"
+    step.intent = "click 'View referral'"
     error, check = _gate(tmp_path, step, res, _button(), pixel_verify_enabled=True)
     assert error is None
     assert check is not None
     assert check.status == "verified" and check.mode == "pixel"
+    # A crop alone gives no row text to count look-alikes with; nothing
+    # consequential follows this step, so it runs and records that.
+    assert check.census_status == "cannot_count"
+
+
+def test_a_verified_crop_with_no_row_count_halts_before_a_write(
+    tmp_path: Path,
+) -> None:
+    """The crop matches, but nothing can say no other row shows the same."""
+
+    step, res = _crop_only_step(tmp_path, _button())
+    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
+    assert check.status == "abstain" and check.ambiguity_halt == "identity_uncounted"
 
 
 def test_crop_that_does_not_match_halts(tmp_path: Path) -> None:
diff --git a/packages/flow/tests/test_loop_authoring.py b/packages/flow/tests/test_loop_authoring.py
index 3ef30bb05..1e9c87ac0 100644
--- a/packages/flow/tests/test_loop_authoring.py
+++ b/packages/flow/tests/test_loop_authoring.py
@@ -544,10 +544,14 @@ def test_poisoned_record_identity_mismatch_halts_no_wrong_click(dirs):
 
     backend = FakeBackend()
     # Consequential identity-gated actions are checked once during ordinary
-    # replay and again against the freshly armed actuation target. Both reads
-    # for iteration 1 match; iteration 2 then names another entity and HALTs.
+    # replay and again against the freshly armed actuation target, and each
+    # verified read is followed by the candidate census's whole-frame read
+    # (runtime.identity_census). Every read for iteration 1 matches; iteration
+    # 2 then names another entity and HALTs.
     vision = _ScriptedIdentityVision(
         bands=[
+            "Jane Sample Knee pain referral High",
+            "Jane Sample Knee pain referral High",
             "Jane Sample Knee pain referral High",
             "Jane Sample Knee pain referral High",
             "Taylor Duplicate Knee pain referral High",
diff --git a/packages/flow/tests/test_qualification_fault_contract.py b/packages/flow/tests/test_qualification_fault_contract.py
index 94b85ba52..e4329df72 100644
--- a/packages/flow/tests/test_qualification_fault_contract.py
+++ b/packages/flow/tests/test_qualification_fault_contract.py
@@ -361,6 +361,14 @@ def drag_structural_guarded(
     def structured_text_at(self, _x: int, _y: int) -> str:
         return "Wrong record" if self.wrong_identity else "Synthetic record"
 
+    def identity_candidates_at(self, x, y):
+        # One record row on this screen (IdentityCandidatesBackend).
+        return {
+            "record_key": None,
+            "incomplete": False,
+            "candidates": [{"self": True, "record_key": None, "dx": 0, "dy": 0}],
+        }
+
     def qualification_environment_identity(self) -> tuple[str, str, str, str]:
         return "https://fixture.example", "1", _SESSION, _ENVIRONMENT
 
diff --git a/packages/flow/tests/test_realism_browser.py b/packages/flow/tests/test_realism_browser.py
index 4c311c96f..8469bdae2 100644
--- a/packages/flow/tests/test_realism_browser.py
+++ b/packages/flow/tests/test_realism_browser.py
@@ -172,9 +172,16 @@ def _recorder(page, root: Path, url: str) -> Recorder:
     )
 
 
-def _replay(browser, url, bundle_dir: Path, run_dir: Path, params=None):
+class _CannotCountBackend(PlaywrightBackend):
+    """A browser backend without the row census, like the native and remote
+    backends: it reads structured text but cannot count look-alike rows."""
+
+    identity_candidates_at = None
+
+
+def _replay(browser, url, bundle_dir: Path, run_dir: Path, params=None, backend=None):
     page = _page(browser, url)
-    report = Replayer(PlaywrightBackend(page), poll_interval_s=0.05).run(
+    report = Replayer((backend or PlaywrightBackend)(page), poll_interval_s=0.05).run(
         Workflow.load(bundle_dir),
         params=params or {},
         bundle_dir=bundle_dir,
@@ -400,3 +407,46 @@ def test_a_unique_row_among_homonyms_replays_after_a_swap(
         assert report.results[0].identity.candidate_count == 1
     finally:
         page.close()
+
+
+def test_a_backend_that_cannot_count_rows_halts_before_the_sign(
+    worklist_bundle, app_url, browser, tmp_path
+) -> None:
+    """No census on this backend: opening a chart that a sign follows halts."""
+
+    report, page = _replay(
+        browser,
+        app_url + "/worklist.html?order=ba",
+        worklist_bundle,
+        tmp_path / "run",
+        backend=_CannotCountBackend,
+    )
+    try:
+        assert not report.success, _summary(report)
+        first = report.results[0]
+        assert not first.ok and first.safety_halt, _summary(report)
+        assert first.identity.ambiguity_halt == "identity_uncounted"
+        assert page.locator("#status").inner_text() == "No chart open"
+        assert page.locator("#signed").inner_text() == "Nothing signed"
+    finally:
+        page.close()
+
+
+def test_a_backend_that_cannot_count_rows_still_runs_a_plain_form(
+    note_bundle, app_url, browser, tmp_path
+) -> None:
+    """The note form has no rows: the same backend saves it as before."""
+
+    report, page = _replay(
+        browser,
+        app_url + "/form.html",
+        note_bundle,
+        tmp_path / "run",
+        params={"patient": "Ada"},
+        backend=_CannotCountBackend,
+    )
+    try:
+        assert report.success, _summary(report)
+        assert page.locator("#status").inner_text() == "Saved Ada"
+    finally:
+        page.close()
diff --git a/packages/flow/tests/test_replayer.py b/packages/flow/tests/test_replayer.py
index 5030a74eb..dc76155a7 100644
--- a/packages/flow/tests/test_replayer.py
+++ b/packages/flow/tests/test_replayer.py
@@ -1518,7 +1518,11 @@ def test_remote_preedge_retry_refuses_a_changed_identity(bundle, run_dir):
         Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95)
         for _ in range(3)
     ]
+    # Each verified read is followed by the candidate census's whole-frame
+    # read (runtime.identity_census); one row each time.
     vision.ocr_results = [
+        [OcrLine("Jane Sample Knee pain referral High")],
+        [OcrLine("Jane Sample Knee pain referral High")],
         [OcrLine("Jane Sample Knee pain referral High")],
         [OcrLine("Jane Sample Knee pain referral High")],
         [OcrLine("Taylor Duplicate Knee pain referral High")],
@@ -2697,6 +2701,9 @@ def test_consequential_remote_reruns_identity_on_fresh_frame(bundle, run_dir):
         Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.99),
     ]
     vision.ocr_results = [
+        [OcrLine("Jane Sample Knee pain referral High")],
+        # The candidate census reads the whole frame once the first read
+        # verifies (runtime.identity_census); one row, so it stays verified.
         [OcrLine("Jane Sample Knee pain referral High")],
         [OcrLine("Taylor Duplicate Knee pain referral High")],
     ]
@@ -2728,6 +2735,9 @@ def test_fresh_pre_actuation_identity_loss_has_typed_refusal_and_zero_input(
         Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.99),
     ]
     vision.ocr_results = [
+        [OcrLine("Jane Sample Knee pain referral High")],
+        # The candidate census reads the whole frame once the first read
+        # verifies (runtime.identity_census); one row, so it stays verified.
         [OcrLine("Jane Sample Knee pain referral High")],
         [OcrLine("Taylor Duplicate Knee pain referral High")],
     ]

From f3f0b80ed6b1f8a50e6326e6542026b2970a2db3 Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 19:36:53 -0400
Subject: [PATCH 4/4] test(flow): a relabel onto Sign is refused as an added
 write

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/flow/tests/test_standard_change.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/packages/flow/tests/test_standard_change.py b/packages/flow/tests/test_standard_change.py
index ba0c911e3..12742c2b2 100644
--- a/packages/flow/tests/test_standard_change.py
+++ b/packages/flow/tests/test_standard_change.py
@@ -864,7 +864,8 @@ def client(attend: bool) -> TestClient:
     [
         ("", "", "the control has no readable label"),
         ("Void encounter", "Void encounter", "the control reads as 'void'"),
-        ("Sign note", "Sign note", "the control reads as 'sign'"),
+        # "Sign" is a write verb (openadapt_flow.risk), so the relabel adds one.
+        ("Sign note", "Sign note", "the repair adds a save"),
     ],
     ids=["unlabeled", "void", "sign"],
 )