diff --git a/packages/flow/docs/JIT.md b/packages/flow/docs/JIT.md index 45bccc037..0b27d69a6 100644 --- a/packages/flow/docs/JIT.md +++ b/packages/flow/docs/JIT.md @@ -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 diff --git a/packages/flow/openadapt_flow/jit/cli.py b/packages/flow/openadapt_flow/jit/cli.py index 2850e5984..c4f7dcb7b 100644 --- a/packages/flow/openadapt_flow/jit/cli.py +++ b/packages/flow/openadapt_flow/jit/cli.py @@ -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)" diff --git a/packages/flow/openadapt_flow/jit/compile_agent.py b/packages/flow/openadapt_flow/jit/compile_agent.py index d86665cde..266e1a62c 100644 --- a/packages/flow/openadapt_flow/jit/compile_agent.py +++ b/packages/flow/openadapt_flow/jit/compile_agent.py @@ -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( diff --git a/packages/flow/openadapt_flow/jit/lifecycle.py b/packages/flow/openadapt_flow/jit/lifecycle.py index 882c4f0cf..66d0627d9 100644 --- a/packages/flow/openadapt_flow/jit/lifecycle.py +++ b/packages/flow/openadapt_flow/jit/lifecycle.py @@ -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, *, diff --git a/packages/flow/tests/test_jit_lifecycle.py b/packages/flow/tests/test_jit_lifecycle.py index 06911447f..ea3e1fc51 100644 --- a/packages/flow/tests/test_jit_lifecycle.py +++ b/packages/flow/tests/test_jit_lifecycle.py @@ -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"