From 8406e0a13104350623e22b58e0195f927fb2401e Mon Sep 17 00:00:00 2001 From: abrichr Date: Sun, 25 Oct 2026 10:19:27 -0400 Subject: [PATCH] verify-run: exact_new_set sets aside a record another user added Co-Authored-By: Claude Opus 5.5 --- packages/flow/openadapt_flow/verify_run.py | 81 +++++++++++++++++ .../tests/test_exact_new_set_attribution.py | 86 +++++++++++++++++++ 2 files changed, 167 insertions(+) create mode 100644 packages/flow/tests/test_exact_new_set_attribution.py diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index 1d9d4100e..278e0a576 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/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 = ( diff --git a/packages/flow/tests/test_exact_new_set_attribution.py b/packages/flow/tests/test_exact_new_set_attribution.py new file mode 100644 index 000000000..76d2ec36d --- /dev/null +++ b/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"