abrichr · Oct 19, 2026 · Oct 19, 2026 · Oct 19, 2026
@@ -226,6 +226,16 @@ for the new value. `accept_params` must name exactly the task's parameters.
 The recompile can renumber the steps, so it comes before the review and
 returns the step ids the review asks for. It refuses once a review exists.
 
+A policy such as `clinical-write` also wants an idempotency key on every
+write, so that a duplicate submission is caught. One run can't show which
+input identifies a record; the person confirming names it:
+`confirm_params(..., idempotency_key="claim_no")`, or
+`openadapt-flow jit review --idempotency-key claim_no`. It must be a task
+parameter that the mined record check already selects the saved record by.
+That check then counts the records that carry the run's value and refuses
+more than one. Name it only when the app keeps that value unique, such as a
+claim number. `status()` lists the keys the program carries.
+
 ## A look-up the agent made
 
 When the agent looks the record up after it saves (types the identifier into

@@ -148,8 +148,16 @@ def confirm(prompt: str) -> str:
                     program,
                     principal_id=args.principal,
                     accept_params=accept,
-                    confirm=confirm(f"Make {accept} the task's parameters?"),
+                    confirm=confirm(
+                        f"Make {accept} the task's parameters"
+                        + (
+                            f", with {args.idempotency_key} identifying one record?"
+                            if args.idempotency_key
+                            else "?"
+                        )
+                    ),
                     key_dir=_key_dir(args),
+                    idempotency_key=args.idempotency_key,
                 )
                 status = lifecycle.admission_status(program, key_dir=_key_dir(args))
             for i, step in enumerate(status["steps"], 1):
@@ -280,6 +288,15 @@ def common(p: argparse.ArgumentParser, *, program: bool = True) -> None:
             "before the steps are shown, because it recompiles the program"
         ),
     )
