Propose composite_key when SoR read has no key field

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,6 +847,32 @@ 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
             ],
@@ -890,6 +918,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=<name>|...``) keeps its route.

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,81 @@ 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,13 +476,20 @@ 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,
                 match=match,
                 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())

packages/flow/tests/test_record_sor.py

@@ -139,6 +139,49 @@ 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():