From 7e263e054ff49e1de977ddecbf43553232080cc2 Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 14:45:31 -0400
Subject: [PATCH 1/6] fix(flow): relabel repair, a duplicate check that reads
 the match list, fresh run ids for re-qualification

- repair/relabel.py: when a save button is renamed, prove from the recorded
  identity hash that only its label changed (same control id, kind and form),
  build a candidate that changes only that label, re-qualify it on the test
  copy, ask a person once, then re-admit by lineage, or, for a control in the
  record check, return a fresh admission for the deployment to pin.
- no_duplicate_candidate postcondition: an app's possible-duplicate list must
  have loaded and name no row with the run's value of any identity parameter
  (for example the last name or date of birth). It fails closed on an unreadable
  list, a list that hasn't loaded, or a missing value. The browser backend reads
  the rows from the DOM (element_texts).
- readmission.requalify names each re-qualification's case runs afresh. A case
  run's authorization id is its run id, and a process refuses an id it has
  already used, so re-qualifying in the process that qualified the parent was
  always refused.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/flow/docs/DUPLICATE_CHECK.md         |  61 ++
 packages/flow/docs/SELF_REPAIR.md             |  50 ++
 .../backends/playwright_backend.py            |  20 +
 packages/flow/openadapt_flow/ir.py            |  20 +
 .../flow/openadapt_flow/repair/readmission.py |   7 +
 .../flow/openadapt_flow/repair/relabel.py     | 563 ++++++++++++++++++
 .../flow/openadapt_flow/runtime/replayer.py   |  53 ++
 .../flow/tests/test_no_duplicate_candidate.py | 107 ++++
 .../flow/tests/test_repair_readmission.py     |  48 +-
 packages/flow/tests/test_repair_relabel.py    | 389 ++++++++++++
 10 files changed, 1308 insertions(+), 10 deletions(-)
 create mode 100644 packages/flow/docs/DUPLICATE_CHECK.md
 create mode 100644 packages/flow/openadapt_flow/repair/relabel.py
 create mode 100644 packages/flow/tests/test_no_duplicate_candidate.py
 create mode 100644 packages/flow/tests/test_repair_relabel.py

