diff --git a/CAPABILITIES.md b/CAPABILITIES.md index e9353b06de..34c71137fe 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -147,7 +147,7 @@ improvising or retrying when something is off. | Judge an "ensure the record exists" write (found, created once with every value, or needs review) through a FHIR server's read path only: searches, AuditEvent and `_history`, never the database; the coverage claim names that read path. An agent's own correction of the record it created counts as one write when the audit trail and history show only that agent, no other reader, and the same record in every version | On main ([#1640](https://github.com/OpenAdaptAI/OpenAdapt/pull/1640), corrections [#1655](https://github.com/OpenAdaptAI/OpenAdapt/pull/1655)) | Real app (Medplum 5.2.8; three blind fault suites: 17/18, 17/20, 19/20; no fault passed in the last two. Corrections, on 15 blind cases and the 37 public ones: false stops 8 to 2 and 9 to 4 for Claude, no harm passed that the old rule caught) | `openadapt_flow/ensure_record.py`, [ENSURE_RECORD.md](packages/flow/docs/ENSURE_RECORD.md) | | Generate an ensure-record contract from three demonstrations, the server's CapabilityStatement and its SearchParameter definitions: field paths and `[system=…]` filters, comparison kinds, value maps, identity, search parameters, the closed record and the change feed. It refuses when unsure, asks a person to confirm an identity for a type with no reviewed template, and self-checks every demonstration plus a perturbation per field | On main ([#1653](https://github.com/OpenAdaptAI/OpenAdapt/pull/1653)) | Real app (Medplum 5.2.8: all 16 hand-written Patient lines regenerated; one blind write type, Immunization: 78 s of person time, 2 lines edited after the person declined the proposed identity; blind suite 23/24, no fault passed) | `openadapt_flow/generate/ensure_contract.py`, [RESULTS.md](packages/flow/benchmark/contract_autogen/results/RESULTS.md) | | A stop card for a stopped write: the exact record difference and who else wrote, with Undo or Fix that reads back, or Open in the app; cost modelled from published operators and timed machine steps, as pre-registered | On main ([#1653](https://github.com/OpenAdaptAI/OpenAdapt/pull/1653)) | Real app (Medplum 5.2.8, blind suite: 16 stops, 14 one-tap with no failed read-back; modelled 42 s per caught harm against 420; one wrong category on a false stop, gone in the re-measure with `void`: 24/24, 15 stops, no wrong category, no failed read-back) | `benchmark/stop_card/`, [PREREG_REVIEW.md](packages/flow/benchmark/stop_card/PREREG_REVIEW.md) | -| Generate `openadapt.verify/v1` contracts from a MySQL, MariaDB or PostgreSQL schema, the kinds of write with their parameter names, and one demonstration write per example: one contract per field set (an update gets one for each set of fields it may change), list parameters by length, a `found` twin for a record already in place, the whole-database frame on every dialect, rows the app adds on its own declared by count, a recheck window the person confirms, and `frame.versions` (a lost update) from a version column or the app's audit trail of saves. Every undecided point is one numbered question; every contract is loaded and dry-read against the live store before it is written; a needed `TRIGGER` grant or view is applied with a flag or the kind is refused. Not yet: OpenAPI metadata | [#1672](https://github.com/OpenAdaptAI/OpenAdapt/pull/1672) | Real app (first study, instance copies; questions for a person: Dolibarr 4, Redmine 2, EspoCRM 3, against 11 hand decisions each; every in-scope planted fault stopped (22/22, 19/19, 21/21); correct writes confirmed 32/43, 32/38, 27/40; evidence in the private study repository) | `python -m openadapt_flow.generate.verify_contract`, [GENERATE_CONTRACTS.md](packages/flow/docs/GENERATE_CONTRACTS.md) | +| Generate `openadapt.verify/v1` contracts from a MySQL, MariaDB or PostgreSQL schema, the kinds of write with their parameter names, and one demonstration write per example: one contract per field set (an update gets one for each set of fields it may change), list parameters by length, a `found` twin for a record already in place, the whole-database frame on every dialect (it compares the contract's own tables too, so an excused column holds only on the rows the write declares, and the recheck sees a late change to a look-alike row), rows the app adds on its own declared by count, a recheck window the person confirms, and `frame.versions` (a lost update) from a version column or the app's audit trail of saves. Every undecided point is one numbered question; every contract is loaded and dry-read against the live store before it is written; a needed `TRIGGER` grant or view is applied with a flag or the kind is refused. Not yet: OpenAPI metadata | [#1672](https://github.com/OpenAdaptAI/OpenAdapt/pull/1672) | Real app (first study, instance copies; questions for a person: Dolibarr 4, Redmine 2, EspoCRM 3, against 11 hand decisions each; every in-scope planted fault stopped (22/22, 19/19, 21/21); correct writes confirmed 32/43, 32/38, 27/40; evidence in the private study repository) | `python -m openadapt_flow.generate.verify_contract`, [GENERATE_CONTRACTS.md](packages/flow/docs/GENERATE_CONTRACTS.md) | | `void` in an ensure-record contract: values that make a record assert nothing (`status="entered-in-error"`). A record holding one is never the person, is not a duplicate, and a created one needs review; it names values to leave out, never values to keep, and can't name an identity field | On main ([#1657](https://github.com/OpenAdaptAI/OpenAdapt/pull/1657)) | Real app (Medplum 5.2.8: 8 live cases written in advance, 8/8 with the line, 5/8 and one silent pass without it; frozen blind Immunization suite 24/24); adversarial review recorded | `openadapt_flow/ensure_record.py`, [RESULTS.md](packages/flow/benchmark/fhir_medplum/void/results/RESULTS.md), [REVIEW_VOID.md](packages/flow/benchmark/fhir_medplum/void/REVIEW_VOID.md) | | Check a patient registration on a commercial EHR through its public FHIR sandbox: Epic (open.epic, R4), read through FHIR search and read with SMART backend auth; the frame is the search set, since the sandbox has no AuditEvent or `_history` | On main ([#1656](https://github.com/OpenAdaptAI/OpenAdapt/pull/1656)) | Real app (Epic public sandbox, August 2026 release; blind round 12/13 with none passed as verified, the miss a false refusal fixed in the contract; caught Epic dropping the phone on a 201 create) | `benchmark/commercial_ehr/`, [FINDINGS.md](packages/flow/benchmark/commercial_ehr/FINDINGS.md) | | A public benchmark a vendor reruns with their own agent: `openadapt bench medplum --agent ` runs 37 written-for-it cases (Patient, AllergyIntolerance, Appointment) against a local Medplum through a fault proxy, and scores the agent's claim, a read-back and OpenAdapt's verifier against an end-state oracle | On main ([#1652](https://github.com/OpenAdaptAI/OpenAdapt/pull/1652)) | Real app (Medplum 5.2.8; Claude Sonnet 5.5: OpenAdapt missed 1 of 9 harms with 4 false stops, all corrections that changed the record's identity, against 8 missed by the claim and the read-back; cold run from clone to scorecard 5 min) | [README](packages/flow/benchmark/fhir_medplum/public/README.md), [RESULTS.md](packages/flow/benchmark/fhir_medplum/public/RESULTS.md) | diff --git a/packages/flow/docs/GENERATE_CONTRACTS.md b/packages/flow/docs/GENERATE_CONTRACTS.md index b4d12cc9a5..3e07fcec29 100644 --- a/packages/flow/docs/GENERATE_CONTRACTS.md +++ b/packages/flow/docs/GENERATE_CONTRACTS.md @@ -241,7 +241,17 @@ refused. A mapping without `accept: true` takes only what it names. in any table that the contract doesn't declare stops the write, except rows added to a table the person named as a log. The main record's table and the tables of rows that hold values are read by the contract's own reads, with -every column. On PostgreSQL the read account also needs `SELECT` on the +every row and column. Those tables are compared too +(`frame.database.judged_by`): the rows the contract's read declares (the +target, or the rows the write adds) are judged by the table, every column +exactly, with only their declared fields allowed to change, and any other +row that moves stops the write. A table's key is read as `id` unless the +table has its own `id` column that isn't the whole key; then it is read as +`oa_key` and `id` as it is. A new declared row may set any column of these +tables (`new_row_fields`), since the contract's read checks each one. An excused column is excused only on the rows the +write declares, never on a look-alike row of the same table. The recheck +compares these tables like any other, so a late change to any of their rows +stops the write. On PostgreSQL the read account also needs `SELECT` on the sequences (see [VERIFY_RUN.md](VERIFY_RUN.md)); without it the dry read fails and no contract is written. diff --git a/packages/flow/docs/VERIFY_RUN.md b/packages/flow/docs/VERIFY_RUN.md index 85d5225c53..4c2ee2aef9 100644 --- a/packages/flow/docs/VERIFY_RUN.md +++ b/packages/flow/docs/VERIFY_RUN.md @@ -300,6 +300,35 @@ refuted. The read account needs `TRIGGER` and `SHOW VIEW` besides `SELECT`: without them a trigger planted in the window can't be seen, and the frame is not checked. +`judged_by` names a table whose rows one of the contract's own reads +declares, such as an update's read of its main record's table: + +```yaml + judged_by: + assets: record # the table, and the surface that declares its rows +``` + +The surface only says which rows are declared: its records, before or after +the save, that an effect routed to it selects, matched to the table by +primary key. The table judges every row itself, every column exactly as +stored, so a lossy read (`substr`, `COALESCE`, a cast) can't hide a change. +A declared row may change only the surface's fields in +`target_fields_by_surface`, and a declared new row may set only the columns +the effects name or `new_row_fields` lists; every other column must hold its +default. Every other row of the table must not move, and the app profile's +bookkeeping rules apply as on any table. A recheck reads no surface, so +there nothing is declared and any change since the post-save read refutes +the write. + +A `judged_by` table is compared, not allow-listed, so it can't also appear +in `allow_tables` or `views`. One surface declares rows of one table only, +and the surface may not ignore a field (`frame.ignore_fields`, or a +`record_field` attribution's `server_set`) that the table doesn't ignore +(`ignore_columns`). When a declared record's key can't be read, nothing is +declared, and the change stops the write. The database is read before the +surfaces, so a change can't land between the surface's read and the read a +recheck compares with. + Every column type is read, and none is dropped. On MySQL and MariaDB each row is read as `JSON_OBJECT`. A type whose value it can't carry as exact text is read another way: binary strings, blobs, `BIT`, geometries and vectors as diff --git a/packages/flow/openadapt_flow/generate/verify_contract.py b/packages/flow/openadapt_flow/generate/verify_contract.py index 2b472d69da..3febfebc7c 100644 --- a/packages/flow/openadapt_flow/generate/verify_contract.py +++ b/packages/flow/openadapt_flow/generate/verify_contract.py @@ -1246,7 +1246,7 @@ def _versions_frame( return { "read": {"kind": "sql", **settings.read, "sql_query": sql}, "record_field": "id", - "target_field": "id", + "target_field": _key_alias(b.schema, plan.anchor), "target_surface": "record", "version_field": "ver", } @@ -1271,7 +1271,7 @@ def cond(c: str, val: Any) -> str: return { "read": {"kind": "sql", **settings.read, "sql_query": sql}, "record_field": "id", - "target_field": "id", + "target_field": _key_alias(b.schema, plan.anchor), "target_surface": "record", "version_field": "ver", "author_field": "author", @@ -1621,15 +1621,28 @@ def sql(self, b: SqlBuilder) -> str: return f"SELECT {cols} FROM {b.qt(self.table)} {self.alias}" +def _key_alias(schema: Schema, table: str) -> str: + """The name a surface reads the table's key under: ``id``, unless the + table has its own ``id`` column that is not the whole primary key (a + ``uuid`` key, a key of ``id`` and ``rev``), which must be read as it is.""" + + t = schema.tables[table] + if t.column("id") is None or t.primary_key == ["id"]: + return "id" + return "oa_key" + + def _base_columns(b: SqlBuilder, table: str, alias: str = "t") -> list[tuple[str, str]]: + """Every column of ``table`` by its own name, and the key under + :func:`_key_alias` when that isn't a column already.""" + t = b.schema.tables[table] cols: list[tuple[str, str]] = [] key = b.key(alias, table) - if key is not None: - cols.append(("id", key)) + name = _key_alias(b.schema, table) + if key is not None and t.column(name) is None: + cols.append((name, key)) for c in t.columns: - if c.name == "id" and key is not None: - continue cols.append((c.name, b.text(alias, table, c.name))) return cols @@ -1724,6 +1737,8 @@ def build_contract( #: Surface -> the fields its declared rows may change. by_surface: dict[str, set[str]] = {} named_tables: set[str] = {anchor} + #: Table -> the surface that reads it in full (``frame.database.judged_by``). + judged: dict[str, str] = {anchor: "record"} # The main record, with every column and the values its parameters reach. main = Surface( @@ -1849,6 +1864,7 @@ def build_contract( sname, settings.views.get(g.table, g.table), scols ) named_tables.add(g.table) + judged.setdefault(g.table, sname) surf = surfaces[sname] items = range(variant.lists[p]) if g.list_param else [None] for i in items: @@ -2025,29 +2041,24 @@ def build_contract( if t in named_tables: continue allow.append({"table": t}) - for t in sorted(named_tables): - if plan.mode == "create" or flavor == "found": - # The contract's own read judges the rows the write adds; the - # whole-database frame still sees any existing row change, - # now and at the recheck (a late edit to a look-alike). - allow.append({"table": t}) - continue - # An update changes a row here, which an append-only table - # can't allow: its own read judges this table, and the recheck - # (database only) doesn't see it. - allow.append( - { - "table": t, - "mode": "skip", - "reason": "read in full, every column, by this contract's own surface", - } - ) - views = {s.table: s.name for s in surfaces.values() if s.name.startswith("raw_")} + # A table the contract's own read covers, every row and column, is + # compared like any other: after the save, the rows that read saw move + # are left to it (the target, or the rows the write adds, each with only + # its declared fields), and every other row of the table must not move. + # A recheck reads no surface, so there any change to the table since the + # post-save read (a late checkout of a look-alike) stops the write. + judged_by = {t: judged[t] for t in sorted(named_tables)} + views = { + s.table: s.name + for s in surfaces.values() + if s.name.startswith("raw_") and s.table not in judged_by + } db: dict[str, Any] = { "dialect": schema.dialect, "read": dict(settings.read), "allow_tables": allow, "views": views, + "judged_by": judged_by, "settle_s": 1, "settle_attempts": 5, } @@ -2056,6 +2067,12 @@ def build_contract( for s in surfaces.values() if s.name.startswith("raw_") and getattr(s, "_new_row_fields", None) } + for t in judged_by: + # A row the contract's own read declares is judged by the table + # too, every column exactly. A new one may set any column: the read + # checks each (placed, pinned, pinned empty, or named unchecked by + # the person). + nrf.setdefault(t, sorted(c.name for c in schema.tables[t].columns)) if nrf: db["new_row_fields"] = nrf if settings.recheck_after_s: diff --git a/packages/flow/openadapt_flow/runtime/effects/frame.py b/packages/flow/openadapt_flow/runtime/effects/frame.py index aa41a4e0c4..1b03c21398 100644 --- a/packages/flow/openadapt_flow/runtime/effects/frame.py +++ b/packages/flow/openadapt_flow/runtime/effects/frame.py @@ -302,6 +302,14 @@ class FrameDatabase(BaseModel): #: Table -> the surface whose records it holds (``patient_data`` -> #: ``patient``): an effect routed to the surface declares rows here too. views: dict[str, str] = Field(default_factory=dict) + #: Table -> a surface that declares its rows (a contract's own read of + #: its main record). After the save, the table's rows that surface's + #: records select (by an effect routed to it, matched by primary key) + #: are declared rows of the table: the table judges them, every column + #: exactly, with the surface's ``target_fields_by_surface`` fields as the + #: ones they may change. A recheck reads no surface, so there nothing is + #: declared and any change since the post-save read refutes. + judged_by: dict[str, str] = Field(default_factory=dict) #: Table -> columns a new declared or side row may set beyond the fields #: the effects name (ids, creation dates, columns the app's form fills #: on its own). Any other column must hold its default. @@ -363,6 +371,8 @@ def _compact(self, handler: Any) -> dict[str, Any]: data.pop("recheck_after_s", None) if data.get("settle_attempts") == 1: data.pop("settle_attempts", None) + if not data.get("judged_by"): + data.pop("judged_by", None) return data @model_validator(mode="after") @@ -785,6 +795,7 @@ def judge_frame( ] ] = None, user_ids: Optional[Mapping[str, str]] = None, + declared_rows: Optional[Mapping[str, Collection[str]]] = None, ) -> FrameVerdict: """Judge the frame of ``effects`` over ``slices``. @@ -1339,13 +1350,18 @@ def attribute( added_by_others: set[tuple[str, str]] = set() views_of = dict(database.views) if database is not None else {} + #: Table -> the surface that declares its rows (``database.judged_by``). + judged_of = dict(database.judged_by) if database is not None else {} + surface_declared = {t: set(k) for t, k in (declared_rows or {}).items()} + def may_change_for(name: str) -> set[str]: by_surface = spec.target_fields_by_surface if by_surface is None: return may_change if name in by_surface: return set(by_surface[name]) - return set(by_surface.get(views_of.get(name, ""), [])) + surface = views_of.get(name) or judged_of.get(name, "") + return set(by_surface.get(surface, [])) declared_keys: set[str] = set() #: What an audit entry names for each compared record (``ref_of``). @@ -1525,8 +1541,10 @@ def partner( before = before_map.get(key) after = after_map.get(key) is_declared = ( - before is not None and declared(before, hashed=hasher is not None) - ) or (after is not None and declared(after)) + (before is not None and declared(before, hashed=hasher is not None)) + or (after is not None and declared(after)) + or str(key) in surface_declared.get(name, ()) + ) refs[(name, str(key))] = ref_of(name, key, before, after) if is_declared: ref = refs[(name, str(key))][0] diff --git a/packages/flow/openadapt_flow/verify_run.py b/packages/flow/openadapt_flow/verify_run.py index 2d4ad30797..32c2d107b6 100644 --- a/packages/flow/openadapt_flow/verify_run.py +++ b/packages/flow/openadapt_flow/verify_run.py @@ -371,6 +371,38 @@ def parse_verify_config(data: Any, *, source: str = "verify config") -> VerifyCo raise VerifyRunError( "frame.database.views names no such surface: " + ", ".join(unknown) ) + unknown = sorted(set(database.judged_by.values()) - surfaces) + if unknown: + raise VerifyRunError( + "frame.database.judged_by names no such surface: " + ", ".join(unknown) + ) + named = list(database.judged_by.values()) + twice = sorted({s for s in named if named.count(s) > 1}) + if twice: + raise VerifyRunError( + "frame.database.judged_by names one surface for two tables: " + + ", ".join(twice) + ) + attribution_spec = cfg.frame.attribution if cfg.frame is not None else None + surface_ignore = set(cfg.frame.ignore_fields if cfg.frame is not None else []) + if attribution_spec is not None and attribution_spec.kind == "record_field": + surface_ignore |= set(attribution_spec.server_set) + for table in sorted(database.judged_by): + loose = sorted(surface_ignore - set(database.ignore_columns_for(table))) + if loose: + raise VerifyRunError( + f"frame.database.judged_by: the surface for {table} ignores " + f"{', '.join(loose)}, which the table doesn't ignore " + "(frame.database.ignore_columns)" + ) + both = sorted( + set(database.judged_by) & (set(database.views) | set(database.modes())) + ) + if both: + raise VerifyRunError( + "frame.database.judged_by names a table that views or " + "allow_tables also names: " + ", ".join(both) + ) for effect in cfg.effects: if effect.needs_operator_confirmation: raise VerifyRunError( @@ -753,6 +785,9 @@ class _DatabaseFrame: post_read_at: str = "" #: The snapshot the rows are judged against. old: Any = None + #: (table, record) -> the row key MAC of the table's row with that + #: record's primary key values, or None when they can't be read from it. + row_key: Any = None def _rules_used(profile: Any, old: Any, new: Any) -> list[str]: @@ -810,7 +845,9 @@ def _database_frame( Snapshot, SnapshotError, cell_hash, + key_text, mac_key, + row_key_hash, settled_state, unpack, unread_reasons, @@ -873,6 +910,15 @@ def _database_frame( key = mac_key(salt) out = _DatabaseFrame({}, {}, [], {}, (old.clock, new.clock), {}, None) out.post, out.post_read_at, out.old = new, new_at, old + + def row_key(table: str, record: Mapping[str, Any]) -> Optional[str]: + t = old.tables.get(table) + pk = list(t.primary_key) if t is not None else [] + if not pk or any(record.get(c) is None for c in pk): + return None + return row_key_hash(key, key_text(pk, record), table) + + out.row_key = row_key if "audit" in new.extra: out.audit_after = list(new.extra["audit"]) if old.objects != new.objects: @@ -1001,6 +1047,48 @@ def _read_audit(verifier: Any) -> Optional[list[dict[str, Any]]]: return None +def _declared_by_surface( + table: str, + surface: str, + state: tuple[EffectState, Optional[list[dict[str, Any]]]], + effects: Sequence[Effect], + routes: Mapping[str, Sequence[str]], + db: _DatabaseFrame, +) -> set[str]: + """Row key MACs of ``table``'s rows that ``surface`` declares + (``frame.database.judged_by``): its records, before or after the save, + that an effect routed to it selects, matched to the table by primary + key. The table judges those rows itself, every column exactly, with the + surface's fields as the ones a declared row may change. Empty when a + declared record's key can't be read, so nothing is declared there and a + declared change stops the write.""" + + from openadapt_flow.runtime.effects.effect import record_matches + from openadapt_flow.runtime.effects.frame import declared_selectors + + before_state, after_records = state + if after_records is None or not before_state.reachable: + return set() + selectors = [ + s + for i, effect in enumerate(effects) + if surface in routes.get(str(i), ()) + for s in declared_selectors(effect) + ] + typed = before_state.substrate == "sql" + keys: set[str] = set() + for record in [*before_state.records, *after_records]: + if not any( + record_matches(dict(record), s, text_numbers=typed) for s in selectors + ): + continue + k = db.row_key(table, record) + if k is None: + return set() + keys.add(k) + return keys + + def judge_run_frame( cfg: VerifyConfig, verifier: Any, @@ -1026,6 +1114,10 @@ def judge_run_frame( from openadapt_flow.runtime.effects.surfaces import surface_of spec = cfg.frame or FrameSpec() + # The database first, then the surfaces: a row a surface judges + # (``database.judged_by``) is read by it no earlier than the settled + # read a recheck compares with, so no change falls between the two. + db = _database_frame(cfg, before, salt) after: Any = None if not database_only: try: @@ -1074,7 +1166,8 @@ def read_of(name: str) -> str: elif not database_only: slices[""] = (before, _post_records(after)) identities: Optional[dict[str, str]] = None - db = _database_frame(cfg, before, salt) + #: Table -> row key MACs its ``judged_by`` surface declares. + declared_rows: dict[str, set[str]] = {} if db is not None and db.still_moving: tries = db.evidence.get("settle_attempts", 1) return FrameVerdict( @@ -1112,6 +1205,12 @@ def read_of(name: str) -> str: for route in routes.values(): route.extend(t for t, surface in db_views.items() if surface in route) slices.update(db.slices) + if spec.database is not None and db.row_key is not None: + for table, surface in spec.database.judged_by.items(): + if table in db.slices and surface in slices: + declared_rows[table] = _declared_by_surface( + table, surface, slices[surface], bound_effects, routes, db + ) identities = {name: KEY for name in db.slices} audit_before: Optional[list[dict[str, Any]]] = None audit_after: Optional[list[dict[str, Any]]] = None @@ -1189,6 +1288,7 @@ def excuse(name: str, before: Any, after: Any, fields: list[str]) -> Any: database_evidence=None if db is None else db.evidence, excuse=excuse, user_ids=None if db is None else db.users_map, + declared_rows=declared_rows, ) profile = db.evidence.get("profile") if db is not None else None if profile is not None: diff --git a/packages/flow/tests/test_frame_judged_by.py b/packages/flow/tests/test_frame_judged_by.py new file mode 100644 index 0000000000..ff7b8a3097 --- /dev/null +++ b/packages/flow/tests/test_frame_judged_by.py @@ -0,0 +1,342 @@ +"""``frame.database.judged_by``: a table the contract's own read covers. + +A contract that reads its main record's table in full (every row and every +column) judges the target there, with the fields the write may change. The +whole-database frame compares that table too, and leaves to the read only +the rows it saw move. A recheck reads no surface, so there the whole table is +compared: a late change to a look-alike row stops the write. Each test is one +case, on SQLite, through the public API. Records are made up. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path +from typing import Any + +import pytest + +from openadapt_flow import verifier +from openadapt_flow.verifier import ( + Actor, + Contract, + VerifierError, + begin_write, + recheck, + verify_write, +) + +BOT = Actor(kind="agent", id="bot") + +SCHEMA = """ +CREATE TABLE assets (id INTEGER PRIMARY KEY, tag TEXT, assigned_to INTEGER, + notes TEXT DEFAULT '', updated_at TEXT); +INSERT INTO assets VALUES (1, 'LT-1', NULL, '', '2026-01-01'), + (2, 'LT-1A', NULL, '', '2026-01-01'); +""" +COLUMNS = ( + "CAST(id AS TEXT) AS id, tag, CAST(assigned_to AS TEXT) AS assigned_to, " + "notes, updated_at" +) +PARAMS = {"tag": "LT-1", "user": "7"} +CHECKOUT = ( + "UPDATE assets SET assigned_to = 7, updated_at = '2026-10-31' WHERE tag = '{}'" +) + + +@pytest.fixture(autouse=True) +def _used_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + 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.0) + + +def _db(tmp_path: Path) -> Path: + tmp_path.mkdir(parents=True, exist_ok=True) + path = tmp_path / "app.sqlite" + con = sqlite3.connect(path) + con.executescript(SCHEMA) + con.commit() + con.close() + return path + + +def _sql(db: Path, *statements: str) -> None: + con = sqlite3.connect(db) + for statement in statements: + con.execute(statement) + con.commit() + con.close() + + +def _contract(db: Path, columns: str = COLUMNS, **database: Any) -> Contract: + return Contract.from_dict( + { + "schema_version": "openadapt.verify/v1", + "verifier": { + "kind": "surfaces", + "surfaces": { + "record": { + "kind": "sql", + "sqlite_database": str(db), + "sql_query": f"SELECT {columns} FROM assets", + } + }, + }, + "effects": [ + { + "kind": "record_written", + "probe": "surface=record|the asset is checked out", + "match": { + "tag": {"param": "tag"}, + "assigned_to": {"param": "user"}, + }, + "expected_count": 1, + } + ], + "frame": { + "target_fields_by_surface": {"record": ["assigned_to", "updated_at"]}, + "database": { + "dialect": "sqlite", + "read": {"sqlite_database": str(db)}, + "judged_by": {"assets": "record"}, + "settle_s": 0, + "recheck_after_s": 60, + **database, + }, + }, + } + ) + + +def _write(contract: Contract, db: Path, *statements: str) -> Any: + # No ledger: some tests make the same write twice on purpose. + pending = begin_write(contract, PARAMS, actor=BOT, ledger=None) + _sql(db, *statements) + return verify_write(pending) + + +def test_a_correct_checkout_holds_and_its_recheck_confirms(tmp_path: Path) -> None: + db = _db(tmp_path) + contract = _contract(db) + result = _write(contract, db, CHECKOUT.format("LT-1")) + assert result.verified, result.reasons + late = recheck(contract, result, params=PARAMS, actor=BOT) + assert late.outcome == "CONFIRMED", late.reasons + + +def test_a_look_alike_checked_out_in_the_window_refutes(tmp_path: Path) -> None: + db = _db(tmp_path) + contract = _contract(db) + result = _write(contract, db, CHECKOUT.format("LT-1"), CHECKOUT.format("LT-1A")) + assert not result.verified + + +def test_a_look_alike_checked_out_after_the_verdict_refutes_at_the_recheck( + tmp_path: Path, +) -> None: + db = _db(tmp_path) + contract = _contract(db) + result = _write(contract, db, CHECKOUT.format("LT-1")) + assert result.verified, result.reasons + _sql(db, CHECKOUT.format("LT-1A")) + late = recheck(contract, result, params=PARAMS, actor=BOT) + assert late.outcome == "REFUTED", late.reasons + + +def test_a_late_change_to_the_target_refutes_at_the_recheck(tmp_path: Path) -> None: + db = _db(tmp_path) + contract = _contract(db) + result = _write(contract, db, CHECKOUT.format("LT-1")) + assert result.verified, result.reasons + _sql(db, "UPDATE assets SET notes = 'moved' WHERE tag = 'LT-1'") + late = recheck(contract, result, params=PARAMS, actor=BOT) + assert late.outcome == "REFUTED", late.reasons + + +def test_a_read_that_misses_a_column_still_has_every_column_compared( + tmp_path: Path, +) -> None: + """The read leaves out ``notes``. It only says which rows are declared; + the table compares every column of them, so a change to ``notes`` on the + target stops the write and the correct checkout alone passes.""" + + narrow = "CAST(id AS TEXT) AS id, tag, CAST(assigned_to AS TEXT) AS assigned_to, updated_at" + db = _db(tmp_path / "a") + assert _write(_contract(db, columns=narrow), db, CHECKOUT.format("LT-1")).verified + db = _db(tmp_path / "b") + stray = "UPDATE assets SET notes = 'moved' WHERE tag = 'LT-1'" + result = _write(_contract(db, columns=narrow), db, CHECKOUT.format("LT-1"), stray) + assert not result.verified + + +def test_without_judged_by_the_table_judges_the_target_itself(tmp_path: Path) -> None: + """The same contract without ``judged_by``: the whole-database frame + sees the target change with nothing declaring it there, and stops.""" + + db = _db(tmp_path) + contract = _contract(db, judged_by={}) + result = _write(contract, db, CHECKOUT.format("LT-1")) + assert not result.verified + + +def test_judged_by_must_name_a_surface(tmp_path: Path) -> None: + db = _db(tmp_path) + with pytest.raises(VerifierError, match="judged_by names no such surface"): + _contract(db, judged_by={"assets": "nope"}) + + +def test_judged_by_cant_also_be_allow_listed(tmp_path: Path) -> None: + db = _db(tmp_path) + with pytest.raises(VerifierError, match="views or allow_tables also names"): + _contract(db, allow_tables=["assets"]) + + +# -- review round 1 (#1681): each attack, as a test ---------------------------- + + +def _raw(db: Path, query: str, **frame: Any) -> dict[str, Any]: + return { + "schema_version": "openadapt.verify/v1", + "verifier": { + "kind": "surfaces", + "surfaces": { + "record": { + "kind": "sql", + "sqlite_database": str(db), + "sql_query": query, + } + }, + }, + "effects": [ + { + "kind": "record_written", + "probe": "surface=record|x", + "match": {"tag": {"param": "tag"}, "assigned_to": {"param": "user"}}, + "expected_count": 1, + } + ], + "frame": { + "target_fields_by_surface": {"record": ["assigned_to", "updated_at"]}, + "database": { + "dialect": "sqlite", + "read": {"sqlite_database": str(db)}, + "judged_by": {"assets": "record"}, + "settle_s": 0, + "recheck_after_s": 60, + }, + **frame, + }, + } + + +LOSSY = ( + "CAST(id AS TEXT) AS id, tag, CAST(assigned_to AS TEXT) AS assigned_to, " + "substr(notes,1,3) AS notes, updated_at" +) + + +@pytest.mark.parametrize("tag", ["LT-1", "LT-1A"]) +def test_b1_a_lossy_read_never_hides_a_change(tmp_path: Path, tag: str) -> None: + """``substr`` reads 'abcdef' and 'abcXXX' alike. The table compares the + stored value, on the target and on the look-alike.""" + + db = _db(tmp_path) + _sql(db, "UPDATE assets SET notes='abcdef' WHERE id IN (1,2)") + stray = f"UPDATE assets SET notes='abcXXX' WHERE tag='{tag}'" + result = _write(_contract(db, columns=LOSSY), db, CHECKOUT.format("LT-1"), stray) + assert not result.verified + + +def test_b1_a_coalesced_read_never_hides_a_null(tmp_path: Path) -> None: + db = _db(tmp_path) + query = ( + "SELECT CAST(id AS TEXT) AS id, tag, CAST(assigned_to AS TEXT) AS assigned_to, " + "COALESCE(notes,'') AS notes, updated_at FROM assets" + ) + contract = Contract.from_dict(_raw(db, query)) + stray = "UPDATE assets SET notes=NULL WHERE tag='LT-1'" + assert not _write(contract, db, CHECKOUT.format("LT-1"), stray).verified + + +def test_b1_a_surface_may_not_ignore_what_the_table_compares(tmp_path: Path) -> None: + db = _db(tmp_path) + raw = _raw(db, f"SELECT {COLUMNS} FROM assets", ignore_fields=["notes"]) + with pytest.raises(VerifierError, match="ignores notes, which the table doesn't"): + Contract.from_dict(raw) + raw["frame"]["database"]["ignore_columns"] = {"assets": ["notes"]} + Contract.from_dict(raw) # the table ignores it too: allowed + + +def test_sf2_a_composite_key_with_id_is_matched_by_both_columns(tmp_path: Path) -> None: + """The generator reads ``id`` and ``rev`` as they are; a correct update + passes, and a late change to the look-alike is refuted at the recheck.""" + + db = tmp_path / "c.sqlite" + _sql( + db, + "CREATE TABLE assets (id INTEGER, rev INTEGER, tag TEXT, assigned_to INTEGER, " + "updated_at TEXT, PRIMARY KEY (id, rev))", + "INSERT INTO assets VALUES (1,1,'LT-1',NULL,'x'),(2,1,'LT-1A',NULL,'x')", + ) + query = ( + "SELECT CAST(t.id || '|' || t.rev AS TEXT) AS oa_key, CAST(t.id AS TEXT) AS id, " + "CAST(t.rev AS TEXT) AS rev, t.tag AS tag, CAST(t.assigned_to AS TEXT) AS assigned_to, " + "t.updated_at AS updated_at FROM assets t" + ) + contract = Contract.from_dict(_raw(db, query)) + result = _write(contract, db, CHECKOUT.format("LT-1")) + assert result.verified, result.reasons + _sql(db, CHECKOUT.format("LT-1A")) + assert recheck(contract, result, params=PARAMS, actor=BOT).outcome == "REFUTED" + + +def test_sf2_a_real_id_column_beside_another_key_is_compared(tmp_path: Path) -> None: + db = tmp_path / "u.sqlite" + _sql( + db, + "CREATE TABLE assets (uuid TEXT PRIMARY KEY, id INTEGER, tag TEXT, " + "assigned_to INTEGER, updated_at TEXT)", + "INSERT INTO assets VALUES ('a',1,'LT-1',NULL,'x'),('b',2,'LT-1A',NULL,'x')", + ) + query = ( + "SELECT t.uuid AS oa_key, t.uuid AS uuid, CAST(t.id AS TEXT) AS id, t.tag AS tag, " + "CAST(t.assigned_to AS TEXT) AS assigned_to, t.updated_at AS updated_at FROM assets t" + ) + contract = Contract.from_dict(_raw(db, query)) + assert _write(contract, db, CHECKOUT.format("LT-1")).verified + _sql(db, "UPDATE assets SET assigned_to=NULL, updated_at='x'") + stray = "UPDATE assets SET id=99 WHERE tag='LT-1'" + assert not _write(contract, db, CHECKOUT.format("LT-1"), stray).verified + + +def _create(db: Path, **database: Any) -> Contract: + raw = _raw(db, f"SELECT {COLUMNS} FROM assets") + raw["effects"][0]["match"] = {"tag": {"literal": "LT-9"}} + raw["frame"]["database"].update(database) + return Contract.from_dict(raw) + + +def test_sf3_a_declared_insert_may_set_only_declared_columns(tmp_path: Path) -> None: + db = _db(tmp_path) + contract = _create(db) + pending = begin_write(contract, {}, actor=BOT, ledger=None) + _sql(db, "INSERT INTO assets (id, tag) VALUES (3, 'LT-9')") + assert verify_write(pending).verified + _sql(db, "DELETE FROM assets WHERE id=3") + pending = begin_write(contract, {}, actor=BOT, ledger=None) + _sql(db, "INSERT INTO assets (id, tag, notes) VALUES (3, 'LT-9', 'evil')") + assert not verify_write(pending).verified + # Listed in new_row_fields, the column may be set. + _sql(db, "DELETE FROM assets WHERE id=3") + contract = _create(db, new_row_fields={"assets": ["notes"]}) + pending = begin_write(contract, {}, actor=BOT, ledger=None) + _sql(db, "INSERT INTO assets (id, tag, notes) VALUES (3, 'LT-9', 'ok')") + assert verify_write(pending).verified + + +def test_sf4_one_surface_cant_judge_two_tables(tmp_path: Path) -> None: + db = _db(tmp_path) + with pytest.raises(VerifierError, match="one surface for two tables"): + _contract(db, judged_by={"assets": "record", "other": "record"}) diff --git a/packages/flow/tests/test_generate_verify_contract.py b/packages/flow/tests/test_generate_verify_contract.py index 3806f18152..dd46c8b0c1 100644 --- a/packages/flow/tests/test_generate_verify_contract.py +++ b/packages/flow/tests/test_generate_verify_contract.py @@ -457,3 +457,143 @@ def test_each_fault_is_stopped(generated: dict[str, Any], name: str) -> None: cur.execute( f"INSERT INTO {t} VALUES ({', '.join(['%s'] * len(r))})", r ) + + +# -- the recheck sees the contract's own tables -------------------------------- + +TABLES = ("issues", "journals", "journal_details", "custom_values", "settings") + + +def _snapshot(app: App) -> dict[str, list[tuple]]: + out = {} + with app.con() as c, c.cursor() as cur: + for t in TABLES: + cur.execute(f"SELECT * FROM {t}") + out[t] = cur.fetchall() + return out + + +def _restore(app: App, snapshot: dict[str, list[tuple]]) -> None: + with app.con() as c, c.cursor() as cur: + for t, rows in snapshot.items(): + cur.execute(f"DELETE FROM {t}") + for r in rows: + cur.execute(f"INSERT INTO {t} VALUES ({', '.join(['%s'] * len(r))})", r) + + +def _write_then_recheck( + generated: dict[str, Any], + monkeypatch: pytest.MonkeyPatch, + kind: str, + values: dict[str, Any], + write: Callable[[], Any], + late: Callable[[], Any], +) -> tuple[str, str]: + """A write, its verdict, a late write after the verdict, and the + recheck's verdict.""" + + from openadapt_flow import verifier + from openadapt_flow.verifier import Actor, begin_write, recheck, verify_write + + monkeypatch.setattr(verifier, "seconds_until", lambda due: 0.0) + actor = Actor("agent", "ops-bot") + contract, params = gen.choose( + generated["out"] / "manifest.yaml", kind, values, {"bot_user_id": str(BOT)} + ) + app = generated["app"] + saved = _snapshot(app) + try: + with tempfile.TemporaryDirectory() as tmp: + pending = begin_write(contract, params, actor=actor, ledger=None, + write_key=f"t-{time.time_ns()}", workdir=tmp) # fmt: skip + write() + result = verify_write(pending) + first = str(getattr(result.outcome, "value", result.outcome)) + if not result.verified: + return first + " " + "; ".join(result.reasons)[:400], "" + late() + again = recheck(contract, result, params=params, actor=actor, ledger=None) + outcome = str(getattr(again.outcome, "value", again.outcome)) + return first, outcome + " " + "; ".join(again.reasons)[:400] + finally: + _restore(app, saved) + + +def test_the_generated_frame_compares_its_own_tables(generated: dict[str, Any]) -> None: + """The main record's table and the tables of rows that hold values are + compared, and left to the contract's own read only for the rows that + read saw move (``judged_by``): never skipped, never append-only.""" + + out = generated["out"] + manifest = yaml.safe_load((out / "manifest.yaml").read_text()) + for kind in ("create_issue", "update_issue"): + for v in manifest["kinds"][kind]["variants"]: + for flavor in ("create", "update", "found"): + if flavor not in v: + continue + db = yaml.safe_load((out / v[flavor]).read_text())["frame"]["database"] + allowed = {t["table"] for t in db["allow_tables"]} + assert db["judged_by"]["issues"] == "record" + assert not set(db["judged_by"]) & allowed, (kind, flavor) + assert not set(db["judged_by"]) & set(db.get("views", {})) + + +def test_a_late_update_of_a_look_alike_is_refuted_at_the_recheck( + generated: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + """A correct update, then a change to the look-alike issue in the other + project after the verdict: the recheck stops it. The columns the update + may change are excused on the target only.""" + + app = generated["app"] + + def target() -> int: + return app.find("Billing Portal", "Refund button missing") + + def legacy() -> int: + return app.find("Billing Portal Legacy", "Refund button missing") + + lates = { + "the same update on the look-alike": lambda: app.update_issue(legacy(), UPDATE), + "only the look-alike's status": lambda: app.q( + "UPDATE issues SET status_id=3, updated_on=now() WHERE id=%s", (legacy(),)), + "a second change to the target": lambda: app.q( + "UPDATE issues SET done_ratio=90 WHERE id=%s", (target(),)), + } # fmt: skip + for name, late_write in lates.items(): + first, late = _write_then_recheck( + generated, monkeypatch, "update_issue", UPDATE, + lambda: app.update_issue(target(), UPDATE), late_write, + ) # fmt: skip + assert first == "CONFIRMED", (name, first) + assert late.startswith("REFUTED"), (name, late) + + +def test_a_late_duplicate_create_is_refuted_at_the_recheck( + generated: dict[str, Any], monkeypatch: pytest.MonkeyPatch +) -> None: + app = generated["app"] + first, late = _write_then_recheck( + generated, monkeypatch, "create_issue", CREATE, + lambda: app.create_issue(CREATE), + lambda: app.create_issue(CREATE, project_id=2), + ) # fmt: skip + assert first == "CONFIRMED", first + assert late.startswith("REFUTED"), late + + +@pytest.mark.parametrize("kind", ["create_issue", "update_issue"]) +def test_a_correct_write_with_nothing_late_is_confirmed_at_the_recheck( + generated: dict[str, Any], monkeypatch: pytest.MonkeyPatch, kind: str +) -> None: + app = generated["app"] + if kind == "create_issue": + values, write = CREATE, lambda: app.create_issue(CREATE) + else: + iid = app.find("Billing Portal", "Refund button missing") + values, write = UPDATE, lambda: app.update_issue(iid, UPDATE) + first, late = _write_then_recheck( + generated, monkeypatch, kind, values, write, lambda: None + ) + assert first == "CONFIRMED", first + assert late.startswith("CONFIRMED"), late