fault gate: admit a record check only after it catches its faults

CAPABILITIES.md

@@ -136,6 +136,7 @@ improvising or retrying when something is off.
 | One check-level scale, 0 to 5, from "unchecked" to "counterparty confirmed" | Released | Unit tests | `openadapt_flow/check_level.py` |
 | Judge any executor's write, including an agent's or a script's, from the system of record | Released | Synthetic | `openadapt-flow verify-run`, [VERIFY_RUN.md](packages/flow/docs/VERIFY_RUN.md) |
 | Prove that nothing outside the declared change moved (frame conditions), and attribute concurrent edits by other users through the app's audit log or a modified-by field; the receipt signs which parts were checked | On main | Synthetic (MockMed, SQLite audit log) | `verify-run`, [VERIFICATION_COVERAGE.md](packages/flow/docs/VERIFICATION_COVERAGE.md) |
+| Admit a record check only after a fault suite written to a store copy shows it catches every fault it claims, and sign its coverage into the receipt | On main | Synthetic (in-memory REST store, SQLite copy) | `openadapt_flow/fault_gate.py`, `scaffold-verifier --fault-gate`, `verify-run` `admission:` |
 | Guarded save: a safety layer any agent wraps around one write (check the record, hold for approval, read back) | On main ([#1258](https://github.com/OpenAdaptAI/OpenAdapt/pull/1258)) | Synthetic | `GUARDED_SAVE.md` |
 | Draft a record check from the demonstration's write steps (`scaffold-verifier`) | Released | Synthetic | `openadapt-flow scaffold-verifier` (a person fills in the draft) |
 | Build a working record check from the reads the app made while the person viewed the saved record, and test it against injected bad writes before anyone reviews it | On main ([#1284](https://github.com/OpenAdaptAI/OpenAdapt/pull/1284)) | Synthetic (MockMed, OpenEMR fixture) | `openadapt-flow scaffold-verifier` |

llms-full.txt

@@ -533,6 +533,7 @@ improvising or retrying when something is off.
 | One check-level scale, 0 to 5, from "unchecked" to "counterparty confirmed" | Released | Unit tests | `openadapt_flow/check_level.py` |
 | Judge any executor's write, including an agent's or a script's, from the system of record | Released | Synthetic | `openadapt-flow verify-run`, [VERIFY_RUN.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/VERIFY_RUN.md) |
 | Prove that nothing outside the declared change moved (frame conditions), and attribute concurrent edits by other users through the app's audit log or a modified-by field; the receipt signs which parts were checked | On main | Synthetic (MockMed, SQLite audit log) | `verify-run`, [VERIFICATION_COVERAGE.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/VERIFICATION_COVERAGE.md) |
+| Admit a record check only after a fault suite written to a store copy shows it catches every fault it claims, and sign its coverage into the receipt | On main | Synthetic (in-memory REST store, SQLite copy) | `openadapt_flow/fault_gate.py`, `scaffold-verifier --fault-gate`, `verify-run` `admission:` |
 | Guarded save: a safety layer any agent wraps around one write (check the record, hold for approval, read back) | On main ([#1258](https://github.com/OpenAdaptAI/OpenAdapt/pull/1258)) | Synthetic | `GUARDED_SAVE.md` |
 | Draft a record check from the demonstration's write steps (`scaffold-verifier`) | Released | Synthetic | `openadapt-flow scaffold-verifier` (a person fills in the draft) |
 | Build a working record check from the reads the app made while the person viewed the saved record, and test it against injected bad writes before anyone reviews it | On main ([#1284](https://github.com/OpenAdaptAI/OpenAdapt/pull/1284)) | Synthetic (MockMed, OpenEMR fixture) | `openadapt-flow scaffold-verifier` |

packages/flow/docs/VERIFY_RUN.md

@@ -169,6 +169,42 @@ the counts behind the frame part (records compared, collateral, concurrent)
 and the digest of the `frame:` section. A receipt issued before the field
 existed has neither and still checks.
 
+## Admission: a check shown to catch its faults
+
+A record check proves little until it has been seen to fail. The fault gate
+(`openadapt_flow/fault_gate.py`, [`VERIFICATION_COVERAGE.md`](VERIFICATION_COVERAGE.md))
+runs the check against a test copy of the system of record. It writes seven
+kinds of bad save to the copy's store, and the check reads each one through
+its own read: a wrong value in each declared field, the values on a
+look-alike record, an edit to another record, a partial write, a duplicate,
+no write, and a write that lands after the check's timeout. A read whose
+scope can't see a fault fails the gate, not only a judge that misses it.
+
+The gate writes `check_coverage.json` (`openadapt.check-coverage/v1`): faults
+killed over faults applicable, per class, bound to the check by its digest
+(the effects, the read and the frame; not where the store is). An
+`admission:` section makes verify-run require it:
+
+```yaml
+admission:
+  coverage: check_coverage.json   # relative to verify.yaml
+  min_coverage: 1.0               # default: every applicable fault caught
+  require_classes: [unrelated_edit]
+  refuse_test_copies: [recorded_states]   # the default
+```
+
+Under a policy, `checked.coverage` in the receipt holds the totals, the
+record's digest and `policy` (`met`, or why not, such as
+`coverage_for_another_check`). `verified` requires `met`. A changed check
+needs a new coverage record. Without `admission:` nothing about coverage is
+required or stated.
+
+`openadapt-flow scaffold-verifier --db app.sqlite --fault-gate` runs the gate
+on a copy of the database and keeps the check a draft when a fault survives.
+Without `--fault-gate` it writes the coverage of its recorded-state test,
+which can't test a wrong value written over the right record or a late
+write, so that record is incomplete and refused by default.
+
 ## Verdicts
 
 | What the store shows | `effect_verdict` | `outcome` |

packages/flow/openadapt_flow/__main__.py

@@ -4843,6 +4843,7 @@ def optional_path(name: str) -> Optional[Path]:
         recording=optional_path("recording"),
         bundle=optional_path("bundle"),
         complete_read=bool(getattr(args, "complete_read", False)),
+        fault_gate=bool(getattr(args, "fault_gate", False)),
     )
     if built is not None:
         print(built.summary)
@@ -9424,6 +9425,15 @@ def build_parser() -> argparse.ArgumentParser:
             "(one store, no other tables). Recorded as declared, not proven"
         ),
     )
+    p.add_argument(
+        "--fault-gate",
+        action="store_true",
+        help=(
+            "With --db: write each bad save to a copy of the database and let "
+            "the check read it through its own read (the fault gate). A fault "
+            "the read can't see keeps the check a draft"
+        ),
+    )
     p.add_argument(
         "--recording",
         default=None,

packages/flow/openadapt_flow/countersign/document.py

@@ -651,12 +651,20 @@ def verify_countersigned(
             report.problems.append("checked_how_mismatch")
     if schema == EFFECT_RECEIPT_SCHEMA and "checked" in receipt:
         parts = receipt.get("checked")
-        if not isinstance(parts, dict) or set(parts) != {"delta", "frame", "source"}:
+        if not isinstance(parts, dict) or set(parts) - {"coverage"} != {
+            "delta",
+            "frame",
+            "source",
+        }:
             report.problems.append("checked_malformed")
         elif receipt.get("outcome") in _SEALABLE and (
             parts.get("delta") != "held"
             or parts.get("frame") != "held"
             or parts.get("source") not in ("held", "instruction")
+            or (
+                "coverage" in parts
+                and (parts.get("coverage") or {}).get("policy") != "met"
+            )
         ):
             report.problems.append("verified_without_every_part")
     # Seal signature: openadapt_types.seal, unchanged.

packages/flow/openadapt_flow/countersign/render.py

@@ -98,6 +98,12 @@ def checked_line(checked: dict[str, Any]) -> str:
             out.append("source: the inputs were the authorized instruction")
         else:
             out.append(f"{part} ({_PART_TEXT[part]}): {state}")
+    cover = checked.get("coverage")
+    if isinstance(cover, dict):
+        out.append(
+            f"fault coverage: {cover.get('killed')} of {cover.get('applicable')} "
+            f"faults caught, admission policy {cover.get('policy')}"
+        )
     return "; ".join(out)