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..698daae1a 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -78,6 +78,28 @@ 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`. + +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` | @@ -85,7 +107,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/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/openadapt_flow/runtime/effects/_common.py b/packages/flow/openadapt_flow/runtime/effects/_common.py index ef4c8adee..212477c07 100644 --- a/packages/flow/openadapt_flow/runtime/effects/_common.py +++ b/packages/flow/openadapt_flow/runtime/effects/_common.py @@ -11,6 +11,8 @@ from __future__ import annotations +from datetime import date, datetime, time +from decimal import Decimal, InvalidOperation from typing import Any, Iterable, Optional from openadapt_flow.runtime.effects.effect import ( @@ -261,6 +263,72 @@ def _judge_exact_new_set( ) +_NUMBER = (int, float, Decimal) +_WHEN = (datetime, date, time) + + +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. + """ + + 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( effect: Effect, before: EffectState, @@ -285,7 +353,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 not _same_value(r.get(k), 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 +412,16 @@ 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 ', ...'}); 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, 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_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 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..0c3461036 --- /dev/null +++ b/packages/flow/tests/test_verify_run_collateral_change.py @@ -0,0 +1,329 @@ +"""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_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 + 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"] + + +# -- 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