From 8b4df94caf53d482a89677a7d5981d4ed054d46c Mon Sep 17 00:00:00 2001 From: abrichr Date: Wed, 4 Nov 2026 21:46:39 -0500 Subject: [PATCH] Recheck provisional writes against complete change history Signed-off-by: abrichr --- packages/flow/docs/VERIFIER_API.md | 30 + .../flow/openadapt_flow/agent/verify_tools.py | 1 + .../openadapt_flow/runtime/effects/frame.py | 11 + .../openadapt_flow/runtime/effects/history.py | 528 ++++++++++++++++++ .../runtime/effects/late_write.py | 12 +- packages/flow/openadapt_flow/verifier.py | 23 +- packages/flow/openadapt_flow/verify_run.py | 171 +++++- packages/flow/openadapt_flow/write_receipt.py | 11 +- packages/flow/tests/test_history_recheck.py | 489 ++++++++++++++++ 9 files changed, 1263 insertions(+), 13 deletions(-) create mode 100644 packages/flow/openadapt_flow/runtime/effects/history.py create mode 100644 packages/flow/tests/test_history_recheck.py diff --git a/packages/flow/docs/VERIFIER_API.md b/packages/flow/docs/VERIFIER_API.md index f34e89a93d..45c4023439 100644 --- a/packages/flow/docs/VERIFIER_API.md +++ b/packages/flow/docs/VERIFIER_API.md @@ -474,3 +474,33 @@ A snapshot doesn't go through the ledger, so move to call without a `write_key` counts as a new write. A cancellation the caller sees isn't affected (`delivered=False`). `write_key` closes the remaining gap for callers that have a stable task id. + +### Final verdicts from history + +A contract can set `frame.history` to read complete change feeds across its +record surfaces and declared collateral scope. Each surface maps to a feed; +additional feeds extend the collateral scope. The quiet window `quiet_s` is +chosen from server behavior observed during sandbox setup, and +`setup_evidence_sha256` pins that observation. + +FHIR feeds read every page of the type-level `_history?_since=t0`, including +deletions and earlier versions. SQL feeds read CDC or audit events and an +independent coverage query in one read-only transaction. The coverage query +returns `retained_since`, `through`, and `event_count`; it must describe the +source's complete retained history, independently of the event query. The +SQL event query returns `event_id`, `record_id`, `operation`, `author`, and +`occurred_at`. Both time columns and event times include a timezone. + +A confirmed initial check has `provisional: true` and `final: false` while its +recheck is owed. A consumer that requires a final verdict calls +`recheck(contract, result, params=params, actor=actor, wait=True)`. The ledger +keeps the recheck owed across restarts. The final verdict covers the earliest +coverage frontier returned by all feeds. A write after that frontier remains +outside the receipt's time window. + +Every version and tombstone is checked, so reverting a late edit does not +remove its history. Only server-controlled provenance can attribute a late +edit to another user. Missing coverage, a vanished or changed history entry, +an unreadable page, or an unresolved actor prevents confirmation. The history +checkpoint stores keyed event digests; record values and actor identities do +not enter the receipt. diff --git a/packages/flow/openadapt_flow/agent/verify_tools.py b/packages/flow/openadapt_flow/agent/verify_tools.py index 1c4106b4d8..c00d0c283b 100644 --- a/packages/flow/openadapt_flow/agent/verify_tools.py +++ b/packages/flow/openadapt_flow/agent/verify_tools.py @@ -455,6 +455,7 @@ def _recheck(self, args: Mapping[str, Any]) -> dict[str, Any]: "verified": late.verified, "outcome": late.outcome, "late": late.late, + "final": late.final, "reasons": list(late.reasons), "recheck_due_at": late.due_at, "read_at": late.read_at, diff --git a/packages/flow/openadapt_flow/runtime/effects/frame.py b/packages/flow/openadapt_flow/runtime/effects/frame.py index dbc92dcf96..98a579491e 100644 --- a/packages/flow/openadapt_flow/runtime/effects/frame.py +++ b/packages/flow/openadapt_flow/runtime/effects/frame.py @@ -85,6 +85,7 @@ selector_value, stable_id, ) +from openadapt_flow.runtime.effects.history import FrameHistory from openadapt_flow.runtime.effects.learned import ( NOT_COMPARED, LearnedFrame, @@ -508,6 +509,8 @@ class FrameSpec(BaseModel): #: (``versions.FrameVersions``): a version in the window that isn't the #: agent's own is a lost update. versions: Optional[FrameVersions] = None + #: Complete type/table change feeds for a provisional then final verdict. + history: Optional[FrameHistory] = None #: Moves a person approved from ``openadapt-flow learn`` #: (``learned.LearnedFrame``): the app's own writes on the acting user's #: row or the write's own rows, by table, column and change. @@ -518,6 +521,8 @@ def _compact(self, handler: Any) -> dict[str, Any]: # Unset, ``versions`` is left out, so a contract that doesn't use it # keeps its digests (``frame_sha256``, the check digest). data: dict[str, Any] = handler(self) + if data.get("history") is None: + data.pop("history", None) if data.get("versions") is None: data.pop("versions", None) if data.get("learned") is None: @@ -529,6 +534,12 @@ def _compact(self, handler: Any) -> dict[str, Any]: @model_validator(mode="after") def _database_attribution(self) -> "FrameSpec": a = self.attribution + if ( + self.history is not None + and self.database is not None + and self.database.recheck_after_s is not None + ): + raise ValueError("choose history or database recheck scheduling, not both") if ( self.database is not None and a is not None diff --git a/packages/flow/openadapt_flow/runtime/effects/history.py b/packages/flow/openadapt_flow/runtime/effects/history.py new file mode 100644 index 0000000000..d1cd810c2c --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/effects/history.py @@ -0,0 +1,528 @@ +"""Complete change feeds for a provisional write's final recheck. + +A feed covers a whole declared resource type or table, including tombstones. +The checkpoint is read before the ordinary post-save checks, so no write +between their last read and the checkpoint can disappear into the baseline. +Only keyed event digests are persisted. Unknown coverage never confirms. +""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import re +from collections.abc import Callable, Mapping +from datetime import datetime, timedelta, timezone +from email.utils import parsedate_to_datetime +from typing import Any, Literal, Optional +from urllib.parse import urljoin, urlsplit + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from openadapt_flow.runtime.effects.effect import ValueExpr + +HISTORY_SCHEMA = "openadapt.history-checkpoint/v1" +DETAIL = "openadapt:frame_history" +_ID = re.compile(r"^[A-Za-z0-9\-.]{1,64}$") + + +class HistoryFeed(BaseModel): + """One complete type/table feed, read independently of the writer. + + SQL queries return event_id, record_id, operation, author and occurred_at. + The coverage query returns exactly one row: retained_since, through and + event_count. Both queries run in one read-only repeatable-read snapshot. + Its retention boundary/count must come from the source's CDC/audit + coverage, independently of the event query (never MIN/COUNT of that + query's possibly filtered result). + """ + + model_config = ConfigDict(extra="forbid") + read: dict[str, Any] + #: A server-controlled history-entry path, never client-authored data. + author_path: Optional[str] = None + author_server_set: bool = False + dialect: Optional[Literal["sqlite", "mysql", "postgresql"]] = None + coverage_query: Optional[str] = None + + @model_validator(mode="after") + def _shape(self) -> "HistoryFeed": + kind = self.read.get("kind") + if kind not in ("fhir", "sql") or self.read.get("surfaces"): + raise ValueError("history feeds require one FHIR or SQL read") + if kind == "fhir": + if self.read.get("search_params") or self.read.get("search_param_exprs"): + raise ValueError("type-level history must not filter candidate records") + if not _ID.fullmatch(str(self.read.get("resource_type") or "")): + raise ValueError("history needs a plain resource type") + if self.coverage_query is not None or self.dialect is not None: + raise ValueError("SQL coverage settings apply only to SQL") + elif self.dialect is None or not self.coverage_query: + raise ValueError("SQL history needs dialect and coverage_query") + if kind == "sql": + driver = ( + "sqlite3" + if self.read.get("sqlite_database") + else self.read.get("sql_driver") + ) + allowed = { + "sqlite": {"sqlite3"}, + "mysql": {"pymysql", "MySQLdb", "mysql.connector"}, + "postgresql": {"psycopg", "psycopg2"}, + } + if driver not in allowed.get(str(self.dialect), set()): + raise ValueError( + "history dialect must match a supported read-only driver" + ) + if self.read.get("sql_query_params"): + raise ValueError("history must not filter by candidate parameters") + if kind == "sql" and self.author_path not in (None, "author"): + raise ValueError( + "SQL history attribution uses its server-set author column" + ) + if bool(self.author_path) != self.author_server_set: + raise ValueError("history attribution requires a server-set author path") + return self + + +class FrameHistory(BaseModel): + """Setup-qualified whole-type/table feeds and the app's quiet window. + + Feed names are opaque contract scope names. ``surfaces`` maps every + verifier surface to its feed. Additional feeds declare collateral scope. + ``setup_evidence_sha256`` pins the sandbox observation used to choose W; + evaluation cases must never choose it. Feeds must include deletions and + every version, and the verifier identity must see the whole declared + scope. These are qualification obligations, not inferred from HTTP 200. + """ + + model_config = ConfigDict(extra="forbid", populate_by_name=True) + feeds: dict[str, HistoryFeed] = Field(min_length=1) + surfaces: dict[str, str] = Field(min_length=1) + quiet_s: float = Field(gt=0, allow_inf_nan=False) + setup_evidence_sha256: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") + self_actors: list[ValueExpr] = Field(alias="self", min_length=1) + max_events: int = Field(default=100000, ge=1) + + @model_validator(mode="before") + @classmethod + def _self(cls, data: Any) -> Any: + if isinstance(data, dict): + data = dict(data) + name = "self" if "self" in data else "self_actors" + if isinstance(data.get(name), list): + data[name] = [ + {"literal": v} if isinstance(v, str) else v for v in data[name] + ] + return data + + @model_validator(mode="after") + def _scope(self) -> "FrameHistory": + if not set(self.surfaces.values()) <= set(self.feeds): + raise ValueError("every history surface must name a declared feed") + return self + + +def instant(value: Any) -> datetime: + if not isinstance(value, str): + raise ValueError("history time is missing") + dt = datetime.fromisoformat(value.replace("Z", "+00:00")) + if dt.tzinfo is None: + raise ValueError("history time has no timezone") + return dt.astimezone(timezone.utc) + + +def _mac(key: bytes, value: Any) -> str: + payload = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + return hmac.new(key, payload.encode(), hashlib.sha256).hexdigest() + + +def _path(row: Mapping[str, Any], path: str) -> Any: + node: Any = row + for part in path.split("."): + if not isinstance(node, Mapping): + return None + node = node.get(part) + return node + + +def _sql( + feed: HistoryFeed, reader: Any, budget: int +) -> tuple[list[dict[str, Any]], datetime, datetime]: + from openadapt_flow.runtime.effects.db_snapshot import _Session + from openadapt_flow.runtime.effects.sql import assert_read_only_sql + + if reader.query_params: + raise ValueError("history queries cannot depend on candidate parameters") + query = assert_read_only_sql(str(feed.coverage_query)) + assert feed.dialect is not None + session = _Session(reader._connect, feed.dialect) + try: + if feed.dialect == "sqlite": + session.cur.execute("PRAGMA query_only = ON") + session.begin() + if feed.dialect == "mysql": + engines = session.rows( + "SELECT ENGINE FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_TYPE = 'BASE TABLE'" + ) + if any(str(row[0]).lower() != "innodb" for row in engines): + raise ValueError( + "history coverage requires transactional source tables" + ) + coverage = session.dicts(query, limit=2) + rows = session.dicts(reader.query, limit=budget + 1) + finally: + session.close() + if len(coverage) != 1 or len(rows) > budget: + raise ValueError("history coverage or event budget is incomplete") + cov = coverage[0] + count = cov.get("event_count") + if isinstance(count, bool) or not isinstance(count, int) or count != len(rows): + raise ValueError("history event count disagrees with independent coverage") + return ( + rows, + instant(str(cov.get("retained_since"))), + instant(str(cov.get("through"))), + ) + + +def _fhir( + feed: HistoryFeed, reader: Any, since: str, budget: int +) -> tuple[list[dict[str, Any]], datetime, datetime]: + """Read every page of type-level history, including resource-less deletes.""" + from openadapt_flow.runtime.effects.fhir import MAX_PAGES, _same_origin + + headers = reader._headers() + if headers is None: + raise ValueError("history authentication is unavailable") + headers = {**headers, "Cache-Control": "no-cache, no-store"} + session = reader._get_session() + kwargs: dict[str, Any] = {"headers": headers, "timeout": reader.timeout_s} + if reader._verify_per_request: + kwargs["verify"] = reader.verify_tls + if (type(session).__module__ or "").startswith("requests"): + kwargs["allow_redirects"] = False + url = f"{reader.base_url}/{reader.resource_type}/_history" + params: Any = {"_since": since, "_count": "1000"} + seen: set[str] = set() + rows: list[dict[str, Any]] = [] + total: Any = None + through: Optional[datetime] = None + event_upper: Optional[datetime] = None + for page in range(MAX_PAGES): + if url in seen or not _same_origin(url, reader.base_url): + raise ValueError("history pagination is cyclic or crosses origins") + seen.add(url) + response = session.get(url, params=params, **kwargs) + if response.status_code == 401 and reader.token_source is not None: + reader.token_source.invalidate() + if response.status_code != 200: + raise ValueError("history read did not succeed") + landed = str(getattr(response, "url", url)) + if not _same_origin(landed, reader.base_url): + raise ValueError("history read redirected across origins") + bundle = response.json() + if ( + not isinstance(bundle, dict) + or bundle.get("resourceType") != "Bundle" + or bundle.get("type") != "history" + ): + raise ValueError("history read returned no history Bundle") + if page == 0: + total = bundle.get("total") + if isinstance(total, bool) or not isinstance(total, int) or total < 0: + raise ValueError("history has no complete event count") + if total > budget: + raise ValueError("history exceeds its event budget") + timestamp = bundle.get("timestamp") + if timestamp: + through = instant(timestamp) + event_upper = through + else: + date = getattr(response, "headers", {}).get("Date") + parsed = parsedate_to_datetime(date) if date else None + if parsed is not None and parsed.tzinfo is None: + raise ValueError("history server date has no timezone") + through = ( + parsed.astimezone(timezone.utc) if parsed is not None else None + ) + # HTTP Date has whole-second precision. Keep that lower + # coverage frontier; versions within its last fractional + # second are positive observations, not future coverage. + event_upper = through + timedelta(seconds=1) if through else None + if through is None: + raise ValueError("history has no server coverage time") + elif bundle.get("total") not in (None, total): + raise ValueError("history count changed while paging") + entries = bundle.get("entry", []) + if not isinstance(entries, list): + raise ValueError("history entries are malformed") + for entry in entries: + if not isinstance(entry, dict): + raise ValueError("history entry is malformed") + resource = entry.get("resource") or {} + request = entry.get("request") or {} + reply = entry.get("response") or {} + if not all(isinstance(v, dict) for v in (resource, request, reply)): + raise ValueError("history entry is malformed") + method = request.get("method") + if method == "GET" and isinstance(resource.get("meta"), dict): + version = resource["meta"].get("versionId") + expected = ( + f"{reader.resource_type}/{resource.get('id')}/_history/{version}" + ) + if target := request.get("url"): + if target == expected and version: + method = ( + "PUT" # a retained version, regardless of original verb + ) + if method not in ("POST", "PUT", "PATCH", "DELETE"): + raise ValueError("history entry has no definite write operation") + record = resource.get("id") + target = request.get("url") + if method == "DELETE": + pieces = urlsplit(str(target or "")).path.strip("/").split("/") + if len(pieces) != 2 or pieces[0] != reader.resource_type: + raise ValueError("history tombstone names no definite record") + record = pieces[1] + elif resource.get("resourceType") != reader.resource_type: + raise ValueError("history entry names another resource type") + if not isinstance(record, str) or not _ID.fullmatch(record): + raise ValueError("history entry names no definite record") + meta = resource.get("meta") or {} + at = reply.get("lastModified") or meta.get("lastUpdated") + if event_upper is None or instant(at) > event_upper: + raise ValueError("history event is newer than its server response") + version = meta.get("versionId") or reply.get("etag") + if not version: + # R4 tombstones need not carry a version id. Their complete + # identity includes source, record, operation and server time. + if method != "DELETE": + raise ValueError("history resource has no version") + version = f"delete:{at}" + rows.append( + { + "event_id": [record, str(version)], + "record_id": record, + "operation": method, + "author": _path(entry, feed.author_path) + if feed.author_path + else None, + "occurred_at": at, + "resource": resource, + "record": reader._flatten(resource) if resource else None, + "contents_sha256": hashlib.sha256( + json.dumps( + entry, sort_keys=True, separators=(",", ":") + ).encode() + ).hexdigest(), + } + ) + if len(rows) > budget: + raise ValueError("history exceeds its event budget") + links = bundle.get("link", []) + if not isinstance(links, list) or any(not isinstance(x, dict) for x in links): + raise ValueError("history links are malformed") + nexts = [x.get("url") for x in links if x.get("relation") == "next"] + if len(nexts) > 1 or (nexts and not isinstance(nexts[0], str)): + raise ValueError("history next link is malformed") + if not nexts: + break + url = urljoin(url, nexts[0]) + params = None + else: + raise ValueError("history pagination is incomplete") + if len(rows) != total: + raise ValueError("history is truncated") + assert through is not None + return rows, instant(since), through + + +def read_history( + spec: FrameHistory, since: str, build: Callable[[dict[str, Any]], Any], key: bytes +) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]: + """Sealed digests plus private in-memory events and version bodies.""" + feeds: dict[str, Any] = {} + raw: dict[str, dict[str, Any]] = {} + for name, feed in spec.feeds.items(): + reader = build(feed.read) + rows, retained, through = ( + _fhir(feed, reader, since, spec.max_events) + if feed.read["kind"] == "fhir" + else _sql(feed, reader, spec.max_events) + ) + if through > datetime.now(timezone.utc): + raise ValueError("history coverage frontier is in the future") + if retained > instant(since) or through < instant(since): + raise ValueError("history does not cover the requested interval") + events: dict[str, str] = {} + for row in rows: + if any( + row.get(k) in (None, "") + for k in ("event_id", "record_id", "operation", "occurred_at") + ): + raise ValueError("history event has no identity, operation or time") + when = instant(row["occurred_at"]) + if feed.read["kind"] == "sql" and when > through: + raise ValueError("history event is newer than its coverage boundary") + if when < instant(since): + continue + event = _mac(key, [name, row["event_id"]]) + if event in events: + raise ValueError("history repeats an event identity") + events[event] = _mac(key, row) + raw[event] = {**row, "scope": name, "authoritative": feed.author_server_set} + feeds[name] = {"through": through.isoformat(), "events": events} + return {"schema_version": HISTORY_SCHEMA, "since": since, "feeds": feeds}, raw + + +def checkpoint_problem(checkpoint: Any) -> Optional[str]: + if ( + not isinstance(checkpoint, dict) + or set(checkpoint) != {"schema_version", "since", "feeds"} + or checkpoint.get("schema_version") != HISTORY_SCHEMA + ): + return "the history checkpoint is malformed" + try: + instant(checkpoint["since"]) + if not isinstance(checkpoint["feeds"], dict) or not checkpoint["feeds"]: + raise ValueError() + for feed in checkpoint["feeds"].values(): + if set(feed) != {"through", "events"} or not isinstance( + feed["events"], dict + ): + raise ValueError() + instant(feed["through"]) + if any( + not re.fullmatch(r"[a-f0-9]{64}", value) + for pair in feed["events"].items() + for value in pair + ): + raise ValueError() + except (ValueError, TypeError, KeyError, AttributeError): + return "the history checkpoint is malformed" + return None + + +def judge_history( + spec: FrameHistory, + checkpoint: Mapping[str, Any], + now: Mapping[str, Any], + raw: Mapping[str, Any], + params: Mapping[str, object], + due: str, + run_actor: str, +) -> Any: + from openadapt_flow.runtime.effects.frame import FrameChange, FrameVerdict + + if set(checkpoint["feeds"]) != set(spec.feeds) or set(now["feeds"]) != set( + spec.feeds + ): + return FrameVerdict("not_checked", "history scope is incomplete") + mine = {str(x.resolve(params) or "") for x in spec.self_actors} + if "" in mine: + return FrameVerdict("not_checked", "history self identity is unresolved") + mine.add(run_actor) + out = FrameVerdict( + "held", + "complete history contains no unexplained late write", + attribution="history", + ) + for scope, old in checkpoint["feeds"].items(): + fresh = now["feeds"][scope] + if instant(fresh["through"]) < max(instant(due), instant(old["through"])): + return FrameVerdict( + "not_checked", "history does not cover the quiet window" + ) + if any(fresh["events"].get(k) != v for k, v in old["events"].items()): + return FrameVerdict("not_checked", "history disappeared or changed") + for event in set(fresh["events"]) - set(old["events"]): + row = raw[event] + author = row.get("author") + other = ( + row["authoritative"] + and isinstance(author, str) + and bool(author) + and author not in mine + ) + kinds: dict[str, Literal["added", "changed", "removed"]] = { + "POST": "added", + "INSERT": "added", + "DELETE": "removed", + "PUT": "changed", + "PATCH": "changed", + "UPDATE": "changed", + } + change = kinds.get(str(row["operation"]).upper()) + if change is None: + return FrameVerdict("not_checked", "history write kind is unknown") + item = FrameChange( + scope, + change, + "other" + if other + else "self" + if isinstance(author, str) and author in mine + else "unattributed", + audit_only=True, + ) + (out.concurrent if other else out.collateral).append(item) + out.records_compared += 1 + if out.collateral: + out.status = "refuted" + out.reason = "complete history contains unexplained late writes" + return out + + +def judge_initial_history( + spec: FrameHistory, + before: Mapping[str, Any], + now: Mapping[str, Any], + raw: Mapping[str, Any], + params: Mapping[str, object], + run_actor: str, + targets: set[tuple[str, str]], +) -> Any: + """Every event since begin: one saved version per confirmed target. + + A correct latest snapshot cannot erase an extra intermediate mutation. + Only the declared record's one write is removed from the late judge; + additional versions or deletes remain collateral even on that record. + """ + pruned = { + **now, + "feeds": { + name: {**feed, "events": dict(feed["events"])} + for name, feed in now["feeds"].items() + }, + } + used: set[tuple[str, str]] = set() + remaining = dict(raw) + for scope, feed in now["feeds"].items(): + old = before["feeds"].get(scope, {}).get("events", {}) + for event in sorted(set(feed["events"]) - set(old)): + row = raw[event] + target = (scope, str(row["record_id"])) + if ( + target in targets + and target not in used + and str(row["operation"]).upper() + in ("POST", "PUT", "PATCH", "INSERT", "UPDATE") + ): + used.add(target) + pruned["feeds"][scope]["events"].pop(event) + elif target in targets: + # Before the provisional check, another version of its + # target is a conflicting write, even by a colleague. + remaining[event] = {**row, "authoritative": False} + if targets - used: + from openadapt_flow.runtime.effects.frame import FrameVerdict + + return FrameVerdict( + "not_checked", "a saved target has no corresponding history event" + ) + boundary = min( + instant(feed["through"]) for feed in before["feeds"].values() + ).isoformat() + return judge_history(spec, before, pruned, remaining, params, boundary, run_actor) diff --git a/packages/flow/openadapt_flow/runtime/effects/late_write.py b/packages/flow/openadapt_flow/runtime/effects/late_write.py index f8a39ed390..bdc71824de 100644 --- a/packages/flow/openadapt_flow/runtime/effects/late_write.py +++ b/packages/flow/openadapt_flow/runtime/effects/late_write.py @@ -119,7 +119,7 @@ def schedule( read_at = str(post_state["read_at"]) return { - "compares": COMPARES, + "compares": "history_since_checkpoint" if "history" in post_state else COMPARES, "after_s": after_s, "post_save_read_at": read_at, "due_at": due_at(read_at, after_s) if due else None, @@ -146,6 +146,16 @@ def state_problem(payload: Any) -> Optional[str]: if not isinstance(payload[name], Mapping): return f"the recheck state's {name} is not an object" post = payload["post_state"] + if "history" in post: + from openadapt_flow.runtime.effects.history import checkpoint_problem + + try: + parse_time(str(post.get("read_at"))) + except ValueError: + return "the recheck history time is malformed" + if set(post) != {"read_at", "history"}: + return "the history post-state has unknown fields" + return checkpoint_problem(post["history"]) if not isinstance(post.get("snapshot"), Mapping): return "the recheck state holds no post-save snapshot" try: diff --git a/packages/flow/openadapt_flow/verifier.py b/packages/flow/openadapt_flow/verifier.py index f48efd0ae8..7e913db07c 100644 --- a/packages/flow/openadapt_flow/verifier.py +++ b/packages/flow/openadapt_flow/verifier.py @@ -641,6 +641,12 @@ def provisional(self) -> bool: return self.verified and bool(self.recheck_due_at) + @property + def final(self) -> bool: + """Whether this result is free of an owed final recheck.""" + + return not self.provisional + @property def verdict(self) -> str: """The outcome under its old name (``VERIFIED``, ``REFUTED``, @@ -686,6 +692,7 @@ def to_dict(self) -> dict[str, Any]: "review": self.review, "recheck_due_at": self.recheck_due_at, "provisional": self.provisional, + "final": self.final, "write_receipt": self.write_receipt, "write_receipt_path": self.write_receipt_path, } @@ -780,6 +787,12 @@ def verified(self) -> bool: return self.outcome == "CONFIRMED" + @property + def final(self) -> bool: + """A completed proof through read_at, with no unread history gap.""" + + return bool(self.read_at) and self.frame in ("held", "refuted") + @property def late(self) -> bool: """True when a change after the verdict refutes it.""" @@ -1033,6 +1046,13 @@ def _capture( database = contract.config.frame.database if contract.config.frame else None if database is not None and FRAME_DB_DETAIL not in (state.get("detail") or {}): raise BaselineUnreadable(_frame_db_unread(state)) + if contract.config.frame is not None and contract.config.frame.history is not None: + from openadapt_flow.runtime.effects.history import DETAIL, checkpoint_problem + + if checkpoint_problem((state.get("detail") or {}).get(DETAIL)) is not None: + raise BaselineUnreadable( + "complete change history could not be read before the write" + ) return pending @@ -1062,6 +1082,7 @@ def _settle( # A recheck confirmed it since: no longer provisional, and no # recheck is due. result["provisional"] = False + result["final"] = True result["recheck_due_at"] = None result["rechecked_at"] = attempt.rechecked.get("at") if not attempt.delivered: @@ -2457,7 +2478,7 @@ def _judge_recheck( ): raise VerifierError( "this write scheduled no recheck: it was not verified, or the " - "contract doesn't set frame.database.recheck_after_s" + "contract schedules no history or database recheck" ) run_record = actor.run_record() cfg = contract.config diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index c7e5c9a121..3f822a3ca5 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -369,6 +369,32 @@ def parse_verify_config(data: Any, *, source: str = "verify config") -> VerifyCo "frame.attribution.read must be one read-only system-of-record " "read (rest, sql, fhir, graphql, ...)" ) + history = cfg.frame.history if cfg.frame is not None else None + if history is not None: + sources = cfg.verifier.surfaces or {"": cfg.verifier} + if set(history.surfaces) != set(sources): + raise VerifyRunError("frame.history must cover every verifier surface") + for name, history_source in sources.items(): + feed = history.feeds[history.surfaces[name]] + read = EffectsConfig.model_validate(feed.read) + if read.candidates or read.surfaces or read.records_drop_fields: + raise VerifyRunError( + "history feeds cannot omit fields or route candidates" + ) + if history_source.kind == "fhir" and ( + read.kind != "fhir" + or read.resource_type != history_source.resource_type + or read.base_url != history_source.base_url + ): + raise VerifyRunError( + "history must cover the same FHIR origin and type as the record read" + ) + for feed in history.feeds.values(): + read = EffectsConfig.model_validate(feed.read) + if read.candidates or read.surfaces or read.records_drop_fields: + raise VerifyRunError( + "history feeds cannot omit fields or route candidates" + ) database = cfg.frame.database if cfg.frame is not None else None if database is not None: surfaces = set(cfg.verifier.surfaces) or {""} @@ -1594,6 +1620,13 @@ def capture_baseline( execution_id = _new_execution_id() _tag_reads(verifier, execution_id) salt = secrets.token_bytes(32) + history_before = None + history_error = None + if cfg.frame is not None and cfg.frame.history is not None: + try: + history_before, _history_rows = _read_history(cfg, params, _now(), salt) + except Exception as exc: + history_error = type(exc).__name__ # The versions first: a save between this read and the record read is # then in the record baseline too, never hidden in the versions one. versions_rows = ( @@ -1605,6 +1638,12 @@ def capture_baseline( if not isinstance(state, EffectState): raise VerifyRunError("verifier returned no readable baseline") detail: dict[str, Any] = {} + if cfg.frame is not None and cfg.frame.history is not None: + from openadapt_flow.runtime.effects.history import DETAIL + + detail[DETAIL] = ( + history_before if history_before is not None else {"error": history_error} + ) entries: Optional[list[dict[str, Any]]] = None snapshot = None if cfg.frame is not None and cfg.frame.database is not None: @@ -1960,6 +1999,29 @@ def verify_run( ) _tag_reads(verifier, execution_id) + history_checkpoint = None + history_problem = None + history_spec = cfg.frame.history if cfg.frame is not None else None + if history_spec is not None: + try: + if baseline is None: + raise ValueError("history requires a baseline") + from openadapt_flow.runtime.effects.history import ( + DETAIL, + checkpoint_problem, + ) + + stored_history = baseline["state"]["detail"].get(DETAIL) + if checkpoint_problem(stored_history) is not None: + raise ValueError("history baseline is unavailable") + history_checkpoint, _history_rows = _read_history( + cfg, params, stored_history["since"], salt + ) + except Exception as exc: # unreadable or incomplete history never confirms + history_problem = ( + f"history checkpoint could not be read ({type(exc).__name__})" + ) + read_at = _now() before: EffectState baseline_sha: Optional[str] = None @@ -2072,6 +2134,46 @@ def verify_run( status="not_checked", reason="no baseline, so what else changed can't be known", ) + if history_spec is not None: + if history_problem is not None and frame.status != "refuted": + frame.status = "not_checked" + frame.reason = history_problem + elif history_checkpoint is not None: + from openadapt_flow.runtime.effects.history import judge_initial_history + from openadapt_flow.runtime.effects.surfaces import surface_of + + targets: set[tuple[str, str]] = set() + for effect, verdict in zip(bound_effects, verdicts): + if effect.kind not in ( + EffectKind.RECORD_WRITTEN, + EffectKind.FIELD_EQUALS, + EffectKind.EXACT_NEW_SET, + ): + continue + scope = history_spec.surfaces[surface_of(effect) or ""] + for target_record in verdict.matched_records: + target_identity = target_record.get( + frame_spec.identity_field or "id" + ) + if target_identity not in (None, ""): + targets.add((scope, str(target_identity))) + history_frame = judge_initial_history( + history_spec, + stored_history, + history_checkpoint, + _history_rows, + params, + str(run_record.get("executor_id") or ""), + targets, + ) + frame.collateral.extend(history_frame.collateral) + frame.concurrent.extend(history_frame.concurrent) + if history_frame.status == "refuted" or ( + history_frame.status == "not_checked" and frame.status == "held" + ): + frame.status = history_frame.status + frame.reason = history_frame.reason + frame.post_state = {"read_at": _now(), "history": history_checkpoint} # A record outside the declared delta moved: the save's effect on the # store is not the declared one, whatever the per-effect reads say. effect_verdict: EffectVerdictLabel = ( @@ -2230,12 +2332,15 @@ def verify_run( "effect_verdict": effect_verdict, } database_spec = frame_spec.database + recheck_after_s = ( + frame_spec.history.quiet_s + if frame_spec.history is not None + else database_spec.recheck_after_s + if database_spec is not None + else None + ) 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 - ): + if frame.post_state is not None and 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 @@ -2243,7 +2348,7 @@ def verify_run( # this one's own window covered. schedule = late_write.schedule( frame.post_state, - database_spec.recheck_after_s, + recheck_after_s, salted_digest(salt, late_write.POST_STATE_LABEL, frame.post_state), due=verified, ) @@ -2384,16 +2489,22 @@ def recheck_problem( 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: + history = cfg.frame.history if cfg.frame is not None else None + after_s = ( + history.quiet_s + if history is not None + else database.recheck_after_s + if database is not None + else None + ) + if not isinstance(schedule, Mapping) or after_s 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 - ): + if dict(schedule) != late_write.schedule(post, after_s, digest, due=is_verified): return ( "the contract's recheck_after_s is not the one the receipt was issued under" ) @@ -2496,6 +2607,23 @@ def _later_window(state: Mapping[str, Any]) -> Optional[Any]: return None +def _read_history( + cfg: VerifyConfig, params: Mapping[str, object], since: str, salt: bytes +) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]: + from openadapt_flow.runtime.effects.history import read_history + + if cfg.frame is None or cfg.frame.history is None: + raise VerifyRunError("no history feeds are configured") + + def build(read: dict[str, Any]) -> Any: + verifier = build_effect_verifier(EffectsConfig.model_validate(read), params) + if verifier is None: + raise VerifyRunError("history built no read-only verifier") + return verifier + + return read_history(cfg.frame.history, since, build, _mac_key(salt)) + + def recheck_run( cfg: VerifyConfig, state: Mapping[str, Any], @@ -2527,6 +2655,29 @@ def recheck_run( """ post = dict(state["post_state"]) + if cfg.frame is not None and cfg.frame.history is not None: + from openadapt_flow.runtime.effects.history import instant, judge_history + + if excused: + raise VerifyRunError( + "history rechecks require source provenance; later-write windows cannot excuse events" + ) + checkpoint = post["history"] + now, rows = _read_history( + cfg, params, checkpoint["since"], bytes.fromhex(str(state["salt"])) + ) + schedule = state["evidence"]["frame"][late_write.EVIDENCE_KEY] + verdict = judge_history( + cfg.frame.history, + checkpoint, + now, + rows, + params, + schedule["due_at"], + str(run_record.get("executor_id") or ""), + ) + horizon = min(instant(feed["through"]) for feed in now["feeds"].values()) + return verdict, late_write.format_time(horizon) since = late_write.parse_time(str(post["read_at"])) windows: list[Any] = [] unjudged = 0 diff --git a/packages/flow/openadapt_flow/write_receipt.py b/packages/flow/openadapt_flow/write_receipt.py index 2e716d1b6a..143481c39b 100644 --- a/packages/flow/openadapt_flow/write_receipt.py +++ b/packages/flow/openadapt_flow/write_receipt.py @@ -685,7 +685,16 @@ def issue_for_recheck( }, source="verifier", verdict=verdict, - scope=recheck_scope(scope_of_result(result), ["frame.database"], when), + scope=recheck_scope( + scope_of_result(result), + ["frame.history"] + if ((result.evidence.get("frame") or {}).get("recheck") or {}).get( + "compares" + ) + == "history_since_checkpoint" + else ["frame.database"], + when, + ), checked={**result.checked, "frame": late.frame}, write_id=result.write_id, verify_run_receipt_sha256=result.receipt_sha256 or None, diff --git a/packages/flow/tests/test_history_recheck.py b/packages/flow/tests/test_history_recheck.py new file mode 100644 index 0000000000..83c880b0af --- /dev/null +++ b/packages/flow/tests/test_history_recheck.py @@ -0,0 +1,489 @@ +"""Synthetic history contract controls; real-app qualification is separate.""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest + +from openadapt_flow import verifier +from openadapt_flow.runtime.effects.fhir import FhirEffectVerifier +from openadapt_flow.runtime.effects.history import ( + FrameHistory, + checkpoint_problem, + judge_history, + judge_initial_history, + read_history, +) +from openadapt_flow.verifier import ( + Actor, + Contract, + MemoryLedger, + begin_write, + recheck, + verify_write, +) + +SINCE = "2026-11-05T00:00:00Z" +DUE = "2026-11-05T00:02:00Z" +KEY = b"history-synthetic-key" * 2 + + +def _spec(**changes: Any) -> FrameHistory: + doc = { + "feeds": { + "thing": { + "read": { + "kind": "fhir", + "base_url": "https://source.test/fhir", + "resource_type": "Basic", + "field_paths": {"id": "id"}, + }, + "author_path": "resource.meta.author.reference", + "author_server_set": True, + } + }, + "surfaces": {"": "thing"}, + "quiet_s": 1, + "setup_evidence_sha256": "sha256:" + "a" * 64, + "self": ["actor/bot"], + } + doc.update(changes) + return FrameHistory.model_validate(doc) + + +def _event( + record: str, + version: str = "1", + author: str = "actor/bot", + method: str = "PUT", + at: str = "2026-11-05T00:00:01Z", +) -> dict[str, Any]: + entry = { + "resource": { + "resourceType": "Basic", + "id": record, + "meta": { + "versionId": version, + "lastUpdated": at, + "author": {"reference": author}, + }, + }, + "request": {"method": method, "url": f"Basic/{record}"}, + "response": {"lastModified": at}, + } + if method == "DELETE": + entry.pop("resource") + return entry + + +def _bundle(entries: list[dict[str, Any]], **changes: Any) -> dict[str, Any]: + return { + "resourceType": "Bundle", + "type": "history", + "total": len(entries), + "timestamp": "2026-11-05T00:03:00Z", + "entry": entries, + **changes, + } + + +class Session: + def __init__(self, bundles: list[dict[str, Any]]) -> None: + self.bundles = bundles + self.calls: list[Any] = [] + + def get(self, url: str, **kw: Any) -> Any: + self.calls.append((url, kw)) + bundle = self.bundles.pop(0) + return SimpleNamespace( + status_code=200, + url=url, + headers={"Date": "Thu, 05 Nov 2026 00:03:00 GMT"}, + json=lambda: bundle, + ) + + +def _read( + spec: FrameHistory, bundles: list[dict[str, Any]] +) -> tuple[Any, Any, Session]: + session = Session(bundles) + reader = FhirEffectVerifier( + "https://source.test/fhir", + resource_type="Basic", + field_paths={"id": "id"}, + session=session, + ) + checkpoint, raw = read_history(spec, SINCE, lambda read: reader, KEY) + return checkpoint, raw, session + + +def test_fhir_reads_whole_type_all_pages_and_seals() -> None: + spec = _spec() + first = _bundle( + [_event("a")], + total=2, + link=[ + { + "relation": "next", + "url": "https://source.test/fhir/Basic/_history?page=2", + } + ], + ) + state, raw, session = _read(spec, [first, _bundle([_event("b")], total=2)]) + assert len(raw) == 2 + assert session.calls[0][0].endswith("/Basic/_history") + assert session.calls[0][1]["params"] == {"_since": SINCE, "_count": "1000"} + assert session.calls[1][1]["params"] is None + assert "actor/bot" not in json.dumps(state) + assert checkpoint_problem(state) is None + + +@pytest.mark.parametrize( + "fault", ["update_other", "new_version", "delete_other", "update_revert"] +) +def test_new_versions_and_resource_less_tombstones_refute(fault: str) -> None: + spec = _spec() + original = _event("saved") + old, _, _ = _read(spec, [_bundle([original])]) + events = [original] + if fault == "delete_other": + events.append(_event("other", method="DELETE", at="2026-11-05T00:01:00Z")) + elif fault == "update_revert": + events.extend([_event("other", "2"), _event("other", "3")]) + else: + events.append(_event("saved" if fault == "new_version" else "other", "2")) + now, raw, _ = _read(spec, [_bundle(events)]) + verdict = judge_history(spec, old, now, raw, {}, DUE, "actor/bot") + assert verdict.status == "refuted" + assert verdict.collateral + + +def test_authoritative_other_user_edit_does_not_refute() -> None: + spec = _spec() + old, _, _ = _read(spec, [_bundle([_event("saved")])]) + now, raw, _ = _read( + spec, [_bundle([_event("saved"), _event("saved", "2", "actor/colleague")])] + ) + result = judge_history(spec, old, now, raw, {}, DUE, "actor/bot") + assert result.held + assert len(result.concurrent) == 1 + + +def test_untrusted_actor_or_executor_alias_never_excuses() -> None: + spec = _spec() + old, _, _ = _read(spec, [_bundle([])]) + now, raw, _ = _read(spec, [_bundle([_event("a", author="actor/executor")])]) + assert ( + judge_history(spec, old, now, raw, {}, DUE, "actor/executor").status + == "refuted" + ) + untrusted = spec.model_copy(deep=True) + untrusted.feeds["thing"].author_server_set = False + now, raw, _ = _read(untrusted, [_bundle([_event("a", author="actor/colleague")])]) + assert ( + judge_history(untrusted, old, now, raw, {}, DUE, "actor/bot").status + == "refuted" + ) + + +@pytest.mark.parametrize( + "mutation", + [ + "truncated", + "searchset", + "no_total", + "cycle", + "cross_origin", + "duplicate", + "bad_next", + "future_event", + ], +) +def test_incomplete_fhir_history_never_confirms(mutation: str) -> None: + bundle = _bundle([_event("a")]) + if mutation == "truncated": + bundle["total"] = 2 + elif mutation == "searchset": + bundle["type"] = "searchset" + elif mutation == "no_total": + bundle.pop("total") + elif mutation == "cycle": + bundle["link"] = [ + {"relation": "next", "url": "https://source.test/fhir/Basic/_history"} + ] + elif mutation == "cross_origin": + bundle["link"] = [ + {"relation": "next", "url": "https://elsewhere.test/Basic/_history"} + ] + elif mutation == "duplicate": + bundle = _bundle([_event("a"), _event("a")]) + elif mutation == "bad_next": + bundle["link"] = [{"relation": "next", "url": None}] + else: + bundle = _bundle([_event("a", at="2026-11-05T00:04:00Z")]) + with pytest.raises((ValueError, TypeError)): + _read(_spec(), [bundle]) + + +def test_disappearing_changed_or_stale_history_never_confirms() -> None: + spec = _spec() + old, _, _ = _read(spec, [_bundle([_event("a")])]) + for bundle in [ + _bundle([]), + _bundle([_event("a", author="actor/other")]), + _bundle([_event("a")], timestamp="2026-11-05T00:01:00Z"), + ]: + now, raw, _ = _read(spec, [bundle]) + assert ( + judge_history(spec, old, now, raw, {}, DUE, "actor/bot").status + == "not_checked" + ) + + +def _sql_contract(db: Path) -> dict[str, Any]: + read = { + "kind": "sql", + "sqlite_database": str(db), + "sql_query": "SELECT * FROM changes", + } + return { + "verifier": { + "kind": "sql", + "sqlite_database": str(db), + "sql_query": "SELECT * FROM records", + }, + "record": {"name": "record", "match": ["value"], "fields": {"value": "value"}}, + "frame": { + "identity_field": "id", + "history": { + "feeds": { + "records": { + "read": read, + "dialect": "sqlite", + "coverage_query": "SELECT retained_since, through, event_count FROM coverage", + "author_path": "author", + "author_server_set": True, + } + }, + "surfaces": {"": "records"}, + "quiet_s": 1, + "setup_evidence_sha256": "sha256:" + "a" * 64, + "self": ["bot"], + }, + }, + } + + +@pytest.fixture +def database(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.setenv("OPENADAPT_LEDGER_DIR", str(tmp_path / "ledger")) + monkeypatch.setenv("OPENADAPT_SEAL_HOME", str(tmp_path / "seal")) + monkeypatch.setattr(verifier, "USED_DIR", tmp_path / "used") + monkeypatch.setattr(verifier, "seconds_until", lambda due: 0) + from openadapt_flow import verify_run + + monkeypatch.setattr(verify_run, "_now", lambda: SINCE) + db = tmp_path / "app.sqlite" + with sqlite3.connect(db) as con: + con.executescript( + "CREATE TABLE records(id INTEGER PRIMARY KEY, value TEXT); CREATE TABLE changes(event_id INTEGER PRIMARY KEY, record_id TEXT, operation TEXT, author TEXT, occurred_at TEXT); CREATE TABLE coverage(retained_since TEXT, through TEXT, event_count INTEGER); INSERT INTO coverage VALUES('2000-01-01T00:00:00Z','2026-11-05T00:03:00Z',0);" + ) + return db + + +def test_sql_public_api_provisional_late_hold(database: Path, tmp_path: Path) -> None: + contract = Contract.from_dict(_sql_contract(database)) + ledger = verifier.Ledger(MemoryLedger()) + actor = Actor("agent", "bot") + pending = begin_write( + contract, + {"value": "expected"}, + actor=actor, + ledger=ledger, + workdir=tmp_path / "writes", + ) + with sqlite3.connect(database) as con: + con.execute("INSERT INTO records VALUES(1,'expected')") + con.execute( + "INSERT INTO changes VALUES(1,'1','INSERT','bot',?)", + ("2026-11-05T00:00:01Z",), + ) + con.execute("UPDATE coverage SET event_count=1") + result = verify_write(pending) + assert result.verified and result.provisional and not result.final + assert result.to_dict()["final"] is False + assert len(ledger.rechecks_owed()) == 1 + assert result.recheck is not None + with sqlite3.connect(database) as con: + con.execute( + "INSERT INTO changes VALUES(2,'2','UPDATE','bot',?)", + ("2026-11-05T00:00:01Z",), + ) + con.execute("UPDATE coverage SET event_count=2") + late = recheck(contract, result, params={"value": "expected"}, actor=actor) + assert late.outcome == "REFUTED" + assert not ledger.rechecks_owed() + assert ledger.state(ledger.latest(result.write_key)) == "held" + + +def test_sql_incomplete_coverage_prevents_provisional_success( + database: Path, tmp_path: Path +) -> None: + contract = Contract.from_dict(_sql_contract(database)) + pending = begin_write( + contract, + {"value": "expected"}, + actor=Actor("agent", "bot"), + ledger=MemoryLedger(), + workdir=tmp_path / "writes", + ) + with sqlite3.connect(database) as con: + con.execute("INSERT INTO records VALUES(1,'expected')") + con.execute("UPDATE coverage SET event_count=1") + result = verify_write(pending) + assert not result.verified and result.outcome.value == "UNCERTAIN" + + +def test_history_contract_requires_every_read_surface(database: Path) -> None: + doc = _sql_contract(database) + doc["frame"]["history"]["surfaces"] = {"wrong": "records"} + with pytest.raises(ValueError, match="every verifier surface"): + Contract.from_dict(doc) + + +def test_version_read_entries_require_matching_version_identity() -> None: + entry = _event("a") + entry["request"] = {"method": "GET", "url": "Basic/a/_history/1"} + old, _, _ = _read(_spec(), [_bundle([])]) + now, raw, _ = _read(_spec(), [_bundle([entry])]) + assert ( + judge_history(_spec(), old, now, raw, {}, DUE, "actor/bot").status == "refuted" + ) + entry["request"]["url"] = "Basic/a/_history/different" + with pytest.raises(ValueError): + _read(_spec(), [_bundle([entry])]) + + +def test_mutated_resource_with_same_version_never_confirms() -> None: + spec = _spec() + entry = _event("a") + old, _, _ = _read(spec, [_bundle([entry])]) + entry["resource"]["code"] = {"text": "changed history"} + now, raw, _ = _read(spec, [_bundle([entry])]) + assert ( + judge_history(spec, old, now, raw, {}, DUE, "actor/bot").status == "not_checked" + ) + + +def test_initial_history_proves_target_and_all_collateral() -> None: + spec = _spec() + old, _, _ = _read(spec, [_bundle([])]) + now, raw, _ = _read(spec, [_bundle([_event("saved")])]) + assert judge_initial_history( + spec, old, now, raw, {}, "actor/bot", {("thing", "saved")} + ).held + now, raw, _ = _read(spec, [_bundle([_event("saved"), _event("other")])]) + assert ( + judge_initial_history( + spec, old, now, raw, {}, "actor/bot", {("thing", "saved")} + ).status + == "refuted" + ) + now, raw, _ = _read(spec, [_bundle([_event("saved"), _event("saved", "2")])]) + assert ( + judge_initial_history( + spec, old, now, raw, {}, "actor/bot", {("thing", "saved")} + ).status + == "refuted" + ) + now, raw, _ = _read(spec, [_bundle([])]) + assert ( + judge_initial_history( + spec, old, now, raw, {}, "actor/bot", {("thing", "saved")} + ).status + == "not_checked" + ) + + +def test_colleague_fix_before_provisional_cannot_hide_wrong_agent_version() -> None: + spec = _spec() + old, _, _ = _read(spec, [_bundle([])]) + now, raw, _ = _read( + spec, [_bundle([_event("saved"), _event("saved", "2", "actor/colleague")])] + ) + assert ( + judge_initial_history( + spec, old, now, raw, {}, "actor/bot", {("thing", "saved")} + ).status + == "refuted" + ) + + +def test_sql_history_restart_and_legitimate_late_target_edit( + database: Path, tmp_path: Path +) -> None: + contract = Contract.from_dict(_sql_contract(database)) + ledger = verifier.Ledger(MemoryLedger()) + actor = Actor("agent", "bot") + pending = begin_write( + contract, + {"value": "expected"}, + actor=actor, + ledger=ledger, + workdir=tmp_path / "writes", + ) + at = "2026-11-05T00:00:01Z" + with sqlite3.connect(database) as con: + con.execute("INSERT INTO records VALUES(1,'expected')") + con.execute("INSERT INTO changes VALUES(1,'1','INSERT','bot',?)", (at,)) + con.execute("UPDATE coverage SET event_count=1") + result = verify_write(pending) + assert result.provisional and result.recheck is not None + path = result.recheck.save(tmp_path / "restart.json") + restored = verifier.RecheckState.load(path, params={"value": "expected"}) + with sqlite3.connect(database) as con: + con.execute("UPDATE records SET value='colleague change' WHERE id=1") + con.execute("INSERT INTO changes VALUES(2,'1','UPDATE','colleague',?)", (at,)) + con.execute("UPDATE coverage SET event_count=2") + late = recheck( + contract, restored, params={"value": "expected"}, actor=actor, ledger=ledger + ) + assert late.verified and late.final + assert not ledger.rechecks_owed() + + +def test_history_baseline_failure_blocks_before_write( + database: Path, tmp_path: Path +) -> None: + with sqlite3.connect(database) as con: + con.execute("UPDATE coverage SET event_count=1") + with pytest.raises(verifier.BaselineUnreadable, match="complete change history"): + begin_write( + Contract.from_dict(_sql_contract(database)), + {"value": "expected"}, + actor=Actor("agent", "bot"), + ledger=MemoryLedger(), + workdir=tmp_path / "writes", + ) + + +def test_http_date_precision_does_not_extend_final_coverage() -> None: + spec = _spec() + bundle = _bundle([_event("a", at="2026-11-05T00:03:00.500Z")]) + bundle.pop("timestamp") + state, _, _ = _read(spec, [bundle]) + assert state["feeds"]["thing"]["through"] == "2026-11-05T00:03:00+00:00" + future = _bundle([_event("a", at="2026-11-05T00:03:02Z")]) + future.pop("timestamp") + with pytest.raises(ValueError, match="newer than its server response"): + _read(spec, [future]) + + +def test_future_source_coverage_never_confirms() -> None: + with pytest.raises(ValueError, match="coverage frontier is in the future"): + _read(_spec(), [_bundle([], timestamp="2200-01-01T00:00:00Z")])