From cfe42f1131bade1e7c44bce9f106952c4ba40874 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 03:48:16 -0500 Subject: [PATCH 1/3] Propose composite_key when SoR read has no key field When propose_effects sees a new record without the single-field key, build a composite_key from parameter-bound match fields plus any record-identity fields the person named (teach expect). Show it in the proposed checks for confirmation; never apply it silently. Confirming checks also drops composite_key fields that were dropped from match. Closes #1864. --- .../openadapt_flow/compile_agent/pipeline.py | 39 +++++++- packages/flow/openadapt_flow/record_sor.py | 90 ++++++++++++++++++- packages/flow/tests/test_record_sor.py | 49 ++++++++++ 3 files changed, 173 insertions(+), 5 deletions(-) diff --git a/packages/flow/openadapt_flow/compile_agent/pipeline.py b/packages/flow/openadapt_flow/compile_agent/pipeline.py index eb0c87ac2b..0405f8775a 100644 --- a/packages/flow/openadapt_flow/compile_agent/pipeline.py +++ b/packages/flow/openadapt_flow/compile_agent/pipeline.py @@ -795,7 +795,9 @@ def _propose( render_contract, ) - proposal = propose_effects(diff_snapshots(before, after), values) + proposal = propose_effects( + diff_snapshots(before, after), values, record_fields=self.expect + ) path = self.out_dir / PROPOSED_FILE path.parent.mkdir(parents=True, exist_ok=True) _write_private(path, render_contract(self.verifier_section, proposal.effects)) @@ -845,9 +847,36 @@ def proposed(self) -> dict[str, Any]: if "expected_count" in e else {} ), + **( + { + "idempotency_key": ( + f"{{param: {e['idempotency_key']['param']}}}" + if isinstance(e.get("idempotency_key"), dict) + and "param" in e["idempotency_key"] + else (e.get("idempotency_key") or {}).get("literal") + ) + } + if e.get("idempotency_key") + else {} + ), + **( + { + "composite_key": { + k: ( + f"{{param: {v['param']}}}" + if "param" in v + else v.get("literal") + ) + for k, v in (e.get("composite_key") or {}).items() + } + } + if e.get("composite_key") + else {} + ), } for e in effects ], + } # -- step 2: a person confirms the checks and the parameters ------------------ @@ -890,6 +919,14 @@ def confirm_checks( for k, v in (effect.get("match") or {}).items() if k not in drop_fields } + if effect.get("composite_key"): + effect["composite_key"] = { + k: v + for k, v in (effect.get("composite_key") or {}).items() + if k not in drop_fields + } + if not effect["composite_key"]: + effect.pop("composite_key", None) if _claims_new_record(effect, count_new_only=False): # A check routed to one read (scaffold-verifier's # ``surface=|...``) keeps its route. diff --git a/packages/flow/openadapt_flow/record_sor.py b/packages/flow/openadapt_flow/record_sor.py index 76e82ddda4..943ce57f1c 100644 --- a/packages/flow/openadapt_flow/record_sor.py +++ b/packages/flow/openadapt_flow/record_sor.py @@ -12,7 +12,9 @@ * a record that appeared becomes a ``record_written`` effect whose ``match`` holds the record's observed fields, with a recorded parameter in place of - the typed value wherever a field equals one; + the typed value wherever a field equals one; when the record has no single + ``key`` field, a ``composite_key`` is also proposed from parameter-bound + match fields plus any record-identity fields the person named; * a field that changed on a record that already existed becomes a ``field_equals`` effect on that record; * a record that disappeared, a nested field that changed, and a field that was @@ -39,9 +41,10 @@ from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path -from typing import Any, Mapping, Optional +from typing import Any, Mapping, Optional, Sequence, Union from openadapt_flow.runtime.effects.effect import ( + DEFAULT_IDEMPOTENCY_KEY_FIELD, Effect, EffectKind, EffectState, @@ -376,13 +379,85 @@ def _describe(expr: ValueExpr) -> str: return f"parameter {expr.param!r}" if expr.param else repr(expr.literal) +def _record_field_names( + record_fields: Optional[Union[Mapping[str, Any], Sequence[str]]], +) -> frozenset[str]: + """Names that identify the written record for a person (``expect`` / + ``record_fields``). Used only when proposing a ``composite_key``.""" + + if not record_fields: + return frozenset() + if isinstance(record_fields, Mapping): + return frozenset(str(name) for name in record_fields) + return frozenset(str(name) for name in record_fields) + + +def _proposed_duplicate_key( + record: Mapping[str, Any], + match: Mapping[str, ValueExpr], + record_field_names: frozenset[str], +) -> tuple[Optional[ValueExpr], dict[str, ValueExpr], list[str]]: + """Idempotency key or composite key for a new ``record_written`` effect. + + Prefer the app's single-field key when the read set carries it. Otherwise + propose a ``composite_key`` from parameter-bound match fields plus the + record-identity fields a person named. Never invent a key that was not + observed or named; the effect stays ``needs_operator_confirmation`` so a + person must confirm before clinical-write policy accepts it. + """ + + notes: list[str] = [] + key_val = record.get(DEFAULT_IDEMPOTENCY_KEY_FIELD) + if key_val is not None and str(key_val).strip() != "": + # Prefer the match binding (parameter when equality told us so). + key_expr = match.get(DEFAULT_IDEMPOTENCY_KEY_FIELD) + if key_expr is None: + key_expr = ValueExpr(literal=str(key_val)) + return key_expr, {}, notes + + composite: dict[str, ValueExpr] = {} + for name, expr in match.items(): + if expr.param is not None or name in record_field_names: + composite[name] = expr + if composite: + notes.append( + "no " + f"{DEFAULT_IDEMPOTENCY_KEY_FIELD!r} field on the new record; " + "proposed composite_key from parameter-bound match fields" + + ( + " plus record identity fields" + if record_field_names + else "" + ) + + " for a person to confirm" + ) + return None, composite, notes + + notes.append( + "no " + f"{DEFAULT_IDEMPOTENCY_KEY_FIELD!r} field on the new record and no " + "parameter-bound or record-identity match fields to build a " + "composite_key from; clinical-write policy will refuse until a person " + "adds an idempotency or composite key by hand" + ) + return None, {}, notes + + def propose_effects( - diff: SnapshotDiff, params: Optional[Mapping[str, str]] = None + diff: SnapshotDiff, + params: Optional[Mapping[str, str]] = None, + record_fields: Optional[Union[Mapping[str, Any], Sequence[str]]] = None, ) -> ContractProposal: - """Effects that would check the change ``diff`` shows, all unconfirmed.""" + """Effects that would check the change ``diff`` shows, all unconfirmed. + + ``record_fields`` are the identity fields a person named for the record + (teach ``expect``). When the read set has no single-field key they join + parameter-bound match fields in a proposed ``composite_key``. + """ lookup, shared = _param_lookup(dict(params or {})) proposal = ContractProposal(not_proposed=list(shared)) + identity_names = _record_field_names(record_fields) unbound: list[str] = [] for record in diff.created: @@ -405,6 +480,10 @@ def propose_effects( "a record appeared with no scalar field to match it by" ) continue + idem_key, composite, key_notes = _proposed_duplicate_key( + record, match, identity_names + ) + proposal.not_proposed.extend(key_notes) proposal.effects.append( Effect( kind=EffectKind.RECORD_WRITTEN, @@ -412,6 +491,9 @@ def propose_effects( expected_count=1, count_new_only=True, needs_operator_confirmation=True, + idempotency_key=idem_key, + key_field=DEFAULT_IDEMPOTENCY_KEY_FIELD, + composite_key=composite, probe=( "exactly one new record with " + ", ".join(f"{k}={_describe(v)}" for k, v in match.items()) diff --git a/packages/flow/tests/test_record_sor.py b/packages/flow/tests/test_record_sor.py index e7f76c0496..05badf8c74 100644 --- a/packages/flow/tests/test_record_sor.py +++ b/packages/flow/tests/test_record_sor.py @@ -139,6 +139,55 @@ def test_new_record_becomes_one_unconfirmed_record_written(): "patient_id": ValueExpr(literal="p1"), } assert any("nested field(s) extra" in n for n in proposal.not_proposed) + # Issue #1864: no single-field key -> propose composite from param-bound + # match fields (patient_id stays out until named as record identity). + assert effect.idempotency_key is None + assert effect.composite_key == {"note": ValueExpr(param="note")} + + +def test_composite_key_includes_named_record_identity_fields(): + # Issue #1864: when the read set has no key field, propose composite_key + # from parameter-bound match fields plus the person's record_fields. + diff = diff_snapshots( + [], + [{"id": 7, "patient_id": "p1", "note": NOTE, "key": None}], + ) + proposal = propose_effects( + diff, {"note": NOTE}, record_fields={"patient_id": "p1"} + ) + [effect] = proposal.effects + assert effect.idempotency_key is None + assert effect.composite_key == { + "note": ValueExpr(param="note"), + "patient_id": ValueExpr(literal="p1"), + } + assert effect.needs_operator_confirmation is True + assert any("proposed composite_key" in n for n in proposal.not_proposed) + + +def test_single_field_key_becomes_idempotency_key_not_composite(): + diff = diff_snapshots( + [], + [{"id": 7, "patient_id": "p1", "note": NOTE, "key": "run-abc"}], + ) + proposal = propose_effects( + diff, {"note": NOTE}, record_fields={"patient_id": "p1"} + ) + [effect] = proposal.effects + assert effect.idempotency_key == ValueExpr(literal="run-abc") + assert effect.composite_key == {} + assert not any("composite_key" in n for n in proposal.not_proposed) + + +def test_no_key_and_no_param_bound_fields_notes_missing_composite(): + diff = diff_snapshots([], [{"id": 3, "status": "open"}]) + proposal = propose_effects(diff, {"note": NOTE}) + [effect] = proposal.effects + assert effect.idempotency_key is None + assert effect.composite_key == {} + assert any( + "clinical-write policy will refuse" in n for n in proposal.not_proposed + ) def test_changed_fields_become_field_equals_on_that_record(): From 460727bf891a58c09ff6fe93cd0fc5b2eea5a684 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 04:01:27 -0500 Subject: [PATCH 2/3] Format record_sor propose_effects call sites for ruff CI failed ruff format --check on the composite_key proposal lines. --- packages/flow/openadapt_flow/record_sor.py | 6 +----- packages/flow/tests/test_record_sor.py | 12 +++--------- 2 files changed, 4 insertions(+), 14 deletions(-) diff --git a/packages/flow/openadapt_flow/record_sor.py b/packages/flow/openadapt_flow/record_sor.py index 943ce57f1c..5f7e0a4f19 100644 --- a/packages/flow/openadapt_flow/record_sor.py +++ b/packages/flow/openadapt_flow/record_sor.py @@ -424,11 +424,7 @@ def _proposed_duplicate_key( "no " f"{DEFAULT_IDEMPOTENCY_KEY_FIELD!r} field on the new record; " "proposed composite_key from parameter-bound match fields" - + ( - " plus record identity fields" - if record_field_names - else "" - ) + + (" plus record identity fields" if record_field_names else "") + " for a person to confirm" ) return None, composite, notes diff --git a/packages/flow/tests/test_record_sor.py b/packages/flow/tests/test_record_sor.py index 05badf8c74..a9b6005764 100644 --- a/packages/flow/tests/test_record_sor.py +++ b/packages/flow/tests/test_record_sor.py @@ -152,9 +152,7 @@ def test_composite_key_includes_named_record_identity_fields(): [], [{"id": 7, "patient_id": "p1", "note": NOTE, "key": None}], ) - proposal = propose_effects( - diff, {"note": NOTE}, record_fields={"patient_id": "p1"} - ) + proposal = propose_effects(diff, {"note": NOTE}, record_fields={"patient_id": "p1"}) [effect] = proposal.effects assert effect.idempotency_key is None assert effect.composite_key == { @@ -170,9 +168,7 @@ def test_single_field_key_becomes_idempotency_key_not_composite(): [], [{"id": 7, "patient_id": "p1", "note": NOTE, "key": "run-abc"}], ) - proposal = propose_effects( - diff, {"note": NOTE}, record_fields={"patient_id": "p1"} - ) + proposal = propose_effects(diff, {"note": NOTE}, record_fields={"patient_id": "p1"}) [effect] = proposal.effects assert effect.idempotency_key == ValueExpr(literal="run-abc") assert effect.composite_key == {} @@ -185,9 +181,7 @@ def test_no_key_and_no_param_bound_fields_notes_missing_composite(): [effect] = proposal.effects assert effect.idempotency_key is None assert effect.composite_key == {} - assert any( - "clinical-write policy will refuse" in n for n in proposal.not_proposed - ) + assert any("clinical-write policy will refuse" in n for n in proposal.not_proposed) def test_changed_fields_become_field_equals_on_that_record(): From 854848240b9496c0d005d2d5f5f0eacf6c2fba57 Mon Sep 17 00:00:00 2001 From: abrichr Date: Thu, 5 Nov 2026 04:06:09 -0500 Subject: [PATCH 3/3] Format pipeline proposed() composite_key block for ruff --- packages/flow/openadapt_flow/compile_agent/pipeline.py | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/flow/openadapt_flow/compile_agent/pipeline.py b/packages/flow/openadapt_flow/compile_agent/pipeline.py index 0405f8775a..f903a2476c 100644 --- a/packages/flow/openadapt_flow/compile_agent/pipeline.py +++ b/packages/flow/openadapt_flow/compile_agent/pipeline.py @@ -876,7 +876,6 @@ def proposed(self) -> dict[str, Any]: } for e in effects ], - } # -- step 2: a person confirms the checks and the parameters ------------------