+    p.add_argument(
+        "--idempotency-key",
+        metavar="PARAM",
+        help=(
+            "The parameter that identifies one saved record (the app keeps it "
+            "unique, such as a claim number). The record check selects by it "
+            "and refuses a duplicate"
+        ),
+    )
     group = p.add_mutually_exclusive_group()
     group.add_argument(
         "--step", action="append", help="A step id you reviewed (repeat)"

@@ -710,9 +710,11 @@ def confirm_params(
         principal_id: Any,
         accept_params: Sequence[str],
         confirm: object = None,
+        idempotency_key: Optional[str] = None,
     ) -> dict[str, Any]:
         """Confirm the task's parameters before the review. Returns the step
-        ids of the recompiled program. See
+        ids of the recompiled program. ``idempotency_key`` names the parameter
+        that identifies one saved record. See
         :func:`openadapt_flow.jit.lifecycle.confirm_params`."""
 
         return lifecycle.confirm_params(
@@ -721,6 +723,7 @@ def confirm_params(
             accept_params=accept_params,
             confirm=confirm,
             key_dir=self.key_dir,
+            idempotency_key=idempotency_key,
         )
 
     def review(

@@ -803,6 +803,9 @@ def admission_status(
             },
         },
         "needs_person": _evidence_at(program / NEEDS_PERSON_FILE, digest),
+        # A policy such as clinical-write needs one on every write; a person
+        # names it when confirming the parameters.
+        "idempotency_keys": idempotency_keys(workflow),
     }
 
 
@@ -845,9 +848,19 @@ def confirm_params(
     accept_params: Sequence[str],
     confirm: object = None,
     key_dir: Optional[Path] = None,
+    idempotency_key: Optional[str] = None,
 ) -> dict[str, Any]:
     """A named person confirms which typed values become the task's parameters.
 
+    ``idempotency_key`` names the parameter that identifies one saved record,
+    such as a claim number the app keeps unique. The person asserts that; it
+    can't be read from one run. The parameter must be one the mined record
+    check already selects the saved record by. It becomes that check's
+    at-most-once key (:attr:`Effect.idempotency_key`), so a run counts the
+    records bearing the run's value and refuses a duplicate. Without a
+    parameter the record check selects by, it refuses and names the ones it
+    could take.
+
     ``accept_params`` must be exactly the task's parameters. They are applied
     by recompiling the recording
     (:func:`~openadapt_flow.compiler.param_confirm.apply_decisions`), which can
@@ -907,18 +920,108 @@ def confirm_params(
             f"the program's parameters {sorted(workflow.params)} do not match "
             f"the task's {task_params}"
         )
+    if idempotency_key is not None and _set_idempotency_key(
+        bundle, workflow, idempotency_key
+    ):
+        # The key is part of the digest; the recorder is unchanged.
+        workflow = _load_workflow(bundle)
+        candidate = sign_recorder(program, candidate, key_dir=key_dir)
+        _write(program / CANDIDATE_FILE, candidate)
     record = {
         "principal_id": name,
         "confirmed_at": _now(),
         "digest": _digest(workflow),
         "params": sorted(accept_params),
         "steps": [s.id for s in workflow.steps],
+        "idempotency_key": idempotency_keys(workflow),
     }
     _write(program / PARAMS_FILE, record)
     _event(program, "params_confirmed", principal_id=name, digest=record["digest"])
     return record
 
 
+def _record_checks(workflow: Any) -> list[tuple[Any, Any]]:
+    """(step, effect) for every mined record-written check."""
+
+    return [
+        (step, effect)
+        for step in workflow.steps
+        for effect in step.effects
+        if effect.kind.value == "record_written"
+        and not effect.needs_operator_confirmation
+    ]
+
+
+def idempotency_key_candidates(workflow: Any) -> list[str]:
+    """Parameters a record-written check selects its record by."""
+
+    return sorted(
+        {
+            expr.param
+            for _, effect in _record_checks(workflow)
+            for expr in (effect.match or {}).values()
+            if getattr(expr, "param", None)
+        }
+    )
+
+
+def idempotency_keys(workflow: Any) -> list[dict[str, Any]]:
+    """The at-most-once keys the record checks carry, for the record."""
+
+    return [
+        {
+            "step_id": step.id,
+            "field": effect.key_field,
+            "param": getattr(effect.idempotency_key, "param", None),
+        }
+        for step, effect in _record_checks(workflow)
+        if effect.idempotency_key is not None
+    ]
+
+
+def _set_idempotency_key(bundle: Path, workflow: Any, param: str) -> bool:
+    """Key every record check that selects by ``param``. True if it changed."""
+
+    from openadapt_flow.runtime.effects import ValueExpr
+
+    if param not in workflow.params:
+        raise AdmissionError(
+            f"idempotency_key {param!r} is not one of the task's parameters "
+            f"{sorted(workflow.params)}"
+        )
+    bound = []
+    for step, effect in _record_checks(workflow):
+        fields = [
+            field
+            for field, expr in (effect.match or {}).items()
+            if getattr(expr, "param", None) == param
+        ]
+        if len(fields) != 1:
+            continue
+        current = effect.idempotency_key
+        if current is not None and (
+            getattr(current, "param", None) != param or effect.key_field != fields[0]
+        ):
+            raise AdmissionError(
+                f"the record check on {step.id} already has the key "
+                f"{effect.key_field}={current}; reject the draft to change it"
+            )
+        bound.append((effect, fields[0]))
+    if not bound:
+        raise AdmissionError(
+            f"no record check selects its record by {param!r}, so it can't be "
+            "the idempotency key; parameters a record check selects by: "
+            f"{idempotency_key_candidates(workflow) or 'none'}"
+        )
+    if all(effect.idempotency_key is not None for effect, _ in bound):
+        return False
+    for effect, field in bound:
+        effect.idempotency_key = ValueExpr(param=param)
+        effect.key_field = field
+    workflow.save(bundle)
+    return True
+
+
 def record_review(
     program_dir: Path | str,
     *,

@@ -694,3 +694,157 @@ def test_cli_review_and_admit_need_a_terminal(
     assert _cli(*admit, "--require-distinct") == 2
     assert _cli(*admit) == 0
     assert "ADMITTED by Dana Reviewer" in capsys.readouterr().out
+
+
+# -- the idempotency key a person names (OpenAdaptAI/OpenAdapt#1329) -------------
+#
+# Shaped like the openIMIS claim program: the mined record check already
+# selects the saved claim by the confirmed claim_no parameter, but carried no
+# at-most-once key, so clinical-write refused it. Synthetic values only.
+
+CLAIM = TaskSignature(
+    "t.claim",
+    "Enter claim {claim_no} for {insurance_no}",
+    {"claim_no": str, "insurance_no": str},
+)
+
+
+def _claim_jit(tmp_path: Path) -> Any:
+    from openadapt_flow.runtime.effects import ValueExpr
+
+    effect = Effect(
+        kind=EffectKind.RECORD_WRITTEN,
+        match={
+            "claim_no": ValueExpr(param="claim_no"),
+            "insurance_no": ValueExpr(param="insurance_no"),
+            "hf_code": ValueExpr(literal="HF001"),
+        },
+    )
+    jit = _jit(tmp_path, task=CLAIM, effect=effect)
+    workflow = Workflow.load(jit.bundle_dir)
+    workflow.params = {"claim_no": "CX001", "insurance_no": "990000001"}
+    workflow.save(jit.bundle_dir)
+    candidate = json.loads((jit.program_dir / "candidate.json").read_text())
+    candidate["params_applied"] = ["claim_no", "insurance_no"]
+    candidate = lifecycle.sign_recorder(jit.program_dir, candidate, key_dir=jit.key_dir)
+    (jit.program_dir / "candidate.json").write_text(json.dumps(candidate))
+    return jit
+
+
+def _confirm(jit: Any, key: str | None = None) -> dict:
+    return jit.confirm_params(
+        principal_id=REVIEWER,
+        accept_params=["claim_no", "insurance_no"],
+        confirm="ok",
+        idempotency_key=key,
+    )
+
+
+def test_a_named_parameter_becomes_the_record_checks_idempotency_key(
+    tmp_path: Path,
+) -> None:
+    from openadapt_flow.policy import effect_has_idempotency_key
+    from openadapt_flow.runtime.effects import EffectState
+    from openadapt_flow.runtime.effects._common import judge_records
+
+    jit = _claim_jit(tmp_path)
+    assert not effect_has_idempotency_key(Workflow.load(jit.bundle_dir).steps[1])
+    record = _confirm(jit, "claim_no")
+    assert record["idempotency_key"] == [
+        {"step_id": "step_001", "field": "claim_no", "param": "claim_no"}
+    ]
+    save = Workflow.load(jit.bundle_dir).steps[1]
+    assert effect_has_idempotency_key(save)
+    assert jit.status()["idempotency_keys"] == record["idempotency_key"]
+    # The review and canary pin the bundle with the key.
+    assert record["digest"] == jit.status()["digest"]
+
+    # At run time the key is the run's own claim number: a second claim with
+    # that number is refused, one is confirmed.
+    effect = save.effects[0].resolve({"claim_no": "CX002", "insurance_no": "1"})
+    row = {"claim_no": "CX002", "insurance_no": "1", "hf_code": "HF001"}
+    before = EffectState(substrate="sql", reachable=True, records=[])
+    one = judge_records(effect, before, [{"id": 1, **row}], substrate="sql")
+    two = judge_records(
+        effect, before, [{"id": 1, **row}, {"id": 2, **row}], substrate="sql"
+    )
+    assert one.verdict.value == "confirmed"
+    assert two.verdict.value == "refuted"
+
+
+def test_without_a_named_key_the_record_check_stays_unkeyed(tmp_path: Path) -> None:
+    from openadapt_flow.policy import effect_has_idempotency_key
+
+    jit = _claim_jit(tmp_path)
+    assert _confirm(jit)["idempotency_key"] == []
+    assert not effect_has_idempotency_key(Workflow.load(jit.bundle_dir).steps[1])
+
+
+@pytest.mark.parametrize(
+    ("key", "why"),
+    [
+        ("note", "not one of the task's parameters"),
+        ("hf_code", "not one of the task's parameters"),
+    ],
+)
+def test_the_key_must_be_a_task_parameter(tmp_path: Path, key: str, why: str) -> None:
+    jit = _claim_jit(tmp_path)
+    with pytest.raises(AdmissionError, match=why):
+        _confirm(jit, key)
+    assert not (jit.program_dir / "params.json").exists()
+
+
+def test_the_key_must_select_the_saved_record(tmp_path: Path) -> None:
+    from openadapt_flow.runtime.effects import ValueExpr
+
+    jit = _claim_jit(tmp_path)
+    workflow = Workflow.load(jit.bundle_dir)
+    workflow.steps[1].effects[0].match = {
+        "insurance_no": ValueExpr(param="insurance_no"),
+        "hf_code": ValueExpr(literal="HF001"),
+    }
+    workflow.save(jit.bundle_dir)
+    candidate = json.loads((jit.program_dir / "candidate.json").read_text())
+    candidate = lifecycle.sign_recorder(jit.program_dir, candidate, key_dir=jit.key_dir)
+    (jit.program_dir / "candidate.json").write_text(json.dumps(candidate))
+    with pytest.raises(
+        AdmissionError,
+        match=r"no record check selects its record by 'claim_no'.*\['insurance_no'\]",
+    ):
+        _confirm(jit, "claim_no")
+
+
+def test_a_placeholder_record_check_takes_no_key(tmp_path: Path) -> None:
+    jit = _claim_jit(tmp_path)
+    workflow = Workflow.load(jit.bundle_dir)
+    workflow.steps[1].effects[0].needs_operator_confirmation = True
+    workflow.save(jit.bundle_dir)
+    candidate = json.loads((jit.program_dir / "candidate.json").read_text())
+    candidate = lifecycle.sign_recorder(jit.program_dir, candidate, key_dir=jit.key_dir)
+    (jit.program_dir / "candidate.json").write_text(json.dumps(candidate))
+    with pytest.raises(AdmissionError, match="no record check selects"):
+        _confirm(jit, "claim_no")
+
+
+@pytest.mark.usefixtures("at_terminal")
+def test_cli_review_names_the_idempotency_key(tmp_path: Path, capsys) -> None:
+    jit = _claim_jit(tmp_path)
+    assert (
+        _cli(
+            "review",
+            str(jit.task_dir),
+            "--principal",
+            REVIEWER,
+            "--accept-param",
+            "claim_no",
+            "--accept-param",
+            "insurance_no",
+            "--idempotency-key",
+            "claim_no",
+            "--all-steps-shown",
+            "--key-dir",
+            str(tmp_path / "keys"),
+        )
+        == 0
+    )
+    assert jit.status()["idempotency_keys"][0]["param"] == "claim_no"