From 48b0ad0d5eb8d5616065df30d859f034229c2a68 Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 14:53:22 -0400 Subject: [PATCH 1/4] verify-run: a record outside the target that changed is collateral, not only one that vanished (#1373) Effect.forbid_collateral_change (record_written only, off by default in a run, bound into the contract hash only when set) refutes when a pre-state record that is not the target is still there but its fields differ. verify-run applies it to every record_written effect when it judges, unless an effect opts out, and leaves the file's contract and digest unchanged so they still equal qualify's; the receipt's evidence records that the check ran. Regression: the OpenEMR collateral_edit case from the scoreboard's claims run. Co-Authored-By: Claude Opus 5.5 --- packages/flow/docs/EFFECT_KIT.md | 4 +- packages/flow/docs/VERIFY_RUN.md | 17 +- .../openadapt_flow/runtime/effects/_common.py | 29 ++- .../openadapt_flow/runtime/effects/effect.py | 23 ++ packages/flow/openadapt_flow/verify_run.py | 34 ++- .../test_verify_run_collateral_change.py | 211 ++++++++++++++++++ 6 files changed, 312 insertions(+), 6 deletions(-) create mode 100644 packages/flow/tests/test_verify_run_collateral_change.py diff --git a/packages/flow/docs/EFFECT_KIT.md b/packages/flow/docs/EFFECT_KIT.md index aa76328c2..412cfdaaf 100644 --- a/packages/flow/docs/EFFECT_KIT.md +++ b/packages/flow/docs/EFFECT_KIT.md @@ -362,7 +362,9 @@ The command looks for a read that returns a record holding them: a recorded value (bound as `{param: NAME}`, so each run is checked against its own values) and on fields the save sent with a fixed value under the same name (`type: Triage`). `forbid_collateral_loss` stops the run if a - record that was there before is gone. + record that was there before is gone. `forbid_collateral_change` (off by + default in a run, on in `verify-run`) also stops it if such a record is + still there but changed. - Recorded values in the read's URL become placeholders (`path_params`, FHIR `search_param_exprs`), so the read is scoped to the run's record. A path segment that looks like one record's id (an MRN, a number, a UUID) diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index 5c5fa8d21..e92d6bef8 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -78,6 +78,21 @@ Optional flags: Both link ids must be opaque (8 to 128 of `A-Za-z0-9._:-`). They are absent unless given, and when present they are signed with the rest of the receipt. +## Records outside the target + +With a baseline, a `record_written` check also refutes when a record that +was there before the agent started and is not the target has changed +(`forbid_collateral_change`), not only when it is gone +(`forbid_collateral_loss`). The agent's write can be right while it also +edited another patient's record in place, and only the baseline shows that. +verify-run turns this on for every `record_written` effect when it judges. +The contract in the file is unchanged, so its digest is still the one qualify +reports, and `evidence.readback.effects[].collateral_change_checked` says the +check ran. To skip it for one effect, for example when the read covers a +table other people edit during the run, set `forbid_collateral_change: false` +on that effect in `effects:`. Fields that change on their own, such as a +modification time, belong in the verifier's `records_drop_fields`. + ## Verdicts | What the store shows | `effect_verdict` | `outcome` | @@ -85,7 +100,7 @@ unless given, and when present they are signed with the rest of the receipt. | The declared record, exactly once, through a system-of-record read | `CONFIRMED` | `verified` | | The same, but the executor approved its own write | `CONFIRMED` | `rejected_policy` | | Nothing that matches | `REFUTED` | `halted_before_effect` | -| A duplicate, a wrong value, or only some of the effects | `REFUTED` | `reconciliation_required` | +| A duplicate, a wrong value, only some of the effects, or another record that vanished or changed | `REFUTED` | `reconciliation_required` | | The store could not be read, no baseline, or a separate-session read | `INDETERMINATE` or `CONFIRMED` | `reconciliation_required` | Only a read through the system of record produces `verified`. A screen diff --git a/packages/flow/openadapt_flow/runtime/effects/_common.py b/packages/flow/openadapt_flow/runtime/effects/_common.py index ef4c8adee..94783edb0 100644 --- a/packages/flow/openadapt_flow/runtime/effects/_common.py +++ b/packages/flow/openadapt_flow/runtime/effects/_common.py @@ -285,7 +285,26 @@ def _judge_record_written( if stable_id(r) not in current_ids: collateral_lost.append(r) - if count_ok and not collateral_lost: + # Collateral change: a non-target record that is still there but whose + # fields differ from the pre-state (an in-place write to the wrong row). + collateral_changed: list[tuple[dict[str, Any], list[str]]] = [] + if effect.forbid_collateral_change and before.reachable: + now = {stable_id(r): r for r in current} + for r in before.records: + if record_matches(r, effect.match): + continue + after = now.get(stable_id(r)) + if after is None: + continue # vanished: the collateral-loss rule's case + fields = sorted( + k + for k in set(r) | set(after) + if str(r.get(k, "")) != str(after.get(k, "")) + ) + if fields: + collateral_changed.append((r, fields)) + + if count_ok and not collateral_lost and not collateral_changed: return EffectVerdict( verdict=Verdict.CONFIRMED, kind=effect.kind, @@ -325,6 +344,14 @@ def _judge_record_written( f"collateral loss (stale / concurrent last-write-wins overwrote " f"another actor's row)" ) + if collateral_changed: + names = sorted({f for _r, fields in collateral_changed for f in fields}) + reasons.append( + f"{len(collateral_changed)} pre-existing record(s) outside the " + f"target changed -- collateral change (field(s): " + f"{', '.join(names[:_SAMPLE_LIMIT])}" + f"{'' if len(names) <= _SAMPLE_LIMIT else ', ...'})" + ) return EffectVerdict( verdict=Verdict.REFUTED, kind=effect.kind, diff --git a/packages/flow/openadapt_flow/runtime/effects/effect.py b/packages/flow/openadapt_flow/runtime/effects/effect.py index d41b099e5..59ebc173d 100644 --- a/packages/flow/openadapt_flow/runtime/effects/effect.py +++ b/packages/flow/openadapt_flow/runtime/effects/effect.py @@ -656,6 +656,16 @@ class Effect(BaseModel): #: vanished is the collateral loss (an exact-set claim about additions #: must not quietly tolerate removals inside the same scope). forbid_collateral_loss: bool = True + #: ``record_written`` only: also REFUTE when a record that existed in the + #: pre-state, does NOT match :attr:`match`, and is still there (same + #: :func:`stable_id`) has CHANGED -- collateral change. The vanished-record + #: rule above misses a write that edits another record in place (a wrong + #: row updated alongside the right one). Records without an ``id`` are + #: identified by their whole content, so for them a change already reads + #: as a vanished record. Default False: a shared table that other actors + #: edit would otherwise refute every save; ``verify-run`` turns it on. + #: Bound into the contract hash only when set, so existing digests hold. + forbid_collateral_change: bool = False #: Consequential-write flag (mirrors ``Step.risk`` / RFC ``State.risk``). #: Compensation (``effects.compensation``) only fires for irreversible #: effects; a reversible effect just halts. @@ -743,6 +753,15 @@ def _count_new_only_scope(self) -> "Effect": ) return self + @model_validator(mode="after") + def _collateral_change_scope(self) -> "Effect": + """``forbid_collateral_change`` is judged on ``record_written`` only.""" + if self.forbid_collateral_change and self.kind is not EffectKind.RECORD_WRITTEN: + raise ValueError( + "forbid_collateral_change applies only to record_written effects" + ) + return self + @model_validator(mode="after") def _exact_new_set_shape(self) -> "Effect": """Refuse an ``exact_new_set`` whose declared set is unusable, and @@ -883,6 +902,8 @@ def value(expr: ValueExpr | None) -> object: } if self.count_new_only: payload["count_new_only"] = True + if self.forbid_collateral_change: + payload["forbid_collateral_change"] = True # Bound ONLY on the new kind, so every pre-existing effect's digest # (and every ledger entry that pins it) stays byte-identical. if self.kind is EffectKind.EXACT_NEW_SET: @@ -953,6 +974,8 @@ def _contract_payload(self) -> dict[str, object]: # its hash — PR #129) keeps its exact digest. if self.count_new_only: payload["count_new_only"] = True + if self.forbid_collateral_change: + payload["forbid_collateral_change"] = True if self.kind is EffectKind.EXACT_NEW_SET: payload["new_records"] = [ {k: str(v) for k, v in sorted(selector.items())} diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index e4218a823..e40332f98 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -68,6 +68,7 @@ from openadapt_flow.runtime.effects.adapter import verifier_identity from openadapt_flow.runtime.effects.effect import ( Effect, + EffectKind, EffectState, EffectVerdict, Verdict, @@ -256,7 +257,13 @@ def parse_verify_config(data: Any, *, source: str = "verify config") -> VerifyCo raise VerifyRunError( "give the save's checks as either effects or record, not both" ) - data["effects"] = [e.model_dump(mode="json") for e in _record_effects(record)] + # Only the fields record_effects set, so an effect's model_fields_set + # says what the file chose (``_audit_effect`` reads it); the values, + # and so the digests, are the same as qualify's. + data["effects"] = [ + e.model_dump(mode="json", exclude_unset=True) + for e in _record_effects(record) + ] try: cfg = VerifyConfig.model_validate(data) except Exception as exc: @@ -291,6 +298,25 @@ def parse_verify_config(data: Any, *, source: str = "verify config") -> VerifyCo return cfg +def _audit_effect(effect: Effect, bound: Effect) -> Effect: + """The effect as verify-run judges it: an audit after the fact. + + A ``record_written`` check also refutes when a record outside the target + changed (``Effect.forbid_collateral_change``), not only when one vanished, + unless the file sets ``forbid_collateral_change: false`` on that effect. + The contract itself is left as the file states it, so its digest stays + the one qualify reports; the receipt's evidence says the check ran. + """ + + if ( + bound.kind is not EffectKind.RECORD_WRITTEN + or not bound.forbid_collateral_loss + or "forbid_collateral_change" in effect.model_fields_set + ): + return bound + return bound.model_copy(update={"forbid_collateral_change": True}) + + # -- hashing ------------------------------------------------------------------ @@ -716,9 +742,10 @@ def verify_run( ) except (KeyError, ValueError) as exc: raise VerifyRunError(f"effect contract cannot bind params: {exc}") from exc - tier = verifier_effect_tier(verifier, bound) + judged = _audit_effect(effect, bound) + tier = verifier_effect_tier(verifier, judged) try: - verdict = verify_effect_without_mutation(verifier, bound, before) + verdict = verify_effect_without_mutation(verifier, judged, before) except EffectContractMutationError as exc: raise VerifyRunError(str(exc)) from exc if tier is None and verdict.verdict is Verdict.CONFIRMED: @@ -746,6 +773,7 @@ def verify_run( "expected_count": verdict.expected_count, "unavailable": verdict.unavailable, "stale": verdict.stale, + "collateral_change_checked": judged.forbid_collateral_change, } ) diff --git a/packages/flow/tests/test_verify_run_collateral_change.py b/packages/flow/tests/test_verify_run_collateral_change.py new file mode 100644 index 000000000..e7670c377 --- /dev/null +++ b/packages/flow/tests/test_verify_run_collateral_change.py @@ -0,0 +1,211 @@ +"""A record outside the target that CHANGED is collateral, not only one that vanished. + +Regression for the OpenEMR ``collateral_edit`` case (scoreboard claims run +``raw/openemr_referral/claims/2026-10-19-a``, items 931 to 935): the agent saved +its own chart correctly while another, pre-existing chart's home phone was +changed. ``verify-run`` said ``verified``, because the collateral rule compared +records by id only. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import requests + +from openadapt_flow.mockmed.fault_server import serve as fault_serve +from openadapt_flow.runtime.effects import Effect, EffectKind, Verdict +from openadapt_flow.runtime.effects._common import judge_records +from openadapt_flow.runtime.effects.effect import EffectState +from openadapt_flow.verify_run import capture_baseline, parse_verify_config, verify_run + +# OpenEMR patient-API shaped records (made up). +EXISTING = { + "id": 1, + "fname": "Wendell", + "lname": "Priorchart", + "phone_home": "555-0188", + "email": "w.priorchart@example.invalid", +} +TARGET = { + "id": 2, + "fname": "Barnaby", + "lname": "ScoreboardV931", + "phone_home": "555-0131", + "email": "barnaby.scoreboard.931@example.invalid", +} + + +def _effect(**extra: object) -> Effect: + return Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"email": TARGET["email"], "lname": TARGET["lname"]}, + expected_count=1, + count_new_only=True, + **extra, + ) + + +def _judge(effect: Effect, after: list[dict]) -> object: + before = EffectState(substrate="rest", reachable=True, records=[dict(EXISTING)]) + return judge_records(effect, before, after, substrate="rest") + + +def test_openemr_collateral_edit_is_refuted() -> None: + edited = {**EXISTING, "phone_home": "555-0199"} + verdict = _judge(_effect(forbid_collateral_change=True), [edited, TARGET]) + assert verdict.verdict is Verdict.REFUTED + assert "collateral change" in verdict.reason + assert "phone_home" in verdict.reason + + +def test_without_the_flag_the_old_rule_still_confirms() -> None: + # Unchanged default: the run engine's contracts behave as before. + edited = {**EXISTING, "phone_home": "555-0199"} + verdict = _judge(_effect(), [edited, TARGET]) + assert verdict.verdict is Verdict.CONFIRMED + + +def test_untouched_neighbours_confirm() -> None: + verdict = _judge(_effect(forbid_collateral_change=True), [dict(EXISTING), TARGET]) + assert verdict.verdict is Verdict.CONFIRMED + + +def test_a_vanished_neighbour_is_still_collateral_loss() -> None: + verdict = _judge(_effect(forbid_collateral_change=True), [TARGET]) + assert verdict.verdict is Verdict.REFUTED + assert "vanished" in verdict.reason + + +def test_the_target_itself_may_change() -> None: + # A pre-existing record inside the target selector is the record the + # action is allowed to write. + effect = Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"email": EXISTING["email"]}, + expected_count=1, + forbid_collateral_change=True, + ) + edited = {**EXISTING, "phone_home": "555-0199"} + assert _judge(effect, [edited]).verdict is Verdict.CONFIRMED + + +def test_flag_only_on_record_written() -> None: + with pytest.raises(ValueError, match="record_written"): + Effect( + kind=EffectKind.FIELD_EQUALS, + match={"email": TARGET["email"]}, + field="phone_home", + value="555-0131", + forbid_collateral_change=True, + ) + + +def test_existing_contract_digests_are_unchanged() -> None: + assert "forbid_collateral_change" not in _effect()._contract_payload() + assert ( + _effect().contract_hash() + == _effect(forbid_collateral_change=False).contract_hash() + ) + assert ( + _effect(forbid_collateral_change=True).contract_hash() + != _effect().contract_hash() + ) + + +def _yaml(base: str, effects: list[dict] | None = None, record: bool = False) -> dict: + doc: dict = { + "schema_version": "openadapt.verify/v1", + "verifier": {"kind": "rest", "base_url": base, "records_path": "/api/db"}, + } + if record: + doc["record"] = { + "name": "encounter", + "match": ["patient_id", "note"], + "fields": {"patient_id": "patient_id", "note": "note", "type": "type"}, + } + else: + doc["effects"] = effects or [ + { + "kind": "record_written", + "match": { + "patient_id": {"param": "patient_id"}, + "type": "Triage", + "note": {"param": "note"}, + }, + "expected_count": 1, + "count_new_only": True, + "timeout_s": 0.5, + } + ] + return doc + + +def test_verify_run_turns_the_check_on_when_it_judges() -> None: + from openadapt_flow.verify_run import _audit_effect + + params = {"patient_id": "p1", "note": "n", "type": "Triage"} + for doc in (_yaml("http://127.0.0.1:1"), _yaml("http://127.0.0.1:1", record=True)): + cfg = parse_verify_config(doc) + for effect in cfg.effects: + # The contract stays as the file states it (qualify's digest)... + assert effect.forbid_collateral_change is False + judged = _audit_effect(effect, effect.resolve(params)) + # ...and verify-run judges record_written with the change check. + assert judged.forbid_collateral_change is ( + effect.kind is EffectKind.RECORD_WRITTEN + ) + + +def test_an_effect_can_opt_out() -> None: + from openadapt_flow.verify_run import _audit_effect + + doc = _yaml("http://127.0.0.1:1") + doc["effects"][0]["forbid_collateral_change"] = False + effect = parse_verify_config(doc).effects[0] + params = {"patient_id": "p1", "note": "n", "type": "Triage"} + assert ( + _audit_effect(effect, effect.resolve(params)).forbid_collateral_change is False + ) + + +@pytest.fixture +def sor(): + url, db, stop = fault_serve() + try: + yield url.rstrip("/"), db + finally: + stop() + + +def test_verify_run_refuses_an_in_place_edit_of_another_record(sor, tmp_path: Path): + base, db = sor + params = {"patient_id": "p1", "note": "Follow-up in two weeks"} + # A concurrent clinician's earlier encounter on another patient. + db.add("p2", "Consult", "Allergy: penicillin") + cfg = parse_verify_config(_yaml(base)) + baseline = tmp_path / "baseline.json" + capture_baseline(cfg, params, baseline) + requests.post( + f"{base}/api/encounter", + json={"patient_id": "p1", "type": "Triage", "note": params["note"]}, + timeout=5, + ) + # The same save also rewrote the other record in place. + with db._lock: + for r in db._records: + if r["patient_id"] == "p2": + r["note"] = "Allergy: none" + result = verify_run( + cfg, + params=params, + run_record={"executor_kind": "agent", "executor_id": "test-agent"}, + authorization_ref="ticket-0000-collateral", + baseline_path=baseline, + private_dir=tmp_path / "out", + ) + receipt = result["receipt"] + assert receipt["effect_verdict"] == "REFUTED" + assert receipt["outcome"] != "verified" + assert result["evidence"]["readback"]["effects"][0]["collateral_change_checked"] From 9850b62747975d3b913efe44e77707a2b2b72f39 Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 15:01:36 -0400 Subject: [PATCH 2/4] verify-run collateral change: compare values in canonical form (a baseline's nested keys come back sorted) A rebuilt OpenEMR clean save was refuted: the baseline file stores records as canonical JSON, so the existing chart's nested addresses came back with their keys in another order and str() saw a change. Values are now compared as sorted-key JSON. Test: nested key order is not a change. Co-Authored-By: Claude Opus 5.5 --- .../openadapt_flow/runtime/effects/_common.py | 14 +++++++++++++- .../tests/test_verify_run_collateral_change.py | 18 ++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/packages/flow/openadapt_flow/runtime/effects/_common.py b/packages/flow/openadapt_flow/runtime/effects/_common.py index 94783edb0..72278304b 100644 --- a/packages/flow/openadapt_flow/runtime/effects/_common.py +++ b/packages/flow/openadapt_flow/runtime/effects/_common.py @@ -11,6 +11,7 @@ from __future__ import annotations +import json from typing import Any, Iterable, Optional from openadapt_flow.runtime.effects.effect import ( @@ -261,6 +262,17 @@ def _judge_exact_new_set( ) +def _same_value(value: Any) -> str: + """A value in a form where key order and number formatting don't matter. + + A baseline round-trips through canonical JSON (sorted keys), so a nested + record (an address list) can come back with its keys in another order; + that is not a change. + """ + + return json.dumps(value, sort_keys=True, default=str) + + def _judge_record_written( effect: Effect, before: EffectState, @@ -299,7 +311,7 @@ def _judge_record_written( fields = sorted( k for k in set(r) | set(after) - if str(r.get(k, "")) != str(after.get(k, "")) + if _same_value(r.get(k)) != _same_value(after.get(k)) ) if fields: collateral_changed.append((r, fields)) diff --git a/packages/flow/tests/test_verify_run_collateral_change.py b/packages/flow/tests/test_verify_run_collateral_change.py index e7670c377..dafb2c09d 100644 --- a/packages/flow/tests/test_verify_run_collateral_change.py +++ b/packages/flow/tests/test_verify_run_collateral_change.py @@ -72,6 +72,24 @@ def test_untouched_neighbours_confirm() -> None: assert verdict.verdict is Verdict.CONFIRMED +def test_nested_key_order_is_not_a_change() -> None: + # A baseline file stores records as canonical JSON (sorted keys); OpenEMR + # returns its nested addresses in another key order. + before = EffectState( + substrate="rest", + reachable=True, + records=[{**EXISTING, "addresses": [{"zip": "02139", "city": "EXAMPLETON"}]}], + ) + after = [ + {**EXISTING, "addresses": [{"city": "EXAMPLETON", "zip": "02139"}]}, + TARGET, + ] + verdict = judge_records( + _effect(forbid_collateral_change=True), before, after, substrate="rest" + ) + assert verdict.verdict is Verdict.CONFIRMED + + def test_a_vanished_neighbour_is_still_collateral_loss() -> None: verdict = _judge(_effect(forbid_collateral_change=True), [TARGET]) assert verdict.verdict is Verdict.REFUTED From 2a302c6d1309aa962802ab4cee9cc34a22e894bb Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 15:38:20 -0400 Subject: [PATCH 3/4] verify-run collateral change: compare driver values with their stored JSON form; say it may be another user's edit Review of #1375: SQL rows come back as raw driver values (Decimal, datetime, bytes) while verify-run's baseline went through JSON, so an unchanged row could read as changed. Field values now compare by type: numbers numerically when either side is a number, dates and times as parsed values (either ISO separator), bytes against their hex or UTF-8 text, nested values element by element; two strings still compare exactly ('02139' to '2139' is a change). The refutation now says the change may be another user's concurrent edit and that a person should reconcile it. Tests: an unchanged SQL row (Decimal, TIMESTAMP, BLOB) read against its own round-tripped baseline confirms, and a real edit to it refutes; pairwise driver-vs-JSON cases; numeric-looking text compares exactly; the reason. Co-Authored-By: Claude Opus 5.5 --- packages/flow/docs/VERIFY_RUN.md | 7 ++ .../openadapt_flow/runtime/effects/_common.py | 76 +++++++++++-- .../test_verify_run_collateral_change.py | 100 ++++++++++++++++++ 3 files changed, 174 insertions(+), 9 deletions(-) diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index e92d6bef8..698daae1a 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -93,6 +93,13 @@ table other people edit during the run, set `forbid_collateral_change: false` on that effect in `effects:`. Fields that change on their own, such as a modification time, belong in the verifier's `records_drop_fields`. +A changed record is not proof that the executor changed it: another user may +have edited it during the run. The verdict says so, and the outcome is +`reconciliation_required`, so a person checks the record before anything is +retried or undone. Values compare by type. A `Decimal` or `datetime` read +from a database equals the JSON form the baseline stored it in, while two +strings compare exactly, so `"02139"` changed to `"2139"` is a change. + ## Verdicts | What the store shows | `effect_verdict` | `outcome` | diff --git a/packages/flow/openadapt_flow/runtime/effects/_common.py b/packages/flow/openadapt_flow/runtime/effects/_common.py index 72278304b..212477c07 100644 --- a/packages/flow/openadapt_flow/runtime/effects/_common.py +++ b/packages/flow/openadapt_flow/runtime/effects/_common.py @@ -11,7 +11,8 @@ from __future__ import annotations -import json +from datetime import date, datetime, time +from decimal import Decimal, InvalidOperation from typing import Any, Iterable, Optional from openadapt_flow.runtime.effects.effect import ( @@ -262,15 +263,70 @@ def _judge_exact_new_set( ) -def _same_value(value: Any) -> str: - """A value in a form where key order and number formatting don't matter. +_NUMBER = (int, float, Decimal) +_WHEN = (datetime, date, time) - A baseline round-trips through canonical JSON (sorted keys), so a nested - record (an address list) can come back with its keys in another order; - that is not a change. + +def _as_number(value: Any) -> Optional[Decimal]: + try: + return Decimal(str(value)) + except (InvalidOperation, ValueError): + return None + + +def _as_when(value: Any) -> Optional[Any]: + if isinstance(value, _WHEN): + return value + if isinstance(value, str): + for parse in (datetime.fromisoformat, date.fromisoformat, time.fromisoformat): + try: + return parse(value) + except ValueError: + continue + return None + + +def _same_value(before: Any, after: Any) -> bool: + """Whether a record field is unchanged between the baseline and now. + + The baseline may have been written to JSON and read back (verify-run's + baseline file), while the current read has the driver's own types. So a + ``Decimal('1.50')`` read now equals the ``"1.50"`` or ``1.5`` it was + stored as, a ``datetime`` equals its ISO string (either separator), and + ``bytes`` equal their hex or UTF-8 text. Nested values compare element by + element, so key order doesn't matter. Two strings compare exactly: a + text field that changed from ``"02139"`` to ``"2139"`` is a change. """ - return json.dumps(value, sort_keys=True, default=str) + if isinstance(before, dict) and isinstance(after, dict): + return set(before) == set(after) and all( + _same_value(before[k], after[k]) for k in before + ) + if isinstance(before, (list, tuple)) and isinstance(after, (list, tuple)): + return len(before) == len(after) and all( + _same_value(b, a) for b, a in zip(before, after) + ) + if before is None or after is None: + return before is None and after is None + if isinstance(before, bool) or isinstance(after, bool): + return before == after + if isinstance(before, (bytes, bytearray)) or isinstance(after, (bytes, bytearray)): + raw, other = ( + (before, after) + if isinstance(before, (bytes, bytearray)) + else (after, before) + ) + if isinstance(other, (bytes, bytearray)): + return bytes(raw) == bytes(other) + text = str(other) + return text == bytes(raw).hex() or text == bytes(raw).decode("utf-8", "replace") + if isinstance(before, _NUMBER) or isinstance(after, _NUMBER): + b, a = _as_number(before), _as_number(after) + return b is not None and a is not None and b == a + if isinstance(before, _WHEN) or isinstance(after, _WHEN): + b, a = _as_when(before), _as_when(after) + return b is not None and a is not None and b == a + return before == after def _judge_record_written( @@ -311,7 +367,7 @@ def _judge_record_written( fields = sorted( k for k in set(r) | set(after) - if _same_value(r.get(k)) != _same_value(after.get(k)) + if not _same_value(r.get(k), after.get(k)) ) if fields: collateral_changed.append((r, fields)) @@ -362,7 +418,9 @@ def _judge_record_written( f"{len(collateral_changed)} pre-existing record(s) outside the " f"target changed -- collateral change (field(s): " f"{', '.join(names[:_SAMPLE_LIMIT])}" - f"{'' if len(names) <= _SAMPLE_LIMIT else ', ...'})" + f"{'' if len(names) <= _SAMPLE_LIMIT else ', ...'}); this may be " + f"another user's concurrent edit, not this action -- a person " + f"should reconcile it before anything is retried or undone" ) return EffectVerdict( verdict=Verdict.REFUTED, diff --git a/packages/flow/tests/test_verify_run_collateral_change.py b/packages/flow/tests/test_verify_run_collateral_change.py index dafb2c09d..0c3461036 100644 --- a/packages/flow/tests/test_verify_run_collateral_change.py +++ b/packages/flow/tests/test_verify_run_collateral_change.py @@ -227,3 +227,103 @@ def test_verify_run_refuses_an_in_place_edit_of_another_record(sor, tmp_path: Pa assert receipt["effect_verdict"] == "REFUTED" assert receipt["outcome"] != "verified" assert result["evidence"]["readback"]["effects"][0]["collateral_change_checked"] + + +# -- driver types against a baseline that went through JSON -------------------- + + +def test_unchanged_sql_row_against_its_round_tripped_baseline(tmp_path: Path): + import sqlite3 + from decimal import Decimal + + from openadapt_flow.runtime.effects import SqlRecordVerifier + + sqlite3.register_adapter(Decimal, str) + sqlite3.register_converter("DECIMAL", lambda b: Decimal(b.decode())) + path = tmp_path / "sor.db" + conn = sqlite3.connect(path) + conn.execute( + "CREATE TABLE claims (id INTEGER PRIMARY KEY, claim_no TEXT, " + "amount DECIMAL, entered TIMESTAMP, scan BLOB, zip TEXT)" + ) + conn.execute( + "INSERT INTO claims (claim_no, amount, entered, scan, zip) VALUES " + "('C-1', ?, '2026-10-19 08:30:00', ?, '02139')", + (Decimal("1.50"), b"scan-1"), + ) + conn.commit() + + def connect() -> sqlite3.Connection: + return sqlite3.connect(path, detect_types=sqlite3.PARSE_DECLTYPES) + + verifier = SqlRecordVerifier( + connect, + "SELECT id, claim_no, amount, entered, scan, zip FROM claims", + poll_interval_s=0.01, + ) + live = verifier.capture_pre_state() + assert isinstance(live.records[0]["amount"], Decimal) + # verify-run's baseline file: JSON out, JSON back in. + before = EffectState.model_validate(live.model_dump(mode="json")) + conn.execute( + "INSERT INTO claims (claim_no, amount, entered, scan, zip) VALUES " + "('C-2', ?, '2026-10-19 09:00:00', ?, '02139')", + (Decimal("20.00"), b"new"), + ) + conn.commit() + effect = Effect( + kind=EffectKind.RECORD_WRITTEN, + match={"claim_no": "C-2"}, + expected_count=1, + count_new_only=True, + forbid_collateral_change=True, + timeout_s=0.05, + ) + verdict = verifier.verify(effect, before) + assert verdict.verdict is Verdict.CONFIRMED, verdict.reason + + # The same read does catch a real change to the old row. + conn.execute("UPDATE claims SET zip = '2139' WHERE claim_no = 'C-1'") + conn.commit() + verdict = verifier.verify(effect, before) + assert verdict.verdict is Verdict.REFUTED + assert "zip" in verdict.reason + conn.close() + + +@pytest.mark.parametrize( + ("stored", "read"), + [ + (1.5, "Decimal"), + ("1.50", "Decimal"), + ("2026-10-19T08:30:00", "datetime"), + ("2026-10-19 08:30:00", "datetime"), + ("00ff", "bytes"), + ], +) +def test_driver_values_equal_their_json_form(stored, read) -> None: + from datetime import datetime + from decimal import Decimal + + from openadapt_flow.runtime.effects._common import _same_value + + value = { + "Decimal": Decimal("1.50"), + "datetime": datetime(2026, 10, 19, 8, 30), + "bytes": b"\x00\xff", + }[read] + assert _same_value(stored, value) + + +def test_text_that_looks_numeric_still_compares_exactly() -> None: + from openadapt_flow.runtime.effects._common import _same_value + + assert not _same_value("02139", "2139") + assert not _same_value("1.50", "1.5") + + +def test_the_reason_says_it_may_be_someone_else() -> None: + edited = {**EXISTING, "phone_home": "555-0199"} + verdict = _judge(_effect(forbid_collateral_change=True), [edited, TARGET]) + assert "another user's concurrent edit" in verdict.reason + assert "a person should reconcile" in verdict.reason From a1f06b18479a0f60e001a00e4dd8e35af4407af8 Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 19:24:59 -0400 Subject: [PATCH 4/4] bundle validation: omit forbid_collateral_change=False from sealed v2 content The new Effect field entered every serialized workflow.json, so bundles sealed before it no longer matched their content digest (registry pull, loop showcase, and the pinned pre-field digest test failed in CI). It is now a reviewed sealed-canonical omission when False, like new_records and identity_field; a contract that turns the check on stays sealed content. Co-Authored-By: Claude Opus 5.5 --- packages/flow/openadapt_flow/bundle_validation.py | 10 ++++++++++ packages/flow/tests/test_bundle_schema_v2.py | 3 +++ 2 files changed, 13 insertions(+) diff --git a/packages/flow/openadapt_flow/bundle_validation.py b/packages/flow/openadapt_flow/bundle_validation.py index 3387d3645..7bbdc5ad6 100644 --- a/packages/flow/openadapt_flow/bundle_validation.py +++ b/packages/flow/openadapt_flow/bundle_validation.py @@ -277,6 +277,16 @@ class _SealedCanonicalOmission: "the whole over-write contract and remains sealed content" ), ), + _SealedCanonicalOmission( + owner=Effect, + field_name="forbid_collateral_change", + omitted_values=(False,), + reason=( + "an effect that does not check unchanged non-target records is " + "what every earlier bundle held; a contract that turns the " + "collateral-change check on remains sealed content" + ), + ), _SealedCanonicalOmission( owner=Effect, field_name="identity_field", diff --git a/packages/flow/tests/test_bundle_schema_v2.py b/packages/flow/tests/test_bundle_schema_v2.py index f2b29a689..bd47f2235 100644 --- a/packages/flow/tests/test_bundle_schema_v2.py +++ b/packages/flow/tests/test_bundle_schema_v2.py @@ -208,6 +208,9 @@ def _pre_frame_path_content(wf: Workflow, shape: str) -> dict: # key name -- see the note above. assert effect.pop("new_records") == [] assert effect.pop("identity_field") == "id" + # Reviewed v2 omission: the collateral-change check postdates these + # seals (#1375). + assert effect.pop("forbid_collateral_change") is False return content