verify-run: exact_new_set sets aside a record another user added

packages/flow/openadapt_flow/verify_run.py

@@ -492,6 +492,79 @@ def read_of(name: str) -> str:
     )
 
 
+def _excuse_attributed_additions(
+    cfg: VerifyConfig,
+    verifier: Any,
+    before: EffectState,
+    effects: Sequence[Effect],
+    verdicts: list[EffectVerdict],
+    rows: list[dict[str, Any]],
+    frame: FrameVerdict,
+) -> None:
+    """Re-judge a refuted ``exact_new_set`` without records another user added.
+
+    ``exact_new_set`` says the save added these records and nothing else. A
+    record the frame's attribution names another actor for (a colleague's
+    new patient in the same window) is not this save's addition. Only such
+    attributed additions are set aside; an unattributed one still refutes.
+    """
+
+    from openadapt_flow.runtime.effects._common import judge_records
+    from openadapt_flow.runtime.effects.effect import stable_id
+    from openadapt_flow.runtime.effects.surfaces import surface_of
+
+    spec = cfg.frame or FrameSpec()
+    excused = {
+        (c.surface, str(c.key))
+        for c in frame.concurrent
+        if c.change == "added" and c.key is not None
+    }
+    if not excused:
+        return
+
+    def key_of(record: Mapping[str, Any]) -> str:
+        if spec.identity_field:
+            return str(record.get(spec.identity_field))
+        ignore = set(spec.ignore_fields)
+        return str(stable_id({k: v for k, v in record.items() if k not in ignore}))
+
+    for i, (effect, verdict) in enumerate(zip(effects, verdicts)):
+        if (
+            effect.kind is not EffectKind.EXACT_NEW_SET
+            or verdict.verdict is not Verdict.REFUTED
+        ):
+            continue
+        try:
+            state = verifier.capture_pre_state()
+        except Exception:  # noqa: BLE001 - unreadable: the refutation stands
+            continue
+        surface = surface_of(effect) or ""
+        routed = isinstance(getattr(verifier, "verifiers", None), dict)
+        if routed and surface:
+            post = (state.detail or {}).get(surface, {})
+            pre = (before.detail or {}).get(surface, {})
+            records = list(post.get("records", [])) if post.get("reachable") else None
+            base = EffectState(
+                substrate=str(pre.get("substrate", "")),
+                reachable=bool(pre.get("reachable", False)),
+                records=list(pre.get("records", [])),
+            )
+            name = surface
+        else:
+            records = list(state.records) if state.reachable else None
+            base, name = before, ""
+        if records is None:
+            continue
+        kept = [r for r in records if (name, key_of(r)) not in excused]
+        if len(kept) == len(records):
+            continue
+        again = judge_records(effect, base, kept, substrate=verdict.substrate)
+        if again.verdict is Verdict.CONFIRMED:
+            verdicts[i] = again
+            rows[i]["verdict"] = _label(again.verdict)
+            rows[i]["observed_count"] = again.observed_count
+
+
 def _target_record(
     effects: Sequence[Effect], verdicts: Sequence[EffectVerdict]
 ) -> Optional[dict[str, Any]]:
@@ -985,6 +1058,7 @@ def verify_run(
     verdicts: list[EffectVerdict] = []
     rows: list[dict[str, Any]] = []
     bound_effects: list[Effect] = []
+    judged_effects: list[Effect] = []
     weakest: Optional[VerificationTier] = None
     tiers: list[Optional[VerificationTier]] = []
     for effect in cfg.effects:
@@ -1007,6 +1081,7 @@ def verify_run(
             else _audit_effect(effect, bound)
         )
         bound_effects.append(bound)
+        judged_effects.append(judged)
         tier = verifier_effect_tier(verifier, judged)
         try:
             verdict = verify_effect_without_mutation(verifier, judged, before)
@@ -1065,6 +1140,12 @@ def verify_run(
             status="not_checked",
             reason="no baseline, so what else changed can't be known",
         )
+    if frame.concurrent and any(c.change == "added" for c in frame.concurrent):
+        _excuse_attributed_additions(
+            cfg, verifier, before, judged_effects, verdicts, rows, frame
+        )
+        delta_verdict = _aggregate(verdicts)
+        observed_effect = _observed(verdicts)
     # A record outside the declared delta moved: the save's effect on the
     # store is not the declared one, whatever the per-effect reads say.
     effect_verdict: EffectVerdictLabel = (

packages/flow/tests/test_exact_new_set_attribution.py

@@ -0,0 +1,86 @@
+"""``exact_new_set`` sets aside a record the frame attributes to another user.
+
+A colleague registering a different patient in the same window is not this
+save's addition. Only an addition the audit log names another actor for is
+set aside; an unattributed one still refutes. Records are made up.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+from openadapt_flow.verify_run import capture_baseline, parse_verify_config, verify_run
+
+
+def _db(path: Path) -> sqlite3.Connection:
+    con = sqlite3.connect(path)
+    con.execute(
+        "CREATE TABLE patient (id INTEGER PRIMARY KEY, pid INTEGER, lname TEXT, email TEXT)"
+    )
+    con.execute(
+        "CREATE TABLE log (id INTEGER PRIMARY KEY, user TEXT, patient_id INTEGER)"
+    )
+    con.execute("INSERT INTO patient VALUES (1, 1, 'Old', 'old@example.test')")
+    con.commit()
+    return con
+
+
+def _run(tmp_path: Path, colleague_logged: bool) -> str:
+    db = tmp_path / "s.db"
+    con = _db(db)
+    cfg = parse_verify_config(
+        {
+            "schema_version": "openadapt.verify/v1",
+            "verifier": {
+                "kind": "sql",
+                "sqlite_database": str(db),
+                "sql_query": "SELECT * FROM patient",
+            },
+            "record": {
+                "name": "patient",
+                "match": ["email"],
+                "exact_new_set": True,
+                "fields": {"email": "email", "lname": "lname"},
+            },
+            "frame": {
+                "identity_field": "pid",
+                "attribution": {
+                    "kind": "audit_read",
+                    "actor_field": "user",
+                    "record_field": "patient_id",
+                    "self": ["bot"],
+                    "read": {
+                        "kind": "sql",
+                        "sqlite_database": str(db),
+                        "sql_query": "SELECT * FROM log",
+                    },
+                },
+            },
+        }
+    )
+    params = {"email": "new@example.test", "lname": "New"}
+    capture_baseline(cfg, params, tmp_path / "b.json")
+    con.execute("INSERT INTO patient VALUES (2, 2, 'New', 'new@example.test')")
+    con.execute("INSERT INTO log (user, patient_id) VALUES ('bot', 2)")
+    con.execute("INSERT INTO patient VALUES (3, 3, 'Colleague', 'c@example.test')")
+    if colleague_logged:
+        con.execute("INSERT INTO log (user, patient_id) VALUES ('clinic2', 3)")
+    con.commit()
+    con.close()
+    receipt = verify_run(
+        cfg,
+        params=params,
+        run_record={"executor_kind": "agent", "executor_id": "bot"},
+        authorization_ref="t",
+        baseline_path=tmp_path / "b.json",
+    )["receipt"]
+    return receipt["outcome"]
+
+
+def test_a_colleagues_new_patient_does_not_refute_the_save(tmp_path: Path) -> None:
+    assert _run(tmp_path, colleague_logged=True) == "verified"
+
+
+def test_an_unattributed_new_record_still_refutes(tmp_path: Path) -> None:
+    assert _run(tmp_path, colleague_logged=False) == "reconciliation_required"