diff --git a/packages/flow/docs/DUPLICATE_CHECK.md b/packages/flow/docs/DUPLICATE_CHECK.md
new file mode 100644
index 000000000..61e151467
--- /dev/null
+++ b/packages/flow/docs/DUPLICATE_CHECK.md
@@ -0,0 +1,61 @@
+# A duplicate check that reads the match list
+
+Many registration forms run a duplicate search before they save. OpenEMR's New
+Patient form, for example, opens a list of existing charts that might be the
+same person. A recorded program usually expects the empty list ("No matches
+were found."), so any listed chart stops the item before the save.
+
+OpenEMR lists a chart that shares only a first name with the new patient. On
+the lifecycle soak's synthetic worklist, most of those stops were false: the
+listed chart was clearly someone else, and a person entered the item by hand.
+
+## The postcondition
+
+`no_duplicate_candidate` replaces the "empty list" check on the step that
+opens the search:
+
+```json
+{
+  "kind": "no_duplicate_candidate",
+  "text": "No matches were found.",
+  "candidate_rows": "tr.oneresult",
+  "candidate_frame_path": ["#modalframe"],
+  "candidate_params": ["lname", "DOB"],
+  "timeout_s": 10.0
+}
+```
+
+It passes only when the list has loaded and names no real candidate:
+
+- A listed row is a **candidate** when it contains every word of this run's
+  value of any `candidate_params` parameter (case-insensitive). With the
+  example above, a chart with the same last name or the same date of birth
+  stops the item. A chart that shares only a first name doesn't.
+- An **empty list** passes only when its "no matches" marker (`text`) is
+  shown, so a list that hasn't loaded yet never passes.
+- It **fails closed**: an unreadable list (frame missing or ambiguous, or a
+  backend that can't read rows), a missing or empty parameter value, or any
+  candidate row all fail the step before anything is saved.
+
+The rule only lets an item continue past rows that share none of the identity
+values. A same-name patient with a mistyped date of birth still stops, and a
+person decides. The rows are read from the DOM (`element_texts` on the browser
+backend), never from OCR.
+
+## Changing a program to use it
+
+The compiler doesn't propose this postcondition yet. Changing it is a program
+change, so the program needs a fresh qualification and admission. The
+lifecycle soak's `production_arm.with_duplicate_list_check` makes the change
+on the OpenEMR referral program, and the soak admits that version afresh
+(`"duplicate_check": true`).
+
+## Limits
+
+- The list must render its rows together, as OpenEMR's server-rendered list
+  does. A list that streams rows in one by one could pass on a partial list.
+- A value is recognized only in the form the run typed it. If the app shows a
+  value in another form (a date as `09/09/1990` for `1990-09-09`), the check
+  misses that candidate and lets the item continue. Name only parameters the
+  list shows exactly as typed, and check this on the real list before
+  admitting the program. OpenEMR's list shows the date of birth as typed.
diff --git a/packages/flow/docs/SELF_REPAIR.md b/packages/flow/docs/SELF_REPAIR.md
index 543c3e86e..7e4521f74 100644
--- a/packages/flow/docs/SELF_REPAIR.md
+++ b/packages/flow/docs/SELF_REPAIR.md
@@ -404,6 +404,56 @@ self-repair runs without a named profile or under Demo, and the record check
 still runs. The Python API can run the Standard profile contract directly, as
 the tutorial does.
 
+### A relabeled control: proof that only the label changed
+
+Self-repair never repairs a stop on a failed identity check, because a
+renamed control usually can't be told from a different control. One case can
+be told apart. When an app release renames a button, the ladder finds the same
+control on the structural rung, and the run stops because the control's
+structured identity (its id, kind, form and label) no longer matches.
+
+The recorded identity is a salted hash, but the bundle also keeps the recorded
+label in plaintext (the step's intent and OCR text).
+`openadapt_flow/repair/relabel.py` rebuilds the recorded identity from the
+observed one with only the label swapped back. If that matches the recorded
+hash, the id, kind and form are proven unchanged and only the label differs.
+It refuses everything else:
+
+- a step that received input, or any earlier save that did (a step that may
+  have written is never retried);
+- an identity check that failed outside the structured tier, or one that names
+  a record value rather than a control;
+- a target found below the structural rung;
+- an identity that doesn't match under any recorded label.
+
+The repair then follows the pinned-bundle path above:
+
+1. The candidate bundle changes only that step's identity hash, intent and
+   OCR text.
+2. It is re-qualified with the parent's own campaign on the test copy, which
+   now runs the new release. Each re-qualification names its case runs
+   afresh, because a process refuses a run's authorization id once it has
+   been used, and the parent may have been qualified in the same process.
+3. A person approves once, with the old and new label and the
+   re-qualification in front of them.
+4. The certified copy runs next:
+   - For a control that is not part of the record check, it is re-admitted
+     with signed lineage, and the pin accepts it through the chain.
+   - A save button's identity is part of the record check, so no lineage hop
+     may carry it (`fixed_contract_refusal`). The person's approval is then a
+     fresh admission of the certified copy, and the deployment pins its new
+     digest. `readmit` returns `fresh_admission` and never re-pins by itself.
+
+The stopped item never wrote, so it runs again on the re-admitted version as a
+retry; `problems_only` sends a retry to a person. Under a production
+qualification admission or a managed dispatch the repair stays refused, as for
+every other repair.
+
+On the local OpenEMR referral (proof-15's configuration), a relabeled save
+button stopped every item until a person re-taught the step. The lifecycle
+soak measures this path: see
+the lifecycle soak (`benchmark/lifecycle`).
+
 ## What is recorded
 
 | Where | What |
diff --git a/packages/flow/openadapt_flow/backends/playwright_backend.py b/packages/flow/openadapt_flow/backends/playwright_backend.py
index 8e95edacb..5778c218b 100644
--- a/packages/flow/openadapt_flow/backends/playwright_backend.py
+++ b/packages/flow/openadapt_flow/backends/playwright_backend.py
@@ -1691,6 +1691,26 @@ def focused_element_matches(self, selector: str) -> Optional[bool]:
         )
         return result if isinstance(result, bool) else None
 
+    def element_texts(
+        self, selector: str, frame_path: tuple[str, ...] = ()
+    ) -> Optional[list[str]]:
+        """The rendered text of every element matching ``selector``.
+
+        Read in the frame ``frame_path`` names (outermost first). ``None``
+        when the frame is missing or ambiguous, or the read fails; an empty
+        list means the frame is there and nothing matches. Read-only; never
+        raises.
+        """
+
+        try:
+            scope = self._resolve_scope(tuple(frame_path))
+            if scope is None:
+                return None
+            texts = scope.locator(selector).all_inner_texts()
+        except Exception:  # noqa: BLE001 - unreadable is None
+            return None
+        return [str(text) for text in texts]
+
     def focused_select_label(self) -> Optional[str]:
         """The focused ``<select>``'s selected option label, or None."""
 
diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py
index 6c88d7982..9a3fd2cbd 100644
--- a/packages/flow/openadapt_flow/ir.py
+++ b/packages/flow/openadapt_flow/ir.py
@@ -474,6 +474,14 @@ class PostconditionKind(str, Enum):
     # The focused dropdown shows the run's value of ``param`` (its selected
     # option's label, whitespace-squashed and case-insensitive).
     FIELD_VALUE = "field_value"
+    # An app's possible-duplicate list (rows matching ``candidate_rows`` in
+    # the frame ``candidate_frame_path``) has finished loading and names no
+    # real candidate: no listed row carries this run's value of any of
+    # ``candidate_params`` (for example the last name or the date of birth).
+    # An empty list passes only with its "no matches" marker (``text``)
+    # shown, so a list that has not loaded yet never passes. Unreadable rows
+    # fail the check.
+    NO_DUPLICATE_CANDIDATE = "no_duplicate_candidate"
 
 
 class Postcondition(BaseModel):
@@ -497,6 +505,15 @@ class Postcondition(BaseModel):
     param: Optional[str] = Field(
         default=None, pattern=r"^[A-Za-z_][A-Za-z0-9_]{0,127}$"
     )
+    #: NO_DUPLICATE_CANDIDATE: the CSS selector of one listed candidate row.
+    candidate_rows: Optional[str] = Field(default=None, min_length=1, max_length=256)
+    #: NO_DUPLICATE_CANDIDATE: the frame chain the list is in (outermost first).
+    candidate_frame_path: Optional[tuple[str, ...]] = Field(default=None, max_length=8)
+    #: NO_DUPLICATE_CANDIDATE: parameters whose value in any row makes it a
+    #: real candidate.
+    candidate_params: Optional[tuple[str, ...]] = Field(
+        default=None, min_length=1, max_length=8
+    )
 
     @model_serializer(mode="wrap")
     def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
@@ -508,6 +525,9 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]:
             data.pop("field_selector", None)
         if self.param is None:
             data.pop("param", None)
+        for name in ("candidate_rows", "candidate_frame_path", "candidate_params"):
+            if getattr(self, name) is None:
+                data.pop(name, None)
         return data
 
 
diff --git a/packages/flow/openadapt_flow/repair/readmission.py b/packages/flow/openadapt_flow/repair/readmission.py
index fc5ad9cd4..dde9079d9 100644
--- a/packages/flow/openadapt_flow/repair/readmission.py
+++ b/packages/flow/openadapt_flow/repair/readmission.py
@@ -34,6 +34,7 @@
 
 import hashlib
 import json
+import uuid
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
@@ -220,6 +221,11 @@ def requalify(
         from openadapt_flow.qualify.campaign import qualify
 
         run = qualify
+    # Each case run's authorization id is its run id, and a process refuses an
+    # authorization id it has already consumed. The parent was qualified with
+    # the campaign's own run ids, possibly in this same process, so a
+    # re-qualification names its runs afresh.
+    run_id_prefix = f"requalify-{uuid.uuid4().hex[:16]}"
     try:
         result = run(
             Path(candidate_bundle),
@@ -229,6 +235,7 @@ def requalify(
             trials=requalification.trials,
             certify=provenance.policy_name,
             hooks=requalification.hooks,
+            run_id_prefix=run_id_prefix,
         )
     except Exception as exc:  # noqa: BLE001 - every campaign failure refuses
         raise ReadmissionRefused(
diff --git a/packages/flow/openadapt_flow/repair/relabel.py b/packages/flow/openadapt_flow/repair/relabel.py
new file mode 100644
index 000000000..744e3f47e
--- /dev/null
+++ b/packages/flow/openadapt_flow/repair/relabel.py
@@ -0,0 +1,563 @@
+"""Repair a control the vendor relabeled, with proof that only its label changed.
+
+When an app release renames a button (for example "Confirm Create New Patient"
+becomes "Save new patient"), the run stops before the save: the ladder finds
+the same control, but its structured identity no longer matches the recorded
+one. Self-repair never repairs a failed identity check, because in general it
+can't tell a renamed control from a different control. This module handles
+the one case where it can.
+
+The recorded structured identity is a salted hash, so the recorded label is
+not readable from it. But the bundle keeps the recorded label in plaintext
+(the step's intent and its anchor's OCR text). :func:`propose` rebuilds the
+recorded identity from the observed one with only the label swapped back, and
+checks it against the recorded hash. A match proves that the control id, the
+form and the control kind are all unchanged, and that only the label differs.
+
+Everything else is refused:
+
+- the halted step acted, or any step before it delivered input to a save or
+  showed an effect (a step that may have written is never retried);
+- the identity check failed on anything but the structured tier, or names a
+  record field (a parameter) rather than a control;
+- the ladder found the target below the structural rung (a structural
+  handle always names exactly one control);
+- the rebuilt identity doesn't match the recorded hash under any recorded
+  label.
+
+A proposal is never applied on its own. :func:`build` writes a candidate
+bundle with the new label in the step's identity template, intent and OCR
+text, and nothing else changed. :func:`readmit` then re-qualifies the
+candidate on a test copy with the parent's own campaign, asks a person once
+(the review shows the old and new label and the re-qualification), and signs
+the lineage that lets a pin to the parent accept the certified copy
+(:mod:`openadapt_flow.repair.readmission`). Any failure leaves the parent
+pinned and saves nothing.
+
+A production qualification admission and a managed dispatch are signed
+outside this machine, so their runs stay refused (docs/COMPATIBILITY.md).
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Callable, Optional
+
+from openadapt_flow.ir import RunReport, Step, Workflow
+
+#: Only a control: an identity naming a record field is never relabeled.
+_CONTROL_KINDS = frozenset({"action"})
+#: The identity key that holds the visible label.
+_LABEL_KEY = "label"
+
+
+class RelabelRefused(RuntimeError):
+    """The relabel repair can't go ahead; nothing was changed or saved."""
+
+
+@dataclass(frozen=True)
+class RelabelProposal:
+    """A control whose label changed, and the proof that nothing else did."""
+
+    step_id: str
+    old_label: str
+    new_label: str
+    #: The observed structured identity, as the run read it (no record values:
+    #: only the control's id, kind, form path and label).
+    observed_identity: str
+    #: Every key of the identity except the label, unchanged by proof.
+    unchanged: dict[str, str]
+
+    def summary(self) -> str:
+        return (
+            f"{self.step_id}: the control now reads {self.new_label!r} instead of "
+            f"{self.old_label!r}; its id, kind and form are unchanged (the "
+            "recorded identity hash matches with only the label swapped back)"
+        )
+
+
+@dataclass
+class RelabelReview:
+    """What the person sees before approving a relabel repair."""
+
+    proposal: RelabelProposal
+    parent_digest: str
+    certified_digest: str
+    requalification: dict[str, Any]
+    #: Set when the relabeled control is part of the record check (a save
+    #: button's identity is): no lineage hop can carry it, so approving it is
+    #: a fresh admission of the certified copy. Says why.
+    fresh_admission: Optional[str] = None
+
+
+@dataclass
+class RelabelResult:
+    """The outcome of :func:`readmit`."""
+
+    #: "readmitted" (lineage hop), "fresh_admission" (a person approved the
+    #: certified copy, which runs once the deployment pins its digest),
+    #: "declined" or "refused".
+    status: str
+    reason: str = ""
+    bundle_dir: Optional[Path] = None
+    digest: Optional[str] = None
+    lineage: Optional[dict[str, Any]] = None
+    approved_by: Optional[str] = None
+    #: Machine and person seconds, by stage, for the operator-time ledger.
+    timings: dict[str, float] = field(default_factory=dict)
+
+
+# -- propose ---------------------------------------------------------------------------
+
+
+def _quoted(text: str) -> Optional[str]:
+    start, end = text.find("'"), text.rfind("'")
+    if 0 <= start < end:
+        return text[start + 1 : end]
+    return None
+
+
+def _recorded_labels(step: Step) -> list[str]:
+    """Every plaintext label the bundle recorded for this step, in order."""
+
+    anchor = step.anchor
+    found: list[str] = []
+    for label in (
+        anchor.ocr_text if anchor is not None else None,
+        _quoted(step.intent or ""),
+        (
+            anchor.structural.name
+            if anchor is not None and anchor.structural is not None
+            else None
+        ),
+    ):
+        if label and label not in found:
+            found.append(label)
+    return found
+
+
+def _certifying_policy(workflow: Workflow) -> Any:
+    """The policy the bundle is certified under, as the approval gate reads it.
+
+    A step the certified qualification classified as not saving (a button
+    that opens a duplicate check, say) counts as a save without it.
+    """
+
+    provenance = workflow.manifest.provenance if workflow.manifest else None
+    if provenance is None or not provenance.certified or not provenance.policy_name:
+        return None
+    from openadapt_flow.policy import load_policy
+
+    try:
+        return load_policy(provenance.policy_name)
+    except Exception:  # noqa: BLE001 - an unknown policy proves nothing
+        return None
+
+
+def propose(report: RunReport, workflow: Workflow) -> "RelabelProposal | str":
+    """A relabel proposal for this halt, or why it isn't one."""
+
+    from openadapt_flow.run_gate import is_consequential
+    from openadapt_flow.runtime.identity_template import verify_structured_template
+
+    steps = {step.id: step for step in workflow.steps}
+    policy = _certifying_policy(workflow)
+    for result in report.results:
+        step = steps.get(result.step_id)
+        if step is None:
+            continue
+        if result.delivery_attempted and is_consequential(
+            step, workflow, certifying_policy=policy
+        ):
+            return f"input reached {step.id}, a step that saves; never repaired"
+        for item in result.effect_evidence:
+            if item.observed_effect not in (None, "absent"):
+                return f"{step.id} has an observed effect; never repaired"
+    if report.halt is None:
+        return "the run did not halt at a step"
+    halted = next(
+        (r for r in reversed(report.results) if not r.ok and not r.skipped), None
+    )
+    if halted is None or halted.step_id != report.halt.state_id:
+        return "the report's halt does not name a failed step"
+    step = steps.get(halted.step_id)
+    if step is None or step.anchor is None:
+        return "the halted step has no on-screen target"
+    if halted.delivery_attempted is not False:
+        return f"input may have reached {step.id}; only a step that never acted is repaired"
+    identity = halted.identity
+    if identity is None or identity.status != "mismatch":
+        return f"{step.id} did not stop on its identity check"
+    if identity.mode != "structured" or identity.param is not None:
+        return f"{step.id}'s identity check failed outside the structured tier"
+    resolution = halted.resolution
+    handle = resolution.structural_handle if resolution is not None else None
+    if resolution is None or resolution.rung != "structural" or handle is None:
+        return f"{step.id}'s target was not found on the structural rung"
+    template = step.anchor.identity_template
+    if template is None or not template.structured:
+        return f"{step.id} has no recorded structured identity to compare"
+    if template.structured_params:
+        return f"{step.id}'s identity carries a record value; never relabeled"
+    observed = identity.observed or ""
+    try:
+        live = json.loads(observed)
+    except ValueError:
+        return f"{step.id}'s observed identity is not a structured control"
+    if (
+        not isinstance(live, dict)
+        or not all(isinstance(v, str) for v in live.values())
+        or str(live.get("target_kind", "")) not in _CONTROL_KINDS
+        or not live.get(_LABEL_KEY)
+    ):
+        return f"{step.id}'s observed identity is not a labeled control"
+    new_label = live[_LABEL_KEY]
+    encoded = json.dumps(new_label)
+    if observed.count(encoded) != 1:
+        return f"{step.id}'s observed label can't be isolated"
+    for old_label in _recorded_labels(step):
+        if old_label == new_label:
+            continue
+        rebuilt = observed.replace(encoded, json.dumps(old_label))
+        check = verify_structured_template(template, rebuilt)
+        if check is not None and check.status == "verified":
+            return RelabelProposal(
+                step_id=step.id,
+                old_label=old_label,
+                new_label=new_label,
+                observed_identity=observed,
+                unchanged={k: v for k, v in live.items() if k != _LABEL_KEY},
+            )
+    return (
+        f"{step.id}'s control changed more than its label (no recorded label "
+        "matches the recorded identity)"
+    )
+
+
+# -- build -----------------------------------------------------------------------------
+
+
+def _relabeled_template(template: Any, observed: str) -> Any:
+    from openadapt_flow.runtime.identity_template import build_identity_template
+
+    fresh = build_identity_template(
+        None, structured_identity=observed, salt_hex=template.salt
+    )
+    assert fresh is not None and fresh.structured
+    signals = {
+        k: v
+        for k, v in (template.signal_hashes or {}).items()
+        if not k.startswith("structured|")
+    }
+    signals.update(
+        {
+            k: v
+            for k, v in (fresh.signal_hashes or {}).items()
+            if k.startswith("structured|")
+        }
+    )
+    return template.model_copy(
+        update={"structured": fresh.structured, "signal_hashes": signals}
+    )
+
+
+def build(
+    bundle_dir: Path | str,
+    proposal: RelabelProposal,
+    out_dir: Path | str,
+    *,
+    key: Optional[str] = None,
+) -> Path:
+    """Write the candidate bundle: the new label, and nothing else changed.
+
+    The candidate is uncertified; only :func:`readmit` can certify and admit
+    it.
+    """
+
+    from openadapt_flow.repair.self_repair import _uncertify_patch
+
+    bundle_dir, out_dir = Path(bundle_dir), Path(out_dir)
+    workflow = Workflow.load(bundle_dir, key=key) if key else Workflow.load(bundle_dir)
+    patched = workflow.model_copy(deep=True)
+    steps = list(patched.steps)
+    index = next((i for i, s in enumerate(steps) if s.id == proposal.step_id), None)
+    if index is None:
+        raise RelabelRefused(f"the bundle has no step {proposal.step_id}")
+    step = steps[index]
+    anchor = step.anchor
+    if anchor is None or anchor.identity_template is None:
+        raise RelabelRefused(f"{step.id} has no identity template")
+    new_anchor = anchor.model_copy(
+        update={
+            "identity_template": _relabeled_template(
+                anchor.identity_template, proposal.observed_identity
+            ),
+            "ocr_text": (
+                proposal.new_label
+                if anchor.ocr_text == proposal.old_label
+                else anchor.ocr_text
+            ),
+        }
+    )
+    steps[index] = step.model_copy(
+        update={
+            "anchor": new_anchor,
+            "intent": (step.intent or "").replace(
+                f"'{proposal.old_label}'", f"'{proposal.new_label}'"
+            ),
+        }
+    )
+    patched.steps = steps
+    _uncertify_patch(patched)
+    if out_dir.exists():
+        shutil.rmtree(out_dir)
+    shutil.copytree(
+        bundle_dir, out_dir, ignore=shutil.ignore_patterns("repair", "manifest.json")
+    )
+    patched.save(out_dir, encrypt=bool(workflow.encrypted), key=key)
+    return out_dir
+
+
+# -- re-qualify, a person's one approval, signed lineage ------------------------------
+
+
+def _rows(result: Any) -> list[dict[str, Any]]:
+    return [
+        {k: row.get(k) for k in ("case_id", "trial", "expected", "transaction_outcome")}
+        for row in (getattr(result, "rows", None) or [])
+        if isinstance(row, dict)
+    ]
+
+
+def _campaigns(rows: list[dict[str, Any]]) -> tuple[Any, Any]:
+    """The re-qualification's cases as the store's replay and fault campaigns."""
+
+    from openadapt_flow.repair.candidate import CampaignCaseResult, CampaignResult
+
+    def campaign(kind: str, picked: list[dict[str, Any]]) -> Any:
+        cases = [
+            CampaignCaseResult(
+                label=f"{r.get('case_id')}#{r.get('trial')}"[:128],
+                kind=str(r.get("case_id") or "case")[:64],
+                passed=True,
+                detail=f"{r.get('transaction_outcome')} as expected"[:512],
+            )
+            for r in picked
+        ]
+        return CampaignResult(
+            campaign=kind,  # type: ignore[arg-type]
+            passed=bool(cases),
+            total=len(cases),
+            failed=0,
+            cases=cases,
+        )
+
+    replay = [r for r in rows if r.get("case_id") == "representative"]
+    fault = [r for r in rows if r.get("case_id") != "representative"]
+    return campaign("replay", replay), campaign("fault", fault)
+
+
+def readmit(
+    *,
+    parent_bundle: Path | str,
+    candidate_bundle: Path | str,
+    proposal: RelabelProposal,
+    store: Any,
+    requalification: Any,
+    ask_person: Callable[[RelabelReview], tuple[str, str]],
+    work_dir: Path | str,
+    evidence_run_dir: Optional[Path | str] = None,
+    key_dir: Optional[Path] = None,
+    clock: Callable[[], float] = time.monotonic,
+) -> RelabelResult:
+    """Re-qualify the candidate, ask a person once, and sign the lineage.
+
+    ``ask_person(review)`` returns ``(decision, approver_id)``, where decision
+    is ``"approve"`` or ``"decline"``. It is asked only after the candidate
+    passed its parent's whole campaign on the test copy. ``store`` is the
+    :class:`~openadapt_flow.repair.lifecycle.RepairStore` that holds both
+    bundles and the lineage record.
+    """
+
+    import dataclasses
+
+    from openadapt_flow.repair.candidate import Actor
+    from openadapt_flow.repair.readmission import (
+        ReadmissionRefused,
+        Requalification,
+        admit,
+        requalify,
+    )
+    from openadapt_flow.repair.registration import build_candidate
+
+    automation = Actor(kind="automation", id="openadapt-relabel-repair")
+    work = Path(work_dir)
+    timings: dict[str, float] = {}
+    captured: dict[str, Any] = {}
+    inner = requalification.run
+    if inner is None:
+        from openadapt_flow.qualify.campaign import qualify as inner
+
+    def run(*args: Any, **kwargs: Any) -> Any:
+        captured["result"] = inner(*args, **kwargs)
+        return captured["result"]
+
+    candidate_digest = Workflow.load(candidate_bundle).manifest.content_digest  # type: ignore[union-attr]
+    parent_digest = Workflow.load(parent_bundle).manifest.content_digest  # type: ignore[union-attr]
+    t0 = clock()
+    try:
+        requalified = dataclasses.replace(
+            requalify(
+                parent_bundle,
+                candidate_bundle,
+                work / "requalify",
+                dataclasses.replace(requalification, run=run)
+                if isinstance(requalification, Requalification)
+                else requalification,
+            ),
+            candidate_digest=candidate_digest,
+        )
+    except ReadmissionRefused as exc:
+        timings["requalify_s"] = round(clock() - t0, 1)
+        result: Any = captured.get("result")
+        if (
+            "fresh admission" not in str(exc)
+            or not getattr(result, "passed", False)
+            or not getattr(result, "certified", False)
+        ):
+            return RelabelResult(status="refused", reason=str(exc), timings=timings)
+        # The campaign passed and certified a copy, but the relabeled control
+        # is part of the fixed contract, so no lineage hop may carry it. A
+        # person may still admit the certified copy afresh.
+        return _fresh_admission(
+            proposal,
+            parent_digest,
+            Path(result.bundle_dir),
+            _rows(result),
+            str(exc),
+            ask_person,
+            timings,
+            clock,
+        )
+    timings["requalify_s"] = round(clock() - t0, 1)
+    rows = _rows(captured.get("result"))
+    candidate = build_candidate(
+        parent_bundle,
+        requalified.bundle_dir,
+        source="manual",
+        evidence_run_dir=evidence_run_dir,
+    )
+    if candidate.state == "rejected":
+        return RelabelResult(
+            status="refused",
+            reason=candidate.rejection_reason
+            or "the certified copy weakens the contract",
+            timings=timings,
+        )
+    candidate = store.add_candidate(candidate)
+    review = RelabelReview(
+        proposal=proposal,
+        parent_digest=parent_digest,
+        certified_digest=requalified.digest,
+        requalification={
+            "policy": requalified.policy_name,
+            "cases": list(requalified.case_ids),
+            "rows": rows,
+        },
+    )
+    t1 = clock()
+    decision, approver = ask_person(review)
+    timings["person_wait_s"] = round(clock() - t1, 1)
+    if decision != "approve":
+        store.reject(
+            candidate.candidate_id,
+            Actor(kind="human", id=approver),
+            "declined by a person (relabel repair)",
+        )
+        return RelabelResult(
+            status="declined", reason=f"declined by {approver}", timings=timings
+        )
+    person = Actor(kind="human", id=approver)
+    t2 = clock()
+    store.review(candidate.candidate_id, person)
+    replay, fault = _campaigns(rows)
+    for result in (replay, fault):
+        store.record_campaign(candidate.candidate_id, result, automation)
+    store.approve(candidate.candidate_id, person)
+    store.stage(candidate.candidate_id, automation)
+    try:
+        record = admit(
+            store.root,
+            parent_bundle=parent_bundle,
+            candidate_digest=candidate_digest,
+            candidate_id=candidate.candidate_id,
+            requalified=requalified,
+            approved_by=approver,
+            route="person",
+            key_dir=key_dir,
+        )
+    except ReadmissionRefused as exc:
+        timings["readmit_s"] = round(clock() - t2, 1)
+        return RelabelResult(status="refused", reason=str(exc), timings=timings)
+    timings["readmit_s"] = round(clock() - t2, 1)
+    return RelabelResult(
+        status="readmitted",
+        bundle_dir=store.bundle_path(requalified.digest),
+        digest=requalified.digest,
+        lineage=record,
+        approved_by=approver,
+        timings=timings,
+    )
+
+
+def _fresh_admission(
+    proposal: RelabelProposal,
+    parent_digest: str,
+    certified_dir: Path,
+    rows: list[dict[str, Any]],
+    why: str,
+    ask_person: Callable[[RelabelReview], tuple[str, str]],
+    timings: dict[str, float],
+    clock: Callable[[], float],
+) -> RelabelResult:
+    """Ask a person to admit the certified copy afresh (no lineage)."""
+
+    certified = Workflow.load(certified_dir)
+    provenance = certified.manifest.provenance if certified.manifest else None
+    if provenance is None or not provenance.certified:
+        return RelabelResult(
+            status="refused",
+            reason="the re-qualified copy is not certified",
+            timings=timings,
+        )
+    digest = certified.manifest.content_digest  # type: ignore[union-attr]
+    review = RelabelReview(
+        proposal=proposal,
+        parent_digest=parent_digest,
+        certified_digest=digest,
+        requalification={
+            "policy": provenance.policy_name,
+            "cases": sorted({str(r.get("case_id")) for r in rows}),
+            "rows": rows,
+        },
+        fresh_admission=why,
+    )
+    t1 = clock()
+    decision, approver = ask_person(review)
+    timings["person_wait_s"] = round(clock() - t1, 1)
+    if decision != "approve":
+        return RelabelResult(
+            status="declined", reason=f"declined by {approver}", timings=timings
+        )
+    return RelabelResult(
+        status="fresh_admission",
+        reason=why,
+        bundle_dir=certified_dir,
+        digest=digest,
+        approved_by=approver,
+        timings=timings,
+    )
diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py
index e5df26bc3..7bcb44129 100644
--- a/packages/flow/openadapt_flow/runtime/replayer.py
+++ b/packages/flow/openadapt_flow/runtime/replayer.py
@@ -1881,6 +1881,9 @@ def _run_unsealed_steps(
         # iteration scope from the exact loop cursor below. Hashing ``params``
         # inside a loop would count each row twice and reject valid runs.
         self._governed_base_params = dict(params)
+        # The run's own parameters, for postconditions that compare the screen
+        # with a value (a program run also tracks them per state).
+        self._run_params: dict[str, RuntimeParamScalar] = dict(params)
         # Preserve the exact caller-owned worklist object for last-common-point
         # authorization checks.  A callback that mutates it after admission
         # must invalidate the authorization before any reacquired input edge.
@@ -17665,6 +17668,51 @@ def _local_edit_held(self, pc: Any, start_state: dict[str, Any]) -> bool:
                 return False
         return True
 
+    def _no_duplicate_candidate(self, pc: Any, frame_png: bytes) -> bool:
+        """The app's possible-duplicate list has loaded and names no candidate.
+
+        A row is a candidate when its words include every word of this run's
+        value of any ``pc.candidate_params`` parameter (case-insensitive).
+        Fails closed: an unreadable list, a missing or empty parameter, or an
+        empty list without its "no matches" marker all fail the check.
+        """
+
+        params = (
+            getattr(self, "_current_params", None)
+            or getattr(self, "_run_params", None)
+            or {}
+        )
+        names = tuple(pc.candidate_params or ())
+        if not pc.candidate_rows or not names or not pc.text:
+            return False
+        reader = getattr(self.backend, "element_texts", None)
+        if not callable(reader):
+            return False
+        rows = reader(pc.candidate_rows, tuple(pc.candidate_frame_path or ()))
+        if rows is None:
+            return False
+
+        def words(value: str) -> list[str]:
+            return re.findall(r"[0-9a-z]+", value.casefold())
+
+        wanted = []
+        for name in names:
+            value = words(runtime_param_text(params[name])) if name in params else []
+            if not value:
+                return False
+            wanted.append(value)
+        if not rows:
+            # Nothing listed: only the list's own "no matches" marker shows
+            # the search finished.
+            if self._dom_text_visible(pc.text):
+                return True
+            return self.vision.text_present(frame_png, pc.text)
+        for row in rows:
+            row_words = set(words(row))
+            if any(all(w in row_words for w in value) for value in wanted):
+                return False
+        return True
+
     def _field_value_shown(self, pc: Any) -> bool:
         """The focused field shows this run's value of ``pc.param``.
 
@@ -17704,6 +17752,9 @@ def _describe_postcondition(pc: Any) -> str:
             return "local_edit (URL, title, page count and focus unchanged)"
         if kind == "field_value":
             return f"field_value (the focused field shows {{{pc.param}}})"
+        if kind == "no_duplicate_candidate":
+            names = ", ".join(pc.candidate_params or ())
+            return f"no_duplicate_candidate (no listed row has this run's {names})"
         if kind in ("text_present", "text_absent"):
             return f"{kind} {pc.text!r}"
         if kind in ("url_changed", "title_changed", "new_tab_opened"):
@@ -17754,6 +17805,8 @@ def _postcondition_passes(
             return self._local_edit_held(pc, start_state or {})
         if kind == "field_value":
             return self._field_value_shown(pc)
+        if kind == "no_duplicate_candidate":
+            return self._no_duplicate_candidate(pc, frame_png)
         if kind in ("url_changed", "title_changed", "new_tab_opened"):
             key = {
                 "url_changed": "url",
diff --git a/packages/flow/tests/test_no_duplicate_candidate.py b/packages/flow/tests/test_no_duplicate_candidate.py
new file mode 100644
index 000000000..ebe1aea3e
--- /dev/null
+++ b/packages/flow/tests/test_no_duplicate_candidate.py
@@ -0,0 +1,107 @@
+"""The possible-duplicate list names no real candidate (no_duplicate_candidate).
+
+A registration program used to stop whenever an app's duplicate check listed
+any chart. OpenEMR lists a chart that shares only a first name, so most of
+those stops were false. This postcondition reads the listed rows and passes
+only when the list has loaded and no row carries this run's value of any
+identity parameter (here the last name or the date of birth). It never passes
+on a list it can't read, on a list that hasn't loaded, or on a missing value.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+import pytest
+
+from openadapt_flow.ir import Postcondition, PostconditionKind
+
+PARAMS = {"lname": "Jesswood", "fname": "Rosalind", "DOB": "1990-09-09"}
+MARKER = "No matches were found."
+
+
+class _Backend:
+    def __init__(self, rows: Optional[list[str]], marker: bool = False) -> None:
+        self.rows, self.marker = rows, marker
+        self.read: list[Any] = []
+        self.url, self.page_title, self.page_count = "http://x/", "x", 1
+
+    def element_texts(self, selector: str, frame_path: tuple[str, ...]) -> Any:
+        self.read.append((selector, frame_path))
+        return self.rows
+
+    def visible_text_present(self, text: str) -> bool:
+        return self.marker and text == MARKER
+
+
+def _check(backend: _Backend, params: dict[str, str] = PARAMS) -> bool:
+    from openadapt_flow.runtime.replayer import Replayer
+    from tests.test_replayer import FakeVision
+
+    replayer = Replayer(backend, vision=FakeVision())
+    replayer._current_params = dict(params)
+    pc = Postcondition(
+        kind=PostconditionKind.NO_DUPLICATE_CANDIDATE,
+        text=MARKER,
+        candidate_rows="tr.oneresult",
+        candidate_frame_path=("#modalframe",),
+        candidate_params=("lname", "DOB"),
+    )
+    return replayer._postcondition_passes(pc, b"", None, {})
+
+
+def test_a_first_name_only_hit_is_not_a_candidate() -> None:
+    backend = _Backend(["3\tOakwood, Rosalind\t1950-04-21\tFemale"])
+    assert _check(backend) is True
+    assert backend.read == [("tr.oneresult", ("#modalframe",))]
+
+
+@pytest.mark.parametrize(
+    "row",
+    [
+        "5\tJesswood, Rosalind\t1990-09-09\tFemale",  # the same person
+        "5\tJESSWOOD, Ana\t1961-01-01\tFemale",  # a namesake: a person decides
+        "5\tOakwood, Ana\t1990-09-09\tFemale",  # the same date of birth
+    ],
+)
+def test_a_real_candidate_always_stops(row: str) -> None:
+    assert _check(_Backend(["3\tOakwood, Rosalind\t1950-04-21\tFemale", row])) is False
+
+
+def test_an_empty_list_passes_only_with_its_no_matches_marker() -> None:
+    assert _check(_Backend([], marker=True)) is True
+    # Not loaded yet: no rows and no marker.
+    assert _check(_Backend([], marker=False)) is False
+
+
+def test_it_fails_closed() -> None:
+    # The list can't be read (frame missing, or no reader on this backend).
+    assert _check(_Backend(None, marker=True)) is False
+    # A value to compare with is missing.
+    assert _check(_Backend([], marker=True), {"lname": "Jesswood"}) is False
+    assert _check(_Backend([], marker=True), {**PARAMS, "DOB": " "}) is False
+
+
+def test_unset_candidate_fields_serialize_as_before() -> None:
+    pc = Postcondition(kind=PostconditionKind.TEXT_PRESENT, text="Saved")
+    assert not {"candidate_rows", "candidate_frame_path", "candidate_params"} & set(
+        pc.model_dump()
+    )
+
+
+def test_a_linear_run_compares_with_its_own_parameters() -> None:
+    from openadapt_flow.runtime.replayer import Replayer
+    from tests.test_replayer import FakeVision
+
+    # A linear (non-program) run keeps its parameters in _run_params only.
+    replayer = Replayer(
+        _Backend(["3\tOakwood, Rosalind\t1950-04-21"]), vision=FakeVision()
+    )
+    replayer._run_params = dict(PARAMS)
+    pc = Postcondition(
+        kind=PostconditionKind.NO_DUPLICATE_CANDIDATE,
+        text=MARKER,
+        candidate_rows="tr.oneresult",
+        candidate_params=("lname", "DOB"),
+    )
+    assert replayer._postcondition_passes(pc, b"", None, {}) is True
diff --git a/packages/flow/tests/test_repair_readmission.py b/packages/flow/tests/test_repair_readmission.py
index bdb07dabc..c626c5f60 100644
--- a/packages/flow/tests/test_repair_readmission.py
+++ b/packages/flow/tests/test_repair_readmission.py
@@ -90,8 +90,17 @@ class Campaign:
     def __init__(self, *, passes: bool = True, field: str = "note") -> None:
         self.passes, self.field, self.calls = passes, field, []
 
-    def __call__(self, candidate, config, *, out, cases, trials, certify, hooks):
-        self.calls.append({"cases": cases, "certify": certify, "trials": trials})
+    def __call__(
+        self, candidate, config, *, out, cases, trials, certify, hooks, run_id_prefix
+    ):
+        self.calls.append(
+            {
+                "cases": cases,
+                "certify": certify,
+                "trials": trials,
+                "run_id_prefix": run_id_prefix,
+            }
+        )
         if not self.passes:
             raise RuntimeError("the representative case halted")
         label = Workflow.load(candidate).steps[0].anchor.ocr_text
@@ -161,14 +170,15 @@ def test_a_repair_is_requalified_and_readmitted_with_lineage(tmp_path, keys):
         tmp_path, store, root, keys, label="Save record", campaign=campaign
     )
 
-    # The parent's own campaign: same policy, same cases.
-    assert campaign.calls == [
-        {
-            "cases": list(requalified.case_ids),
-            "certify": "clinical-write",
-            "trials": 1,
-        }
-    ]
+    # The parent's own campaign: same policy, same cases, under fresh run ids
+    # (each case's authorization id is its run id, consumed once per process).
+    [call] = campaign.calls
+    assert call.pop("run_id_prefix").startswith("requalify-")
+    assert call == {
+        "cases": list(requalified.case_ids),
+        "certify": "clinical-write",
+        "trials": 1,
+    }
     assert requalified.digest != _digest(root)
     assert record["hop"] == 1 and record["parent_digest"] == _digest(root)
     assert record["policy_name"] == "clinical-write"
@@ -446,3 +456,21 @@ def test_the_readmission_step_signs_lineage_after_approval(tmp_path, keys):
     assert (
         lineage_refusal(store, _digest(root), requalified.digest, key_dir=keys) is None
     )
+
+
+def test_each_requalification_names_its_runs_afresh(tmp_path, keys):
+    # A process that qualified the parent refuses its run ids again, so two
+    # re-qualifications must never share a run-id prefix.
+    root = _certified(tmp_path / "root")
+    campaign = Campaign()
+    for label in ("Save record", "Store record"):
+        _repair(
+            tmp_path,
+            tmp_path / f"store-{label}",
+            root,
+            keys,
+            label=label,
+            campaign=campaign,
+        )
+    prefixes = [call["run_id_prefix"] for call in campaign.calls]
+    assert len(set(prefixes)) == 2
diff --git a/packages/flow/tests/test_repair_relabel.py b/packages/flow/tests/test_repair_relabel.py
new file mode 100644
index 000000000..134a4dfff
--- /dev/null
+++ b/packages/flow/tests/test_repair_relabel.py
@@ -0,0 +1,389 @@
+"""A relabeled control: proof that only its label changed, then re-admission.
+
+openadapt_flow/repair/relabel.py. When a vendor renames a save button, the
+run stops on the structured identity check. The relabel repair proposes a fix
+only when the recorded identity hash matches the observed identity with the
+label swapped back, so the control id, kind and form are proven unchanged.
+Every other identity stop, and any stop after input reached a save, is
+refused. The candidate changes the label and nothing else, and is admitted
+only after re-qualification and a person's approval, through signed lineage.
+
+The workflow and reports are made-up; the observed identity has the shape the
+OpenEMR referral's save button showed in proof-15.
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+from openadapt_flow.ir import (
+    ActionKind,
+    Anchor,
+    RunReport,
+    Step,
+    StructuralLocator,
+    Workflow,
+)
+from openadapt_flow.repair import relabel
+from openadapt_flow.runtime.effects import Effect, EffectKind, ValueExpr
+from openadapt_flow.runtime.identity_template import (
+    build_identity_template,
+    verify_structured_template,
+)
+
+OLD = "Confirm Create New Patient"
+NEW = "Save new patient"
+
+
+def _identity(label: str, **changes: str) -> str:
+    value = {
+        "form_path": "/interface/new/new_comprehensive.php",
+        "label": label,
+        "target_id": "confirm_create",
+        "target_kind": "action",
+        **changes,
+    }
+    return json.dumps(value, sort_keys=True, separators=(",", ":"))
+
+
+def _workflow(*, open_saves: bool = False) -> Workflow:
+    template = build_identity_template(None, structured_identity=_identity(OLD))
+    return Workflow(
+        name="referral",
+        params={"email": "a@example.invalid", "lname": "Example"},
+        steps=[
+            # A step before the save: typing a value, or (open_saves) a click
+            # the fail-closed classifier counts as a save.
+            Step(
+                id="open",
+                intent="click 'Next'",
+                action=ActionKind.CLICK,
+                anchor=Anchor(
+                    template="templates/open.png",
+                    region=(0, 0, 40, 20),
+                    click_point=(20, 10),
+                    ocr_text="Next",
+                ),
+            )
+            if open_saves
+            else Step(
+                id="open", intent="type <lname>", action=ActionKind.TYPE, text="x"
+            ),
+            Step(
+                id="save",
+                intent=f"click '{OLD}'",
+                action=ActionKind.CLICK,
+                anchor=Anchor(
+                    template="templates/save.png",
+                    region=(520, 248, 160, 64),
+                    click_point=(600, 280),
+                    ocr_text=OLD,
+                    structural=StructuralLocator(selector="#confirmCreate"),
+                    identity_template=template,
+                ),
+                effects=[
+                    Effect(
+                        kind=EffectKind.FIELD_EQUALS,
+                        match={"email": ValueExpr(param="email")},
+                        field="lname",
+                        value=ValueExpr(param="lname"),
+                        risk="irreversible",
+                    )
+                ],
+                risk="irreversible",
+                identity_armed=True,
+            ),
+        ],
+    )
+
+
+def _result(step_id: str, **fields: Any) -> dict[str, Any]:
+    return {"step_id": step_id, "intent": step_id, **fields}
+
+
+def _halted(
+    observed: str,
+    *,
+    rung: str = "structural",
+    mode: str = "structured",
+    acted: bool = False,
+) -> RunReport:
+    open_result = _result("open", ok=True, delivery_attempted=True)
+    return RunReport.model_validate(
+        {
+            "workflow_name": "referral",
+            "started_at": "2026-10-24T00:00:00+00:00",
+            "success": False,
+            "transaction_outcome": "REJECTED_POLICY",
+            "results": [
+                open_result,
+                _result(
+                    "save",
+                    ok=False,
+                    delivery_attempted=acted,
+                    resolution={
+                        "rung": rung,
+                        "point": [600, 280],
+                        "confidence": 1.0,
+                        "elapsed_ms": 50.0,
+                        "structural_handle": {
+                            "point": [600, 280],
+                            "confidence": 1.0,
+                            "region": [526, 261, 149, 38],
+                            "candidate_count": 1,
+                        },
+                    },
+                    identity={
+                        "status": "mismatch",
+                        "mode": mode,
+                        "expected": "<structured identity template>",
+                        "observed": observed,
+                    },
+                    error="Identity check failed for step 'save'",
+                ),
+            ],
+            "halt": {
+                "state_id": "save",
+                "intent": f"click '{OLD}'",
+                "reason": "Identity check failed",
+            },
+        }
+    )
+
+
+def test_a_relabeled_save_is_proposed_with_proof() -> None:
+    proposal = relabel.propose(_halted(_identity(NEW)), _workflow())
+    assert isinstance(proposal, relabel.RelabelProposal)
+    assert (proposal.old_label, proposal.new_label) == (OLD, NEW)
+    assert proposal.unchanged["target_id"] == "confirm_create"
+
+
+@pytest.mark.parametrize(
+    "observed",
+    [
+        # another control on the same form, with the new label
+        _identity(NEW, target_id="cancel"),
+        # the same id on another form
+        _identity(NEW, form_path="/interface/patient_file/delete.php"),
+        # a record field, not a control
+        _identity(NEW, target_kind="field"),
+    ],
+)
+def test_a_control_that_changed_more_than_its_label_is_refused(observed: str) -> None:
+    assert isinstance(relabel.propose(_halted(observed), _workflow()), str)
+
+
+def test_unstructured_or_unproven_stops_are_refused() -> None:
+    wf = _workflow()
+    assert isinstance(relabel.propose(_halted("Save new patient"), wf), str)
+    assert isinstance(relabel.propose(_halted(_identity(NEW), mode="context"), wf), str)
+    assert isinstance(
+        relabel.propose(_halted(_identity(NEW), rung="geometry"), wf), str
+    )
+
+
+def test_a_step_that_may_have_written_is_never_relabeled() -> None:
+    wf = _workflow()
+    # The save itself received input.
+    assert "save" in str(relabel.propose(_halted(_identity(NEW), acted=True), wf))
+    # An earlier step that saves received input.
+    refused = relabel.propose(_halted(_identity(NEW)), _workflow(open_saves=True))
+    assert isinstance(refused, str) and "open" in refused
+
+
+def test_the_candidate_changes_only_the_label(tmp_path: Path) -> None:
+    wf = _workflow()
+    bundle = tmp_path / "bundle"
+    wf.save(bundle)
+    proposal = relabel.propose(_halted(_identity(NEW)), Workflow.load(bundle))
+    assert isinstance(proposal, relabel.RelabelProposal)
+    out = relabel.build(bundle, proposal, tmp_path / "candidate")
+    before, after = Workflow.load(bundle), Workflow.load(out)
+    old, new = before.steps[1], after.steps[1]
+    assert new.intent == f"click '{NEW}'" and new.anchor.ocr_text == NEW
+    template = new.anchor.identity_template
+    assert verify_structured_template(template, _identity(NEW)).status == "verified"
+    assert verify_structured_template(template, _identity(OLD)).status == "mismatch"
+    # The salt and every other part of the step are unchanged.
+    assert template.salt == old.anchor.identity_template.salt
+    assert new.effects == old.effects and new.risk == old.risk
+    assert new.anchor.structural == old.anchor.structural
+    assert after.steps[0] == before.steps[0]
+    assert after.params == before.params
+
+
+class _Store:
+    """Records the store calls readmit makes, in order."""
+
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self.calls: list[str] = []
+
+    def add_candidate(self, candidate: Any) -> Any:
+        self.calls.append("add")
+        return candidate
+
+    def __getattr__(self, name: str) -> Any:
+        def call(*args: Any, **kwargs: Any) -> Any:
+            self.calls.append(name)
+            return SimpleNamespace(state="ok")
+
+        return call
+
+    def bundle_path(self, digest: str) -> Path:
+        return self.root / "bundles" / digest
+
+
+def test_a_failed_requalification_asks_nobody(tmp_path: Path, monkeypatch) -> None:
+    from openadapt_flow.repair import readmission
+
+    def failing(*args: Any, **kwargs: Any) -> Any:
+        raise readmission.ReadmissionRefused("the representative case halted")
+
+    monkeypatch.setattr(readmission, "requalify", failing)
+    wf = _workflow()
+    bundle = tmp_path / "bundle"
+    wf.save(bundle)
+    proposal = relabel.propose(_halted(_identity(NEW)), wf)
+    assert isinstance(proposal, relabel.RelabelProposal)
+    candidate = relabel.build(bundle, proposal, tmp_path / "candidate")
+    asked: list[Any] = []
+    store = _Store(tmp_path / "store")
+    result = relabel.readmit(
+        parent_bundle=bundle,
+        candidate_bundle=candidate,
+        proposal=proposal,
+        store=store,
+        requalification=readmission.Requalification(config=None),
+        ask_person=lambda review: asked.append(review) or ("approve", "p"),
+        work_dir=tmp_path / "work",
+    )
+    assert result.status == "refused" and "halted" in result.reason
+    assert asked == [] and store.calls == []
+
+
+def test_a_declined_relabel_admits_nothing(tmp_path: Path, monkeypatch) -> None:
+    from openadapt_flow.repair import readmission, registration
+
+    wf = _workflow()
+    bundle = tmp_path / "bundle"
+    wf.save(bundle)
+    proposal = relabel.propose(_halted(_identity(NEW)), wf)
+    assert isinstance(proposal, relabel.RelabelProposal)
+    candidate = relabel.build(bundle, proposal, tmp_path / "candidate")
+    certified = tmp_path / "certified"
+    shutil.copytree(candidate, certified)
+    monkeypatch.setattr(
+        readmission,
+        "requalify",
+        lambda *a, **k: readmission.Requalified(
+            bundle_dir=certified,
+            digest="c" * 64,
+            policy_name="clinical-write",
+            policy_contract_sha256="d" * 64,
+            case_ids=("representative",),
+            parent_report_sha256="e" * 64,
+            report_sha256="f" * 64,
+            result_sha256=None,
+        ),
+    )
+    monkeypatch.setattr(
+        registration,
+        "build_candidate",
+        lambda *a, **k: SimpleNamespace(state="candidate", candidate_id="cand-1"),
+    )
+    admitted: list[Any] = []
+    monkeypatch.setattr(readmission, "admit", lambda *a, **k: admitted.append(k))
+    store = _Store(tmp_path / "store")
+    result = relabel.readmit(
+        parent_bundle=bundle,
+        candidate_bundle=candidate,
+        proposal=proposal,
+        store=store,
+        requalification=readmission.Requalification(config=None),
+        ask_person=lambda review: ("decline", "operator-1"),
+        work_dir=tmp_path / "work",
+    )
+    assert result.status == "declined"
+    assert admitted == [] and "approve" not in store.calls
+    assert store.calls == ["add", "reject"]
+
+
+def test_a_relabeled_save_is_admitted_afresh_not_by_lineage(
+    tmp_path: Path, monkeypatch
+) -> None:
+    """A save button's identity is part of the record check, so no lineage hop
+    can carry its relabel. Once the campaign passed, a person may admit the
+    certified copy afresh; nothing is admitted without that approval."""
+
+    from openadapt_flow.repair import readmission
+    from tests.test_repair_readmission import _certified
+
+    wf = _workflow()
+    bundle = tmp_path / "bundle"
+    wf.save(bundle)
+    proposal = relabel.propose(_halted(_identity(NEW)), wf)
+    assert isinstance(proposal, relabel.RelabelProposal)
+    candidate = relabel.build(bundle, proposal, tmp_path / "candidate")
+    certified = _certified(tmp_path / "certified")
+    rows = [
+        {"case_id": "representative", "trial": 1, "transaction_outcome": "VERIFIED"}
+    ]
+
+    def requalify(parent, cand, out, requalification):
+        requalification.run(
+            cand,
+            None,
+            out=out,
+            cases=["representative"],
+            trials=1,
+            certify="clinical-write",
+            hooks=None,
+            run_id_prefix="requalify-x",
+        )
+        raise readmission.ReadmissionRefused(
+            "the repair changes the program's record check; it needs a person's "
+            "fresh admission"
+        )
+
+    monkeypatch.setattr(readmission, "requalify", requalify)
+    campaign = lambda *a, **k: SimpleNamespace(  # noqa: E731
+        passed=True, certified=True, bundle_dir=certified, rows=rows
+    )
+    reviews: list[Any] = []
+    store = _Store(tmp_path / "store")
+
+    def ask(review: Any) -> tuple[str, str]:
+        reviews.append(review)
+        return ("approve", "operator-1")
+
+    result = relabel.readmit(
+        parent_bundle=bundle,
+        candidate_bundle=candidate,
+        proposal=proposal,
+        store=store,
+        requalification=readmission.Requalification(config=None, run=campaign),
+        ask_person=ask,
+        work_dir=tmp_path / "work",
+    )
+    assert result.status == "fresh_admission" and result.bundle_dir == certified
+    assert result.lineage is None and store.calls == []
+    [review] = reviews
+    assert "fresh admission" in review.fresh_admission
+    # Declined: nothing to admit.
+    declined = relabel.readmit(
+        parent_bundle=bundle,
+        candidate_bundle=candidate,
+        proposal=proposal,
+        store=store,
+        requalification=readmission.Requalification(config=None, run=campaign),
+        ask_person=lambda review: ("decline", "operator-1"),
+        work_dir=tmp_path / "work2",
+    )
+    assert declined.status == "declined" and declined.bundle_dir is None

From 970f7447dcbf9115427afaa59e6ced3399f6e898 Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 16:35:33 -0400
Subject: [PATCH 2/6] fix(flow): a switched-off problems_only gate asks a
 person before the save

The switch check runs at the save, after the last point a person can be
asked, so once the switch turned off every later item stopped unasked and
was entered by hand. The gate now asks at that last point while the switch
log reads off, as each_item does, when an approval store is wired.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/flow/docs/APPROVAL_MODES.md          |  4 ++
 .../flow/openadapt_flow/approval_mode/gate.py | 27 ++++++++-
 .../flow/tests/test_approval_mode_replayer.py | 56 +++++++++++++++++++
 3 files changed, 84 insertions(+), 3 deletions(-)

diff --git a/packages/flow/docs/APPROVAL_MODES.md b/packages/flow/docs/APPROVAL_MODES.md
index 79c86b118..963612848 100644
--- a/packages/flow/docs/APPROVAL_MODES.md
+++ b/packages/flow/docs/APPROVAL_MODES.md
@@ -229,6 +229,10 @@ and a missing or broken log also reads as off.
 These thresholds are the public defaults. A deployment can only make them
 stricter.
 
+While the switch reads off, each item waits for a person to approve it before
+the save, as in `each_item`, when the run has an approval store. Without one,
+the item stops before the save and nothing is saved.
+
 ## Requirements
 
 `evaluate_requirements` in `openadapt_flow/approval_mode/requirements.py`
diff --git a/packages/flow/openadapt_flow/approval_mode/gate.py b/packages/flow/openadapt_flow/approval_mode/gate.py
index df3ec9a33..a081e698e 100644
--- a/packages/flow/openadapt_flow/approval_mode/gate.py
+++ b/packages/flow/openadapt_flow/approval_mode/gate.py
@@ -569,12 +569,33 @@ def _needs_approval(self) -> bool:
         if self.mode is ApprovalMode.EACH_ITEM:
             return True
         if self.mode is ApprovalMode.PROBLEMS_ONLY:
-            return self._item_fallback is not None or (
-                self._decision is not None
-                and self._decision.kind is ItemDecisionKind.APPROVAL_REQUIRED
+            return (
+                self._item_fallback is not None
+                or (
+                    self._decision is not None
+                    and self._decision.kind is ItemDecisionKind.APPROVAL_REQUIRED
+                )
+                or self._switch_reads_off()
             )
         return False
 
+    def _switch_reads_off(self) -> bool:
+        """The local switch log already reads off (P1 would fail at the save).
+
+        A switched-off mode waits for a person to approve the item, as in
+        ``each_item``. P1 runs at the save, after the last point a person can
+        be asked, so the ask happens here; without this, every item after the
+        switch turned off stopped at the save and nobody was asked. A missing
+        or unreadable log reads as off, as P1 does.
+        """
+
+        if self.record is None or self.log is None:
+            return True
+        try:
+            return bool(self.log.state().is_off)
+        except (SwitchLogError, OSError):
+            return True
+
     def _before_step(self, ctx: StepContext) -> GateVerdict:
         if not self._began:
             raise RuntimeError("begin_item was not called for this run")
diff --git a/packages/flow/tests/test_approval_mode_replayer.py b/packages/flow/tests/test_approval_mode_replayer.py
index 9901aec68..30c3b18b1 100644
--- a/packages/flow/tests/test_approval_mode_replayer.py
+++ b/packages/flow/tests/test_approval_mode_replayer.py
@@ -431,3 +431,59 @@ def test_outcome_words_are_identical_with_and_without_the_gate(tmp_path, answer)
     without = plain.run(gate=False)
     assert with_gate.transaction_outcome == without.transaction_outcome
     assert with_gate.execution_outcome == without.execution_outcome
+
+
+def test_a_switched_off_mode_asks_a_person_when_an_approval_store_is_wired(tmp_path):
+    """Switched off, problems_only waits for a person as each_item does.
+
+    The switch check (P1) runs at the save, after the last point a person can
+    be asked. With an approval store wired, the gate must ask there instead of
+    stopping every later item unasked.
+    """
+
+    from openadapt_flow.approval_mode.item_approval import ItemApprovalStore
+    from tests.test_approval_mode_each_item import APPROVE, FormBackend, Person
+
+    person = Person([APPROVE])
+    store = ItemApprovalStore(tmp_path / "run")
+    person.store = store
+    item = Item(
+        tmp_path,
+        approval_store=store,
+        approval_wait_s=30.0,
+        approval_poll_s=1.0,
+        sleep=person.sleep,
+        monotonic=person.monotonic,
+    )
+    item.backend = FormBackend()
+    turn_off(item.log, OffReason.TURNED_OFF)
+    report = item.run()
+
+    assert len(person.seen) == 1  # one request, answered by the person
+    assert _save_delivered(item.backend)
+    assert report.transaction_outcome == "VERIFIED"
+    assert [a.decision for a in report.approval.approvals] == ["approved"]
+
+
+def test_a_switched_off_mode_without_an_answer_still_saves_nothing(tmp_path):
+    from openadapt_flow.approval_mode.item_approval import ItemApprovalStore
+    from tests.test_approval_mode_each_item import FormBackend, Person
+
+    person = Person([None])  # nobody answers
+    store = ItemApprovalStore(tmp_path / "run")
+    person.store = store
+    item = Item(
+        tmp_path,
+        approval_store=store,
+        approval_wait_s=5.0,
+        approval_poll_s=1.0,
+        sleep=person.sleep,
+        monotonic=person.monotonic,
+    )
+    item.backend = FormBackend()
+    turn_off(item.log, OffReason.TURNED_OFF)
+    report = item.run()
+
+    assert not _save_delivered(item.backend)
+    assert report.transaction_outcome == "HALTED_BEFORE_EFFECT"
+    assert report.approval.needs_person

From 7aa07f00a3dd90611d9fb1b9a6df231f40e432fa Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 19:10:54 -0400
Subject: [PATCH 3/6] docs: capabilities rows for relabel repair, the
 duplicate-list check and the switched-off gate

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

diff --git a/CAPABILITIES.md b/CAPABILITIES.md
index e541d413f..f66900f16 100644
--- a/CAPABILITIES.md
+++ b/CAPABILITIES.md
@@ -227,6 +227,9 @@ improvising or retrying when something is off.
 | Pre-approved repairs: a signed policy lists low-risk changes that apply without a person, a test-record run before each queue, and one approval for every item halted the same way | On main ([#1294](https://github.com/OpenAdaptAI/OpenAdapt/pull/1294)) | Synthetic | standard changes |
 | Repair a pinned bundle: the fix runs the pinned bundle's own qualification campaign on a test copy, keeps every write, effect, identity rule and record check, and is re-admitted with signed lineage (at most three repairs, no revoked key) that the pin verifies | On main | Synthetic | [SELF_REPAIR.md](packages/flow/docs/SELF_REPAIR.md#a-pinned-bundle-re-qualify-and-re-admit) |
 | Repair by default: once a repair agent is configured, every stop that wrote nothing goes to it without anyone starting it; a stop after a possible write goes to reconciliation; every stop is a row in a ledger with the funnel from stop to resumed item | On main ([#1360](https://github.com/OpenAdaptAI/OpenAdapt/pull/1360)) | Synthetic (MockMed, 22 drift modes, real Claude agent: 6 of 21 repairable stops resumed and verified, 0 wrong saves in 66 runs); routing over recorded OpenEMR stops; live openIMIS relabel: 3 of 3 reached repair, 0 resumed; silent-saves 1.1.0 UI-change blocks: 18 of 28 items that always stopped now correct, 0 wrong | `run` (on by default, `--no-auto-repair`), `repair:` in `deploy.yaml`, [benchmark/auto_repair](packages/flow/benchmark/auto_repair/README.md) |
+| Repair a relabeled save button: prove from the recorded identity hash that only the label changed (same control id, kind and form), re-qualify the fix on the new release, and have a person admit the certified copy afresh, since a save button's identity is part of the record check | On main | Real app (local OpenEMR, synthetic data): after the save button was relabeled, the stopped item re-ran and saved once after 8.3 operator minutes and 763 s of re-qualification; 896 s from stop to done ([#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513)) | `openadapt_flow/repair/relabel.py`, [SELF_REPAIR.md](packages/flow/docs/SELF_REPAIR.md#a-relabeled-control-proof-that-only-the-label-changed) |
+| A duplicate check that reads the app's match list (`no_duplicate_candidate`): stop only on a listed record with the item's last name or date of birth, fail closed on an unreadable or unloaded list | On main | Real app (local OpenEMR, synthetic data): false stops per 100 items 70.4 before, 16.4 after; 0 silent wrongs in 30 saves ([#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513)) | [DUPLICATE_CHECK.md](packages/flow/docs/DUPLICATE_CHECK.md) |
+| A switched-off `problems_only` gate asks a person before the save instead of stopping the item | On main | Synthetic; real app in [#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513) | [APPROVAL_MODES.md](packages/flow/docs/APPROVAL_MODES.md#when-the-mode-turns-itself-off) |
 | Fleet drift patches: a verified fix with no patient data, shared across sites under policy | Released, local reference hub | Simulator | `openadapt-flow fleet`, [FLEET_DRIFT_PATCHES.md](packages/flow/docs/FLEET_DRIFT_PATCHES.md) |
 
 ## Computer-use agents
diff --git a/llms-full.txt b/llms-full.txt
index 02ba1a00d..acb6358d3 100644
--- a/llms-full.txt
+++ b/llms-full.txt
@@ -563,6 +563,7 @@ improvising or retrying when something is off.
 | Learn one program with branches from two or more recordings, or say which recordings are missing | Released | Synthetic | `openadapt-flow induce`, `disambiguate`; [CAPABILITIES.md (flow)](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/CAPABILITIES.md) |
 | Learn a guarded branch from several halts fixed the same way | Released | Synthetic | `openadapt-flow teach-branch`, `learning/branch_teach.py` |
 | Run a program over a worklist | Released | Real app | `openadapt-flow for-each` |
+| Learn a worklist loop from recordings that search a record, open it, work it and go back to the list, even when the last row skips the return or rows differ (a field some rows lack, a write some rows get behind a screen check a person accepts). Each pass selects the row's own record by its values, verifies it, and checks that row's write. Certified and governed runs refuse it until qualification covers it | On main ([#1435](https://github.com/OpenAdaptAI/OpenAdapt/pull/1435)) | Synthetic (held-out rows run through the interpreter against an in-memory app) | `openadapt-flow induce`; `Step.target_param` |
 | Typed business decisions inside a run | Released | Synthetic | `business_decision` states, `openadapt-flow business-decisions` |
 | Watch many real runs in the background and mine candidate branches (shadow mode) | On main ([#1212](https://github.com/OpenAdaptAI/OpenAdapt/pull/1212)) | Synthetic | [SHADOW_MINING.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/capture/docs/SHADOW_MINING.md), [LEARN_FROM_MANY_RUNS.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/LEARN_FROM_MANY_RUNS.md) |
 | Fill a workflow's inputs from a referral PDF, fax image or email | Released | Synthetic | `openadapt-flow intake`, [INTAKE.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/INTAKE.md) |
@@ -623,6 +624,9 @@ improvising or retrying when something is off.
 | Pre-approved repairs: a signed policy lists low-risk changes that apply without a person, a test-record run before each queue, and one approval for every item halted the same way | On main ([#1294](https://github.com/OpenAdaptAI/OpenAdapt/pull/1294)) | Synthetic | standard changes |
 | Repair a pinned bundle: the fix runs the pinned bundle's own qualification campaign on a test copy, keeps every write, effect, identity rule and record check, and is re-admitted with signed lineage (at most three repairs, no revoked key) that the pin verifies | On main | Synthetic | [SELF_REPAIR.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/SELF_REPAIR.md#a-pinned-bundle-re-qualify-and-re-admit) |
 | Repair by default: once a repair agent is configured, every stop that wrote nothing goes to it without anyone starting it; a stop after a possible write goes to reconciliation; every stop is a row in a ledger with the funnel from stop to resumed item | On main ([#1360](https://github.com/OpenAdaptAI/OpenAdapt/pull/1360)) | Synthetic (MockMed, 22 drift modes, real Claude agent: 6 of 21 repairable stops resumed and verified, 0 wrong saves in 66 runs); routing over recorded OpenEMR stops; live openIMIS relabel: 3 of 3 reached repair, 0 resumed; silent-saves 1.1.0 UI-change blocks: 18 of 28 items that always stopped now correct, 0 wrong | `run` (on by default, `--no-auto-repair`), `repair:` in `deploy.yaml`, [benchmark/auto_repair](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/benchmark/auto_repair/README.md) |
+| Repair a relabeled save button: prove from the recorded identity hash that only the label changed (same control id, kind and form), re-qualify the fix on the new release, and have a person admit the certified copy afresh, since a save button's identity is part of the record check | On main | Real app (local OpenEMR, synthetic data): after the save button was relabeled, the stopped item re-ran and saved once after 8.3 operator minutes and 763 s of re-qualification; 896 s from stop to done ([#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513)) | `openadapt_flow/repair/relabel.py`, [SELF_REPAIR.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/SELF_REPAIR.md#a-relabeled-control-proof-that-only-the-label-changed) |
+| A duplicate check that reads the app's match list (`no_duplicate_candidate`): stop only on a listed record with the item's last name or date of birth, fail closed on an unreadable or unloaded list | On main | Real app (local OpenEMR, synthetic data): false stops per 100 items 70.4 before, 16.4 after; 0 silent wrongs in 30 saves ([#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513)) | [DUPLICATE_CHECK.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/DUPLICATE_CHECK.md) |
+| A switched-off `problems_only` gate asks a person before the save instead of stopping the item | On main | Synthetic; real app in [#1513](https://github.com/OpenAdaptAI/OpenAdapt/pull/1513) | [APPROVAL_MODES.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/APPROVAL_MODES.md#when-the-mode-turns-itself-off) |
 | Fleet drift patches: a verified fix with no patient data, shared across sites under policy | Released, local reference hub | Simulator | `openadapt-flow fleet`, [FLEET_DRIFT_PATCHES.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/FLEET_DRIFT_PATCHES.md) |
 
 ## Computer-use agents

From 45bdae38bbe7d41cef6f0f786ee9dd41fc898386 Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 19:23:46 -0400
Subject: [PATCH 4/6] fix(flow): refuse a typed date parameter that is not a
 canonical ISO date before any step

A referral that wrote the date of birth as MM/DD/YYYY was typed into the form
and saved as the app's guess. The record check caught it afterwards, but a
wrong chart landed and the problems_only switch turned off for the rest of
the run. A run now refuses before any step when a ParamKind.DATE value is not
a canonical ISO date, as hosted dispatch already did.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/flow/docs/PARAMETERS.md              |  6 +++
 .../flow/openadapt_flow/runtime/replayer.py   | 46 +++++++++++++++++++
 packages/flow/tests/test_program_ir_phase1.py | 41 +++++++++++++++++
 3 files changed, 93 insertions(+)

diff --git a/packages/flow/docs/PARAMETERS.md b/packages/flow/docs/PARAMETERS.md
index 1ae1ef02b..703cb0350 100644
--- a/packages/flow/docs/PARAMETERS.md
+++ b/packages/flow/docs/PARAMETERS.md
@@ -28,6 +28,12 @@ one config.
 To fill parameters from a referral PDF, fax image, or email, see
 [`INTAKE.md`](INTAKE.md).
 
+A parameter declared as a `date` (`ParamSpec.type`) must be a canonical ISO
+date (`1990-03-14`) before a run starts. A date written another way, such as
+`03/14/1990`, refuses the run before any step, so it is never typed into a
+form and saved as whatever the app makes of it. Hosted dispatch applies the
+same check to every typed parameter.
+
 ## Choose an execution profile
 
 Select `--profile regulated` for encrypted, fail-closed production execution,
diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py
index 7bcb44129..c674ad1df 100644
--- a/packages/flow/openadapt_flow/runtime/replayer.py
+++ b/packages/flow/openadapt_flow/runtime/replayer.py
@@ -113,6 +113,7 @@
     LoopIterationBinding,
     LoopSpec,
     ObservedSurfaceEnvironment,
+    ParamKind,
     PixelIdentityEvidence,
     Point,
     Postcondition,
@@ -2325,6 +2326,28 @@ def acknowledge_delivery(self_nonlocal) -> None:
             report.total_ms = (time.monotonic() - t_run) * 1000.0
             return self._finalize_report(report, workflow, run_dir)
 
+        # A typed date parameter must be a canonical ISO date before any step
+        # runs: a date written another way (03/14/1990) would otherwise be
+        # typed into the form and saved as whatever the app makes of it.
+        malformed = self._malformed_date_params(workflow, params)
+        if malformed:
+            report.results.append(
+                StepResult(
+                    step_id="<params>",
+                    intent="validate typed workflow parameters",
+                    ok=False,
+                    failure_category="governed_refusal",
+                    error=(
+                        "Typed workflow parameter(s) are not valid: "
+                        + "; ".join(malformed)
+                        + " — refusing to run; nothing was typed; run aborted"
+                    ),
+                )
+            )
+            report.success = False
+            report.total_ms = (time.monotonic() - t_run) * 1000.0
+            return self._finalize_report(report, workflow, run_dir)
+
         if self._approval_refusal is not None:
             # Approval modes take values only from the caller: an input the
             # run would fill from the recording is refused before any step.
@@ -17668,6 +17691,29 @@ def _local_edit_held(self, pc: Any, start_state: dict[str, Any]) -> bool:
                 return False
         return True
 
+    @staticmethod
+    def _malformed_date_params(workflow: Workflow, params: dict[str, Any]) -> list[str]:
+        """Why each supplied ``date`` parameter is not a canonical ISO date.
+
+        The same check hosted dispatch applies (``runner.inputs``). An empty
+        optional value is left to the required-parameter check.
+        """
+
+        from openadapt_flow.runner.inputs import AdmittedInputError, _validate_value
+
+        problems: list[str] = []
+        for name, spec in sorted(workflow.param_specs.items()):
+            if spec.type is not ParamKind.DATE or name not in params:
+                continue
+            value = params[name]
+            if value == "" and not spec.required:
+                continue
+            try:
+                _validate_value(spec, value)
+            except AdmittedInputError as exc:
+                problems.append(str(exc))
+        return problems
+
     def _no_duplicate_candidate(self, pc: Any, frame_png: bytes) -> bool:
         """The app's possible-duplicate list has loaded and names no candidate.
 
diff --git a/packages/flow/tests/test_program_ir_phase1.py b/packages/flow/tests/test_program_ir_phase1.py
index 7f19ac290..9e215044e 100644
--- a/packages/flow/tests/test_program_ir_phase1.py
+++ b/packages/flow/tests/test_program_ir_phase1.py
@@ -182,6 +182,47 @@ def test_missing_required_param_fails_fast_naming_it(bundle, run_dir):
     assert backend.actions == []  # nothing ran
 
 
+def _date_param_workflow() -> Workflow:
+    return Workflow(
+        name="wf",
+        param_specs={
+            "DOB": ParamSpec(name="DOB", type=ParamKind.DATE, example="1985-02-03")
+        },
+        steps=[Step(id="t1", intent="type DOB", action=ActionKind.TYPE, param="DOB")],
+    )
+
+
+@pytest.mark.parametrize(
+    "value", ["04/20/1944", "1944-4-20", "1944-02-30", "", 19440420]
+)
+def test_a_malformed_date_param_halts_before_anything_is_typed(bundle, run_dir, value):
+    """A typed date must be a canonical ISO date before any step runs, so a
+    date written another way is never typed into the form and saved as
+    whatever the app makes of it."""
+    backend = FakeBackend()
+    report = Replayer(backend, vision=FakeVision()).run(
+        _date_param_workflow(),
+        params={"DOB": value},
+        bundle_dir=bundle,
+        run_dir=run_dir,
+    )
+    assert report.success is False
+    assert [r.step_id for r in report.results] == ["<params>"]
+    assert backend.actions == []  # nothing was typed
+
+
+def test_an_iso_date_param_runs(bundle, run_dir):
+    backend = FakeBackend()
+    report = Replayer(backend, vision=FakeVision(), poll_interval_s=0.01).run(
+        _date_param_workflow(),
+        params={"DOB": "1944-04-20"},
+        bundle_dir=bundle,
+        run_dir=run_dir,
+    )
+    assert report.success is True
+    assert backend.actions == [("type", "1944-04-20")]
+
+
 @pytest.mark.parametrize(
     ("kind", "value"),
     [(ParamKind.BOOLEAN, False), (ParamKind.NUMBER, 0)],

From b944909f30e7a2bb3b53a7bb03939cc3ffdde7f5 Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 19:28:40 -0400
Subject: [PATCH 5/6] feat(flow): intake converts a date to ISO when its order
 can be told, and records the conversion

A date of birth arrives however the sender writes dates. Intake now converts
it to ISO, with the original text and the conversion kept in the provenance
file, when the order can be told: ISO already, a spelled-out month,
YYYY/MM/DD, a numeric date with a part over 12 or equal parts, or a
numeric date whose order is declared (--date-format DOB=MM/DD/YYYY).
A numeric date that reads either way with no declared format (03/04/1970)
and a date that doesn't exist now go to a person; intake used to read every
numeric date as MM/DD/YYYY. The run's ISO check stays as the last guard.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/flow/docs/INTAKE.md                  |  13 +-
 packages/flow/openadapt_flow/intake/cli.py    |  31 +++-
 .../flow/openadapt_flow/intake/extract.py     | 147 +++++++++++++++---
 packages/flow/openadapt_flow/provenance.py    |  17 +-
 packages/flow/tests/test_intake_dates.py      | 103 ++++++++++++
 packages/flow/tests/test_provenance.py        |   8 +-
 6 files changed, 292 insertions(+), 27 deletions(-)
 create mode 100644 packages/flow/tests/test_intake_dates.py

diff --git a/packages/flow/docs/INTAKE.md b/packages/flow/docs/INTAKE.md
index 9230d3304..9d21cd775 100644
--- a/packages/flow/docs/INTAKE.md
+++ b/packages/flow/docs/INTAKE.md
@@ -20,7 +20,18 @@ openadapt-flow run ./new-referral --params-file params.json
    `Insurance No:` and `Insurance Number:`. Add more labels with
    `--alias insurance_no="Member ID"`.
 3. Validates each value against the parameter's type (`date`, `number`,
-   `boolean`, `enum`, `string`) and normalizes it (dates become `YYYY-MM-DD`).
+   `boolean`, `enum`, `string`) and normalizes it. Dates become `YYYY-MM-DD`
+   only when their order can be told:
+   - an ISO date, a spelled-out month (`25 Dec 1970`) or `YYYY/MM/DD`;
+   - a numeric date with a part over 12 (`12/25/1970`), or equal parts;
+   - a numeric date whose order you declare with
+     `--date-format DOB=MM/DD/YYYY` (or `DD/MM/YYYY`).
+
+   A numeric date that could be read either way (`03/04/1970`) with no
+   declared format, or a date that doesn't exist (`1970-02-30`), needs a
+   person. A converted value records its original text and the conversion in
+   the provenance file (`raw_value`, `conversion`). The run checks again that
+   every `date` parameter is an ISO date before its first step.
 4. Holds back any field that is missing, invalid, ambiguous (two different
    values), or below the confidence threshold (default 0.9). On a terminal it
    asks you to confirm or correct each one. Otherwise it prints the pending
diff --git a/packages/flow/openadapt_flow/intake/cli.py b/packages/flow/openadapt_flow/intake/cli.py
index a9a3dd3da..7c8ac878a 100644
--- a/packages/flow/openadapt_flow/intake/cli.py
+++ b/packages/flow/openadapt_flow/intake/cli.py
@@ -40,6 +40,20 @@ def _parse_aliases(pairs: list[str]) -> dict[str, list[str]]:
     return out
 
 
+def _parse_date_formats(pairs: list[str]) -> dict[str, str]:
+    from openadapt_flow.intake.extract import DATE_ORDERS
+
+    out: dict[str, str] = {}
+    for pair in pairs:
+        name, sep, fmt = pair.partition("=")
+        if not sep or fmt.upper() not in DATE_ORDERS:
+            raise SystemExit(
+                f"--date-format expects NAME=MM/DD/YYYY or NAME=DD/MM/YYYY, got {pair!r}"
+            )
+        out[name.strip()] = fmt.upper()
+    return out
+
+
 def _prompt(result: IntakeResult) -> None:
     for name in result.needs_confirmation():
         f = result.fields[name]
@@ -65,7 +79,11 @@ def cmd_intake(args: argparse.Namespace) -> int:
         result = apply_agent_response(doc, specs, response, threshold=args.threshold)
     else:
         result = extract_fields(
-            doc, specs, aliases=_parse_aliases(args.alias), threshold=args.threshold
+            doc,
+            specs,
+            aliases=_parse_aliases(args.alias),
+            threshold=args.threshold,
+            date_formats=_parse_date_formats(args.date_format),
         )
     for name in args.accept:
         result.confirm(name)
@@ -123,6 +141,17 @@ def add_parser(sub: argparse._SubParsersAction) -> None:
         default=0.9,
         help="Fields below this confidence need confirmation",
     )
+    p.add_argument(
+        "--date-format",
+        action="append",
+        default=[],
+        metavar="NAME=MM/DD/YYYY",
+        help=(
+            "How the source writes a date parameter's numeric dates "
+            "(MM/DD/YYYY or DD/MM/YYYY). Without it, a date such as 03/04/1970 "
+            "that could be read either way needs a person (repeatable)"
+        ),
+    )
     p.add_argument("--no-input", action="store_true", help="Never prompt")
     p.add_argument(
         "--agent-request",
diff --git a/packages/flow/openadapt_flow/intake/extract.py b/packages/flow/openadapt_flow/intake/extract.py
index 7831b6c6b..ffbdf25f7 100644
--- a/packages/flow/openadapt_flow/intake/extract.py
+++ b/packages/flow/openadapt_flow/intake/extract.py
@@ -13,15 +13,17 @@
 #: Fields whose confidence is below this need a person to confirm them.
 DEFAULT_CONFIRM_THRESHOLD = 0.9
 
-_DATE_FORMATS = (
-    "%Y-%m-%d",
-    "%m/%d/%Y",
-    "%m-%d-%Y",
+#: Dates whose month is spelled out: never ambiguous.
+_NAMED_MONTH_FORMATS = (
     "%d %b %Y",
+    "%d %B %Y",
     "%b %d, %Y",
     "%B %d, %Y",
-    "%Y/%m/%d",
+    "%b %d %Y",
+    "%B %d %Y",
 )
+#: A source's declared order for numeric dates such as 03/04/1970.
+DATE_ORDERS = ("MM/DD/YYYY", "DD/MM/YYYY")
 _TRUE = {"yes", "y", "true", "1", "x", "checked"}
 _FALSE = {"no", "n", "false", "0", "unchecked"}
 _SEP = r"\s*(?:[:#=]|-\s)\s*"
@@ -40,6 +42,10 @@ class ExtractedField:
     label: Optional[str] = None
     error: Optional[str] = None
     confirmed: bool = False
+    #: The text as the document wrote it, when it differs from ``value``.
+    raw: Optional[str] = None
+    #: How ``raw`` became ``value`` (for example a date's conversion to ISO).
+    conversion: Optional[str] = None
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -50,6 +56,8 @@ def to_dict(self) -> dict[str, Any]:
             "label": self.label,
             "error": self.error,
             "confirmed": self.confirmed,
+            "raw": self.raw,
+            "conversion": self.conversion,
         }
 
 
@@ -62,6 +70,8 @@ class IntakeResult:
     threshold: float = DEFAULT_CONFIRM_THRESHOLD
     mode: str = "local"
     notes: list[str] = field(default_factory=list)
+    #: Declared order of numeric dates, per parameter (``DATE_ORDERS``).
+    date_formats: dict[str, str] = field(default_factory=dict)
 
     def needs_confirmation(self) -> list[str]:
         """Names of fields a person must confirm or correct before running."""
@@ -86,10 +96,14 @@ def confirm(self, name: str, value: Optional[str] = None) -> ExtractedField:
         candidate = f.value if value is None else value
         if candidate is None:
             raise ValueError(f"{name}: no value to confirm")
-        normalized, error = validate_value(self.specs[name], candidate)
+        normalized, error, conversion = convert_value(
+            self.specs[name], candidate, date_format=self.date_formats.get(name)
+        )
         if error:
             raise ValueError(f"{name}: {error}")
         f.value, f.error, f.confirmed, f.confidence = normalized, None, True, 1.0
+        f.raw = candidate.strip() if conversion else None
+        f.conversion = conversion
         return f
 
     def to_params(self) -> dict[str, str]:
@@ -109,20 +123,95 @@ def to_dict(self) -> dict[str, Any]:
         }
 
 
-def validate_value(spec: ParamSpec, raw: str) -> tuple[Optional[str], Optional[str]]:
+def normalize_date(
+    raw: str, date_format: Optional[str] = None
+) -> tuple[Optional[str], Optional[str], Optional[str]]:
+    """A date as a canonical ISO date: ``(iso, error, conversion)``.
+
+    - An ISO date is checked and kept (no conversion).
+    - A date with the month spelled out converts.
+    - A numeric date (``03/04/1970``, ``03-04-1970``) converts when its order
+      is declared (``date_format``, one of ``DATE_ORDERS``), or can be told
+      from the date itself: one part over 12 is the day, and equal parts read
+      the same either way. Otherwise it is ambiguous and refused.
+    - ``YYYY/MM/DD`` converts.
+    - Anything else, or a date that doesn't exist (1970-02-30), is refused.
+    """
+
+    value = raw.strip()
+    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
+        try:
+            return date.fromisoformat(value).isoformat(), None, None
+        except ValueError:
+            return None, f"not a valid date: {value!r}", None
+    numeric = re.fullmatch(r"(\d{1,2})([/-])(\d{1,2})\2(\d{4})", value)
+    if numeric:
+        a, b, year = int(numeric.group(1)), int(numeric.group(3)), int(numeric.group(4))
+        if date_format is not None:
+            if date_format not in DATE_ORDERS:
+                return None, f"unknown declared date format {date_format!r}", None
+            month_first, why = date_format == "MM/DD/YYYY", "declared"
+        elif a > 12 >= b:
+            month_first, why = False, f"{a} can only be the day"
+        elif b > 12 >= a:
+            month_first, why = True, f"{b} can only be the day"
+        elif a == b:
+            month_first, why = True, "day and month are equal"
+        else:
+            return (
+                None,
+                f"ambiguous date {value!r}: day and month can't be told apart; "
+                "declare the source's format (MM/DD/YYYY or DD/MM/YYYY)",
+                None,
+            )
+        month, day = (a, b) if month_first else (b, a)
+        try:
+            parsed = date(year, month, day)
+        except ValueError:
+            return None, f"not a valid date: {value!r}", None
+        order = "MM/DD/YYYY" if month_first else "DD/MM/YYYY"
+        return parsed.isoformat(), None, f"{order} to ISO ({why})"
+    ymd = re.fullmatch(r"(\d{4})/(\d{1,2})/(\d{1,2})", value)
+    if ymd:
+        try:
+            parsed = date(int(ymd.group(1)), int(ymd.group(2)), int(ymd.group(3)))
+        except ValueError:
+            return None, f"not a valid date: {value!r}", None
+        return parsed.isoformat(), None, "YYYY/MM/DD to ISO"
+    for fmt in _NAMED_MONTH_FORMATS:
+        try:
+            named: date = datetime.strptime(value, fmt).date()
+        except ValueError:
+            continue
+        return named.isoformat(), None, "month name to ISO"
+    return None, f"not a recognized date: {value!r}", None
+
+
+def convert_value(
+    spec: ParamSpec, raw: str, *, date_format: Optional[str] = None
+) -> tuple[Optional[str], Optional[str], Optional[str]]:
+    """Normalize ``raw`` for ``spec``: ``(value, error, conversion)``."""
+
+    if spec.type == ParamKind.DATE and raw.strip():
+        iso, error, conversion = normalize_date(raw, date_format)
+        if error:
+            return raw.strip(), error, None
+        return iso, None, conversion
+    value, error = validate_value(spec, raw)
+    return value, error, None
+
+
+def validate_value(
+    spec: ParamSpec, raw: str, *, date_format: Optional[str] = None
+) -> tuple[Optional[str], Optional[str]]:
     """Normalize ``raw`` for ``spec``. Returns ``(value, error)``."""
     value = raw.strip()
     if not value:
         return None, "empty value"
     kind = spec.type
     if kind == ParamKind.DATE:
-        for fmt in _DATE_FORMATS:
-            try:
-                parsed: date = datetime.strptime(value, fmt).date()
-            except ValueError:
-                continue
-            return parsed.isoformat(), None
-        return value, f"not a recognized date: {value!r}"
+        iso, error, _conversion = normalize_date(value, date_format)
+        return (value, error) if error else (iso, None)
     if kind == ParamKind.NUMBER:
         cleaned = value.replace(",", "")
         try:
@@ -184,9 +273,15 @@ def extract_fields(
     *,
     aliases: Optional[Mapping[str, list[str]]] = None,
     threshold: float = DEFAULT_CONFIRM_THRESHOLD,
+    date_formats: Optional[Mapping[str, str]] = None,
 ) -> IntakeResult:
-    """Local, deterministic extraction by label matching. No model call."""
+    """Local, deterministic extraction by label matching. No model call.
+
+    ``date_formats`` declares the order of numeric dates per parameter
+    (``DATE_ORDERS``); without it an ambiguous date needs a person.
+    """
     aliases = aliases or {}
+    formats = dict(date_formats or {})
     fields: dict[str, ExtractedField] = {}
     for name, spec in specs.items():
         hits: list[tuple[float, int, str, str]] = []
@@ -203,15 +298,25 @@ def extract_fields(
         best = max(h[0] for h in hits)
         top = [h for h in hits if h[0] == best]
         conf, line_no, label, raw = top[0]
-        value, error = validate_value(spec, raw)
-        distinct = {validate_value(spec, h[3])[0] for h in top}
+        fmt = formats.get(name)
+        value, error, conversion = convert_value(spec, raw, date_format=fmt)
+        distinct = {convert_value(spec, h[3], date_format=fmt)[0] for h in top}
         if len(distinct) > 1:
             conf = min(conf, 0.5)
             error = error or f"{len(distinct)} different values found"
         if error:
             conf = 0.0
-        fields[name] = ExtractedField(name, value, conf, line_no + 1, label, error)
-    return IntakeResult(fields, dict(specs), threshold, "local")
+        fields[name] = ExtractedField(
+            name,
+            value,
+            conf,
+            line_no + 1,
+            label,
+            error,
+            raw=raw.strip() if conversion else None,
+            conversion=conversion,
+        )
+    return IntakeResult(fields, dict(specs), threshold, "local", date_formats=formats)
 
 
 def build_agent_request(
@@ -294,7 +399,7 @@ def apply_agent_response(
         # default threshold unless the label matched exactly on that line.
         if not label or _match_label(text, label) is None:
             conf = min(conf, 0.7)
-        value, error = validate_value(spec, raw)
+        value, error, conversion = convert_value(spec, raw)
         fields[name] = ExtractedField(
             name,
             value,
@@ -302,5 +407,7 @@ def apply_agent_response(
             line,
             label or None,
             error,
+            raw=raw.strip() if conversion else None,
+            conversion=conversion,
         )
     return IntakeResult(fields, dict(specs), threshold, "agent")
diff --git a/packages/flow/openadapt_flow/provenance.py b/packages/flow/openadapt_flow/provenance.py
index 8fcf5907e..8af4abbbd 100644
--- a/packages/flow/openadapt_flow/provenance.py
+++ b/packages/flow/openadapt_flow/provenance.py
@@ -99,6 +99,12 @@ class ParamProvenance(BaseModel):
     extractor: str = Field(min_length=1)
     #: ``text`` | ``code`` | ``date`` | ``currency``. Unset: inferred.
     value_kind: Optional[Literal["text", "code", "date", "currency"]] = None
+    #: The text as the source wrote it, when intake converted it (``value`` is
+    #: the converted form, for example a date as ISO).
+    raw_value: Optional[str] = None
+    #: How ``raw_value`` became ``value``, for example
+    #: ``MM/DD/YYYY to ISO (25 can only be the day)``.
+    conversion: Optional[str] = None
 
 
 class ProvenanceFile(BaseModel):
@@ -160,6 +166,8 @@ def from_intake(
                 label=fld.label,
             ),
             extractor="person" if fld.confirmed and fld.confidence >= 1.0 else name_of,
+            raw_value=getattr(fld, "raw", None),
+            conversion=getattr(fld, "conversion", None),
         )
     return ProvenanceFile(params=params)
 
@@ -189,6 +197,8 @@ class LabelExtractor:
     specs: Mapping[str, Any]
     aliases: Mapping[str, list[str]] = field(default_factory=dict)
     name: str = "intake:local"
+    #: Declared order of numeric dates, per parameter (as in intake).
+    date_formats: Mapping[str, str] = field(default_factory=dict)
 
     def read(self, document: Any, name: str, ref: SourceRef) -> Optional[str]:
         from openadapt_flow.intake.extract import extract_fields
@@ -199,7 +209,12 @@ def read(self, document: Any, name: str, ref: SourceRef) -> Optional[str]:
         aliases = dict(self.aliases)
         if ref.label:
             aliases[name] = [*aliases.get(name, []), ref.label]
-        found = extract_fields(document, {name: spec}, aliases=aliases).fields[name]
+        found = extract_fields(
+            document,
+            {name: spec},
+            aliases=aliases,
+            date_formats={k: v for k, v in self.date_formats.items() if k == name},
+        ).fields[name]
         if found.error or found.value is None:
             return None
         return str(found.value)
diff --git a/packages/flow/tests/test_intake_dates.py b/packages/flow/tests/test_intake_dates.py
new file mode 100644
index 000000000..77356eb6c
--- /dev/null
+++ b/packages/flow/tests/test_intake_dates.py
@@ -0,0 +1,103 @@
+"""Dates are converted to ISO at intake, never guessed.
+
+A referral writes a date of birth however its sender writes dates. Intake
+converts it to the canonical ISO date the program types, and records the
+conversion as provenance, when the order can be told: the source's declared
+format, a part over 12, equal parts, a spelled-out month, or ISO already. A
+numeric date that could be read either way, with no declared format, and a
+date that doesn't exist both go to a person. The run's own ISO check stays as
+the last guard (tests/test_program_ir_phase1.py).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from openadapt_flow.intake.document import IntakeDocument
+from openadapt_flow.intake.extract import extract_fields, normalize_date
+from openadapt_flow.ir import ParamKind, ParamSpec
+from openadapt_flow.provenance import from_intake, sha256_bytes
+
+SPECS = {"DOB": ParamSpec(name="DOB", type=ParamKind.DATE)}
+
+
+def _doc(dob: str) -> IntakeDocument:
+    return IntakeDocument("referral.txt", "text", ["Last name: Okafor", f"DOB: {dob}"])
+
+
+@pytest.mark.parametrize(
+    "raw, iso",
+    [
+        ("12/25/1970", "1970-12-25"),  # 25 can only be the day
+        ("25/12/1970", "1970-12-25"),
+        ("07/07/1970", "1970-07-07"),  # reads the same either way
+        ("12-25-1970", "1970-12-25"),
+        ("1970-12-25", "1970-12-25"),
+        ("25 Dec 1970", "1970-12-25"),
+        ("December 25, 1970", "1970-12-25"),
+        ("1970/12/25", "1970-12-25"),
+    ],
+)
+def test_a_date_whose_order_can_be_told_converts(raw: str, iso: str) -> None:
+    value, error, _conversion = normalize_date(raw)
+    assert (value, error) == (iso, None)
+
+
+def test_an_ambiguous_date_halts_without_a_declared_format() -> None:
+    value, error, conversion = normalize_date("03/04/1970")
+    assert value is None and conversion is None
+    assert error is not None and "ambiguous" in error
+
+
+@pytest.mark.parametrize(
+    "declared, iso", [("MM/DD/YYYY", "1970-03-04"), ("DD/MM/YYYY", "1970-04-03")]
+)
+def test_an_ambiguous_date_converts_with_a_declared_format(declared, iso) -> None:
+    value, error, conversion = normalize_date("03/04/1970", declared)
+    assert (value, error) == (iso, None)
+    assert conversion == f"{declared} to ISO (declared)"
+
+
+@pytest.mark.parametrize(
+    "raw", ["1970-02-30", "02/30/1970", "13/13/1970", "1970-2-3", "next Tuesday"]
+)
+def test_an_invalid_date_halts(raw: str) -> None:
+    value, error, _ = normalize_date(raw)
+    assert value is None and error
+
+
+def test_a_declared_format_still_refuses_a_date_that_does_not_exist() -> None:
+    assert normalize_date("02/30/1970", "MM/DD/YYYY")[0] is None
+
+
+def test_intake_records_the_conversion_as_provenance() -> None:
+    doc = _doc("12/25/1970")
+    result = extract_fields(doc, SPECS)
+    field = result.fields["DOB"]
+    assert field.value == "1970-12-25" and field.error is None
+    assert field.raw == "12/25/1970"
+    assert field.conversion == "MM/DD/YYYY to ISO (25 can only be the day)"
+    prov = from_intake(result, sha256_bytes(doc.text.encode("utf-8")))
+    entry = prov.params["DOB"]
+    assert (entry.value, entry.raw_value) == ("1970-12-25", "12/25/1970")
+    assert entry.conversion == field.conversion
+    assert result.to_params() == {"DOB": "1970-12-25"}
+
+
+def test_intake_sends_an_ambiguous_date_to_a_person() -> None:
+    result = extract_fields(_doc("03/04/1970"), SPECS)
+    assert result.needs_confirmation() == ["DOB"]
+    assert "ambiguous" in (result.fields["DOB"].error or "")
+    # With the source's format declared, it converts and needs nobody.
+    declared = extract_fields(
+        _doc("03/04/1970"), SPECS, date_formats={"DOB": "DD/MM/YYYY"}
+    )
+    assert declared.needs_confirmation() == []
+    assert declared.to_params() == {"DOB": "1970-04-03"}
+
+
+def test_an_iso_date_records_no_conversion() -> None:
+    field = extract_fields(_doc("1970-12-25"), SPECS).fields["DOB"]
+    assert (
+        field.value == "1970-12-25" and field.raw is None and field.conversion is None
+    )
diff --git a/packages/flow/tests/test_provenance.py b/packages/flow/tests/test_provenance.py
index f2711d5b6..d8f856ec5 100644
--- a/packages/flow/tests/test_provenance.py
+++ b/packages/flow/tests/test_provenance.py
@@ -28,7 +28,7 @@
     [
         "Referral to: Example Family Clinic",
         "Last name: Okafor",
-        "Date of birth: 03/07/1991",
+        "Date of birth: 03/27/1991",
         "Postal code: 02139",
         "Home phone: 555-0142",
     ],
@@ -64,7 +64,7 @@ def _prov(**values: str) -> ProvenanceFile:
 
 GOOD = {
     "lname": "Okafor",
-    "DOB": "1991-03-07",
+    "DOB": "1991-03-27",
     "postal_code": "02139",
     "phone_home": "555-0142",
 }
@@ -75,7 +75,7 @@ def test_intake_keeps_the_line_it_read() -> None:
     prov = from_intake(result, DOC)
     assert prov.params["postal_code"].source.line == 4
     assert prov.params["postal_code"].extractor == "intake:local"
-    assert prov.params["DOB"].value == "1991-03-07"
+    assert prov.params["DOB"].value == "1991-03-27"
 
 
 def test_two_methods_that_agree_let_the_save_go() -> None:
@@ -138,7 +138,7 @@ def test_round_trip_holds_for_a_faithful_record() -> None:
     record = {
         "id": 7,
         "lname": "Okafor",
-        "DOB": "1991-03-07",
+        "DOB": "1991-03-27",
         "postal_code": "02139",
         "phone_home": "555-0142",
     }

From 1485eb851e356c52bf4c6a9af45141d273cec2d5 Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sat, 24 Oct 2026 19:32:37 -0400
Subject: [PATCH 6/6] fix(flow): the duplicate check compares dates as dates,
 in any format the row shows

A chart that shares only the date of birth (a married-name change) must
stop the item. The rows showed the date in the app's own format, so a word
match against the run's ISO date missed 'Apr 20, 1944' or '20.04.1944'. Each
date in a row is now read the way intake reads dates and compared as a
date; a row date that reads either way counts when either reading matches.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/flow/docs/DUPLICATE_CHECK.md         | 18 +++---
 .../flow/openadapt_flow/runtime/replayer.py   | 63 ++++++++++++++++++-
 .../flow/tests/test_no_duplicate_candidate.py | 38 +++++++++++
 3 files changed, 109 insertions(+), 10 deletions(-)

diff --git a/packages/flow/docs/DUPLICATE_CHECK.md b/packages/flow/docs/DUPLICATE_CHECK.md
index 61e151467..02fffc8ae 100644
--- a/packages/flow/docs/DUPLICATE_CHECK.md
+++ b/packages/flow/docs/DUPLICATE_CHECK.md
@@ -28,9 +28,11 @@ opens the search:
 It passes only when the list has loaded and names no real candidate:
 
 - A listed row is a **candidate** when it contains every word of this run's
-  value of any `candidate_params` parameter (case-insensitive). With the
-  example above, a chart with the same last name or the same date of birth
-  stops the item. A chart that shares only a first name doesn't.
+  value of any `candidate_params` parameter (case-insensitive), or, for a
+  date, shows the same date in any format. With the example above, a chart
+  with the same last name or the same date of birth stops the item, even
+  under another last name (a married-name change). A chart that shares only
+  a first name doesn't.
 - An **empty list** passes only when its "no matches" marker (`text`) is
   shown, so a list that hasn't loaded yet never passes.
 - It **fails closed**: an unreadable list (frame missing or ambiguous, or a
@@ -54,8 +56,8 @@ on the OpenEMR referral program, and the soak admits that version afresh
 
 - The list must render its rows together, as OpenEMR's server-rendered list
   does. A list that streams rows in one by one could pass on a partial list.
-- A value is recognized only in the form the run typed it. If the app shows a
-  value in another form (a date as `09/09/1990` for `1990-09-09`), the check
-  misses that candidate and lets the item continue. Name only parameters the
-  list shows exactly as typed, and check this on the real list before
-  admitting the program. OpenEMR's list shows the date of birth as typed.
+- A date (a value that is an ISO date) is compared as a date. Every date the
+  row shows is read the way intake reads dates (`Apr 20, 1944`, `20.04.1944`,
+  `04/20/1944`, ISO), and a row date that reads either way counts when either
+  reading matches. Other values are matched word for word, so name only
+  parameters the list shows as they were typed.
diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py
index c674ad1df..a841389f8 100644
--- a/packages/flow/openadapt_flow/runtime/replayer.py
+++ b/packages/flow/openadapt_flow/runtime/replayer.py
@@ -17718,7 +17718,11 @@ def _no_duplicate_candidate(self, pc: Any, frame_png: bytes) -> bool:
         """The app's possible-duplicate list has loaded and names no candidate.
 
         A row is a candidate when its words include every word of this run's
-        value of any ``pc.candidate_params`` parameter (case-insensitive).
+        value of any ``pc.candidate_params`` parameter (case-insensitive). A
+        value that is an ISO date is also compared as a date: every date the
+        row shows, in any format intake reads (``Apr 20, 1944``,
+        ``20.04.1944``, ``04/20/1944``), is normalized and compared, and a
+        row date that reads either way counts when either reading matches.
         Fails closed: an unreadable list, a missing or empty parameter, or an
         empty list without its "no matches" marker all fail the check.
         """
@@ -17742,11 +17746,15 @@ def words(value: str) -> list[str]:
             return re.findall(r"[0-9a-z]+", value.casefold())
 
         wanted = []
+        wanted_dates: set[str] = set()
         for name in names:
-            value = words(runtime_param_text(params[name])) if name in params else []
+            text = runtime_param_text(params[name]) if name in params else ""
+            value = words(text)
             if not value:
                 return False
             wanted.append(value)
+            if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text.strip()):
+                wanted_dates.add(text.strip())
         if not rows:
             # Nothing listed: only the list's own "no matches" marker shows
             # the search finished.
@@ -17757,6 +17765,8 @@ def words(value: str) -> list[str]:
             row_words = set(words(row))
             if any(all(w in row_words for w in value) for value in wanted):
                 return False
+            if wanted_dates and wanted_dates & _row_dates(row):
+                return False
         return True
 
     def _field_value_shown(self, pc: Any) -> bool:
@@ -18397,3 +18407,52 @@ def _save_step_png(run_dir: Path, step_id: str, suffix: str, png: bytes) -> str:
         rel = f"steps/{step_id}_{suffix}.png"
         _sealed.write_bytes(Path(run_dir) / rel, _scrub_png(png))
         return rel
+
+
+#: Date-looking text in a listed row: ISO or Y/M/D, numeric with / . or -,
+#: and a spelled-out month either side of the day.
+_ROW_DATE_PATTERNS = (
+    r"\b\d{4}[-/]\d{1,2}[-/]\d{1,2}\b",
+    r"\b\d{1,2}[./-]\d{1,2}[./-]\d{4}\b",
+    r"\b\d{1,2}\s+[A-Za-z]{3,9}\.?,?\s+\d{4}\b",
+    r"\b[A-Za-z]{3,9}\.?\s+\d{1,2},?\s+\d{4}\b",
+)
+
+
+def _row_dates(row: str) -> set[str]:
+    """Every ISO date a listed row could be showing.
+
+    Each date-looking substring is read the way intake reads a date
+    (``intake.extract.normalize_date``). One that reads either way (03/04/1970)
+    contributes both readings; one that isn't a date contributes nothing.
+    """
+
+    from datetime import date as _date
+
+    from openadapt_flow.intake.extract import normalize_date
+
+    found: set[str] = set()
+    for pattern in _ROW_DATE_PATTERNS:
+        for match in re.finditer(pattern, row):
+            text = match.group(0)
+            ymd = re.fullmatch(r"(\d{4})[-/](\d{1,2})[-/](\d{1,2})", text)
+            if ymd:
+                try:
+                    found.add(
+                        _date(
+                            int(ymd.group(1)), int(ymd.group(2)), int(ymd.group(3))
+                        ).isoformat()
+                    )
+                except ValueError:
+                    pass
+                continue
+            text = re.sub(r"(\d)\.(?=\d)", r"\1/", text).replace(".", "")
+            iso, error, _ = normalize_date(text)
+            if iso is not None:
+                found.add(iso)
+            elif error and "ambiguous" in error:
+                for order in ("MM/DD/YYYY", "DD/MM/YYYY"):
+                    reading = normalize_date(text, order)[0]
+                    if reading is not None:
+                        found.add(reading)
+    return found
diff --git a/packages/flow/tests/test_no_duplicate_candidate.py b/packages/flow/tests/test_no_duplicate_candidate.py
index ebe1aea3e..faaa2d8ba 100644
--- a/packages/flow/tests/test_no_duplicate_candidate.py
+++ b/packages/flow/tests/test_no_duplicate_candidate.py
@@ -105,3 +105,41 @@ def test_a_linear_run_compares_with_its_own_parameters() -> None:
         candidate_params=("lname", "DOB"),
     )
     assert replayer._postcondition_passes(pc, b"", None, {}) is True
+
+
+DOB_PARAMS = {"lname": "Brennan", "fname": "Nadia", "DOB": "1944-04-20"}
+
+
+@pytest.mark.parametrize(
+    "row",
+    [
+        # A married-name change: another last name, the same date of birth,
+        # shown however the app formats dates.
+        "7\tOkafor, Nadia\tApr 20, 1944\tFemale",
+        "7\tOkafor, Nadia\t04/20/1944\tFemale",
+        "7\tOkafor, Nadia\t20.04.1944\tFemale",
+        "7\tOkafor, Nadia\t20 April 1944\tFemale",
+        # A row date that reads either way counts when either reading matches.
+        "7\tOkafor, Nadia\t04/20/1944 (20/04/1944)\tFemale",
+    ],
+)
+def test_a_shared_date_of_birth_in_any_format_stops(row: str) -> None:
+    assert _check(_Backend([row]), DOB_PARAMS) is False
+
+
+def test_an_ambiguous_row_date_counts_when_either_reading_matches() -> None:
+    params = {**DOB_PARAMS, "DOB": "1970-04-03"}
+    # 03/04/1970 is 4 March or 3 April: the 3 April reading is this patient's.
+    assert _check(_Backend(["7\tOkafor, Nadia\t03/04/1970"]), params) is False
+
+
+@pytest.mark.parametrize(
+    "row",
+    [
+        "7\tOkafor, Nadia\tApr 21, 1944\tFemale",  # an unrelated date
+        "7\tOkafor, Nadia\t1950-04-21\tFemale",
+        "7\tOkafor, Nadia\tno date shown\tFemale",  # nothing to compare
+    ],
+)
+def test_a_row_with_an_unrelated_date_or_none_proceeds(row: str) -> None:
+    assert _check(_Backend([row]), DOB_PARAMS) is True