diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 0e164b7bea..ebe7fec5ec 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -149,6 +149,7 @@ improvising or retrying when something is off. | Check that the saved record matches its source document, not only the parameters: intake keeps each value's line, a second reading must agree before a guarded save, and a round trip compares the read-back with the document | On main | Synthetic (MockMed, text referral) | `openadapt_flow/provenance.py`, `intake`, `verify-run --provenance`, guarded save | | Check one write from Python, without the CLI: one API (`openadapt_flow.verifier`: `begin_write`, `verify_write`; 2.4.1's `capture` form still works, deprecated) with one outcome (`CONFIRMED`, `REFUTED`, `UNCERTAIN`), the receipt, which parts were checked, and the result's scope (read path, stores read, time window, what it doesn't cover) | On main | Synthetic (SQLite; MockMed's fault server over real HTTP) | `openadapt_flow/verifier.py`, [VERIFIER_API.md](packages/flow/docs/VERIFIER_API.md) | | Never send a write twice while its outcome is unknown: an append-only ledger on local disk (pluggable backend), keyed by the write's identity, that every adapter, the MCP tools and the CLI consult before sending; a read-back against the attempt's own baseline resolves it, and a refuted write waits for a person (`openadapt verify ledger list`, `reconcile`, `release`) | On main | Synthetic (MockMed's fault server: a worker killed between the send and the read-back, killed during a timed-out send, killed before the send, a late write, and two worker processes started together on one write) | `openadapt_flow/write_ledger.py`, [VERIFIER_API.md](packages/flow/docs/VERIFIER_API.md#the-uncertain-write-ledger) | +| Catch a write that lands after the verdict (a background job, a trigger, a delayed side record): the receipt says when a recheck is due, and the recheck reads the whole database again and refutes the earlier result when a change since the post-save read has no explanation (`frame.database.recheck_after_s`, `verifier.recheck`) | On main ([#1639](https://github.com/OpenAdaptAI/OpenAdapt/pull/1639)) | Synthetic (SQLite); opt-in live OpenEMR test | `openadapt_flow/verifier.py`, `runtime/effects/late_write.py`, [VERIFY_RUN.md](packages/flow/docs/VERIFY_RUN.md#writes-that-land-later) | | Guarded save: a safety layer any agent wraps around one write (check the record, hold for approval, read back) | On main ([#1258](https://github.com/OpenAdaptAI/OpenAdapt/pull/1258)) | Synthetic | `GUARDED_SAVE.md` | | A person's approval in Desktop is signed with a key kept behind Touch ID on a Mac: one-prompt setup in the app, a prompt for every answer, and Flow never sees the key | On main ([#1500](https://github.com/OpenAdaptAI/OpenAdapt/pull/1500)) | Unit tests; opt-in test on a real Secure Enclave | `packages/desktop/openadapt_desktop/approver_keychain.py`, [APPROVAL_MODES.md](packages/flow/docs/APPROVAL_MODES.md#the-approver-key) | | The approver root Desktop trusts can be pinned by an administrator, in a file this user can't write (macOS, Linux, Windows); Desktop checks the owner, mode or ACL and refuses links. A Mac without one uses the root it set up behind Touch ID, read with Keychain prompts off; the two disagreeing is refused | On main | Unit tests per attack on real files; Windows ACL rules unit-tested, real-ACL test on Windows CI; opt-in test against the real login Keychain | `openadapt-desktop approver-pin`, `packages/desktop/openadapt_desktop/approver_anchor.py`, [APPROVAL_MODES.md](packages/flow/docs/APPROVAL_MODES.md#the-approver-key) | diff --git a/llms-full.txt b/llms-full.txt index 63753acae1..f6407df0d2 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -546,6 +546,7 @@ improvising or retrying when something is off. | Check that the saved record matches its source document, not only the parameters: intake keeps each value's line, a second reading must agree before a guarded save, and a round trip compares the read-back with the document | On main | Synthetic (MockMed, text referral) | `openadapt_flow/provenance.py`, `intake`, `verify-run --provenance`, guarded save | | Check one write from Python, without the CLI: one API (`openadapt_flow.verifier`: `begin_write`, `verify_write`; 2.4.1's `capture` form still works, deprecated) with one outcome (`CONFIRMED`, `REFUTED`, `UNCERTAIN`), the receipt, which parts were checked, and the result's scope (read path, stores read, time window, what it doesn't cover) | On main | Synthetic (SQLite; MockMed's fault server over real HTTP) | `openadapt_flow/verifier.py`, [VERIFIER_API.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/VERIFIER_API.md) | | Never send a write twice while its outcome is unknown: an append-only ledger on local disk (pluggable backend), keyed by the write's identity, that every adapter, the MCP tools and the CLI consult before sending; a read-back against the attempt's own baseline resolves it, and a refuted write waits for a person (`openadapt verify ledger list`, `reconcile`, `release`) | On main | Synthetic (MockMed's fault server: a worker killed between the send and the read-back, killed during a timed-out send, killed before the send, a late write, and two worker processes started together on one write) | `openadapt_flow/write_ledger.py`, [VERIFIER_API.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/VERIFIER_API.md#the-uncertain-write-ledger) | +| Catch a write that lands after the verdict (a background job, a trigger, a delayed side record): the receipt says when a recheck is due, and the recheck reads the whole database again and refutes the earlier result when a change since the post-save read has no explanation (`frame.database.recheck_after_s`, `verifier.recheck`) | On main ([#1639](https://github.com/OpenAdaptAI/OpenAdapt/pull/1639)) | Synthetic (SQLite); opt-in live OpenEMR test | `openadapt_flow/verifier.py`, `runtime/effects/late_write.py`, [VERIFY_RUN.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/VERIFY_RUN.md#writes-that-land-later) | | Guarded save: a safety layer any agent wraps around one write (check the record, hold for approval, read back) | On main ([#1258](https://github.com/OpenAdaptAI/OpenAdapt/pull/1258)) | Synthetic | `GUARDED_SAVE.md` | | A person's approval in Desktop is signed with a key kept behind Touch ID on a Mac: one-prompt setup in the app, a prompt for every answer, and Flow never sees the key | On main ([#1500](https://github.com/OpenAdaptAI/OpenAdapt/pull/1500)) | Unit tests; opt-in test on a real Secure Enclave | `packages/desktop/openadapt_desktop/approver_keychain.py`, [APPROVAL_MODES.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/APPROVAL_MODES.md#the-approver-key) | | The approver root Desktop trusts can be pinned by an administrator, in a file this user can't write (macOS, Linux, Windows); Desktop checks the owner, mode or ACL and refuses links. A Mac without one uses the root it set up behind Touch ID, read with Keychain prompts off; the two disagreeing is refused | On main | Unit tests per attack on real files; Windows ACL rules unit-tested, real-ACL test on Windows CI; opt-in test against the real login Keychain | `openadapt-desktop approver-pin`, `packages/desktop/openadapt_desktop/approver_anchor.py`, [APPROVAL_MODES.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/APPROVAL_MODES.md#the-approver-key) | diff --git a/packages/flow/CHANGELOG.md b/packages/flow/CHANGELOG.md index 549c1a1caa..c0c7a8b828 100644 --- a/packages/flow/CHANGELOG.md +++ b/packages/flow/CHANGELOG.md @@ -15,6 +15,12 @@ +## Unreleased + +### Features + +- A write that lands after the verdict can now be caught. With `frame.database.recheck_after_s`, a verified write's receipt says when a recheck is due (`recheck_due_at`), and `openadapt_flow.verifier.recheck` reads the whole database again at or after that time. It judges every change since the post-save read with the frame's rules: the app profile's bookkeeping, other users' writes the audit log attributes, and changes a later write by the same actor under the same contract saw in its own window (`excused_by`, any outcome). Anything else turns the earlier result into `REFUTED` (late). A change that may be what refuted a later write, a database that can't be read, or a saved state that wasn't issued here or doesn't match its receipt gives `UNCERTAIN`. The saved state holds keyed hashes, not values. `frame.database.settle_attempts` lets the reads after a save, and at a recheck, retry within a stated budget until two in a row match; rows added to append-only tables between them don't count as movement. Without these settings, nothing changes ([#1639](https://github.com/OpenAdaptAI/OpenAdapt/pull/1639)). + ## v2.4.1 (2026-10-28) Flow runs without its benchmark package and accepts a pinned config with Windows line endings. Several checks now halt or refute where 2.4.0 could act on the wrong row or pass a weaker check, and the frame can compare the whole database. diff --git a/packages/flow/docs/VERIFIER_API.md b/packages/flow/docs/VERIFIER_API.md index 3ed6217282..d3278e89be 100644 --- a/packages/flow/docs/VERIFIER_API.md +++ b/packages/flow/docs/VERIFIER_API.md @@ -111,7 +111,10 @@ The read path comes from each verifier's substrate: `sql` is `database`; `not_covered` always lists: -- writes after `read_at` (delayed or queued writes); +- writes after `read_at` (delayed or queued writes). When the contract + sets `frame.database.recheck_after_s` and the result is `CONFIRMED`, this + line names the recheck's due time instead: the result is provisional + until the recheck runs; - stores the contract doesn't read, by name when known: email, other services, files; - for a contract with no `frame.database`, tables outside the read set. @@ -119,6 +122,28 @@ The read path comes from each verifier's substrate: `sql` is `database`; `result.to_dict()` is the JSON reply for agents and the CLI. It holds closed fields only: no record values. +## Writes that land after the result + +With `frame.database.recheck_after_s`, a `CONFIRMED` result has +`recheck_due_at` (also in the receipt), and `result.recheck` holds the private +state the recheck compares with. `verify_write` saves it as `recheck.json` in +the write's folder, beside `baseline.json` and `effect-receipt.json` (mode +0600, keyed hashes only), and records it as issued under the verifier's state +directory. + +```python +late = recheck(contract, result, params=params, actor=actor, + excused_by=[later_result, ...]) # later writes, any outcome +late.outcome # CONFIRMED, REFUTED (late) or UNCERTAIN +``` + +The caller can schedule it from `recheck_due_at` and keep working: before +then, `recheck` raises `RecheckNotDue` (`wait=True` sleeps instead). It reads +the whole database again and judges every change since the post-save read. +A later write passed in `excused_by` accounts only for what changed in its +own window. A change that may be what refuted that later write gives +`UNCERTAIN`. See [VERIFY_RUN.md](VERIFY_RUN.md#writes-that-land-later). + ## Before the write `begin_write` raises a subclass of `WriteBlocked` when the write must not be diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index ec0791ead6..110af6436f 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -314,6 +314,104 @@ Without `--fault-gate` it writes the coverage of its recorded-state test, which can't test a wrong value written over the right record or a late write, so that record is incomplete and refused by default. +## Writes that land later + +With `frame.database`, the database is read twice after the save, +`frame.database.settle_s` seconds apart (default 1). When the two reads +differ, the frame is not checked. A write the app makes later than the +second read, such as a background job, a trigger-driven integration or a +delayed duplicate side record, lands after the verdict, and that check +doesn't see it. + +A live app moves on its own: a screen that polls, a scheduler tick. Set +`frame.database.settle_attempts` (default 1) to take up to that many more +reads, each `settle_s` after the last, until two in a row match. Two reads +also match when they differ only by rows added to append-only tables, which +the frame excuses anyway. Any other difference, including an edited row of +an append-only table, keeps it waiting. The earlier read of the matching +pair is judged. The budget is `settle_attempts * settle_s` seconds; when no +pair matches within it, the frame is not checked. +`evidence.frame.database.settle_attempts` says how many reads it took. The +same budget applies after the save and at a recheck. + +Set `frame.database.recheck_after_s` to read the database once more after a +verified save: + +```yaml +frame: + database: + dialect: mysql + read: {...} + settle_s: 1 + recheck_after_s: 300 # pick it for the app's own jobs; there is no default +``` + +When it is set and the run is verified: + +- The receipt has `recheck_due_at` (whole-second UTC): `recheck_after_s` + after the second read, never early (the read time is truncated to the + second, so one second is added before rounding up). A caller can schedule + the recheck without waiting on it. +- A verified receipt that carries `recheck_due_at` is provisional until the + recheck runs: a write that lands in the window can still refute it. +- `evidence.frame.recheck` says what the recheck compares + (`database_since_post_save`), when the post-save read was taken, and a + salted digest of the state it compares with. +- The caller keeps that state privately (`WriteResult.recheck`). Through + `begin_write` and `verify_write`, it is saved as `recheck.json` in the + write's folder, beside `baseline.json` (mode 0600); `RecheckState.save` + writes it elsewhere. It holds the run's baseline and its + post-save snapshot as keyed MACs (no values, as the baseline does), the + audit entries seen then as keyed hashes, what the run's frame reported as + collateral (row key MACs and entry hashes), and the run's salt. +- `verify_write` records each state it issues under the verifier's state + directory (`OPENADAPT_VERIFIER_STATE_DIR`, in `rechecks/`, named by the + state's digest). A state's salt and digests can be recomputed by whoever + can write the file, so a state without that record is never used: as the + write being rechecked it gives `UNCERTAIN`, and as an excuse it is + refused. As with snapshots, the writer being judged must not be able to + write the state directory. + +`openadapt_flow.verifier.recheck` reads the whole database at or after the +due time (twice, `settle_s` apart) and judges every change since the +post-save read with the frame's own rules: the profile's bookkeeping, and a +write that the audit log attributes to another actor. + +Later writes under the same contract by the same actor can be passed as +`excused_by`, with any outcome, so the caller can keep working on the next +item while the window runs. A later write accounts only for its own window, +the time between its baseline and its post-save read, where its own frame +judged every change: + +- A row that changed in that window, and still holds what that write's + post-save read saw, is the later write's. A row that matches its + selector but lands after its window (a delayed duplicate, or a save that + landed late), or a row changed again since, is judged like any other. +- A removal counts when that window recorded it, matched by the row's link + MAC: a per-host keyed hash of the row key, kept only when + `recheck_after_s` is set. +- An audit entry logged in that window is the later write's. +- When the later write was not verified and the change is one its frame + reported as collateral, or its delta failed, the change may be what + refuted it, and the harm may be the earlier write's. It doesn't refute + the earlier write, but the recheck gives `UNCERTAIN`: reconcile both. + +| Since the post-save read | Recheck `outcome` | +|---|---| +| Nothing moved that these rules don't explain | `CONFIRMED`: the verdict stands | +| A row moved that nothing explains | `REFUTED`: the earlier `verified` no longer stands (late) | +| A change lies in a later write's window and may be what refuted it | `UNCERTAIN` | +| The database couldn't be read or was still moving, or the state wasn't issued here or doesn't match its receipt, the contract, the actor or the params | `UNCERTAIN`: never a pass | + +A recheck before the due time raises `RecheckNotDue`; `wait=True` sleeps +until then instead. Limits: + +- A row of a table without a primary key (its rows are keyed by a content + fingerprint) is never excused by a later write's window. +- A later write under another contract doesn't excuse anything. +- A later write whose database never settled after its save has no state, + so it can't be passed as `excused_by`. Reconcile it first. + ## Verdicts | What the store shows | `effect_verdict` | `outcome` | @@ -354,7 +452,8 @@ and still checks. every field of the Execute receipt (`openadapt.execute-evidence-receipt/v1`) plus `executor_kind`, `executor_ref_sha256`, `effect_verdict`, and `verification_tier`, and `checked_how` (and `remediates` and -`correlation_id` when given), +`correlation_id` when given, and `recheck_due_at` when a verified run +schedules a recheck), under the schema `openadapt.effect-receipt/v1`. `evidence.readback` lists each effect's contract hash, verdict, and counts. `receipt.evidence_digest` is the SHA-256 of `evidence`. Neither part holds record values, URLs, or the diff --git a/packages/flow/openadapt_flow/runtime/effects/db_snapshot.py b/packages/flow/openadapt_flow/runtime/effects/db_snapshot.py index 9860626f05..5147f3563d 100644 --- a/packages/flow/openadapt_flow/runtime/effects/db_snapshot.py +++ b/packages/flow/openadapt_flow/runtime/effects/db_snapshot.py @@ -145,6 +145,17 @@ def _mac(key: bytes, text: str) -> str: ).hexdigest() +def link_key() -> bytes: + """The host's row-link key: one per host, never per run, derived from + the host secret (not in any snapshot). A row's link MAC + (:attr:`Table.links`) is the same in every snapshot taken on this host, + which shows that two snapshots hold the same row key and nothing else.""" + + return hashlib.blake2b( + b"openadapt:row-link/v1", key=_host_secret()[:64], digest_size=32 + ).digest() + + def cell_hash(key: bytes, value: Any, table: str = "", column: str = "") -> str: """A cell's keyed MAC, bound to its table and column, so equal values in different places don't share one.""" @@ -209,6 +220,12 @@ class Table: values: dict[str, dict[str, Any]] = field(default_factory=dict, repr=False) #: The engine keeps no snapshot isolation for it (MyISAM, MEMORY). non_transactional: bool = False + #: Row key MAC -> the row key's link MAC (:func:`link_key`), for a table + #: with a primary key, when the read asked for links (a late-write + #: recheck). One row has one link MAC in every snapshot on this host, so + #: two runs' snapshots can tell they hold the same row (a removal one of + #: them saw) without values. Empty otherwise. + links: dict[str, str] = field(default_factory=dict) @dataclass @@ -248,6 +265,7 @@ def to_json(self) -> dict[str, Any]: "cell_columns": t.cell_columns, "rows": t.rows, "non_transactional": t.non_transactional, + **({"links": t.links} if t.links else {}), } for name, t in self.tables.items() }, @@ -276,6 +294,7 @@ def from_json(cls, data: Mapping[str, Any]) -> "Snapshot": cell_columns=list(t.get("cell_columns", [])), rows={k: list(v) for k, v in (t.get("rows") or {}).items()}, non_transactional=bool(t.get("non_transactional")), + links={str(k): str(v) for k, v in (t.get("links") or {}).items()}, ) for name, t in data["tables"].items() }, @@ -595,6 +614,7 @@ def take_snapshot( ref_columns: Sequence[str] = (), schemas: Sequence[str] = (), optional_modes: Mapping[str, TableMode] = {}, + links: bool = False, ) -> Snapshot: """Read every table in one read-only transaction. @@ -616,6 +636,7 @@ def take_snapshot( optional_modes: Like ``modes``, for tables an app profile names that this database may not have (a profile covers several versions). ``ignore`` may name ``*`` for every table. + links: Keep each keyed row's link MAC (:attr:`Table.links`). Raises: SnapshotError: the database can't be read as the frame needs it. @@ -637,6 +658,7 @@ def take_snapshot( ref_columns, schemas, optional_modes, + link_key() if links else None, ) except SnapshotError: raise @@ -657,6 +679,7 @@ def _snapshot( ref_columns: Sequence[str], schemas: Sequence[str], optional_modes: Mapping[str, TableMode] = {}, + links: Optional[bytes] = None, ) -> Snapshot: dialect = s.dialect counters: dict[str, int] = {} @@ -761,7 +784,10 @@ def _snapshot( skip = [*ignore.get(t, ()), *ignore.get("*", ())] seen_keys: dict[str, int] = {} for row in _read_rows(s, t, d) if table.count else []: - k = row_key_hash(key, _row_key(d, row, seen_keys, skip), t) + text = _row_key(d, row, seen_keys, skip) + k = row_key_hash(key, text, t) + if links is not None and d.primary_key and mode == "compare": + table.links[k] = row_key_hash(links, text, t) mac = row_hash(key, t, row, skip) table.rows[k] = [ mac, @@ -860,6 +886,34 @@ def diff( return out +def settled_state(a: Snapshot, b: Snapshot) -> bool: + """Do two reads taken after the save show the same database, apart from + rows added to append-only tables (the app's logs, which the frame + excuses)? An append-only table's existing rows must not change, and + every other table must be the same, as in :func:`same_state`.""" + + if (a.objects, a.code_visible) != (b.objects, b.code_visible): + return False + if set(a.tables) != set(b.tables): + return False + for name, x in a.tables.items(): + y = b.tables[name] + if (x.mode, x.schema) != (y.mode, y.schema): + return False + if x.mode == "append_only": + if any(y.rows.get(k) != v for k, v in x.rows.items()): + return False + continue + if (x.count, x.counter, x.max_auto, x.rows) != ( + y.count, + y.counter, + y.max_auto, + y.rows, + ): + return False + return True + + def same_state(a: Snapshot, b: Snapshot) -> bool: """Do two reads taken after the save show the same database?""" diff --git a/packages/flow/openadapt_flow/runtime/effects/frame.py b/packages/flow/openadapt_flow/runtime/effects/frame.py index d46247a0e5..e08c67d5bf 100644 --- a/packages/flow/openadapt_flow/runtime/effects/frame.py +++ b/packages/flow/openadapt_flow/runtime/effects/frame.py @@ -70,7 +70,7 @@ from dataclasses import dataclass, field from typing import Any, Literal, Optional -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, model_serializer, model_validator from openadapt_flow.app_profiles import Owner, table_name from openadapt_flow.runtime.effects.effect import ( @@ -318,7 +318,41 @@ class FrameDatabase(BaseModel): #: Seconds between the two reads after the save. They must match: a #: write the app makes a moment later (a background job) shows as a #: difference, and the frame is not checked until the database settles. + #: A write later than the second read is not seen: see + #: ``recheck_after_s``. settle_s: float = Field(default=1.0, ge=0) + #: How many reads after the first one, each ``settle_s`` after the last, + #: may be taken to find two in a row that match, after the save and at a + #: recheck. A live app moves on its own (a screen polling, a scheduler + #: tick), so one pair often differs. Two reads match when they are the + #: same, or differ only by rows added to append-only tables (the app's + #: logs, which the frame excuses anyway). The rows judged are the earlier + #: read of the matching pair. The budget is ``settle_attempts * + #: settle_s`` seconds; when no pair matches within it, the frame is not + #: checked. Default 1: one pair, as before. + settle_attempts: int = Field(default=1, ge=1, le=60) + #: Seconds after the settled read at which the database is read once + #: more (``openadapt_flow.verifier.recheck``). A write that lands later + #: than ``settle_s`` (a background job, a trigger-driven integration, a + #: delayed side record) lands after the verdict; the recheck judges every + #: change since that read with the same rules, and an unexplained one + #: turns the verdict into REFUTED (late). Unset: no recheck is scheduled, + #: and a write that lands after the verdict goes unseen. No default: the + #: right wait depends on the app's own jobs, so the contract's author + #: picks it. + recheck_after_s: Optional[float] = Field(default=None, gt=0) + + @model_serializer(mode="wrap") + def _compact(self, handler: Any) -> dict[str, Any]: + # Unset (or at its default), a field added later is left out, so a + # contract that doesn't use it keeps the digests (``frame_sha256``, + # the check digest) it had. + data: dict[str, Any] = handler(self) + if data.get("recheck_after_s") is None: + data.pop("recheck_after_s", None) + if data.get("settle_attempts") == 1: + data.pop("settle_attempts", None) + return data @model_validator(mode="after") def _pinned_profile(self) -> "FrameDatabase": @@ -537,6 +571,8 @@ class FrameChange: other_on_target: bool = False #: In memory only, for the operator's console; never put in a receipt. key: Any = None + #: In memory only: the audit entry behind an ``audit_only`` change. + entry: Any = None @dataclass @@ -555,6 +591,12 @@ class FrameVerdict: added_by_others: set[tuple[str, str]] = field(default_factory=set) #: In memory only: app profile rules that excused a row change. excused_by: set[str] = field(default_factory=set) + #: In memory only, set when the frame held and ``recheck_after_s`` is + #: set: what a later recheck compares with (the settled post-save + #: snapshot as keyed MACs, the audit entries seen then as keyed hashes, + #: and when it was read). Never put in a receipt; only its salted digest + #: is. + post_state: Optional[dict[str, Any]] = None @property def held(self) -> bool: @@ -1683,7 +1725,9 @@ def names_moved(entry_ref: Any) -> bool: continue logged.add(str(ref)) collateral.append( - FrameChange("", "changed", "self", audit_only=True, key=ref) + FrameChange( + "", "changed", "self", audit_only=True, key=ref, entry=entry + ) ) verdict = FrameVerdict( diff --git a/packages/flow/openadapt_flow/runtime/effects/late_write.py b/packages/flow/openadapt_flow/runtime/effects/late_write.py new file mode 100644 index 0000000000..8b749879dc --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/effects/late_write.py @@ -0,0 +1,304 @@ +"""Late writes: judge the database again some time after a verified save. + +The whole-database frame (``frame.database``) reads the database twice after +the save, ``settle_s`` apart, and judges what moved. A write that lands later +(a background job, a trigger-driven integration, a delayed duplicate side +record) lands after that verdict, and nothing sees it. With +``frame.database.recheck_after_s`` set, a VERIFIED write also schedules a +recheck: + +- the receipt says when it is due (``recheck_due_at``), and its evidence + says what it compares and binds a salted digest of the state it compares + with (``evidence.frame.recheck``), so a caller can schedule it without + blocking; +- the caller keeps a private recheck state: the settled post-save snapshot + (keyed MACs, never values, as the baseline keeps them since #1599), the + audit entries seen then (keyed hashes), the salt, the receipt and its + evidence; +- at or after the due time the database is read again (twice, ``settle_s`` + apart, as after the save) and every change since the post-save read is + judged by the frame's own rules: the profile's bookkeeping, another actor's + write the audit log attributes, and what later writes under the same + contract by the same actor account for (``excused_by``, any outcome): + their declared effects and side records, and, where their own frame was + judged, every row and audit entry in their own window, which they already + explained (VERIFIED) or reported against themselves (REFUTED). Anything + else unexplained is collateral, and the earlier verdict becomes REFUTED + (late). + +This module holds the parts that need no read: the state's shape, the due +time and the outcome vocabulary. ``verify_run.recheck_run`` does the read and +the judgment, and ``openadapt_flow.verifier.recheck`` is the public entry. + +Fail-closed: a state that is malformed, doesn't match its receipt, or was +taken under another contract, and a database that can't be read or is still +moving, give UNCERTAIN, never CONFIRMED. +""" + +from __future__ import annotations + +import math +import re +from collections.abc import Mapping +from datetime import datetime, timedelta, timezone +from typing import Any, Literal, Optional + +#: The private recheck state's schema. +RECHECK_SCHEMA = "openadapt.verify-recheck/v1" +#: Exactly these keys: nothing rides along unbound. +STATE_KEYS: tuple[str, ...] = ( + "schema_version", + "execution_id", + "salt", + "verifier_identity", + "baseline", + "receipt", + "evidence", + "post_state", +) +#: The key under ``evidence.frame`` that says a recheck is due. +EVIDENCE_KEY = "recheck" +#: What the recheck compares: the whole database now against the settled +#: post-save read. +COMPARES = "database_since_post_save" +#: The salted-digest label of the post-save state. +POST_STATE_LABEL = "recheck_post_state" +#: The rule name a change gets when a later write's own window covers it. +LATER_WRITE_RULE = "later_write_window" +#: The same, when the change may be what refuted that later write. +LATER_WRITE_DISPUTED = "later_write_window_disputed" + +#: ``CONFIRMED``: nothing unexplained moved since the post-save read. +#: ``REFUTED``: something did (late collateral). ``UNCERTAIN``: the database, +#: or the state, couldn't be read or doesn't match. +LateOutcome = Literal["CONFIRMED", "REFUTED", "UNCERTAIN"] + +_ISO_SECONDS = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$") + + +def parse_time(value: str) -> datetime: + """Whole-second UTC (``2026-10-29T12:00:00Z``).""" + + if not isinstance(value, str) or not _ISO_SECONDS.match(value): + raise ValueError(f"{value!r} is not whole-second UTC") + return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + + +def format_time(when: datetime) -> str: + return when.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def due_at(read_at: str, after_s: float) -> str: + """When the recheck is due: ``after_s`` after the post-save read, never + early. ``read_at`` is truncated to the second, so the read happened up + to a second after it: one second is added before rounding up.""" + + return format_time(parse_time(read_at) + timedelta(seconds=math.ceil(after_s + 1))) + + +def seconds_until(due: str, now: Optional[datetime] = None) -> float: + """Seconds until ``due`` (0 or less when it is due).""" + + current = now or datetime.now(timezone.utc) + return (parse_time(due) - current).total_seconds() + + +def schedule( + post_state: Mapping[str, Any], + after_s: float, + post_state_sha256: str, + *, + due: bool, +) -> dict[str, Any]: + """The ``evidence.frame.recheck`` block: closed fields only. ``due``: the + write was verified, so a recheck is due (``due_at``); otherwise + ``due_at`` is None and the block only binds the post-save state.""" + + read_at = str(post_state["read_at"]) + return { + "compares": COMPARES, + "after_s": after_s, + "post_save_read_at": read_at, + "due_at": due_at(read_at, after_s) if due else None, + "post_state_sha256": post_state_sha256, + } + + +def state_problem(payload: Any) -> Optional[str]: + """A plain sentence when ``payload`` is not a recheck state's shape.""" + + if not isinstance(payload, Mapping): + return "the recheck state is not a JSON object" + if payload.get("schema_version") != RECHECK_SCHEMA: + return f"the recheck state is not an {RECHECK_SCHEMA} document" + if set(payload) != set(STATE_KEYS): + return "the recheck state must hold exactly " + ", ".join(STATE_KEYS) + try: + salt = bytes.fromhex(str(payload["salt"])) + except ValueError: + return "the recheck state's salt is not hex" + if len(salt) != 32: + return "the recheck state's salt is not 32 bytes" + for name in ("baseline", "receipt", "evidence", "post_state"): + if not isinstance(payload[name], Mapping): + return f"the recheck state's {name} is not an object" + post = payload["post_state"] + if not isinstance(post.get("snapshot"), Mapping): + return "the recheck state holds no post-save snapshot" + try: + parse_time(str(post.get("read_at"))) + except ValueError: + return "the recheck state's post-save read time is not whole-second UTC" + audit = post.get("audit") + if audit is not None and not ( + isinstance(audit, Mapping) + and isinstance(audit.get("reachable"), bool) + and isinstance(audit.get("ids"), list) + ): + return "the recheck state's audit entries are malformed" + try: + from openadapt_flow.runtime.effects.db_snapshot import Snapshot + + Snapshot.from_json(post["snapshot"]) + except (KeyError, TypeError, ValueError, AttributeError): + return "the recheck state's post-save snapshot can't be read" + return None + + +def outcome_of(frame_status: str) -> LateOutcome: + """The frame's status since the post-save read, as the recheck's outcome.""" + + if frame_status == "held": + return "CONFIRMED" + if frame_status == "refuted": + return "REFUTED" + return "UNCERTAIN" + + +class LaterWindow: + """What one later write's own frame already judged: every row that + changed between its baseline and its settled post-save read, and every + audit entry logged in that window. + + A change there was judged by that write's frame: explained, if that + write was verified; otherwise it may be what refuted it. An earlier + write's recheck therefore treats a change in a later write's window as + that write's (``clean``) only when the later write was verified, or when + the change was neither its reported collateral nor possibly behind a + failed delta. Otherwise the change is ``disputed``: it doesn't refute the + earlier write, but it doesn't confirm it either, since the harm may be + the earlier write's. + + Only the window counts. A row that matches the later write's selector + but lands after its post-save read, or a row changed again since, is + judged like any other. A row counts while it still holds the value the + post-save read saw (matched by key and contents MAC under the later + write's own key); a removal counts when that write's window recorded it + (matched by the row's link MAC, ``db_snapshot.link_key``). A table + without a primary key (rows keyed by a content fingerprint) is never + excused this way. + """ + + def __init__( + self, + key: bytes, + baseline: Mapping[str, Any], + post: Mapping[str, Any], + *, + verified: bool, + delta_held: bool, + collateral_rows: set[tuple[str, str]], + collateral_entries: set[str], + unmapped: bool, + audit_before: Optional[set[str]] = None, + audit_after: Optional[set[str]] = None, + entry_id: Optional[Any] = None, + ) -> None: + from openadapt_flow.runtime.effects.db_snapshot import Snapshot + + self.key = key + self.baseline = Snapshot.from_json(baseline) + self.post = Snapshot.from_json(post) + self.verified = verified + self.delta_held = delta_held + self.collateral_rows = collateral_rows + self.collateral_entries = collateral_entries + self.unmapped = unmapped + self.audit_before = audit_before + self.audit_after = audit_after + self.entry_id = entry_id + self._base_links = { + name: {link: k for k, link in t.links.items()} + for name, t in self.baseline.tables.items() + } + self._post_links = { + name: set(t.links.values()) for name, t in self.post.tables.items() + } + + def _status(self, collateral: bool) -> str: + if self.verified: + return "clean" + if collateral or self.unmapped or not self.delta_held: + return "disputed" + return "clean" + + def row( + self, + table: str, + before: Optional[Mapping[str, Any]], + after: Optional[Mapping[str, Any]], + skip: list[str], + links: Mapping[str, str], + ) -> Optional[str]: + """``clean`` or ``disputed`` when this row reached its present state + (or its removal) inside the window; None when it didn't. + + ``before``/``after``: the row as the recheck's judge sees it (the + earlier write's post-save MACs with ``KEY``, the plain row now). + ``links``: the earlier post-save snapshot's link MACs of ``table``. + """ + + from openadapt_flow.runtime.effects.db_snapshot import ( + KEY, + ROW, + canonical, + row_hash, + row_key_hash, + ) + + now = self.post.tables.get(table) + if now is None or now.mode != "compare" or not now.primary_key: + return None # no key to look the row up by: judged as usual + old = self.baseline.tables.get(table) + if after is not None: + row = {c: v for c, v in after.items() if c not in (KEY, ROW)} + text = "|".join(canonical(row.get(c)) for c in now.primary_key) + k = row_key_hash(self.key, text, table) + stored = now.rows.get(k) + if stored is None or stored[0] != row_hash(self.key, table, row, skip): + return None # not what that write's post-save read saw + was = None if old is None else old.rows.get(k) + if was is not None and was[0] == stored[0]: + return None # unchanged in that window + else: + if before is None: + return None + link = links.get(str(before.get(KEY))) + found = self._base_links.get(table, {}).get(link) if link else None + if found is None or link in self._post_links.get(table, set()): + return None # that window didn't record this removal + k = found + return self._status((table, k) in self.collateral_rows) + + def entry(self, entry: Mapping[str, Any]) -> Optional[str]: + """``clean`` or ``disputed`` when this audit entry was logged inside + the window; None when it wasn't (or the log wasn't read there).""" + + if self.audit_before is None or self.audit_after is None: + return None + if self.entry_id is None: + return None + h = self.entry_id(entry, self.key) + if h not in self.audit_after or h in self.audit_before: + return None + return self._status(h in self.collateral_entries) diff --git a/packages/flow/openadapt_flow/verifier.py b/packages/flow/openadapt_flow/verifier.py index 652cda825d..0119f8bc57 100644 --- a/packages/flow/openadapt_flow/verifier.py +++ b/packages/flow/openadapt_flow/verifier.py @@ -24,6 +24,29 @@ raises a :class:`WriteBlocked` when the write must not be sent. :func:`verify_write` records each outcome there. A failed check never raises. No model calls. See ``docs/VERIFIER_API.md``. + +Writes that land later: the frame reads the database twice after the save, +``settle_s`` apart, so a write the app makes later than that (a background +job, a trigger-driven integration, a delayed side record) lands after the +verdict. With ``frame.database.recheck_after_s`` set, a confirmed result +carries a private :class:`RecheckState` (saved as ``recheck.json`` in the +write's folder) and says when the recheck is due +(``WriteResult.recheck_due_at``, the receipt's ``recheck_due_at``):: + + result = verify_write(pending) + ... # the caller works on the next item + late = recheck(contract, result, params=params, actor=actor, + excused_by=[next_result]) # later writes, any outcome + if late.outcome == "REFUTED": # changed after the verdict + print(late.reasons) + +:func:`recheck` reads the whole database again at or after the due time +and judges every change since the post-save read with the frame's rules. +Before the due time it raises :class:`RecheckNotDue` (``wait=True`` sleeps +until then instead). Its outcome is ``CONFIRMED``, ``REFUTED`` (late: the +earlier CONFIRMED no longer stands) or ``UNCERTAIN`` (the database or the +state couldn't be read or doesn't match: never a pass). A confirmed result +that carries ``recheck_due_at`` is provisional until the recheck runs. """ from __future__ import annotations @@ -46,6 +69,13 @@ from openadapt_flow.runtime.effects.db_snapshot import Snapshot as DatabaseSnapshot from openadapt_flow.runtime.effects.db_snapshot import TableDiff, diff, take_snapshot +from openadapt_flow.runtime.effects.late_write import ( + EVIDENCE_KEY, + LATER_WRITE_DISPUTED, + LateOutcome, + seconds_until, + state_problem, +) from openadapt_flow.verify_run import ( FRAME_DB_DETAIL, VerifyConfig, @@ -54,6 +84,9 @@ capture_baseline, parse_verify_config, receipt_sha256, + recheck_problem, + recheck_run, + recheck_state, verify_run, write_result, ) @@ -81,6 +114,9 @@ "MemoryLedger", "Outcome", "PendingWrite", + "RecheckNotDue", + "RecheckResult", + "RecheckState", "Scope", "Snapshot", "Source", @@ -94,6 +130,7 @@ "capture", "capture_database", "diff_database", + "recheck", "reconcile", "verified_write", "verify_write", @@ -522,6 +559,16 @@ class WriteResult: #: A person should look at the store: shown by ``openadapt verify ledger #: list`` until released, though it doesn't stop a retry being handed back. review: bool = False + #: When the contract sets ``frame.database.recheck_after_s`` and the + #: database settled after the save: the private state :func:`recheck` + #: needs (saved as ``recheck.json`` in the write's folder). A confirmed + #: write's is what its recheck compares with; any write's can excuse its + #: own window in an earlier write's recheck. + recheck: Optional["RecheckState"] = field(default=None, repr=False, compare=False) + #: When a confirmed write's recheck is due (whole-second UTC), as the + #: receipt's ``recheck_due_at`` says; None when no recheck is due. Until + #: the recheck runs, the result is provisional. + recheck_due_at: Optional[str] = None @property def verified(self) -> bool: @@ -570,9 +617,114 @@ def to_dict(self) -> dict[str, Any]: "seal_path": self.seal_path, "landed": self.landed, "review": self.review, + "recheck_due_at": self.recheck_due_at, } +@dataclass +class RecheckState: + """What a recheck compares with: the settled post-save snapshot as keyed + MACs (no values), the audit entries seen then as keyed hashes, the run's + salt, and the receipt and evidence it is bound to. + + Private, like a :class:`Snapshot`: it holds the salt, so :meth:`save` + writes it with mode 0600, and the writer being judged must not be able + to write it. Its salt and digests can be recomputed by whoever writes + the file, so :func:`verify_write` records each state it issues under + the state directory (``rechecks/``, by digest), and :func:`recheck` uses + only a state with that record. ``params`` (the write's inputs) stay in memory only; a + loaded state needs them back (:meth:`load`) to excuse a write + (``excused_by``). + """ + + payload: dict[str, Any] = field(repr=False) + params: Optional[Mapping[str, object]] = field(default=None, repr=False) + path: Optional[Path] = field(default=None, repr=False) + + @property + def due_at(self) -> str: + """When the recheck is due (whole-second UTC).""" + + frame = (self.payload.get("evidence") or {}).get("frame") or {} + return str((frame.get(EVIDENCE_KEY) or {}).get("due_at") or "") + + def save(self, path: str | Path) -> Path: + out = Path(path) + _write_private(out, self.payload) + return out + + @classmethod + def load( + cls, path: str | Path, *, params: Optional[Mapping[str, object]] = None + ) -> "RecheckState": + """Read a state :meth:`save` wrote. Raises :class:`VerifierError` + when it can't be read as one; :func:`recheck` checks the rest.""" + + source = Path(path) + try: + payload = json.loads(source.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise VerifierError(f"can't read the recheck state: {exc}") from exc + problem = state_problem(payload) + if problem is not None: + raise VerifierError(problem) + return cls(payload, params=params, path=source) + + +class RecheckNotDue(VerifierError): + """The recheck is not due yet: a read now would end the window early.""" + + def __init__(self, due_at: str, wait_s: float) -> None: + super().__init__(f"the recheck is due at {due_at}, in {wait_s:.0f} s") + self.due_at = due_at + self.wait_s = wait_s + + +@dataclass(frozen=True) +class RecheckResult: + #: ``CONFIRMED`` (nothing unexplained moved since the post-save read), + #: ``REFUTED`` (something did: the earlier CONFIRMED no longer stands), or + #: ``UNCERTAIN`` (the database or the state couldn't be read or doesn't + #: match; never a pass). + outcome: LateOutcome + #: The frame since the post-save read: ``held``, ``refuted`` or + #: ``not_checked``. + frame: str + #: Plain sentences, without record values. + reasons: list[str] = field(default_factory=list) + #: The earlier write's receipt digest. + receipt_sha256: str = "" + post_save_read_at: str = "" + due_at: str = "" + #: When the database was read for the recheck ("" when it wasn't). + read_at: str = "" + #: How many later writes were passed as excuses (``excused_by``). + excused: int = 0 + #: The frame's closed evidence since the post-save read (counts and + #: names, no values). + evidence: dict[str, Any] = field(default_factory=dict) + + @property + def verified(self) -> bool: + """True only when the earlier verdict still stands.""" + + return self.outcome == "CONFIRMED" + + @property + def late(self) -> bool: + """True when a change after the verdict refutes it.""" + + return self.outcome == "REFUTED" + + @property + def status(self) -> str: + """The outcome under :class:`WriteResult`'s names.""" + + return {"CONFIRMED": "verified", "REFUTED": "refuted"}.get( + self.outcome, "unconfirmed" + ) + + _NO_LEDGER: Any = object() # sentinel: the caller passed no ``ledger`` @@ -1088,6 +1240,21 @@ def _downgrade( ) -> WriteResult: import dataclasses + if result.recheck is not None and result.outcome is Outcome.CONFIRMED: + # Its receipt says verified, but the write is no longer CONFIRMED: a + # recheck of it must not say "the earlier verdict still stands". It + # still excuses its own window for later writes (its frame held). + try: + _withdraw_recheck(result.recheck.payload) + except VerifierError: + # verify_write never raises: un-issue the state instead, so a + # recheck of it is UNCERTAIN ("not issued") all the same. + try: + ( + _issued_dir("rechecks") / _state_digest(result.recheck.payload) + ).unlink() + except (OSError, VerifierError): + pass return dataclasses.replace( result, outcome=Outcome.UNCERTAIN, @@ -1095,6 +1262,7 @@ def _downgrade( retry_safe=False, landed=landed, review=review, + recheck_due_at=None, ) @@ -1214,7 +1382,9 @@ def _judge( "declared_by": pending.declared_by, "authorization": "operator" if pending.authorized else "unbound", } + post_state: dict[str, Any] = {} try: + baseline = json.loads(pending.baseline_path.read_text(encoding="utf-8")) doc = verify_run( pending.contract.config, params=pending.params, @@ -1227,6 +1397,7 @@ def _judge( write_attempted=write_attempted, provenance=None if source is None else source.provenance, documents=None if source is None else source.documents, + post_state_out=post_state, ) path: Optional[str] = str(write_result(doc, pending.workdir)) except Exception as exc: # noqa: BLE001 - unreadable store, timeout, used baseline @@ -1247,6 +1418,9 @@ def _judge( seal_path, seal_state = None, "off" if outcome is Outcome.CONFIRMED and seal and path: seal_path, seal_state = _seal(path, pending.write_id) + state = _issue_state(baseline, doc, post_state, pending.params) + if state is not None: + state.path = state.save(pending.workdir / "recheck.json") return WriteResult( outcome=outcome, scope=_scope(pending.contract, evidence, receipt), @@ -1261,10 +1435,28 @@ def _judge( evidence=evidence, seal=seal_state, seal_path=seal_path, + recheck=state, + recheck_due_at=receipt.get("recheck_due_at"), **base, ) +def _issue_state( + baseline: Mapping[str, Any], + doc: Mapping[str, Any], + post_state: Mapping[str, Any], + params: Mapping[str, Any], +) -> Optional["RecheckState"]: + """The run's recheck state, recorded as issued here, or None when its + evidence binds no post-save state.""" + + payload = recheck_state(baseline, doc, post_state) + if payload is None: + return None + _issue_recheck(payload) + return RecheckState(payload, params=dict(params)) + + def classify( receipt_outcome: str, checked: Mapping[str, Any], evidence: Mapping[str, Any] ) -> tuple[Outcome, bool]: @@ -1344,8 +1536,13 @@ def _scope( paths.append(path) contracts = (receipt or {}).get("contracts") or {} strength = str(contracts.get("observed_effect_strength") or "unchecked") + schedule = (evidence.get("frame") or {}).get(EVIDENCE_KEY) or {} + due = schedule.get("due_at") if isinstance(schedule, Mapping) else None not_covered = [ - "writes that land after read_at (delayed or queued writes)", + f"writes that land after read_at until the recheck due at {due} " + "(the result is provisional until it runs), and after that recheck" + if due + else "writes that land after read_at (delayed or queued writes)", "stores the contract doesn't read: email, other services, files", "whether the write was authorized; the verifier doesn't judge it", ] @@ -1468,12 +1665,12 @@ def load(cls, path: str | Path) -> "Snapshot": USED_DIR: Optional[Path] = None -def _issued_dir() -> Path: +def _issued_dir(name: str = "issued") -> Path: root = USED_DIR or Path( os.environ.get("OPENADAPT_VERIFIER_STATE_DIR") or Path.home() / ".cache" / "openadapt" / "verifier" ) - issued = Path(root) / "issued" + issued = Path(root) / name try: issued.mkdir(mode=0o700, parents=True, exist_ok=True) Path(root).chmod(0o700) # created above, or already this user's own @@ -1539,6 +1736,54 @@ def _consume(payload: Any) -> None: raise VerifierError(f"can't record the snapshot's use: {exc}") from exc +def _state_digest(payload: Any) -> str: + body = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(body.encode("utf-8")).hexdigest() + + +def _issue_recheck(payload: Mapping[str, Any]) -> None: + """Record a recheck state as issued here: an empty file named by the + digest of its whole payload, under the state directory's ``rechecks``. + A state is used only when its record is there, so one written by hand + (whose salt and digests anyone holding the file could recompute) is + never taken for a verified write's.""" + + path = _issued_dir("rechecks") / _state_digest(payload) + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT, 0o600) + except OSError as exc: + raise VerifierError(f"can't record the recheck state: {exc}") from exc + os.close(fd) + + +def _withdraw_recheck(payload: Mapping[str, Any]) -> None: + """Mark an issued recheck state as not to be rechecked: the result it + came with was downgraded from CONFIRMED.""" + + path = _issued_dir("rechecks") / f"{_state_digest(payload)}.withdrawn" + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT, 0o600) + except OSError as exc: + raise VerifierError(f"can't withdraw the recheck state: {exc}") from exc + os.close(fd) + + +def _recheck_withdrawn(payload: Any) -> bool: + try: + return ( + _issued_dir("rechecks") / f"{_state_digest(payload)}.withdrawn" + ).exists() + except (VerifierError, TypeError, ValueError): + return True + + +def _recheck_issued(payload: Any) -> bool: + try: + return (_issued_dir("rechecks") / _state_digest(payload)).is_file() + except (VerifierError, TypeError, ValueError): + return False + + def capture(contract: Contract, params: Mapping[str, object]) -> Snapshot: """Deprecated (#1588): read the system of record before the write. @@ -1586,6 +1831,7 @@ def _verify_snapshot( before.used = True _consume(before.payload) # Judge the payload whose salt was just used, never the file as it is now. + post_state: dict[str, Any] = {} with tempfile.TemporaryDirectory(prefix="openadapt-verify-") as tmp: try: doc = verify_run( @@ -1597,6 +1843,7 @@ def _verify_snapshot( baseline_path=before.save(Path(tmp) / "baseline.json"), provenance=None if source is None else source.provenance, documents=None if source is None else source.documents, + post_state_out=post_state, ) except (VerifyRunError, OSError) as exc: raise VerifierError(str(exc)) from exc @@ -1616,6 +1863,8 @@ def _verify_snapshot( receipt=receipt, receipt_sha256=receipt_sha256(receipt), evidence=evidence, + recheck=_issue_state(before.payload, doc, post_state, params), + recheck_due_at=receipt.get("recheck_due_at"), ) @@ -1683,6 +1932,161 @@ def reconcile( lease.release() +# -- late writes: the recheck -------------------------------------------------- + + +def recheck( + contract: Contract, + write: WriteResult | RecheckState, + *, + params: Mapping[str, object], + actor: Actor, + excused_by: Sequence[WriteResult | RecheckState] = (), + wait: bool = False, +) -> RecheckResult: + """Read the whole database again and judge every change since a verified + write's post-save read. + + Args: + contract: The write's contract; it must set + ``frame.database.recheck_after_s``. + write: The verified :class:`WriteResult`, or its saved + :class:`RecheckState`. + params: The write's inputs, as given to :func:`verify_write`. + actor: Who made the write. + excused_by: Later writes under this contract by the same actor, + each begun after this write's post-save read, of any outcome, + so the caller can keep working while the window runs. Where its + own frame was judged (held or refuted), a row or audit entry + that changed in its own window (between its baseline and its + post-save read) and still holds what that read saw is its own; + a row matching its selector that lands later is not. A change + that may be what refuted it gives ``UNCERTAIN``. Each must have + been issued by :func:`verify_write` under this state directory; + a loaded state needs its params (:meth:`RecheckState.load`). + wait: Before the due time, sleep until it instead of raising + :class:`RecheckNotDue`. + + Changes the frame excuses after a save are excused here too: the app + profile's bookkeeping, and another actor's write the audit log + attributes. Anything else unexplained is ``REFUTED`` (late). A state + that doesn't match its receipt, the contract, the actor or the params, + and a database that can't be read or is still moving, give + ``UNCERTAIN``. Raises :class:`VerifierError` when ``write`` scheduled no + recheck or an ``excused_by`` entry doesn't qualify, and + :class:`RecheckNotDue` before the due time. + """ + + state = write.recheck if isinstance(write, WriteResult) else write + receipt = state.payload.get("receipt") if state is not None else None + if ( + state is None + or not isinstance(receipt, dict) + or receipt.get("outcome") != "verified" + ): + raise VerifierError( + "this write scheduled no recheck: it was not verified, or the " + "contract doesn't set frame.database.recheck_after_s" + ) + run_record = actor.run_record() + cfg = contract.config + spec = cfg.frame + post = state.payload.get("post_state") + base = RecheckResult( + outcome="UNCERTAIN", + frame="not_checked", + receipt_sha256=receipt_sha256(receipt), + due_at=state.due_at, + post_save_read_at=str(post.get("read_at") or "") + if isinstance(post, dict) + else "", + ) + problem = ( + "the recheck state was not issued under this state directory" + if not _recheck_issued(state.payload) + else "the write's result was downgraded from CONFIRMED to UNCERTAIN" + if _recheck_withdrawn(state.payload) + else None + ) or recheck_problem(cfg, state.payload, params=params, run_record=run_record) + if problem is not None: + return _replace(base, reasons=[f"{problem}, so the write can't be rechecked"]) + wait_s = seconds_until(state.due_at) + if wait_s > 0: + if not wait: + raise RecheckNotDue(state.due_at, wait_s) + time.sleep(wait_s) + excused: list[tuple[Mapping[str, Any], Mapping[str, object]]] = [] + for i, other in enumerate(excused_by): + later = other.recheck if isinstance(other, WriteResult) else other + if later is None: + raise VerifierError( + f"excused_by[{i}] has no recheck state: its database didn't " + "settle after its save, so what it wrote can't be told apart; " + "reconcile it before rechecking this write" + ) + if not _recheck_issued(later.payload): + raise VerifierError( + f"excused_by[{i}] was not issued under this state directory" + ) + if later.params is None: + raise VerifierError( + f"excused_by[{i}] has no params: load it with RecheckState.load" + "(path, params=...)" + ) + excused.append((later.payload, later.params)) + try: + verdict, read_at = recheck_run( + cfg, state.payload, params=params, run_record=run_record, excused=excused + ) + except VerifyRunError as exc: + raise VerifierError(str(exc)) from exc + except Exception as exc: # noqa: BLE001 - an unreadable store is not a pass + return _replace( + base, + reasons=[f"the database could not be read again ({type(exc).__name__})"], + ) + evidence = verdict.evidence(spec) if spec is not None else {} + reasons: list[str] = [] + status = verdict.status + if status == "held" and LATER_WRITE_DISPUTED in verdict.excused_by: + # A change in a later write's window may be what refuted that write; + # the harm may be this write's, so this verdict can't stand as is. + status = "not_checked" + reasons.append( + "a change since the verdict lies in a later write that was not " + "verified and may be what refuted it; reconcile both" + ) + if status == "refuted": + kinds = ", ".join(sorted({c.change for c in verdict.collateral})) + reasons.append( + f"{len(verdict.collateral)} record(s) moved after the verdict that " + f"nothing explains ({kinds}); the earlier CONFIRMED no longer stands" + ) + elif status != "held" and not reasons: + reasons.append(verdict.reason or "the database could not be judged again") + return _replace( + base, + outcome=_late_outcome(status), + frame=verdict.status, + reasons=reasons, + read_at=read_at, + excused=len(excused), + evidence=evidence, + ) + + +def _late_outcome(status: str) -> LateOutcome: + from openadapt_flow.runtime.effects.late_write import outcome_of + + return outcome_of(status) + + +def _replace(result: RecheckResult, **changes: Any) -> RecheckResult: + import dataclasses + + return dataclasses.replace(result, **changes) + + # -- the database frame, without a contract (from #1588) ---------------------- #: The key the snapshots of this process share, when the caller gives none: diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index a2c6876b6c..af56a71987 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -74,6 +74,7 @@ load_coverage, ) from openadapt_flow.qualify.result import verify_config_sha256 +from openadapt_flow.runtime.effects import late_write from openadapt_flow.runtime.effects.adapter import verifier_identity from openadapt_flow.runtime.effects.effect import ( Effect, @@ -171,6 +172,10 @@ "verification_tier", ) _OPTIONAL_RECEIPT_FIELDS = ("remediates", "correlation_id") +#: Optional too: when a verified write's recheck is due (whole-second UTC), +#: present only when the contract sets ``frame.database.recheck_after_s``. +#: It equals ``evidence.frame.recheck.due_at``. +RECHECK_RECEIPT_FIELD = "recheck_due_at" #: ``checked_how`` is optional too, but a label, not an opaque id. _SIGNATURE_FIELDS = frozenset( { @@ -538,6 +543,7 @@ def _read_database( ref_columns=_ref_columns(cfg), schemas=database.schemas, optional_modes=database.profile_modes(), # type: ignore[arg-type] + links=database.recheck_after_s is not None, ) except Exception: # noqa: BLE001 - unreadable -> the frame is not checked return None @@ -593,6 +599,12 @@ class _DatabaseFrame: #: Every user's id by sign-in name (in memory only), to resolve an #: audit entry's actor to an owner column (owner-scoped attribution). users_map: dict[str, str] = field(default_factory=dict) + #: The settled read after the save, and when it was taken: what a later + #: recheck compares with (``recheck_after_s``). + post: Any = None + post_read_at: str = "" + #: The snapshot the rows are judged against. + old: Any = None def _rules_used(profile: Any, old: Any, new: Any) -> list[str]: @@ -651,7 +663,7 @@ def _database_frame( SnapshotError, cell_hash, mac_key, - same_state, + settled_state, unpack, unread_reasons, ) @@ -677,15 +689,34 @@ def _database_frame( new = _read_database(cfg, salt, before=old) if new is None: return unreachable - # Read again a moment later: the two reads must match, or the app is - # still writing and what the save caused can't be told yet. - time.sleep(database.settle_s) - again = _read_database(cfg, salt, before=old) - if again is None: - return unreachable - if not same_state(new, again): + new_at = _now() + # Read again a moment later: two reads in a row must match, or the app + # is still writing and what the save caused can't be told yet. A live + # app moves on its own, so up to ``settle_attempts`` more reads are + # taken, each ``settle_s`` after the last; the earlier read of the first + # matching pair is judged. Rows added to append-only tables between the + # two don't count: the frame excuses them anyway. + attempts = database.settle_attempts + settled = False + used = 0 + for used in range(1, attempts + 1): + time.sleep(database.settle_s) + again = _read_database(cfg, salt, before=old) + if again is None: + return unreachable + if settled_state(new, again): + settled = True + break + if used < attempts: + new, new_at = again, _now() + if not settled: moving = _DatabaseFrame({}, {}, [], {}, ("", ""), {}, None) moving.still_moving = True + moving.evidence = { + "settle_attempts": used, + "settle_attempts_allowed": attempts, + "settled": False, + } return moving if not (old.code_visible and new.code_visible): blind = _DatabaseFrame({}, {}, [], {}, ("", ""), {}, None) @@ -693,6 +724,7 @@ def _database_frame( return blind key = mac_key(salt) out = _DatabaseFrame({}, {}, [], {}, (old.clock, new.clock), {}, None) + out.post, out.post_read_at, out.old = new, new_at, old if "audit" in new.extra: out.audit_after = list(new.extra["audit"]) if old.objects != new.objects: @@ -742,6 +774,8 @@ def _database_frame( ), "schemas": list(new.schemas), "allow_list_sha256": database.allow_list_sha256(), + "settle_attempts": used, + "settle_attempts_allowed": attempts, } profile = database.loaded_profile() if profile is not None: @@ -766,6 +800,34 @@ def excuse(name: str, before: Any, after: Any, fields: list[str]) -> Any: return out +def _later_status(found: Sequence[Optional[str]]) -> Optional[str]: + """``clean`` when a later write's window covers a change and none calls + it disputed; ``disputed`` when one does; None when none covers it.""" + + hits = [f for f in found if f is not None] + if not hits: + return None + return "disputed" if "disputed" in hits else "clean" + + +def _collateral_record(verdict: FrameVerdict, db: Any, salt: bytes) -> dict[str, Any]: + rows: list[list[str]] = [] + entries: list[Mapping[str, Any]] = [] + unmapped = False + for change in verdict.collateral: + if change.audit_only and isinstance(change.entry, Mapping): + entries.append(change.entry) + elif change.surface in db.slices and isinstance(change.key, str): + rows.append([change.surface, change.key]) + else: + unmapped = True # a change a recheck can't match by key + return { + "rows": sorted(rows), + "entries": sorted(_audit_ids_each(entries, salt)), + "unmapped": unmapped or bool(verdict.collateral and not db.slices), + } + + def _read_audit(verifier: Any) -> Optional[list[dict[str, Any]]]: if verifier is None: return None @@ -785,19 +847,30 @@ def judge_run_frame( audit_verifier: Any = None, run_actors: Sequence[str] = (), salt: bytes = b"", + database_only: bool = False, + later: Sequence[Any] = (), ) -> FrameVerdict: - """Read the slice after the save and judge the frame against ``before``.""" + """Read the slice after the save and judge the frame against ``before``. + + ``database_only`` (a recheck): judge ``frame.database`` and the audit + log only; the verifier's own read set is not read, and is used only to + route effects to their tables. ``later``: the windows of later writes + (``late_write.LaterWindow``) whose own frames already judged the rows + and audit entries in them. + """ from openadapt_flow.runtime.effects.surfaces import surface_of spec = cfg.frame or FrameSpec() - try: - capture = getattr(verifier, "capture_post_state", None) or ( - verifier.capture_pre_state - ) - after = capture() - except Exception: # noqa: BLE001 - unreadable -> not checked - after = None + after: Any = None + if not database_only: + try: + capture = getattr(verifier, "capture_post_state", None) or ( + verifier.capture_pre_state + ) + after = capture() + except Exception: # noqa: BLE001 - unreadable -> not checked + after = None routed = getattr(verifier, "verifiers", None) slices: dict[str, tuple[EffectState, Optional[list[dict[str, Any]]]]] = {} routes: Optional[dict[str, list[str]]] = None @@ -819,7 +892,7 @@ def read_of(name: str) -> str: for i, effect in enumerate(bound_effects): own = surface_of(effect) or "" routes[str(i)] = views.get(read_of(own), [own]) if own in routed else [own] - for name in routed: + for name in [] if database_only else routed: sub = before.detail.get(name, {}) if isinstance(before.detail, dict) else {} post = ( after.detail.get(name, {}) @@ -834,18 +907,21 @@ def read_of(name: str) -> str: ), list(post.get("records", [])) if post.get("reachable") else None, ) - else: + elif not database_only: slices[""] = (before, _post_records(after)) identities: Optional[dict[str, str]] = None db = _database_frame(cfg, before, salt) if db is not None and db.still_moving: + tries = db.evidence.get("settle_attempts", 1) return FrameVerdict( status="not_checked", reason=( - "the database was still changing after the save (two reads " - "differed); what the save caused can't be told yet" + "the database was still changing after the save (no two reads " + f"in a row matched in {tries} attempt(s)); what the save " + "caused can't be told yet" ), attribution=spec.attribution.kind if spec.attribution else "none", + database=dict(db.evidence), ) if db is not None and db.blind: return FrameVerdict( @@ -880,25 +956,57 @@ def read_of(name: str) -> str: if isinstance(before.detail, dict) else None ) - if isinstance(stored, dict) and stored.get("reachable"): + entries_now: Optional[list[dict[str, Any]]] = None + #: A change a later write's window covers that may be what refuted it. + disputed: list[bool] = [] + if isinstance(stored, dict): entries_now = ( db.audit_after if db is not None and db.audit_after is not None else _read_audit(audit_verifier) ) + if isinstance(stored, dict) and stored.get("reachable"): if entries_now is not None: if "ids" in stored: # The window: entries now that the baseline didn't hold. seen = set(stored["ids"]) audit_before = [] - audit_after = [ - e - for e, h in zip(entries_now, _audit_ids_each(entries_now, salt)) - if h not in seen - ] + audit_after = [] + for e, h in zip(entries_now, _audit_ids_each(entries_now, salt)): + if h in seen: + continue + status = _later_status([w.entry(e) for w in later]) + if status == "disputed": + disputed.append(True) + if status is None: + audit_after.append(e) else: audit_before = list(stored.get("records", [])) audit_after = entries_now + excuse = None if db is None else db.excuse + if db is not None and later and spec.database is not None: + from openadapt_flow.runtime.effects.db_snapshot import KEY + + db_spec = spec.database + profile_excuse = db.excuse + + old_tables = db.old.tables if db.old is not None else {} + + def excuse(name: str, before: Any, after: Any, fields: list[str]) -> Any: + skip = db_spec.ignore_columns_for(name) + table = old_tables.get(name) + links = table.links if table is not None else {} + status = _later_status( + [w.row(name, before, after, skip, links) for w in later] + ) + if status == "disputed": + disputed.append(True) + if status is not None: + return [late_write.LATER_WRITE_RULE] + if profile_excuse is None: + return None + return profile_excuse(name, before, after, fields) + verdict = judge_frame( spec, bound_effects, @@ -914,14 +1022,42 @@ def read_of(name: str) -> str: hashers=None if db is None else db.hashers, clock=("", "") if db is None else db.clock, database_evidence=None if db is None else db.evidence, - excuse=None if db is None else db.excuse, + excuse=excuse, user_ids=None if db is None else db.users_map, ) profile = db.evidence.get("profile") if db is not None else None if profile is not None: # The rules that actually excused something: unconditional ones seen # in the snapshots, conditional ones as the judge applied them. - profile["rules_used"] = sorted(set(profile["rules_used"]) | verdict.excused_by) + profile["rules_used"] = sorted( + (set(profile["rules_used"]) | verdict.excused_by) + - {late_write.LATER_WRITE_RULE} + ) + if disputed: + verdict.excused_by.add(late_write.LATER_WRITE_DISPUTED) + database = spec.database + if ( + db is not None + and db.post is not None + and database is not None + and database.recheck_after_s is not None + ): + # What a later recheck compares with (and, for a later write, what + # its own window covered): MACs and keyed hashes only. + verdict.post_state = { + "snapshot": db.post.to_json(), + "read_at": db.post_read_at, + # What this run's frame reported, as row key MACs and keyed + # entry hashes: a later recheck of an earlier write can tell a + # change this run was refuted for from one it explained. + "collateral": _collateral_record(verdict, db, salt), + "audit": None + if not isinstance(stored, dict) + else { + "reachable": entries_now is not None, + "ids": _audit_ids(entries_now or [], salt), + }, + } return verdict @@ -1403,6 +1539,7 @@ def verify_run( provenance: Optional[Any] = None, documents: Optional[Mapping[str, Any]] = None, second_extractor: Optional[Any] = None, + post_state_out: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """Read the system of record and return ``{"receipt", "evidence"}``. @@ -1421,6 +1558,12 @@ def verify_run( guarded save sets it after its commit step). Then a store that shows nothing is ``reconciliation_required``, not ``halted_before_effect``: the write may still land, so it must not be retried. + + With ``frame.database.recheck_after_s``, a verified run's evidence says a + recheck is due (``evidence.frame.recheck``), and ``post_state_out`` (a + dict the caller passes) receives what the recheck compares with: the + settled post-save snapshot as keyed MACs. It is private, like the + baseline (``runtime/effects/late_write.py``, :func:`recheck_run`). """ from openadapt_types.execute import ( @@ -1731,6 +1874,28 @@ def verify_run( }, "effect_verdict": effect_verdict, } + database_spec = frame_spec.database + recheck_due_at: Optional[str] = None + if ( + frame.post_state is not None + and database_spec is not None + and database_spec.recheck_after_s is not None + ): + # A write that lands after the settled read lands after this verdict: + # say when the database is read again (a verified write only), and + # bind what it compares with. A write of any outcome binds its + # post-save state, so a recheck of an earlier write can tell what + # this one's own window covered. + schedule = late_write.schedule( + frame.post_state, + database_spec.recheck_after_s, + salted_digest(salt, late_write.POST_STATE_LABEL, frame.post_state), + due=verified, + ) + cast(dict[str, Any], evidence["frame"])[late_write.EVIDENCE_KEY] = schedule + recheck_due_at = schedule["due_at"] + if post_state_out is not None: + post_state_out.update(frame.post_state) contracts = ExecuteEvidenceContractV1( # A reference was bound and the approver is not the executor; # verify-run did not evaluate a policy (evidence.authorization says so). @@ -1778,6 +1943,8 @@ def verify_run( receipt["remediates"] = remediates if correlation_id is not None: receipt["correlation_id"] = correlation_id + if recheck_due_at is not None: + receipt[RECHECK_RECEIPT_FIELD] = recheck_due_at problem = receipt_problem(receipt) or evidence_problem(evidence, receipt) if problem is not None: # pragma: no cover - a producer bug, never input raise VerifyRunError(f"verify-run built an invalid receipt: {problem}") @@ -1788,6 +1955,263 @@ def verify_run( } +# -- late writes ---------------------------------------------------------------- + + +def recheck_state( + baseline: Mapping[str, Any], + result: Mapping[str, Any], + post_state: Mapping[str, Any], +) -> Optional[dict[str, Any]]: + """The private recheck state of a run, or None when its evidence binds + no post-save state (the contract sets no ``recheck_after_s``, or the + database didn't settle after the save). ``baseline`` is the payload + :func:`capture_baseline` wrote, ``result`` what :func:`verify_run` + returned, ``post_state`` its ``post_state_out``. + + A verified run's state is what its recheck compares with. Any run's + state can excuse its own window in an earlier write's recheck.""" + + evidence = dict(result["evidence"]) + if not post_state or late_write.EVIDENCE_KEY not in (evidence.get("frame") or {}): + return None + return { + "schema_version": late_write.RECHECK_SCHEMA, + "execution_id": str(baseline["execution_id"]), + "salt": str(baseline["salt"]), + "verifier_identity": str(baseline["verifier_identity"]), + "baseline": {k: baseline[k] for k in _BASELINE_BOUND_KEYS}, + "receipt": dict(result["receipt"]), + "evidence": evidence, + "post_state": dict(post_state), + } + + +def recheck_problem( + cfg: VerifyConfig, + state: Any, + *, + params: Mapping[str, object], + run_record: Mapping[str, Any], + verified: bool = True, +) -> Optional[str]: + """A plain sentence when ``state`` can't be used as this contract's + write by this executor with these params; None when it can. + ``verified``: the write must also have been verified (the write being + rechecked; a later write that excuses its window may have any outcome). + + Everything the recheck relies on is bound to the receipt: the receipt + and its evidence agree, the salt is the one the evidence commits to, the + baseline and the post-save state are the ones their digests name, and + the contract, the read, the executor and each bound parameter are the + ones the receipt was issued for. + """ + + problem = late_write.state_problem(state) + if problem is not None: + return problem + receipt, evidence = dict(state["receipt"]), dict(state["evidence"]) + if receipt_problem(receipt) or evidence_problem(evidence, receipt): + return "the recheck state's receipt and evidence don't match" + if verified and receipt.get("outcome") != "verified": + return "the write was not verified, so there is nothing to recheck" + if receipt.get("execution_id") != state["execution_id"]: + return "the recheck state belongs to another run" + salt = bytes.fromhex(str(state["salt"])) + if (evidence.get("salting") or {}).get("salt_commitment") != salt_commitment(salt): + return "the recheck state's salt is not the one its receipt commits to" + readback = evidence.get("readback") or {} + baseline = dict(state["baseline"]) + if set(baseline) != set(_BASELINE_BOUND_KEYS) or readback.get( + "baseline_sha256" + ) != baseline_digest(baseline, salt): + return "the recheck state's baseline is not the one its receipt names" + frame = evidence.get("frame") or {} + schedule = frame.get(late_write.EVIDENCE_KEY) + database = cfg.frame.database if cfg.frame is not None else None + if not isinstance(schedule, Mapping) or database is None: + return "the receipt binds no post-save state" + post = dict(state["post_state"]) + digest = salted_digest(salt, late_write.POST_STATE_LABEL, post) + if schedule.get("post_state_sha256") != digest: + return "the recheck state is not the post-save state its receipt names" + is_verified = receipt.get("outcome") == "verified" + if database.recheck_after_s is None or dict(schedule) != late_write.schedule( + post, database.recheck_after_s, digest, due=is_verified + ): + return ( + "the contract's recheck_after_s is not the one the receipt was issued under" + ) + try: + verifier = _build_verifier(cfg, params) + identity = verifier_identity(verifier) + executor_kind, executor_ref, _build = _executor(run_record) + except (VerifyRunError, KeyError, ValueError) as exc: + return f"the contract can't be bound for the recheck: {exc}" + if ( + identity != state["verifier_identity"] + or identity != readback.get("verifier_identity") + or identity != baseline.get("verifier_identity") + ): + return "the recheck state was taken by a different verifier" + contract = evidence.get("contract") or {} + if ( + contract.get("contract_sha256") + != verify_config_sha256(identity, list(cfg.effects)) + or frame.get("frame_sha256") != (cfg.frame or FrameSpec()).digest() + ): + return "the recheck state was taken under a different contract" + if ( + receipt.get("executor_kind") != executor_kind + or receipt.get("executor_ref_sha256") != executor_ref + ): + return "the recheck state's write was made by a different actor" + rows = readback.get("effects") or [] + if len(rows) != len(cfg.effects): + return "the recheck state was taken under a different contract" + for effect, row in zip(cfg.effects, rows): + try: + salted = { + name: _salted_hex(salt, f"effect_param:{name}", params[name]) + for name in sorted(effect.referenced_params()) + } + effect_digest = effect.resolved_contract_hash( + {}, opaque_param_sha256=salted + ) + except (KeyError, ValueError): + return "the params don't bind the contract" + if ( + not isinstance(row, Mapping) + or row.get("effect_contract_hash") != effect_digest + ): + return "the params are not the ones the write was verified with" + return None + + +def _entry_id(entry: Mapping[str, Any], key: bytes) -> str: + from openadapt_flow.runtime.effects.effect import stable_id + + return hmac.new( + key, repr(stable_id(dict(entry))).encode(), hashlib.sha256 + ).hexdigest() + + +def _later_window(state: Mapping[str, Any]) -> Optional[Any]: + """A later write's own window, when its frame judged it (held or + refuted); None when its frame wasn't checked, so nothing in its window + was reported and nothing there is excused.""" + + evidence = state["evidence"] + frame = (evidence.get("frame") or {}).get("status") + if frame not in ("held", "refuted"): + return None + record = state["post_state"].get("collateral") + if not isinstance(record, Mapping): + return None + detail = (state["baseline"].get("state") or {}).get("detail") or {} + snapshot = detail.get(FRAME_DB_DETAIL) + if not isinstance(snapshot, dict): + return None + post = state["post_state"] + before_audit = detail.get(FRAME_AUDIT_DETAIL) + after_audit = post.get("audit") + readable = bool( + isinstance(before_audit, Mapping) + and before_audit.get("reachable") + and isinstance(after_audit, Mapping) + and after_audit.get("reachable") + ) + ids_before = set(before_audit["ids"]) if readable and before_audit else None + ids_after = set(after_audit["ids"]) if readable and after_audit else None + try: + return late_write.LaterWindow( + _mac_key(bytes.fromhex(str(state["salt"]))), + snapshot, + post["snapshot"], + verified=state["receipt"].get("outcome") == "verified", + delta_held=(evidence.get("checked") or {}).get("delta") == "held", + collateral_rows={(str(t), str(k)) for t, k in record.get("rows") or []}, + collateral_entries={str(h) for h in record.get("entries") or []}, + unmapped=bool(record.get("unmapped", True)), + audit_before=ids_before, + audit_after=ids_after, + entry_id=_entry_id, + ) + except (KeyError, TypeError, ValueError): + return None + + +def recheck_run( + cfg: VerifyConfig, + state: Mapping[str, Any], + *, + params: Mapping[str, object], + run_record: Mapping[str, Any], + excused: Sequence[tuple[Mapping[str, Any], Mapping[str, object]]] = (), +) -> tuple[FrameVerdict, str]: + """Judge every change since a verified write's post-save read. + + ``state`` must have passed :func:`recheck_problem`. ``excused``: the + recheck states and params of later writes under this contract by this + executor, of any outcome, each begun after ``state``'s post-save read + (each is checked here as ``state`` was). Where a later write's frame + was judged (held or refuted), each row and audit entry that changed in + its own window, between its baseline and its post-save read, and still + holds what that read saw, is accounted for by it + (``late_write.LaterWindow``): its declared effects and side rows as they + landed then, never a row matching its selector that lands later. A + change that may be what refuted that later write is ``disputed``: the + verdict's ``excused_by`` then holds ``LATER_WRITE_DISPUTED``. The + database is read twice, ``settle_s`` apart, as after the save. Returns + the frame verdict and when the database was read. + + Raises :class:`VerifyRunError` for an excuse that doesn't qualify. + """ + + post = dict(state["post_state"]) + since = late_write.parse_time(str(post["read_at"])) + windows: list[Any] = [] + for i, (other, other_params) in enumerate(excused): + problem = recheck_problem( + cfg, other, params=other_params, run_record=run_record, verified=False + ) + if problem is not None: + raise VerifyRunError(f"excused_by[{i}] is not a later write: {problem}") + if other["execution_id"] == state["execution_id"]: + raise VerifyRunError(f"excused_by[{i}] is the write being rechecked") + started = other["baseline"].get("baseline_read_at") + try: + later = late_write.parse_time(str(started)) >= since + except ValueError: + later = False + if not later: + raise VerifyRunError( + f"excused_by[{i}] started before this write's post-save read" + ) + window = _later_window(other) + if window is not None: + windows.append(window) + salt = bytes.fromhex(str(state["salt"])) + detail: dict[str, Any] = {FRAME_DB_DETAIL: dict(post["snapshot"])} + if post.get("audit") is not None: + detail[FRAME_AUDIT_DETAIL] = dict(post["audit"]) + before = EffectState(substrate="database", reachable=True, detail=detail) + verifier = _build_verifier(cfg, params) + verdict = judge_run_frame( + cfg, + verifier, + before, + [], # no declared delta: only the later writes' windows excuse rows + params, + audit_verifier=_audit_verifier(cfg, params), + run_actors=[str(run_record.get("executor_id") or "")], + salt=salt, + database_only=True, + later=windows, + ) + return verdict, _now() + + def write_result(result: Mapping[str, Any], out_dir: Path) -> Path: out = Path(out_dir) / "effect-receipt.json" _write_private(out, result) @@ -1812,7 +2236,7 @@ def receipt_problem(receipt: Mapping[str, Any]) -> Optional[str]: v1_names | set(_NEW_RECEIPT_FIELDS) | set(_OPTIONAL_RECEIPT_FIELDS) - | {"checked_how", "checked"} + | {"checked_how", "checked", RECHECK_RECEIPT_FIELD} ) if set(receipt) - allowed: return "unknown_field" @@ -1841,6 +2265,11 @@ def receipt_problem(receipt: Mapping[str, Any]) -> Optional[str]: for name in _OPTIONAL_RECEIPT_FIELDS: if name in receipt and not _OPAQUE.match(str(receipt[name])): return name + if RECHECK_RECEIPT_FIELD in receipt and ( + receipt.get("outcome") != "verified" + or not _ISO_SECONDS.match(str(receipt[RECHECK_RECEIPT_FIELD])) + ): + return RECHECK_RECEIPT_FIELD # Optional: receipts issued before RFC-0002 open question 5 lack it. if "checked_how" in receipt and ( receipt["checked_how"] not in CHECKED_HOW_VALUES @@ -1921,6 +2350,12 @@ def evidence_problem( or frame.get("status") != receipt["checked"].get("frame") ): return "evidence_mismatch" + schedule = ( + frame.get(late_write.EVIDENCE_KEY) if isinstance(frame, Mapping) else None + ) + due = schedule.get("due_at") if isinstance(schedule, Mapping) else None + if receipt.get(RECHECK_RECEIPT_FIELD) != due: + return "evidence_mismatch" if salting.get("scheme") != SALT_SCHEME or not _DIGEST.match( str(salting.get("salt_commitment", "")) ): diff --git a/packages/flow/tests/test_late_write_recheck_live_openemr.py b/packages/flow/tests/test_late_write_recheck_live_openemr.py new file mode 100644 index 0000000000..934502a2da --- /dev/null +++ b/packages/flow/tests/test_late_write_recheck_live_openemr.py @@ -0,0 +1,390 @@ +"""Live test of the late-write recheck (``frame.database.recheck_after_s``) on +a real OpenEMR. + +The whole-database frame reads the database twice after a save, ``settle_s`` +apart. A write that lands later lands after the verdict. This module saves +made-up patients through OpenEMR's Standard REST API (the fixture's +``actor`` OAuth client), verifies each save with ``openadapt_flow.verifier`` +(the record read back with the read-only ``oracle`` client, the frame over +the whole MariaDB database through a read-only account), and then: + +1. **late collateral:** after the first patient's save is VERIFIED, a direct + SQL ``UPDATE`` changes another chart (one this test made before the + save). The recheck must turn the verdict into REFUTED (late). +2. **clean control:** after the first patient's save is VERIFIED, only a + second, verified save happens. The recheck, given the second save as + ``excused_by``, must stay CONFIRMED. + +Each run uses a fresh random last name, so it needs no reset. The patients +stay in the database until the fixture is reset. + +The module is skipped unless ``OPENEMR_LATE_WRITE_LIVE=1``. It needs the local +fixture up and bootstrapped (``benchmark/openemr_local``; it does not start +containers), a DB-API driver for MariaDB (``pymysql`` by default), and a +read-only database account the test can reach (``SELECT`` on the ``openemr`` +schema and ``TRIGGER``, so the frame can see triggers). Run it from +``packages/flow``:: + + OPENEMR_LATE_WRITE_LIVE=1 OPENEMR_DB_USER=frame_reader \\ + OPENEMR_DB_PASSWORD=... OPENEMR_DB_PORT=3306 PYTHONPATH=../types:. \\ + python -m pytest tests/test_late_write_recheck_live_openemr.py -v \\ + -p no:cacheprovider + +Environment: + +- ``OPENEMR_DB_HOST`` (default ``127.0.0.1``), ``OPENEMR_DB_PORT`` (default + ``3306``), ``OPENEMR_DB_USER`` (required), ``OPENEMR_DB_PASSWORD`` + (required; read through ``sql_password_env``), ``OPENEMR_DB_NAME`` + (default ``openemr``), ``OPENEMR_DB_DRIVER`` (default ``pymysql``); +- ``OPENEMR_LATE_WRITE_RECHECK_S``: the test's ``recheck_after_s`` (default + 2). A test value only: the right wait for a deployment depends on the + app's own jobs, and the contract's author picks it; +- ``OPENEMR_LATE_WRITE_SETTLE_ATTEMPTS``: the test's ``settle_attempts`` + (default 5), also a test value; +- ``OPENEMR_LATE_WRITE_FRAME_JSON``: a JSON object merged into the + contract's ``frame`` (for example ``new_row_fields`` or ``side_records``, + if this OpenEMR build's patient insert sets columns the default doesn't + list), so the first save can be tuned to VERIFIED without editing the test; +- as for ``test_composite_key_live_openemr.py``: ``OPENEMR_FIXTURE_STATE``, + ``OPENEMR_FIXTURE_LOCK``, ``OPENEMR_API_URL``; +- ``OPENEMR_LATE_WRITE_RESET=1``: restore the fixture's baseline database + afterwards (only for the session that holds the fixture). +""" + +from __future__ import annotations + +import json +import os +import time +import uuid +from typing import Any + +import pytest + +pytestmark = pytest.mark.skipif( + os.environ.get("OPENEMR_LATE_WRITE_LIVE") != "1", + reason=( + "live OpenEMR: set OPENEMR_LATE_WRITE_LIVE=1 with the local fixture up " + "(benchmark/openemr_local) and a read-only database account; see this " + "module's docstring" + ), +) + +#: Who writes: the fixture's actor OAuth client, through this test. +ACTOR_ID = "openemr-late-write-live" +FIELDS = ( + "title", + "fname", + "lname", + "DOB", + "sex", + "street", + "city", + "postal_code", + "phone_home", + "email", + "country_code", +) + + +#: OpenEMR's audit log as the frame's attribution needs it: who (``user``), +#: which chart (``patient_id``), the kind of write (``ev``: the event's last +#: part, ``insert``, ``update``, ``delete`` or a read) and the table (``tbl``: +#: the third word of the logged statement, the second for an update). Only +#: entries that name a chart: one that names none can't attribute a row, and +#: the whole-database diff still shows every row it wrote. The read-only +#: guard refuses write keywords even inside literals, so the update case is +#: told by its first letter. ``OPENEMR_LATE_WRITE_AUDIT_SQL`` replaces it. +AUDIT_SQL = ( + "SELECT id, user, patient_id, " + "LOWER(SUBSTRING_INDEX(event, '-', -1)) AS ev, " + "LOWER(TRIM(BOTH '`' FROM SUBSTRING_INDEX(SUBSTRING_INDEX(SUBSTRING_INDEX(" + "TRIM(comments), ' ', IF(LEFT(SUBSTRING_INDEX(event, '-', -1), 1) = 'u', " + "2, 3)), ' ', -1), '(', 1))) AS tbl " + "FROM log WHERE patient_id > 0" +) +#: Where the test puts the oracle client's bearer token for the REST read. +#: verify-run drops a ``verifier.session`` hook (a qualify-only setting, so +#: the read went out unauthenticated: the 401), so the read uses +#: ``auth.bearer_env``. +TOKEN_ENV = "OPENEMR_LATE_WRITE_ORACLE_TOKEN" + + +def _email_literal(email: str) -> str: + """The test's own made-up email as a SQL literal (the fixture's quoting + refuses ``@``).""" + + import re + + if not re.fullmatch(r"[A-Za-z0-9_.+-]+@[A-Za-z0-9.-]+", email): + raise ValueError("unexpected email") + return "'" + email + "'" + + +@pytest.fixture(scope="module") +def openemr() -> Any: + from benchmark.openemr_local.qualify_hooks import fixture + + fx = fixture() + yield fx + if os.environ.get("OPENEMR_LATE_WRITE_RESET") == "1": + fx.reset() + + +@pytest.fixture(scope="module", autouse=True) +def _oracle_token(openemr: Any) -> Any: + """The read-only oracle client's token, for the REST read; issued once, + before any baseline, so its OAuth bookkeeping isn't in a window.""" + + if os.environ.get("OPENEMR_LATE_WRITE_BEARER_ENV"): + yield + return + session = openemr.token_session("oracle") + token = str(session.headers["Authorization"]).split(" ", 1)[1] + os.environ[TOKEN_ENV] = token + yield + os.environ.pop(TOKEN_ENV, None) + + +def _warm(fx: Any, actor: Any) -> None: + """Use both tokens once, then wait: OpenEMR writes the OAuth bookkeeping + after the token response, and it must land before the baseline.""" + + response = actor.get(f"{fx.api_base_url}/apis/default/api/patient?_limit=1") + assert response.status_code == 200, f"actor read: HTTP {response.status_code}" + time.sleep(2) + + +def _actor_user(fx: Any) -> str: + """The OpenEMR user both OAuth clients sign in as: the bot's own id in + the audit log.""" + + return os.environ.get("OPENEMR_LATE_WRITE_SELF") or str( + fx._runtime_values()["OPENEMR_ACTOR_USER"] + ) + + +@pytest.fixture(autouse=True) +def _state_dir(tmp_path: Any, monkeypatch: pytest.MonkeyPatch) -> None: + from openadapt_flow import verifier + + monkeypatch.setattr(verifier, "USED_DIR", tmp_path / "verifier-state") + monkeypatch.setenv("OPENADAPT_LEDGER_DIR", str(tmp_path / "ledger")) + monkeypatch.setenv("OPENADAPT_SEAL_HOME", str(tmp_path / "seal")) + + +def _lname() -> str: + letters = "".join(chr(ord("a") + int(c, 16) % 26) for c in uuid.uuid4().hex[:8]) + return "Latewrite" + letters + + +def _patient(lname: str, n: int) -> dict[str, str]: + return { + "title": "Ms.", + "fname": f"Ada{chr(ord('a') + n)}", + "lname": lname, + "DOB": f"198{n}-03-07", + "sex": "Female", + "street": "101 SYNTHETIC WAY", + "city": "EXAMPLETON", + "state": "MA", + "postal_code": "02139", + "phone_home": f"555-010{n}", + "email": f"{lname.lower()}.{n}@example.invalid", + "country_code": "USA", + } + + +def _contract(fx: Any, lname: str) -> Any: + from openadapt_flow.app_profiles import load_profile + from openadapt_flow.verifier import Contract + + if not os.environ.get("OPENEMR_DB_USER") or not os.environ.get( + "OPENEMR_DB_PASSWORD" + ): + pytest.fail("set OPENEMR_DB_USER and OPENEMR_DB_PASSWORD (read-only account)") + read = { + "sql_driver": os.environ.get("OPENEMR_DB_DRIVER", "pymysql"), + "sql_connect_args": { + "host": os.environ.get("OPENEMR_DB_HOST", "127.0.0.1"), + "port": int(os.environ.get("OPENEMR_DB_PORT", "3306")), + "user": os.environ["OPENEMR_DB_USER"], + "database": os.environ.get("OPENEMR_DB_NAME", "openemr"), + }, + "sql_password_env": "OPENEMR_DB_PASSWORD", + } + frame: dict[str, Any] = { + "identity_field": "pid", + # No side record: OpenEMR 8.0.0.3's REST patient insert writes no + # history_data row (the UI's new-patient form does). + # Other users' writes, and the bot's own per-user bookkeeping (the + # profile's own-row rules need its sign-in name), through OpenEMR's + # audit log, read on the frame's own read-only connection. + "attribution": { + "kind": "audit_read", + "actor_field": "user", + "self": [_actor_user(fx)], + "record_field": "patient_id", + "record_key": "pid", + "event_field": "ev", + "write_events": ["insert", "update", "delete"], + "insert_events": ["insert"], + "delete_events": ["delete"], + "table_field": "tbl", + "read": { + "kind": "sql", + **read, + "sql_query": os.environ.get("OPENEMR_LATE_WRITE_AUDIT_SQL", AUDIT_SQL), + }, + }, + "database": { + "dialect": "mysql", + "read": dict(read), + "profile": "openemr", + "profile_sha256": load_profile("openemr").sha256(), + # The REST read set is one surface (""): the chart rows it shows. + "views": {"patient_data": ""}, + # What OpenEMR's patient insert fills on its own. + "new_row_fields": { + "patient_data": [ + "id", + "pid", + "pubpid", + "uuid", + "date", + "regdate", + "created_by", + "updated_by", + "last_updated", + "state", + ], + }, + # A live OpenEMR moves on its own (a UI session polling): a test + # value, as recheck_after_s is. + "settle_attempts": int( + os.environ.get("OPENEMR_LATE_WRITE_SETTLE_ATTEMPTS", "5") + ), + "recheck_after_s": float( + os.environ.get("OPENEMR_LATE_WRITE_RECHECK_S", "2") + ), + }, + } + extra = os.environ.get("OPENEMR_LATE_WRITE_FRAME_JSON") + if extra: + merged = json.loads(extra) + database = {**frame["database"], **merged.pop("database", {})} + frame = {**frame, **merged, "database": database} + return Contract.from_dict( + { + "schema_version": "openadapt.verify/v1", + "environment_id": "openemr-local-fixture-late-write", + "minimum_effect_strength": "independent_system_of_record", + "verifier": { + "kind": "rest", + # The fixture's HTTPS port has a self-signed certificate, + # which the verifier rightly refuses: OPENEMR_LATE_WRITE_API_URL + # can point at its plain-HTTP port instead. + "base_url": os.environ.get( + "OPENEMR_LATE_WRITE_API_URL", fx.ui_base_url + ), + "records_path": f"/apis/default/api/patient?lname={lname}&_limit=100", + "records_key": "data", + "auth": { + "bearer_env": os.environ.get( + "OPENEMR_LATE_WRITE_BEARER_ENV", TOKEN_ENV + ) + }, + }, + "record": { + "name": "patient", + "match": ["email", "lname"], + "duplicate_key": "email", + "fields": {name: name for name in FIELDS}, + }, + "frame": frame, + } + ) + + +def _create(fx: Any, actor: Any, body: dict[str, str]) -> None: + response = actor.post(f"{fx.api_base_url}/apis/default/api/patient", json=body) + assert response.status_code == 201, ( + f"OpenEMR refused the patient write: HTTP {response.status_code}" + ) + + +def _verified_save(fx: Any, actor: Any, contract: Any, body: dict[str, str]) -> Any: + from openadapt_flow.verifier import Actor, begin_write, verify_write + + params = {name: body[name] for name in FIELDS} + time.sleep(2) # the previous request's bookkeeping lands first + pending = begin_write(contract, params, actor=Actor("script", ACTOR_ID)) + _create(fx, actor, body) + return verify_write(pending) + + +def _recheck(contract: Any, result: Any, body: dict[str, str], **kw: Any) -> Any: + from openadapt_flow.verifier import Actor, recheck + + params = {name: body[name] for name in FIELDS} + return recheck( + contract, + result, + params=params, + actor=Actor("script", ACTOR_ID), + wait=True, + **kw, + ) + + +def _pid(fx: Any, email: str) -> int: + rows = fx._db_lines( + f"SELECT pid FROM openemr.patient_data WHERE email = {_email_literal(email)}" + ) + assert len(rows) == 1, f"expected one chart for the made-up email, got {len(rows)}" + return int(rows[0]) + + +def test_late_collateral_after_a_verified_save_is_refuted(openemr: Any) -> None: + lname = _lname() + actor = openemr.token_session("actor") + _warm(openemr, actor) + contract = _contract(openemr, lname) + + other = _patient(lname, 1) # another chart, made before the save + _create(openemr, actor, other) + other_pid = _pid(openemr, other["email"]) + + first = _patient(lname, 2) + result = _verified_save(openemr, actor, contract, first) + assert result.verified, (result.reasons, result.evidence.get("frame")) + assert result.recheck_due_at == result.receipt["recheck_due_at"] + + # After the verdict: a change to another chart that no write declares, + # made straight in the database (a job or an integration would do this). + openemr._db( + "UPDATE openemr.patient_data SET phone_cell = '555-0199' " + f"WHERE pid = {other_pid}" + ) + late = _recheck(contract, result, first) + assert late.outcome == "REFUTED", (late.reasons, late.evidence) + assert late.late and late.evidence["collateral_count"] >= 1 + + +def test_only_a_later_verified_save_does_not_refute(openemr: Any) -> None: + lname = _lname() + actor = openemr.token_session("actor") + _warm(openemr, actor) + contract = _contract(openemr, lname) + + first = _patient(lname, 2) + result = _verified_save(openemr, actor, contract, first) + assert result.verified, (result.reasons, result.evidence.get("frame")) + + second = _patient(lname, 3) + later = _verified_save(openemr, actor, contract, second) + assert later.verified, (later.reasons, later.evidence.get("frame")) + + late = _recheck(contract, result, first, excused_by=[later]) + assert late.outcome == "CONFIRMED", (late.reasons, late.evidence) + assert late.excused == 1 diff --git a/packages/flow/tests/test_verifier_api.py b/packages/flow/tests/test_verifier_api.py index 3afd1891f0..2bdc4f6e5b 100644 --- a/packages/flow/tests/test_verifier_api.py +++ b/packages/flow/tests/test_verifier_api.py @@ -20,12 +20,15 @@ Contract, MemoryLedger, Outcome, + RecheckResult, + RecheckState, VerifierError, WriteResult, begin_write, capture_database, classify, diff_database, + recheck, verify_write, ) @@ -269,6 +272,9 @@ def test_the_public_api_is_stable() -> None: "MemoryLedger", "Outcome", "PendingWrite", + "RecheckNotDue", + "RecheckResult", + "RecheckState", "Scope", "Snapshot", "Source", @@ -282,6 +288,7 @@ def test_the_public_api_is_stable() -> None: "capture", "capture_database", "diff_database", + "recheck", "reconcile", "verified_write", "verify_write", @@ -333,6 +340,33 @@ def params(fn: Any) -> list[tuple[str, str]]: "reasons", "retry_safe", ] + assert list(WriteResult.__dataclass_fields__)[-2:] == [ + "recheck", + "recheck_due_at", + ] + assert params(recheck) == [ + ("contract", "POSITIONAL_OR_KEYWORD"), + ("write", "POSITIONAL_OR_KEYWORD"), + ("params", "KEYWORD_ONLY"), + ("actor", "KEYWORD_ONLY"), + ("excused_by", "KEYWORD_ONLY"), + ("wait", "KEYWORD_ONLY"), + ] + assert [f for f in RecheckResult.__dataclass_fields__] == [ + "outcome", + "frame", + "reasons", + "receipt_sha256", + "post_save_read_at", + "due_at", + "read_at", + "excused", + "evidence", + ] + assert list(inspect.signature(RecheckState.load).parameters) == [ + "path", + "params", + ] def test_diff_database_applies_the_profile_and_before(tmp_path: Path) -> None: diff --git a/packages/flow/tests/test_verifier_recheck.py b/packages/flow/tests/test_verifier_recheck.py new file mode 100644 index 0000000000..ab82b6e188 --- /dev/null +++ b/packages/flow/tests/test_verifier_recheck.py @@ -0,0 +1,805 @@ +"""Late writes: the recheck after a verified save (``recheck_after_s``). + +The frame reads the database twice after the save, ``settle_s`` apart. A +write that lands later lands after the verdict. With +``frame.database.recheck_after_s`` set, a verified write schedules a recheck, +and :func:`openadapt_flow.verifier.recheck` judges every change since the +post-save read. Each test below is one case, on SQLite, through the public +API. Records are made up. +""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import time +from pathlib import Path +from typing import Any + +import pytest + +from openadapt_flow import verifier +from openadapt_flow.runtime.effects.frame import FrameSpec +from openadapt_flow.verifier import ( + Actor, + Contract, + RecheckNotDue, + RecheckResult, + RecheckState, + VerifierError, + WriteResult, + begin_write, + recheck, + verify_write, +) + +BOT = Actor(kind="agent", id="bot") + +SCHEMA = """ +CREATE TABLE patient_data (id INTEGER PRIMARY KEY, pid INTEGER, fname TEXT, + lname TEXT, email TEXT, phone_cell TEXT DEFAULT '', regdate TEXT); +CREATE TABLE history_data (id INTEGER PRIMARY KEY, pid INTEGER, date TEXT, + tobacco TEXT DEFAULT ''); +CREATE TABLE calendar (id INTEGER PRIMARY KEY AUTOINCREMENT, pid INTEGER, title TEXT); +CREATE TABLE api_log (id INTEGER PRIMARY KEY AUTOINCREMENT, note TEXT); +CREATE TABLE log (id INTEGER PRIMARY KEY, user TEXT, patient_id INTEGER, + event TEXT, tbl TEXT); +INSERT INTO patient_data (id, pid, fname, lname, email, regdate) + VALUES (1, 1, 'Wendell', 'Prior', 'w@example.test', '2026-01-01'); +INSERT INTO history_data VALUES (1, 1, '2026-01-01', ''); +""" + + +def _params(n: int) -> dict[str, str]: + return {"fname": "Ada", "lname": f"Quill{n}", "email": f"ada{n}@example.test"} + + +@pytest.fixture(autouse=True) +def _used_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Each test keeps its ledger, Seal key and issuance records apart.""" + + monkeypatch.setenv("OPENADAPT_LEDGER_DIR", str(tmp_path / "ledger")) + monkeypatch.setenv("OPENADAPT_SEAL_HOME", str(tmp_path / "seal")) + used = tmp_path / "used" + monkeypatch.setattr(verifier, "USED_DIR", used) + return used + + +@pytest.fixture +def due_now(monkeypatch: pytest.MonkeyPatch) -> None: + """The recheck is due: the tests don't wait out the window.""" + + monkeypatch.setattr(verifier, "seconds_until", lambda due: 0.0) + + +def _db(tmp_path: Path) -> Path: + path = tmp_path / "app.sqlite" + con = sqlite3.connect(path) + con.executescript(SCHEMA) + con.commit() + con.close() + return path + + +def _doc( + db: Path, recheck_after_s: float | None = 60, settle_attempts: int = 1 +) -> dict[str, Any]: + database: dict[str, Any] = { + "dialect": "sqlite", + "read": {"sqlite_database": str(db)}, + "allow_tables": ["log", "api_log"], + "views": {"patient_data": ""}, + "new_row_fields": { + "patient_data": ["pid", "regdate"], + "history_data": ["date"], + }, + "settle_s": 0, + } + if recheck_after_s is not None: + database["recheck_after_s"] = recheck_after_s + if settle_attempts != 1: + database["settle_attempts"] = settle_attempts + return { + "schema_version": "openadapt.verify/v1", + "verifier": { + "kind": "sql", + "sqlite_database": str(db), + "sql_query": "SELECT * FROM patient_data", + }, + "record": { + "name": "patient", + "match": ["email", "lname"], + "fields": {"fname": "fname", "lname": "lname", "email": "email"}, + }, + "frame": { + "identity_field": "id", + "side_records": [ + { + "surface": "history_data", + "link_field": "pid", + "of_surface": "patient_data", + "of_field": "pid", + } + ], + "attribution": { + "kind": "audit_read", + "actor_field": "user", + "self": ["bot"], + "record_field": "patient_id", + "record_key": "pid", + "event_field": "event", + "write_events": ["insert", "update", "delete"], + "insert_events": ["insert"], + "delete_events": ["delete"], + "table_field": "tbl", + "read": { + "kind": "sql", + "sqlite_database": str(db), + "sql_query": "SELECT * FROM log", + }, + }, + "database": database, + }, + } + + +def _sql(db: Path, *statements: str) -> None: + con = sqlite3.connect(db) + for statement in statements: + con.execute(statement) + con.commit() + con.close() + + +def _save_sql(n: int) -> list[str]: + """The app's own save of patient ``n``: the chart, its history row and + the bot's log lines.""" + + p = _params(n) + return [ + "INSERT INTO patient_data (id, pid, fname, lname, email, regdate) VALUES " + f"({n}, {n}, '{p['fname']}', '{p['lname']}', '{p['email']}', '2026-10-29')", + f"INSERT INTO history_data (id, pid, date) VALUES ({n}, {n}, '2026-10-29')", + "INSERT INTO log (user, patient_id, event, tbl) VALUES " + f"('bot', {n}, 'insert', 'patient_data'), " + f"('bot', {n}, 'insert', 'history_data')", + ] + + +def _save(contract: Contract, db: Path, n: int, *stray: str) -> WriteResult: + before = begin_write(contract, _params(n), actor=BOT) + _sql(db, *_save_sql(n), *stray) + return verify_write(before) + + +def _first(tmp_path: Path, **doc: Any) -> tuple[Contract, Path, WriteResult]: + db = _db(tmp_path) + contract = Contract.from_dict(_doc(db, **doc)) + result = _save(contract, db, 2) + assert result.verified, result.reasons + return contract, db, result + + +def _recheck( + contract: Contract, result: WriteResult | RecheckState, n: int = 2, **kw: Any +) -> RecheckResult: + return recheck(contract, result, params=_params(n), actor=BOT, **kw) + + +# -- scheduling ---------------------------------------------------------------- + + +def test_a_verified_write_says_when_its_recheck_is_due(tmp_path: Path) -> None: + _, _, result = _first(tmp_path) + schedule = result.evidence["frame"]["recheck"] + assert result.recheck is not None + assert result.recheck_due_at == result.receipt["recheck_due_at"] + assert result.recheck_due_at == schedule["due_at"] == result.recheck.due_at + assert schedule["compares"] == "database_since_post_save" + assert schedule["after_s"] == 60 + saved = result.recheck.save(tmp_path / "recheck.json") + assert os.stat(saved).st_mode & 0o777 == 0o600 + text = saved.read_text() + # Keyed MACs only: no record values reach the disk. + for value in ("Quill2", "ada2@example.test", "Wendell", "w@example.test"): + assert value not in text + assert "Quill2" not in repr(result.recheck) + + +def test_without_recheck_after_s_nothing_changes(tmp_path: Path) -> None: + db = _db(tmp_path) + contract = Contract.from_dict(_doc(db, recheck_after_s=None)) + result = _save(contract, db, 2) + assert result.verified + assert result.recheck is None and result.recheck_due_at is None + assert "recheck_due_at" not in result.receipt + assert "recheck" not in result.evidence["frame"] + # Unset, the field stays out of the frame's digest. + spec = contract.config.frame + assert spec is not None and spec.database is not None + assert "recheck_after_s" not in spec.database.model_dump() + with pytest.raises(VerifierError, match="no recheck"): + _recheck(contract, result) + + +def test_a_write_that_was_not_verified_has_nothing_to_recheck(tmp_path: Path) -> None: + db = _db(tmp_path) + contract = Contract.from_dict(_doc(db)) + stray = "UPDATE patient_data SET phone_cell = '555-0100' WHERE pid = 1" + result = _save(contract, db, 2, stray) + assert result.status == "refuted" + assert result.recheck_due_at is None and "recheck_due_at" not in result.receipt + with pytest.raises(VerifierError, match="no recheck"): + _recheck(contract, result) + + +def test_a_recheck_before_its_due_time_is_refused(tmp_path: Path) -> None: + contract, _, result = _first(tmp_path, recheck_after_s=3600) + with pytest.raises(RecheckNotDue) as raised: + _recheck(contract, result) + assert raised.value.due_at == result.recheck_due_at + assert raised.value.wait_s > 3000 + + +def test_wait_sleeps_until_the_recheck_is_due(tmp_path: Path) -> None: + contract, _, result = _first(tmp_path, recheck_after_s=1) + late = _recheck(contract, result, wait=True) + assert late.outcome == "CONFIRMED", late.reasons + assert late.read_at >= late.due_at + + +# -- what a recheck catches and excuses --------------------------------------------- + + +def test_nothing_moved_since_the_verdict_confirms( + tmp_path: Path, due_now: None +) -> None: + contract, _, result = _first(tmp_path) + late = _recheck(contract, result) + assert late.outcome == "CONFIRMED" and late.verified and not late.late + assert late.status == "verified" + assert late.receipt_sha256 == result.receipt_sha256 + + +@pytest.mark.parametrize( + "collateral", + [ + "UPDATE patient_data SET email = 'x@example.test' WHERE pid = 1", + "UPDATE history_data SET tobacco = 'Never' WHERE pid = 2", + "INSERT INTO calendar (pid, title) VALUES (2, 'New Patient')", + "INSERT INTO history_data (id, pid, date) VALUES (9, 2, '2026-10-29')", + "DELETE FROM history_data WHERE pid = 1", + ], +) +def test_late_collateral_refutes_the_earlier_verdict( + tmp_path: Path, due_now: None, collateral: str +) -> None: + contract, db, result = _first(tmp_path) + _sql(db, collateral) # after the verdict: a background job, a trigger + late = _recheck(contract, result) + assert late.outcome == "REFUTED", collateral + assert late.late and not late.verified and late.status == "refuted" + assert late.evidence["collateral_count"] >= 1 + assert "no longer stands" in late.reasons[0] + + +def test_a_later_verified_write_excuses_its_own_rows( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + second = _save(contract, db, 3) + assert second.verified, second.reasons + # Without the excuse, the next patient's rows are unexplained. + assert _recheck(contract, first).outcome == "REFUTED" + late = _recheck(contract, first, excused_by=[second]) + assert late.outcome == "CONFIRMED", late.reasons + assert late.excused == 1 + + +def test_a_later_write_excuses_only_its_own_rows(tmp_path: Path, due_now: None) -> None: + contract, db, first = _first(tmp_path) + second = _save(contract, db, 3) + assert second.verified + _sql(db, "UPDATE patient_data SET phone_cell = '555-0101' WHERE pid = 1") + late = _recheck(contract, first, excused_by=[second]) + assert late.outcome == "REFUTED" + + +def test_what_refuted_a_later_write_leaves_the_earlier_one_uncertain( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + stray = "UPDATE patient_data SET phone_cell = '555-0102' WHERE pid = 1" + second = _save(contract, db, 3, stray) + assert second.status == "refuted" # reported, and stopped, against it + # Without the later write, its rows and the stray refute the earlier one. + assert _recheck(contract, first).outcome == "REFUTED" + # With it, the stray is what refuted the later write, so it doesn't also + # refute the earlier one; but the harm may be the earlier write's. + late = _recheck(contract, first, excused_by=[second]) + assert late.outcome == "UNCERTAIN", late.reasons + assert "may be what refuted it" in late.reasons[0] + # A change no later write accounts for still refutes the earlier one. + _sql(db, "INSERT INTO calendar (pid, title) VALUES (1, 'Follow-up')") + assert _recheck(contract, first, excused_by=[second]).outcome == "REFUTED" + + +def test_p1_a_duplicate_after_the_later_writes_window_refutes( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + second = _save(contract, db, 3) + assert second.verified + # After the later write's verdict: a delayed duplicate chart that + # matches its selector. Its window never saw it. + p = _params(3) + _sql( + db, + "INSERT INTO patient_data (id, pid, fname, lname, email, regdate) VALUES " + f"(30, 3, '{p['fname']}', '{p['lname']}', '{p['email']}', '2026-10-29')", + ) + assert _recheck(contract, first, excused_by=[second]).outcome == "REFUTED" + + +def test_p2_a_later_write_that_landed_late_excuses_nothing( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + before = begin_write(contract, _params(3), actor=BOT) + second = verify_write(before) + assert second.status == "unconfirmed" # its save didn't land in its window + assert second.recheck is not None + _sql(db, *_save_sql(3)) # its rows land after its window + assert _recheck(contract, first, excused_by=[second]).outcome == "REFUTED" + with pytest.raises(VerifierError): + recheck(contract, second, params=_params(3), actor=BOT) + + +def test_p3_the_earlier_writes_late_harm_inside_a_later_window_is_uncertain( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + before = begin_write(contract, _params(3), actor=BOT) + # The first write's background job lands while the second write runs. + harm = "UPDATE history_data SET tobacco = 'Current' WHERE pid = 2" + _sql(db, *_save_sql(3), harm) + second = verify_write(before) + assert second.status == "refuted" + late = _recheck(contract, first, excused_by=[second]) + assert late.outcome == "UNCERTAIN", late.reasons + + +def test_p4_a_removal_a_later_window_reported_is_uncertain( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + before = begin_write(contract, _params(3), actor=BOT) + _sql(db, *_save_sql(3), "DELETE FROM history_data WHERE pid = 1") + second = verify_write(before) + assert second.status == "refuted" + late = _recheck(contract, first, excused_by=[second]) + assert late.outcome == "UNCERTAIN", late.reasons + + +def test_a_removal_a_verified_later_window_explained_is_excused( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + before = begin_write(contract, _params(3), actor=BOT) + _sql( + db, + *_save_sql(3), + "DELETE FROM history_data WHERE pid = 1", + "INSERT INTO log (user, patient_id, event, tbl) " + "VALUES ('nurse', 1, 'delete', 'history_data')", + ) + second = verify_write(before) + assert second.verified, second.reasons # the nurse's delete, attributed + assert _recheck(contract, first).outcome == "REFUTED" + late = _recheck(contract, first, excused_by=[second]) + assert late.outcome == "CONFIRMED", late.reasons + + +def test_a_removal_no_later_window_recorded_refutes( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + second = _save(contract, db, 3) + assert second.verified + _sql(db, "DELETE FROM history_data WHERE pid = 1") # after its window + assert _recheck(contract, first, excused_by=[second]).outcome == "REFUTED" + + +def test_a_state_not_issued_here_is_uncertain_and_never_excuses( + tmp_path: Path, due_now: None, monkeypatch: pytest.MonkeyPatch +) -> None: + contract, db, first = _first(tmp_path) + second = _save(contract, db, 3) + assert first.recheck is not None and second.recheck is not None + # Another state directory: no issuance record for either state. + monkeypatch.setattr(verifier, "USED_DIR", tmp_path / "elsewhere") + late = _recheck(contract, first) + assert late.outcome == "UNCERTAIN" + assert "not issued" in late.reasons[0] + monkeypatch.setattr(verifier, "USED_DIR", tmp_path / "used") + assert _recheck(contract, first, excused_by=[second]).outcome == "CONFIRMED" + record = tmp_path / "used" / "rechecks" + (record / verifier._state_digest(second.recheck.payload)).unlink() + with pytest.raises(VerifierError, match="not issued"): + _recheck(contract, first, excused_by=[second]) + + +def test_the_due_time_is_never_early(tmp_path: Path) -> None: + from openadapt_flow.runtime.effects.late_write import due_at + + # The read time is truncated to the second; the read may be up to a + # second later, so the due time adds that second. + assert due_at("2026-10-29T12:00:00Z", 60) == "2026-10-29T12:01:01Z" + assert due_at("2026-10-29T12:00:00Z", 0.5) == "2026-10-29T12:00:02Z" + + +def test_a_row_changed_again_after_the_later_write_is_not_excused( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + stray = "UPDATE patient_data SET phone_cell = '555-0103' WHERE pid = 1" + second = _save(contract, db, 3, stray) + assert second.status == "refuted" + _sql(db, "UPDATE patient_data SET phone_cell = '555-0104' WHERE pid = 1") + assert _recheck(contract, first, excused_by=[second]).outcome == "REFUTED" + + +def test_another_actors_attributed_write_is_excused( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + _sql( + db, + "UPDATE patient_data SET phone_cell = '555-0105' WHERE pid = 1", + "INSERT INTO log (user, patient_id, event, tbl) " + "VALUES ('nurse', 1, 'update', 'patient_data')", + ) + late = _recheck(contract, first) + assert late.outcome == "CONFIRMED", late.reasons + assert late.evidence["concurrent_count"] == 1 + + +def test_the_bots_own_unattributed_write_is_not_excused( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + _sql( + db, + "UPDATE patient_data SET phone_cell = '555-0106' WHERE pid = 1", + "INSERT INTO log (user, patient_id, event, tbl) " + "VALUES ('bot', 1, 'update', 'patient_data')", + ) + assert _recheck(contract, first).outcome == "REFUTED" + + +def test_reviewed_bookkeeping_is_excused(tmp_path: Path, due_now: None) -> None: + contract, db, first = _first(tmp_path) + _sql(db, "INSERT INTO api_log (note) VALUES ('token refresh')") + late = _recheck(contract, first) + assert late.outcome == "CONFIRMED", late.reasons + + +# -- fail-closed ------------------------------------------------------------------ + + +def test_an_unreadable_database_is_uncertain(tmp_path: Path, due_now: None) -> None: + contract, db, first = _first(tmp_path) + db.rename(db.with_suffix(".moved")) + late = _recheck(contract, first) + assert late.outcome == "UNCERTAIN" and not late.verified + assert late.status == "unconfirmed" + assert late.frame == "not_checked" + + +def test_a_database_still_moving_is_uncertain( + tmp_path: Path, due_now: None, monkeypatch: pytest.MonkeyPatch +) -> None: + from openadapt_flow import verify_run + + contract, db, first = _first(tmp_path) + real = verify_run._read_database + calls = {"n": 0} + + def moving(cfg: Any, salt: bytes, before: Any = None) -> Any: + calls["n"] += 1 + if calls["n"] == 2: # between the two reads, a job writes + _sql(db, "INSERT INTO calendar (pid, title) VALUES (1, 'x')") + return real(cfg, salt, before=before) + + monkeypatch.setattr(verify_run, "_read_database", moving) + assert _recheck(contract, first).outcome == "UNCERTAIN" + + +def test_a_saved_state_round_trips(tmp_path: Path, due_now: None) -> None: + contract, db, first = _first(tmp_path) + assert first.recheck is not None + path = first.recheck.save(tmp_path / "first.json") + second = _save(contract, db, 3) + assert second.recheck is not None + second_path = second.recheck.save(tmp_path / "second.json") + loaded = RecheckState.load(path) + later = RecheckState.load(second_path, params=_params(3)) + assert _recheck(contract, loaded, excused_by=[later]).outcome == "CONFIRMED" + with pytest.raises(VerifierError, match="params"): + _recheck(contract, loaded, excused_by=[RecheckState.load(second_path)]) + + +@pytest.mark.parametrize( + "tamper", + [ + lambda s: s["post_state"].update(read_at="2001-01-01T00:00:00Z"), + lambda s: s["post_state"]["snapshot"]["tables"].pop("calendar"), + lambda s: s.update(salt="00" * 32), + lambda s: s["receipt"].update(outcome="verified", effect_verdict="REFUTED"), + lambda s: s["evidence"]["frame"]["recheck"].update(due_at=None), + lambda s: s["baseline"].update(baseline_read_at="2001-01-01T00:00:00Z"), + ], +) +def test_a_mismatched_state_is_uncertain( + tmp_path: Path, due_now: None, tamper: Any +) -> None: + contract, _, first = _first(tmp_path) + assert first.recheck is not None + payload = json.loads(json.dumps(first.recheck.payload)) + tamper(payload) + late = _recheck(contract, RecheckState(payload)) + assert late.outcome == "UNCERTAIN", late.reasons + assert "can't be rechecked" in late.reasons[0] + + +def test_another_contract_actor_or_params_is_uncertain( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + other = Contract.from_dict(_doc(db, recheck_after_s=120)) + assert _recheck(other, first).outcome == "UNCERTAIN" + assert _recheck(contract, first, n=7).outcome == "UNCERTAIN" + assert ( + recheck( + contract, first, params=_params(2), actor=Actor("agent", "someone-else") + ).outcome + == "UNCERTAIN" + ) + + +def test_an_unreadable_state_file_raises(tmp_path: Path) -> None: + path = tmp_path / "broken.json" + path.write_text("{not json") + with pytest.raises(VerifierError): + RecheckState.load(path) + path.write_text(json.dumps({"schema_version": "openadapt.verify-recheck/v1"})) + with pytest.raises(VerifierError): + RecheckState.load(path) + + +def test_an_excuse_that_began_before_the_post_save_read_is_refused( + tmp_path: Path, due_now: None +) -> None: + db = _db(tmp_path) + contract = Contract.from_dict(_doc(db)) + early = begin_write( + contract, _params(3), actor=BOT + ) # before the first write's verdict + time.sleep(1.1) # whole-second times: a clear second apart + first = _save(contract, db, 2) + assert first.verified + _sql(db, *_save_sql(3)) + second = verify_write(early) + with pytest.raises(VerifierError, match="started before"): + _recheck(contract, first, excused_by=[second]) + + +def test_an_excuse_by_another_actor_is_refused(tmp_path: Path, due_now: None) -> None: + contract, db, first = _first(tmp_path) + before = begin_write(contract, _params(3), actor=Actor("agent", "other-bot")) + _sql(db, *_save_sql(3)) + theirs = verify_write(before) + assert theirs.recheck is not None + with pytest.raises(VerifierError, match="different actor"): + _recheck(contract, first, excused_by=[theirs]) + + +def test_frame_digest_is_unchanged_without_the_field() -> None: + spec = FrameSpec.model_validate( + {"database": {"dialect": "sqlite", "read": {"sqlite_database": "x"}}} + ) + with_field = FrameSpec.model_validate( + { + "database": { + "dialect": "sqlite", + "read": {"sqlite_database": "x"}, + "recheck_after_s": 30, + } + } + ) + assert spec.digest() != with_field.digest() + assert "recheck_after_s" not in json.dumps(spec.model_dump()) + + +def test_the_state_is_saved_beside_the_writes_ledger_folder(tmp_path: Path) -> None: + _, _, result = _first(tmp_path) + assert result.recheck is not None and result.recheck.path is not None + assert result.recheck.path.name == "recheck.json" + assert result.receipt_path is not None + assert result.recheck.path.parent == Path(result.receipt_path).parent + assert os.stat(result.recheck.path).st_mode & 0o777 == 0o600 + assert result.to_dict()["recheck_due_at"] == result.recheck_due_at + # The scope says the result is provisional until the recheck runs. + assert any( + f"recheck due at {result.recheck_due_at}" in line + for line in result.scope.not_covered + ) + + +def test_without_a_recheck_the_scope_keeps_delayed_writes_uncovered( + tmp_path: Path, +) -> None: + db = _db(tmp_path) + contract = Contract.from_dict(_doc(db, recheck_after_s=None)) + result = _save(contract, db, 2) + assert "writes that land after read_at (delayed or queued writes)" in ( + result.scope.not_covered + ) + + +def test_the_deprecated_capture_form_schedules_a_recheck_too( + tmp_path: Path, due_now: None +) -> None: + db = _db(tmp_path) + contract = Contract.from_dict(_doc(db)) + with pytest.warns(DeprecationWarning): + before = verifier.capture(contract, _params(2)) + _sql(db, *_save_sql(2)) + result = verify_write(contract, before, params=_params(2), actor=BOT) + assert result.verified and result.recheck is not None + _sql(db, "UPDATE patient_data SET email = 'x@example.test' WHERE pid = 1") + assert _recheck(contract, result).outcome == "REFUTED" + + +def test_a_confirmed_write_downgraded_to_uncertain_cannot_be_rechecked_as_verified( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + assert first.recheck is not None and first.recheck_due_at + # As when another attempt with the same values was open (_check_overlap). + down = verifier._downgrade(first, "another attempt with these values was open") + assert down.outcome.value == "UNCERTAIN" and down.recheck_due_at is None + for state in (down, first, first.recheck): + late = _recheck(contract, state) + assert late.outcome == "UNCERTAIN", late.reasons + assert "downgraded" in late.reasons[0] + + +def test_a_downgraded_later_write_still_excuses_its_own_window( + tmp_path: Path, due_now: None +) -> None: + contract, db, first = _first(tmp_path) + second = _save(contract, db, 3) + assert second.verified and second.recheck is not None + down = verifier._downgrade(second, "another attempt with these values was open") + # Its frame held, so its own rows still don't refute the earlier write. + assert _recheck(contract, first, excused_by=[down]).outcome == "CONFIRMED" + assert _recheck(contract, first).outcome == "REFUTED" + + +# -- settling: a live app moves on its own ------------------------------------------ + + +def _moving( + monkeypatch: pytest.MonkeyPatch, db: Path, writes: dict[int, list[str]] +) -> dict[str, int]: + """Run ``writes[n]`` just before the recheck's n-th database read.""" + + from openadapt_flow import verify_run + + real = verify_run._read_database + calls = {"n": 0} + + def read(cfg: Any, salt: bytes, before: Any = None) -> Any: + calls["n"] += 1 + if writes.get(calls["n"]) or writes.get(0): + _sql(db, *(writes.get(calls["n"]) or writes[0])) + return real(cfg, salt, before=before) + + monkeypatch.setattr(verify_run, "_read_database", read) + return calls + + +def test_a_recheck_that_settles_on_a_later_attempt_gives_the_normal_verdict( + tmp_path: Path, due_now: None, monkeypatch: pytest.MonkeyPatch +) -> None: + contract, db, first = _first(tmp_path, settle_attempts=3) + # Between the first two reads, another user edits a chart (attributed); + # then the database is still, so the second and third reads match. + _moving( + monkeypatch, + db, + { + 2: [ + "UPDATE patient_data SET phone_cell = '555-0107' WHERE pid = 1", + "INSERT INTO log (user, patient_id, event, tbl) " + "VALUES ('nurse', 1, 'update', 'patient_data')", + ] + }, + ) + late = _recheck(contract, first) + assert late.outcome == "CONFIRMED", late.reasons + assert late.evidence["database"]["settle_attempts"] == 2 + assert late.evidence["concurrent_count"] == 1 + + +def test_a_recheck_that_never_settles_is_uncertain( + tmp_path: Path, due_now: None, monkeypatch: pytest.MonkeyPatch +) -> None: + contract, db, first = _first(tmp_path, settle_attempts=3) + calls = _moving( + monkeypatch, db, {0: ["INSERT INTO calendar (pid, title) VALUES (1, 'x')"]} + ) + late = _recheck(contract, first) + assert late.outcome == "UNCERTAIN" + assert calls["n"] == 4 # the first read and 3 more, then it gives up + assert late.evidence["database"] == { + "settle_attempts": 3, + "settle_attempts_allowed": 3, + "settled": False, + } + assert "3 attempt(s)" in late.reasons[0] + + +def test_bookkeeping_only_movement_counts_as_settled( + tmp_path: Path, due_now: None, monkeypatch: pytest.MonkeyPatch +) -> None: + contract, db, first = _first(tmp_path) # one pair only + # The app's own log grows on every read (an append-only table). + _moving(monkeypatch, db, {0: ["INSERT INTO api_log (note) VALUES ('poll')"]}) + late = _recheck(contract, first) + assert late.outcome == "CONFIRMED", late.reasons + assert late.evidence["database"]["settle_attempts"] == 1 + + +def test_a_changed_log_row_is_still_movement( + tmp_path: Path, due_now: None, monkeypatch: pytest.MonkeyPatch +) -> None: + contract, db, first = _first(tmp_path) + _sql(db, "INSERT INTO api_log (note) VALUES ('seed')") + # An existing append-only row edited between the reads is not + # bookkeeping: the pair doesn't match. + _moving(monkeypatch, db, {2: ["UPDATE api_log SET note = 'edited'"]}) + assert _recheck(contract, first).outcome == "UNCERTAIN" + + +def test_the_save_itself_settles_within_the_same_budget( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + db = _db(tmp_path) + contract = Contract.from_dict(_doc(db, settle_attempts=2)) + before = begin_write(contract, _params(2), actor=BOT) + _sql(db, *_save_sql(2)) + # The app's scheduler writes an append-only row while verify reads. + _moving(monkeypatch, db, {0: ["INSERT INTO api_log (note) VALUES ('tick')"]}) + result = verify_write(before) + assert result.verified, result.reasons + assert result.evidence["frame"]["database"]["settle_attempts"] == 1 + + +def test_settle_attempts_at_its_default_stays_out_of_the_digest() -> None: + plain = FrameSpec.model_validate( + {"database": {"dialect": "sqlite", "read": {"sqlite_database": "x"}}} + ) + explicit = FrameSpec.model_validate( + { + "database": { + "dialect": "sqlite", + "read": {"sqlite_database": "x"}, + "settle_attempts": 1, + } + } + ) + assert plain.digest() == explicit.digest() + assert "settle_attempts" not in json.dumps(plain.model_dump())